"""Persistenzschicht: Nutzer und Sessions. Ein abstraktes Store-Interface mit SQLite-Default (stdlib). Spaetere Backends (Postgres/Redis) koennen dasselbe Interface implementieren, ohne die App zu aendern. """ from __future__ import annotations import json import hashlib import secrets import sqlite3 import uuid from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path ANONYMOUS_USER_ID = "anonymous" def hash_token(raw_token: str) -> str: return hashlib.sha256(raw_token.encode("utf-8")).hexdigest() def _now() -> str: return datetime.now(timezone.utc).isoformat() @dataclass class User: id: str display_name: str prefs: dict = field(default_factory=dict) created_at: str = "" external_id: str | None = None # SSO-/Proxy-Identitaet (Forward-Auth) is_admin: bool = False # transient, aus ADMIN_USERS abgeleitet @dataclass class Session: id: str user_id: str data: dict = field(default_factory=dict) @dataclass class Memory: id: int content: str created_at: str = "" class SessionOwnershipError(Exception): """Eine Session gehoert einem anderen Nutzer (-> HTTP 403).""" class Store(ABC): @abstractmethod def create_user(self, display_name: str) -> tuple[User, str]: """Legt einen Nutzer an und liefert (User, Klartext-Token). Token nur hier sichtbar.""" @abstractmethod def get_user_by_token(self, raw_token: str) -> User | None: ... @abstractmethod def get_user(self, user_id: str) -> User | None: ... @abstractmethod def set_user_prefs(self, user_id: str, prefs: dict) -> User: ... @abstractmethod def ensure_anonymous_user(self) -> User: ... @abstractmethod def list_users(self) -> list[User]: ... @abstractmethod def get_user_by_external_id(self, external_id: str) -> User | None: ... @abstractmethod def get_or_create_user_by_external_id( self, external_id: str, display_name: str | None = None ) -> User: ... @abstractmethod def get_session(self, session_id: str) -> Session | None: ... @abstractmethod def update_session(self, session_id: str, user_id: str, values: dict) -> Session: """Erstellt/aktualisiert eine Session des Nutzers. Fremde Session -> SessionOwnershipError.""" @abstractmethod def append_message(self, session_id: str, user_id: str, role: str, content: str) -> None: """Haengt eine Nachricht an die Session an. Fremde Session -> SessionOwnershipError.""" @abstractmethod def get_recent_messages(self, session_id: str, limit: int) -> list[dict]: """Liefert die letzten `limit` Nachrichten chronologisch ([{'role','content'}, ...]).""" @abstractmethod def add_memory(self, user_id: str, content: str) -> Memory: """Speichert eine dauerhafte Erinnerung (Fakt/Vorliebe) zum Nutzer.""" @abstractmethod def get_memories(self, user_id: str) -> list[Memory]: """Liefert alle Erinnerungen des Nutzers (chronologisch).""" @abstractmethod def delete_memory(self, user_id: str, memory_id: int) -> bool: """Loescht eine Erinnerung des Nutzers. True, wenn etwas geloescht wurde.""" @abstractmethod def get_request_count(self, user_id: str, day: str | None = None) -> int: """Anzahl der Anfragen des Nutzers am angegebenen Tag (Default: heute, UTC).""" @abstractmethod def add_usage(self, user_id: str, units: int = 0, day: str | None = None) -> int: """Zaehlt eine Anfrage (+units) und liefert die neue Tages-Anfragezahl.""" @abstractmethod def log_emergency(self, user_id: str, category: str, snippet: str) -> None: """Protokolliert ein erkanntes Notfall-Signal (sensibel!).""" class SQLiteStore(Store): def __init__(self, db_path: str): self.db_path = db_path Path(db_path).parent.mkdir(parents=True, exist_ok=True) self._init_schema() def _connect(self) -> sqlite3.Connection: conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") return conn def _init_schema(self) -> None: with self._connect() as conn: conn.executescript( """ CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, display_name TEXT NOT NULL, token_hash TEXT NOT NULL UNIQUE, prefs_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL, external_id TEXT ); CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, data_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, id); CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_memories_user ON memories(user_id, id); CREATE TABLE IF NOT EXISTS usage ( user_id TEXT NOT NULL, day TEXT NOT NULL, requests INTEGER NOT NULL DEFAULT 0, units INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (user_id, day) ); CREATE TABLE IF NOT EXISTS emergency_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, category TEXT NOT NULL, snippet TEXT NOT NULL, created_at TEXT NOT NULL ); """ ) # Migration fuer bestehende DBs: external_id ergaenzen (falls noch nicht da). cols = {row["name"] for row in conn.execute("PRAGMA table_info(users)")} if "external_id" not in cols: conn.execute("ALTER TABLE users ADD COLUMN external_id TEXT") # NULLs gelten in SQLite als verschieden -> Alt-Nutzer ohne external_id ok. conn.execute( "CREATE UNIQUE INDEX IF NOT EXISTS idx_users_external" " ON users(external_id)" ) # ----- Nutzer ----------------------------------------------------------- def _row_to_user(self, row: sqlite3.Row) -> User: keys = row.keys() return User( id=row["id"], display_name=row["display_name"], prefs=json.loads(row["prefs_json"] or "{}"), created_at=row["created_at"], external_id=row["external_id"] if "external_id" in keys else None, ) def create_user(self, display_name: str) -> tuple[User, str]: raw_token = secrets.token_urlsafe(32) user = User(id=uuid.uuid4().hex, display_name=display_name, prefs={}, created_at=_now()) with self._connect() as conn: conn.execute( "INSERT INTO users (id, display_name, token_hash, prefs_json, created_at)" " VALUES (?, ?, ?, ?, ?)", (user.id, user.display_name, hash_token(raw_token), "{}", user.created_at), ) return user, raw_token def get_user_by_token(self, raw_token: str) -> User | None: with self._connect() as conn: row = conn.execute( "SELECT * FROM users WHERE token_hash = ?", (hash_token(raw_token),) ).fetchone() return self._row_to_user(row) if row else None def get_user(self, user_id: str) -> User | None: with self._connect() as conn: row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone() return self._row_to_user(row) if row else None def set_user_prefs(self, user_id: str, prefs: dict) -> User: with self._connect() as conn: conn.execute( "UPDATE users SET prefs_json = ? WHERE id = ?", (json.dumps(prefs), user_id), ) row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone() if row is None: raise KeyError(f"Unbekannter Nutzer: {user_id}") return self._row_to_user(row) def ensure_anonymous_user(self) -> User: existing = self.get_user(ANONYMOUS_USER_ID) if existing: return existing with self._connect() as conn: conn.execute( "INSERT OR IGNORE INTO users (id, display_name, token_hash, prefs_json, created_at)" " VALUES (?, ?, ?, ?, ?)", (ANONYMOUS_USER_ID, "Anonymous", f"anon-{ANONYMOUS_USER_ID}", "{}", _now()), ) return self.get_user(ANONYMOUS_USER_ID) def list_users(self) -> list[User]: with self._connect() as conn: rows = conn.execute( "SELECT * FROM users WHERE id != ? ORDER BY created_at", (ANONYMOUS_USER_ID,), ).fetchall() return [self._row_to_user(row) for row in rows] def get_user_by_external_id(self, external_id: str) -> User | None: with self._connect() as conn: row = conn.execute( "SELECT * FROM users WHERE external_id = ?", (external_id,) ).fetchone() return self._row_to_user(row) if row else None def get_or_create_user_by_external_id( self, external_id: str, display_name: str | None = None ) -> User: """Findet den Nutzer zur SSO-/Proxy-Identitaet oder legt ihn an (Forward-Auth).""" existing = self.get_user_by_external_id(external_id) if existing: return existing user = User( id=uuid.uuid4().hex, display_name=display_name or external_id, prefs={}, created_at=_now(), external_id=external_id, ) with self._connect() as conn: conn.execute( "INSERT INTO users (id, display_name, token_hash, prefs_json, created_at," " external_id) VALUES (?, ?, ?, ?, ?, ?)", # token_hash ist NOT NULL UNIQUE -> synthetischer, kollisionsfreier Platzhalter # (SSO-Nutzer authentifizieren sich nicht ueber ein Token). (user.id, user.display_name, f"ext:{external_id}", "{}", user.created_at, external_id), ) return user # ----- Sessions --------------------------------------------------------- def get_session(self, session_id: str) -> Session | None: with self._connect() as conn: row = conn.execute( "SELECT * FROM sessions WHERE id = ?", (session_id,) ).fetchone() if row is None: return None return Session(id=row["id"], user_id=row["user_id"], data=json.loads(row["data_json"] or "{}")) def update_session(self, session_id: str, user_id: str, values: dict) -> Session: existing = self.get_session(session_id) if existing and existing.user_id != user_id: raise SessionOwnershipError( f"Session {session_id!r} gehoert einem anderen Nutzer" ) data = dict(existing.data) if existing else {} data.update({k: v for k, v in values.items() if v is not None}) payload = json.dumps(data) now = _now() with self._connect() as conn: if existing: conn.execute( "UPDATE sessions SET data_json = ?, updated_at = ? WHERE id = ?", (payload, now, session_id), ) else: conn.execute( "INSERT INTO sessions (id, user_id, data_json, created_at, updated_at)" " VALUES (?, ?, ?, ?, ?)", (session_id, user_id, payload, now, now), ) return Session(id=session_id, user_id=user_id, data=data) # ----- Nachrichten / Gespraechsverlauf ---------------------------------- def append_message(self, session_id: str, user_id: str, role: str, content: str) -> None: existing = self.get_session(session_id) if existing and existing.user_id != user_id: raise SessionOwnershipError( f"Session {session_id!r} gehoert einem anderen Nutzer" ) now = _now() with self._connect() as conn: if not existing: conn.execute( "INSERT INTO sessions (id, user_id, data_json, created_at, updated_at)" " VALUES (?, ?, ?, ?, ?)", (session_id, user_id, "{}", now, now), ) conn.execute( "INSERT INTO messages (session_id, role, content, created_at)" " VALUES (?, ?, ?, ?)", (session_id, role, content, now), ) def get_recent_messages(self, session_id: str, limit: int) -> list[dict]: if limit <= 0: return [] with self._connect() as conn: rows = conn.execute( "SELECT role, content FROM messages WHERE session_id = ?" " ORDER BY id DESC LIMIT ?", (session_id, limit), ).fetchall() return [{"role": row["role"], "content": row["content"]} for row in reversed(rows)] # ----- Langzeit-Erinnerungen -------------------------------------------- def add_memory(self, user_id: str, content: str) -> Memory: now = _now() with self._connect() as conn: cur = conn.execute( "INSERT INTO memories (user_id, content, created_at) VALUES (?, ?, ?)", (user_id, content, now), ) memory_id = cur.lastrowid return Memory(id=memory_id, content=content, created_at=now) def get_memories(self, user_id: str) -> list[Memory]: with self._connect() as conn: rows = conn.execute( "SELECT id, content, created_at FROM memories WHERE user_id = ? ORDER BY id", (user_id,), ).fetchall() return [ Memory(id=row["id"], content=row["content"], created_at=row["created_at"]) for row in rows ] def delete_memory(self, user_id: str, memory_id: int) -> bool: with self._connect() as conn: cur = conn.execute( "DELETE FROM memories WHERE id = ? AND user_id = ?", (memory_id, user_id), ) return cur.rowcount > 0 # ----- Nutzung / Quota -------------------------------------------------- @staticmethod def _today() -> str: return datetime.now(timezone.utc).date().isoformat() def get_request_count(self, user_id: str, day: str | None = None) -> int: day = day or self._today() with self._connect() as conn: row = conn.execute( "SELECT requests FROM usage WHERE user_id = ? AND day = ?", (user_id, day), ).fetchone() return int(row["requests"]) if row else 0 def add_usage(self, user_id: str, units: int = 0, day: str | None = None) -> int: day = day or self._today() with self._connect() as conn: conn.execute( "INSERT INTO usage (user_id, day, requests, units) VALUES (?, ?, 1, ?)" " ON CONFLICT(user_id, day) DO UPDATE SET" " requests = requests + 1, units = units + excluded.units", (user_id, day, units), ) row = conn.execute( "SELECT requests FROM usage WHERE user_id = ? AND day = ?", (user_id, day), ).fetchone() return int(row["requests"]) # ----- Notfall-Protokoll ------------------------------------------------ def log_emergency(self, user_id: str, category: str, snippet: str) -> None: with self._connect() as conn: conn.execute( "INSERT INTO emergency_events (user_id, category, snippet, created_at)" " VALUES (?, ?, ?, ?)", (user_id, category, snippet, _now()), )