File size: 2,688 Bytes
e00e8a8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | # 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 |