chat_sessions.py

# 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