"""会话持久化服务:sessions + session_messages 表(持久卷 /data/xtc.db)。 手表端可保存对话历史,重装不丢(持久卷保证)。 """ from __future__ import annotations import json import time import uuid from typing import Optional from ..database import get_conn def _now_ts() -> int: return int(time.time()) def create_session( *, access_key: str, title: Optional[str] = None, provider: Optional[str] = None, model: Optional[str] = None, ) -> dict: sid = uuid.uuid4().hex now = _now_ts() with get_conn() as conn: conn.execute( "INSERT INTO sessions(id, title, provider, model, access_key, created_at, updated_at) " "VALUES(?,?,?,?,?,?,?)", (sid, title or "新会话", provider, model, access_key, now, now), ) sess = {"id": sid, "title": title or "新会话", "provider": provider, "model": model, "created_at": now, "updated_at": now} # Webhook 通知 try: from ..services import webhook_store webhook_store.notify_fire_and_forget("session.created", {"session": sess, "access_key": access_key}) except Exception: pass return sess def list_sessions( *, access_key: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> list[dict]: sql = "SELECT id, title, provider, model, access_key, created_at, updated_at FROM sessions" args: list = [] if access_key: sql += " WHERE access_key = ?" args.append(access_key) sql += " ORDER BY updated_at DESC LIMIT ? OFFSET ?" args += [limit, offset] with get_conn() as conn: rows = conn.execute(sql, args).fetchall() return [dict(r) for r in rows] def get_session(session_id: str, *, access_key: Optional[str] = None) -> Optional[dict]: with get_conn() as conn: row = conn.execute( "SELECT * FROM sessions WHERE id = ?" + (" AND access_key = ?" if access_key else ""), (session_id,) if not access_key else (session_id, access_key), ).fetchone() return dict(row) if row else None def append_message( *, session_id: str, role: str, content: str, thought: Optional[str] = None, provider: Optional[str] = None, model: Optional[str] = None, file_keys: Optional[list[str]] = None, ) -> dict: now = _now_ts() with get_conn() as conn: # 取下一个 seq row = conn.execute( "SELECT COALESCE(MAX(seq), 0) AS max_seq FROM session_messages WHERE session_id = ?", (session_id,), ).fetchone() next_seq = int(row["max_seq"]) + 1 conn.execute( "INSERT INTO session_messages(session_id, seq, role, content, thought, provider, model, ts, file_keys) " "VALUES(?,?,?,?,?,?,?,?,?)", ( session_id, next_seq, role, content, thought, provider, model, now, json.dumps(file_keys) if file_keys else None, ), ) # 更新 session 时间 new_title = None if role == "user" and next_seq == 1: # 第一条 user 消息自动作为标题 new_title = content[:40] + ("..." if len(content) > 40 else "") conn.execute( "UPDATE sessions SET updated_at = ?, title = ? WHERE id = ?", (now, new_title, session_id), ) else: conn.execute( "UPDATE sessions SET updated_at = ? WHERE id = ?", (now, session_id), ) return { "session_id": session_id, "seq": next_seq, "role": role, "content": content, "thought": thought, "provider": provider, "model": model, "ts": now, "file_keys": file_keys, } def list_messages(session_id: str, *, access_key: Optional[str] = None) -> Optional[list[dict]]: sess = get_session(session_id, access_key=access_key) if not sess: return None with get_conn() as conn: rows = conn.execute( "SELECT seq, role, content, thought, provider, model, ts, file_keys FROM session_messages " "WHERE session_id = ? ORDER BY seq ASC", (session_id,), ).fetchall() out = [] for r in rows: item = dict(r) if r["file_keys"]: try: item["file_keys"] = json.loads(r["file_keys"]) except Exception: item["file_keys"] = [] else: item["file_keys"] = [] out.append(item) return out def delete_session(session_id: str, *, access_key: Optional[str] = None) -> bool: sess = get_session(session_id, access_key=access_key) if not sess: return False with get_conn() as conn: conn.execute("DELETE FROM session_messages WHERE session_id = ?", (session_id,)) conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,)) # Webhook 通知 try: from ..services import webhook_store webhook_store.notify_fire_and_forget("session.deleted", {"session_id": session_id, "access_key": access_key}) except Exception: pass return True def update_session( session_id: str, *, title: Optional[str] = None, access_key: Optional[str] = None, ) -> Optional[dict]: sess = get_session(session_id, access_key=access_key) if not sess: return None sets = [] args: list = [] if title is not None: sets.append("title = ?") args.append(title) if not sets: return sess sets.append("updated_at = ?") args.append(_now_ts()) args.append(session_id) with get_conn() as conn: conn.execute( f"UPDATE sessions SET {', '.join(sets)} WHERE id = ?", tuple(args), ) return get_session(session_id)