my_voice_assistant_v3/app/store.py

"""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 = ""


@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 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
                );
                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
                );
                """
            )

    # ----- Nutzer -----------------------------------------------------------
    def _row_to_user(self, row: sqlite3.Row) -> User:
        return User(
            id=row["id"],
            display_name=row["display_name"],
            prefs=json.loads(row["prefs_json"] or "{}"),
            created_at=row["created_at"],
        )

    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)

    # ----- 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()),
            )