From 91a0879912b5451a86cfcc04478d6dd7d1825c7e Mon Sep 17 00:00:00 2001 From: Ivan Date: Mon, 13 Apr 2026 20:55:32 -0500 Subject: [PATCH] feat(stickers): implement sticker management features including creation, deletion, listing, and import/export functionality, along with database schema updates for user stickers --- meshchatx/meshchat.py | 174 +++++++++++++- meshchatx/src/backend/async_utils.py | 8 +- meshchatx/src/backend/config_manager.py | 5 + meshchatx/src/backend/database/__init__.py | 2 + meshchatx/src/backend/database/messages.py | 27 ++- meshchatx/src/backend/database/schema.py | 38 ++- meshchatx/src/backend/database/stickers.py | 229 +++++++++++++++++++ meshchatx/src/backend/nomadnet_downloader.py | 3 +- meshchatx/src/backend/sticker_utils.py | 179 +++++++++++++++ 9 files changed, 655 insertions(+), 10 deletions(-) create mode 100644 meshchatx/src/backend/database/stickers.py create mode 100644 meshchatx/src/backend/sticker_utils.py diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py index 3f47c34..abd4d6b 100644 --- a/meshchatx/meshchat.py +++ b/meshchatx/meshchat.py @@ -25,6 +25,7 @@ import time import traceback import webbrowser import io +import subprocess import zipfile import fnmatch from datetime import UTC, datetime, timedelta @@ -103,6 +104,12 @@ from meshchatx.src.backend.persistent_log_handler import PersistentLogHandler from meshchatx.src.backend.recovery import CrashRecovery, HealthMonitor from meshchatx.src.backend.rnprobe_handler import RNProbeHandler from meshchatx.src.backend.sideband_commands import SidebandCommands +from meshchatx.src.backend.sticker_utils import ( + build_export_document, + mime_for_image_type, + sanitize_sticker_name, + validate_export_document, +) from meshchatx.src.backend.telemetry_utils import Telemeter from meshchatx.src.backend.web_audio_bridge import WebAudioBridge from meshchatx.src.version import __version__ as app_version @@ -4831,6 +4838,12 @@ class ReticulumMeshChat: self.database.misc.delete_all_user_icons() return web.json_response({"message": "All LXMF icons cleared"}) + @routes.delete("/api/v1/maintenance/stickers") + async def maintenance_clear_stickers(request): + identity_hash = self.identity.hash.hex() + n = self.database.stickers.delete_all_for_identity(identity_hash) + return web.json_response({"message": "Stickers cleared", "deleted": n}) + # maintenance - export messages @routes.get("/api/v1/maintenance/messages/export") async def maintenance_export_messages(request): @@ -9170,6 +9183,110 @@ class ReticulumMeshChat: self.database.map_drawings.update_drawing(drawing_id, name, drawing_data) return web.json_response({"message": "Drawing updated successfully"}) + @routes.get("/api/v1/stickers") + async def stickers_list(request): + identity_hash = self.identity.hash.hex() + rows = self.database.stickers.list_for_identity(identity_hash) + return web.json_response({"stickers": [dict(r) for r in rows]}) + + @routes.post("/api/v1/stickers") + async def stickers_create(request): + identity_hash = self.identity.hash.hex() + try: + data = await request.json() + except (json.JSONDecodeError, ValueError): + return web.json_response({"error": "invalid_json"}, status=400) + image_b64 = data.get("image_bytes") + if not isinstance(image_b64, str) or not image_b64.strip(): + return web.json_response({"error": "missing_image_bytes"}, status=400) + try: + raw = base64.b64decode(image_b64.strip(), validate=True) + except (ValueError, TypeError): + return web.json_response({"error": "invalid_base64"}, status=400) + name = sanitize_sticker_name(data.get("name")) + image_type = data.get("image_type") + src = data.get("source_message_hash") + src = src if isinstance(src, str) else None + try: + row = self.database.stickers.insert( + identity_hash, + name, + image_type, + raw, + src, + ) + except ValueError as e: + return web.json_response({"error": str(e)}, status=400) + if row is None: + return web.json_response({"error": "duplicate_sticker"}, status=409) + return web.json_response({"sticker": row}) + + @routes.delete("/api/v1/stickers/{sticker_id}") + async def stickers_delete(request): + identity_hash = self.identity.hash.hex() + sticker_id = int(request.match_info.get("sticker_id", "0")) + ok = self.database.stickers.delete(sticker_id, identity_hash) + if not ok: + return web.json_response({"error": "not_found"}, status=404) + return web.json_response({"message": "deleted"}) + + @routes.patch("/api/v1/stickers/{sticker_id}") + async def stickers_patch(request): + identity_hash = self.identity.hash.hex() + sticker_id = int(request.match_info.get("sticker_id", "0")) + try: + data = await request.json() + except (json.JSONDecodeError, ValueError): + return web.json_response({"error": "invalid_json"}, status=400) + if "name" not in data: + return web.json_response({"error": "missing_name"}, status=400) + name = sanitize_sticker_name(data.get("name")) + ok = self.database.stickers.update_name(sticker_id, identity_hash, name) + if not ok: + return web.json_response({"error": "not_found"}, status=404) + return web.json_response({"message": "updated"}) + + @routes.get("/api/v1/stickers/{sticker_id}/image") + async def stickers_get_image(request): + identity_hash = self.identity.hash.hex() + sticker_id = int(request.match_info.get("sticker_id", "0")) + row = self.database.stickers.get_row(sticker_id, identity_hash) + if row is None: + return web.json_response({"error": "not_found"}, status=404) + ct = mime_for_image_type(row["image_type"]) + return web.Response(body=row["image_blob"], content_type=ct) + + @routes.get("/api/v1/stickers/export") + async def stickers_export(request): + identity_hash = self.identity.hash.hex() + payloads = self.database.stickers.export_payloads_for_identity( + identity_hash + ) + doc = build_export_document( + payloads, + datetime.now(UTC).isoformat(), + ) + return web.json_response(doc) + + @routes.post("/api/v1/stickers/import") + async def stickers_import(request): + identity_hash = self.identity.hash.hex() + try: + data = await request.json() + except (json.JSONDecodeError, ValueError): + return web.json_response({"error": "invalid_json"}, status=400) + replace = bool(data.get("replace_duplicates", False)) + try: + items = validate_export_document(data) + except ValueError as e: + return web.json_response({"error": str(e)}, status=400) + result = self.database.stickers.import_payloads( + identity_hash, + items, + replace_duplicates=replace, + ) + return web.json_response(result) + # get latest telemetry for all peers @routes.get("/api/v1/telemetry/peers") async def get_all_latest_telemetry(request): @@ -10264,6 +10381,11 @@ class ReticulumMeshChat: data["message_failed_bubble_color"] ) + if "message_waiting_bubble_color" in data: + self.config.message_waiting_bubble_color.set( + data["message_waiting_bubble_color"] + ) + # update desktop settings if "desktop_open_calls_in_separate_window" in data: self.config.desktop_open_calls_in_separate_window.set( @@ -11429,6 +11551,7 @@ class ReticulumMeshChat: "message_outbound_bubble_color": ctx.config.message_outbound_bubble_color.get(), "message_inbound_bubble_color": ctx.config.message_inbound_bubble_color.get(), "message_failed_bubble_color": ctx.config.message_failed_bubble_color.get(), + "message_waiting_bubble_color": ctx.config.message_waiting_bubble_color.get(), "translator_enabled": ctx.config.translator_enabled.get(), "libretranslate_url": ctx.config.libretranslate_url.get(), "desktop_open_calls_in_separate_window": ctx.config.desktop_open_calls_in_separate_window.get(), @@ -11778,6 +11901,47 @@ class ReticulumMeshChat: except Exception: return False + def _convert_webm_opus_to_ogg(self, audio_bytes: bytes) -> bytes: + """Convert WebM/Opus audio to OGG/Opus using ffmpeg. + + Browser MediaRecorder outputs Opus in a WebM container, but LXMF + AM_OPUS_OGG expects an OGG container. If ffmpeg is unavailable or + the input is already OGG, the original bytes are returned as-is. + """ + if audio_bytes[:4] == b"OggS": + return audio_bytes + + ffmpeg_path = shutil.which("ffmpeg") + if ffmpeg_path is None: + return audio_bytes + + try: + result = subprocess.run( # noqa: S603 + [ + ffmpeg_path, + "-i", + "pipe:0", + "-c:a", + "libopus", + "-b:a", + "24k", + "-vbr", + "on", + "-f", + "ogg", + "pipe:1", + ], + input=audio_bytes, + capture_output=True, + timeout=30, + ) + if result.returncode == 0 and len(result.stdout) > 0: + return result.stdout + except Exception as e: + print(f"WebM to OGG conversion failed: {e}") + + return audio_bytes + # check if a destination is blocked def is_destination_blocked(self, destination_hash: str, context=None) -> bool: ctx = context or self.current_context @@ -12444,6 +12608,11 @@ class ReticulumMeshChat: lxmf_message.fields = {} + lxmf_message.fields[LXMF.FIELD_RENDERER] = LXMF.RENDERER_MARKDOWN + + if self._is_contact(destination_hash, context=ctx): + lxmf_message.include_ticket = True + # add file attachments field if file_attachments_field is not None: # create array of [[file_name, file_bytes], [file_name, file_bytes], ...] @@ -12464,9 +12633,12 @@ class ReticulumMeshChat: # add audio field if audio_field is not None: + audio_bytes = audio_field.audio_bytes + if audio_field.audio_mode == LXMF.AM_OPUS_OGG: + audio_bytes = self._convert_webm_opus_to_ogg(audio_bytes) lxmf_message.fields[LXMF.FIELD_AUDIO] = [ audio_field.audio_mode, - audio_field.audio_bytes, + audio_bytes, ] # add telemetry field diff --git a/meshchatx/src/backend/async_utils.py b/meshchatx/src/backend/async_utils.py index 4625e4d..e33ad3b 100644 --- a/meshchatx/src/backend/async_utils.py +++ b/meshchatx/src/backend/async_utils.py @@ -62,11 +62,15 @@ class AsyncUtils: """ if AsyncUtils.main_loop and AsyncUtils.main_loop.is_running(): future = asyncio.run_coroutine_threadsafe( - coroutine, AsyncUtils.main_loop, + coroutine, + AsyncUtils.main_loop, ) with AsyncUtils._futures_lock: AsyncUtils._pending_futures.append(future) - if len(AsyncUtils._pending_futures) >= AsyncUtils._FUTURES_SWEEP_THRESHOLD: + if ( + len(AsyncUtils._pending_futures) + >= AsyncUtils._FUTURES_SWEEP_THRESHOLD + ): AsyncUtils._pending_futures = [ f for f in AsyncUtils._pending_futures if not f.done() ] diff --git a/meshchatx/src/backend/config_manager.py b/meshchatx/src/backend/config_manager.py index b301210..6385a3a 100644 --- a/meshchatx/src/backend/config_manager.py +++ b/meshchatx/src/backend/config_manager.py @@ -317,6 +317,11 @@ class ConfigManager: "message_failed_bubble_color", "#ef4444", ) + self.message_waiting_bubble_color = self.StringConfig( + self, + "message_waiting_bubble_color", + "#e5e7eb", + ) # announce caps: max rows stored per aspect (oldest dropped). Default 1000. self.announce_max_stored_lxmf_delivery = self.IntConfig( diff --git a/meshchatx/src/backend/database/__init__.py b/meshchatx/src/backend/database/__init__.py index aee0668..2b1f574 100644 --- a/meshchatx/src/backend/database/__init__.py +++ b/meshchatx/src/backend/database/__init__.py @@ -18,6 +18,7 @@ from .misc import MiscDAO from .provider import DatabaseProvider from .ringtones import RingtoneDAO from .schema import DatabaseSchema +from .stickers import UserStickersDAO from .telemetry import TelemetryDAO from .telephone import TelephoneDAO from .voicemails import VoicemailDAO @@ -42,6 +43,7 @@ class Database: self.ringtones = RingtoneDAO(self.provider) self.contacts = ContactsDAO(self.provider) self.map_drawings = MapDrawingsDAO(self.provider) + self.stickers = UserStickersDAO(self.provider) self.debug_logs = DebugLogsDAO(self.provider) self.access_attempts = AccessAttemptsDAO(self.provider) self.crash_history = CrashHistoryDAO(self.provider) diff --git a/meshchatx/src/backend/database/messages.py b/meshchatx/src/backend/database/messages.py index 792eb4b..00d2689 100644 --- a/meshchatx/src/backend/database/messages.py +++ b/meshchatx/src/backend/database/messages.py @@ -59,9 +59,17 @@ class MessageDAO: self.provider.execute(query, params) - def update_lxmf_message_state(self, message_hash, state, progress, - delivery_attempts, next_delivery_attempt_at, - rssi=None, snr=None, quality=None): + def update_lxmf_message_state( + self, + message_hash, + state, + progress, + delivery_attempts, + next_delivery_attempt_at, + rssi=None, + snr=None, + quality=None, + ): """Lightweight update for delivery-state changes only. Avoids re-serializing the full message (including base64 attachment @@ -73,8 +81,17 @@ class MessageDAO: "delivery_attempts = ?, next_delivery_attempt_at = ?, " "rssi = ?, snr = ?, quality = ?, updated_at = ? " "WHERE hash = ?", - (state, progress, delivery_attempts, next_delivery_attempt_at, - rssi, snr, quality, now, message_hash), + ( + state, + progress, + delivery_attempts, + next_delivery_attempt_at, + rssi, + snr, + quality, + now, + message_hash, + ), ) def get_lxmf_message_by_hash(self, message_hash): diff --git a/meshchatx/src/backend/database/schema.py b/meshchatx/src/backend/database/schema.py index b79cc35..28cf8bc 100644 --- a/meshchatx/src/backend/database/schema.py +++ b/meshchatx/src/backend/database/schema.py @@ -13,7 +13,7 @@ def _validate_identifier(name: str, label: str = "identifier") -> str: class DatabaseSchema: - LATEST_VERSION = 43 + LATEST_VERSION = 44 def __init__(self, provider: DatabaseProvider): self.provider = provider @@ -417,6 +417,20 @@ class DatabaseSchema: updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """, + "user_stickers": """ + CREATE TABLE IF NOT EXISTS user_stickers ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + identity_hash TEXT NOT NULL, + name TEXT, + image_type TEXT NOT NULL, + image_blob BLOB NOT NULL, + content_hash TEXT NOT NULL, + source_message_hash TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + UNIQUE(identity_hash, content_hash) + ) + """, "lxmf_last_sent_icon_hashes": """ CREATE TABLE IF NOT EXISTS lxmf_last_sent_icon_hashes ( destination_hash TEXT PRIMARY KEY, @@ -1119,6 +1133,28 @@ class DatabaseSchema: "CREATE INDEX IF NOT EXISTS idx_lxmf_conversation_pins_pinned_at ON lxmf_conversation_pins(pinned_at)", ) + if current_version < 44: + self._safe_execute(""" + CREATE TABLE IF NOT EXISTS user_stickers ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + identity_hash TEXT NOT NULL, + name TEXT, + image_type TEXT NOT NULL, + image_blob BLOB NOT NULL, + content_hash TEXT NOT NULL, + source_message_hash TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + UNIQUE(identity_hash, content_hash) + ) + """) + self._safe_execute( + "CREATE INDEX IF NOT EXISTS idx_user_stickers_identity ON user_stickers(identity_hash)", + ) + self._safe_execute( + "CREATE INDEX IF NOT EXISTS idx_user_stickers_identity_updated ON user_stickers(identity_hash, updated_at)", + ) + # Update version in config self._safe_execute( """ diff --git a/meshchatx/src/backend/database/stickers.py b/meshchatx/src/backend/database/stickers.py new file mode 100644 index 0000000..ee5744e --- /dev/null +++ b/meshchatx/src/backend/database/stickers.py @@ -0,0 +1,229 @@ +import base64 +import sqlite3 +import time + +from meshchatx.src.backend import sticker_utils + + +class UserStickersDAO: + def __init__(self, provider): + self.provider = provider + + def count_for_identity(self, identity_hash: str) -> int: + row = self.provider.fetchone( + "SELECT COUNT(*) AS c FROM user_stickers WHERE identity_hash = ?", + (identity_hash,), + ) + return int(row["c"]) if row else 0 + + def list_for_identity(self, identity_hash: str): + return self.provider.fetchall( + """ + SELECT id, identity_hash, name, image_type, length(image_blob) AS image_size, + content_hash, source_message_hash, created_at, updated_at + FROM user_stickers + WHERE identity_hash = ? + ORDER BY updated_at DESC, id DESC + """, + (identity_hash,), + ) + + def get_row(self, sticker_id: int, identity_hash: str): + return self.provider.fetchone( + """ + SELECT id, identity_hash, name, image_type, image_blob, content_hash, + source_message_hash, created_at, updated_at + FROM user_stickers + WHERE id = ? AND identity_hash = ? + """, + (sticker_id, identity_hash), + ) + + def delete(self, sticker_id: int, identity_hash: str) -> bool: + cur = self.provider.execute( + "DELETE FROM user_stickers WHERE id = ? AND identity_hash = ?", + (sticker_id, identity_hash), + ) + return cur.rowcount > 0 + + def delete_all_for_identity(self, identity_hash: str) -> int: + cur = self.provider.execute( + "DELETE FROM user_stickers WHERE identity_hash = ?", + (identity_hash,), + ) + return cur.rowcount + + def update_name( + self, sticker_id: int, identity_hash: str, name: str | None + ) -> bool: + now = time.time() + cur = self.provider.execute( + """ + UPDATE user_stickers + SET name = ?, updated_at = ? + WHERE id = ? AND identity_hash = ? + """, + (name, now, sticker_id, identity_hash), + ) + return cur.rowcount > 0 + + def insert( + self, + identity_hash: str, + name: str | None, + image_type: str, + image_bytes: bytes, + source_message_hash: str | None = None, + ) -> dict | None: + """ + Insert a sticker. Returns summary dict or None if duplicate (same content_hash). + """ + if ( + self.count_for_identity(identity_hash) + >= sticker_utils.MAX_STICKERS_PER_IDENTITY + ): + msg = "sticker_limit_reached" + raise ValueError(msg) + + nt, ch = sticker_utils.validate_sticker_payload(image_bytes, image_type) + now = time.time() + try: + cur = self.provider.execute( + """ + INSERT INTO user_stickers ( + identity_hash, name, image_type, image_blob, content_hash, + source_message_hash, created_at, updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + identity_hash, + name, + nt, + image_bytes, + ch, + source_message_hash, + now, + now, + ), + ) + except sqlite3.IntegrityError: + return None + + new_id = cur.lastrowid + row = self.provider.fetchone( + """ + SELECT id, identity_hash, name, image_type, length(image_blob) AS image_size, + content_hash, source_message_hash, created_at, updated_at + FROM user_stickers + WHERE id = ? + """, + (new_id,), + ) + return dict(row) if row else None + + def export_payloads_for_identity(self, identity_hash: str) -> list[dict]: + rows = self.provider.fetchall( + """ + SELECT name, image_type, image_blob, source_message_hash + FROM user_stickers + WHERE identity_hash = ? + ORDER BY id ASC + """, + (identity_hash,), + ) + out = [] + for r in rows: + blob = r["image_blob"] + b64 = base64.b64encode(blob).decode("ascii") + out.append( + { + "name": r["name"], + "image_type": r["image_type"], + "image_bytes": b64, + "source_message_hash": r["source_message_hash"], + }, + ) + return out + + def import_payloads( + self, + identity_hash: str, + items: list[dict], + *, + replace_duplicates: bool, + ) -> dict: + imported = 0 + skipped_duplicates = 0 + skipped_invalid = 0 + errors: list[str] = [] + + for i, item in enumerate(items): + name = sticker_utils.sanitize_sticker_name(item.get("name")) + it = item.get("image_type") + b64 = item.get("image_bytes_b64") + src = item.get("source_message_hash") + try: + raw = base64.b64decode(b64, validate=False) + except (ValueError, TypeError): + skipped_invalid += 1 + errors.append(f"decode_failed_at_{i}") + continue + try: + nt, ch = sticker_utils.validate_sticker_payload(raw, it) + except ValueError: + skipped_invalid += 1 + errors.append(f"invalid_payload_at_{i}") + continue + + existing = self.provider.fetchone( + "SELECT id FROM user_stickers WHERE identity_hash = ? AND content_hash = ?", + (identity_hash, ch), + ) + if existing: + if not replace_duplicates: + skipped_duplicates += 1 + continue + self.provider.execute( + "DELETE FROM user_stickers WHERE identity_hash = ? AND content_hash = ?", + (identity_hash, ch), + ) + + if ( + self.count_for_identity(identity_hash) + >= sticker_utils.MAX_STICKERS_PER_IDENTITY + ): + errors.append("sticker_limit_reached") + break + + now = time.time() + try: + self.provider.execute( + """ + INSERT INTO user_stickers ( + identity_hash, name, image_type, image_blob, content_hash, + source_message_hash, created_at, updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + identity_hash, + name, + nt, + raw, + ch, + src if isinstance(src, str) else None, + now, + now, + ), + ) + imported += 1 + except sqlite3.IntegrityError: + skipped_duplicates += 1 + + return { + "imported": imported, + "skipped_duplicates": skipped_duplicates, + "skipped_invalid": skipped_invalid, + "errors": errors, + } diff --git a/meshchatx/src/backend/nomadnet_downloader.py b/meshchatx/src/backend/nomadnet_downloader.py index d052824..7f131c3 100644 --- a/meshchatx/src/backend/nomadnet_downloader.py +++ b/meshchatx/src/backend/nomadnet_downloader.py @@ -35,7 +35,8 @@ def sweep_stale_links(): """Evict all non-ACTIVE links from the global cache.""" with _nomadnet_links_lock: stale = [ - k for k, v in nomadnet_cached_links.items() + k + for k, v in nomadnet_cached_links.items() if v.status is not RNS.Link.ACTIVE ] for k in stale: diff --git a/meshchatx/src/backend/sticker_utils.py b/meshchatx/src/backend/sticker_utils.py new file mode 100644 index 0000000..0891507 --- /dev/null +++ b/meshchatx/src/backend/sticker_utils.py @@ -0,0 +1,179 @@ +"""Validation and hashing for user sticker images stored per identity.""" + +from __future__ import annotations + +import hashlib +import base64 + +MAX_STICKER_BYTES = 512 * 1024 +MAX_STICKERS_PER_IDENTITY = 2000 + +_ALLOWED_TYPES = frozenset({"png", "jpeg", "jpg", "gif", "webp", "bmp"}) + +_TYPE_ALIASES = { + "jpeg": "jpeg", + "jpg": "jpeg", + "pjpeg": "jpeg", + "png": "png", + "gif": "gif", + "webp": "webp", + "bmp": "bmp", +} + + +def normalize_image_type(image_type: str | None) -> str | None: + if not image_type: + return None + t = str(image_type).strip().lower() + t = t.removeprefix("image/") + t = _TYPE_ALIASES.get(t, t) + return t if t in _ALLOWED_TYPES else None + + +def content_hash_hex(image_bytes: bytes) -> str: + return hashlib.sha256(image_bytes).hexdigest() + + +def detect_image_format_from_magic(image_bytes: bytes) -> str | None: + """ + Identify image format from file signature (magic bytes). Returns normalized + type key (png, jpeg, gif, webp, bmp) or None if unknown / too short / not allowed. + """ + if not isinstance(image_bytes, (bytes, bytearray)) or len(image_bytes) < 4: + return None + b = bytes(image_bytes) + if len(b) >= 8 and b[:8] == b"\x89PNG\r\n\x1a\n": + return "png" + if len(b) >= 3 and b[0:3] == b"\xff\xd8\xff": + return "jpeg" + if len(b) >= 6 and b[0:6] in (b"GIF87a", b"GIF89a"): + return "gif" + if len(b) >= 12 and b[0:4] == b"RIFF" and b[8:12] == b"WEBP": + return "webp" + if len(b) >= 2 and b[0:2] == b"BM": + return "bmp" + return None + + +def validate_sticker_payload( + image_bytes: bytes, + image_type: str | None, +) -> tuple[str, str]: + """ + Returns (normalized_image_type, content_hash_hex). + + Declared image_type must match the format detected from magic bytes; stored + type is the normalized detected format. + + Raises ValueError with a short reason on invalid input. + """ + if not isinstance(image_bytes, (bytes, bytearray)): + msg = "invalid_image_bytes" + raise ValueError(msg) + if len(image_bytes) == 0: + msg = "empty_image" + raise ValueError(msg) + if len(image_bytes) > MAX_STICKER_BYTES: + msg = "image_too_large" + raise ValueError(msg) + + nt = normalize_image_type(image_type) + if not nt: + msg = "invalid_image_type" + raise ValueError(msg) + + detected = detect_image_format_from_magic(image_bytes) + if not detected: + msg = "invalid_image_signature" + raise ValueError(msg) + if detected != nt: + msg = "magic_type_mismatch" + raise ValueError(msg) + + h = content_hash_hex(bytes(image_bytes)) + return detected, h + + +_EXPORT_FORMAT = "meshchatx-stickers" +_EXPORT_VERSION = 1 + + +def validate_export_document(data: object) -> list[dict]: + """ + Parse and lightly validate an import JSON document. + Returns a list of sticker dicts with keys: name, image_type, image_bytes (str base64), + source_message_hash (optional). + """ + if not isinstance(data, dict): + msg = "invalid_document" + raise ValueError(msg) + if data.get("format") != _EXPORT_FORMAT: + msg = "invalid_format" + raise ValueError(msg) + try: + ver = int(data["version"]) + except (KeyError, TypeError, ValueError) as exc: + msg = "unsupported_version" + raise ValueError(msg) from exc + if ver != _EXPORT_VERSION: + msg = "unsupported_version" + raise ValueError(msg) + stickers = data.get("stickers") + if not isinstance(stickers, list): + msg = "invalid_stickers_array" + raise ValueError(msg) + out: list[dict] = [] + for i, item in enumerate(stickers): + if not isinstance(item, dict): + msg = f"invalid_sticker_at_{i}" + raise ValueError(msg) + name = item.get("name") + image_type = item.get("image_type") + image_b64 = item.get("image_bytes") + if not isinstance(image_b64, str) or not image_b64.strip(): + msg = f"missing_image_bytes_at_{i}" + raise ValueError(msg) + try: + base64.b64decode(image_b64.strip(), validate=False) + except (ValueError, TypeError) as exc: + msg = f"invalid_base64_at_{i}" + raise ValueError(msg) from exc + src = item.get("source_message_hash") + out.append( + { + "name": name if isinstance(name, str) else None, + "image_type": image_type, + "image_bytes_b64": image_b64.strip(), + "source_message_hash": src if isinstance(src, str) else None, + }, + ) + return out + + +def build_export_document(stickers: list[dict], exported_at_iso: str) -> dict: + """stickers: rows with name, image_type, image_bytes (base64 str), source_message_hash.""" + return { + "format": _EXPORT_FORMAT, + "version": _EXPORT_VERSION, + "exported_at": exported_at_iso, + "stickers": stickers, + } + + +def mime_for_image_type(normalized_type: str) -> str: + return { + "jpeg": "image/jpeg", + "png": "image/png", + "gif": "image/gif", + "webp": "image/webp", + "bmp": "image/bmp", + }.get(normalized_type, "application/octet-stream") + + +def sanitize_sticker_name(name: str | None) -> str | None: + if name is None: + return None + s = "".join(ch for ch in str(name).strip() if ch.isprintable()) + if not s: + return None + return s[:128]