# Persists gemini_webapi ChatSession metadata (the opaque [cid, rid, rcid, ...] # list — see ChatSession.metadata in gemini_webapi/client.py) per AI Assistant # conversation, so /editor/chat can reconstruct real Gemini-side conversation # memory across requests instead of resending the full history as text every # turn (see routers/editor.py's editor_chat). # # Storage is a plain JSON file per chat_id under static/chat_sessions/ — no DB, # matching this project's existing "small file-based helpers" style (see # srt_utils.py, thumbnailStore-equivalent patterns). This disk is EPHEMERAL on # a Hugging Face Space (wiped on restart/redeploy), which is fine BY DESIGN: # editor_chat treats a missing/corrupt session file as "start fresh, resend # history" rather than an error — this module only ever degrades gracefully, # never raises. import os import json import re CHAT_SESSIONS_DIR = os.path.join(os.path.dirname(__file__), "static", "chat_sessions") # chat_id comes straight from the client (EditorChatRequest.chat_id) and is # used to build a filesystem path — validate it's a safe token (matches the # frontend's crypto.randomUUID()-style ids) before ever touching disk, so a # malicious chat_id can't path-traverse out of CHAT_SESSIONS_DIR. _SAFE_CHAT_ID = re.compile(r'^[a-zA-Z0-9_-]{1,64}$') def _session_path(chat_id: str): if not chat_id or not _SAFE_CHAT_ID.match(chat_id): return None return os.path.join(CHAT_SESSIONS_DIR, f"{chat_id}.json") def load_session(chat_id: str): """Returns the persisted ChatSession metadata list for `chat_id`, or None if there isn't one (never seen before, invalid id, or the file is missing/corrupt) — callers should treat None as "start a fresh session".""" path = _session_path(chat_id) if not path or not os.path.exists(path): return None try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) metadata = data.get("metadata") return metadata if isinstance(metadata, list) else None except Exception: return None def save_session(chat_id: str, metadata: list) -> None: """Best-effort persist — a write failure (e.g. read-only filesystem) should never break the chat turn that already succeeded, so this never raises; the next request just falls back to a fresh session.""" path = _session_path(chat_id) if not path or not isinstance(metadata, list): return try: os.makedirs(CHAT_SESSIONS_DIR, exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump({"metadata": metadata}, f, ensure_ascii=False) except Exception: pass