"""SQLite persistence for users, translation sessions, and one prior revision.""" import json import os import sqlite3 import threading import uuid from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterator from dotenv import load_dotenv from backend.models import TranslationSession PROJECT_ROOT = Path(__file__).parent.parent load_dotenv(PROJECT_ROOT / ".env") class SessionConflictError(Exception): """Raised when an optimistic session update uses a stale version.""" class SessionArchivedError(Exception): """Raised when content mutation is attempted on an archived session.""" class LastAdminError(Exception): """Raised when an update would remove the final active administrator.""" def _now() -> str: return datetime.now(timezone.utc).isoformat() def _default_title(source_text: str) -> str: first_line = next((line.strip() for line in source_text.splitlines() if line.strip()), "") return first_line[:80] or "새 번역" class Database: def __init__(self, path: str | Path | None = None): configured = path or os.getenv( "DATABASE_PATH", str(PROJECT_ROOT / "data" / "translator.db"), ) self.path = Path(configured) self.path.parent.mkdir(parents=True, exist_ok=True) self._lock = threading.RLock() self.initialize() @contextmanager def connect(self) -> Iterator[sqlite3.Connection]: connection = sqlite3.connect(self.path, timeout=10, isolation_level=None) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") connection.execute("PRAGMA busy_timeout = 10000") try: yield connection finally: connection.close() def initialize(self) -> None: with self._lock, self.connect() as connection: connection.execute("PRAGMA journal_mode = WAL") connection.executescript( """ CREATE TABLE IF NOT EXISTS schema_meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, password_hash TEXT NOT NULL, is_admin INTEGER NOT NULL DEFAULT 0, is_active INTEGER NOT NULL DEFAULT 1, token_version INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS translation_sessions ( id TEXT PRIMARY KEY, owner_id TEXT NOT NULL REFERENCES users(id) ON DELETE RESTRICT, title TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'draft', current_phase INTEGER NOT NULL DEFAULT 0, version INTEGER NOT NULL DEFAULT 0, data_json TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, completed_at TEXT, archived_at TEXT ); CREATE INDEX IF NOT EXISTS idx_sessions_owner_updated ON translation_sessions(owner_id, archived_at, updated_at DESC); CREATE TABLE IF NOT EXISTS session_revisions ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL UNIQUE REFERENCES translation_sessions(id) ON DELETE CASCADE, snapshot_json TEXT NOT NULL, reason TEXT NOT NULL, created_at TEXT NOT NULL ); """ ) connection.execute( "INSERT OR REPLACE INTO schema_meta(key, value) VALUES('schema_version', '1')" ) # Users def count_users(self) -> int: with self.connect() as connection: return int(connection.execute("SELECT COUNT(*) FROM users").fetchone()[0]) def count_admins(self) -> int: with self.connect() as connection: return int( connection.execute( "SELECT COUNT(*) FROM users WHERE is_admin = 1 AND is_active = 1" ).fetchone()[0] ) def get_user(self, user_id: str) -> dict[str, Any] | None: with self.connect() as connection: row = connection.execute( "SELECT * FROM users WHERE id = ?", (user_id,) ).fetchone() return dict(row) if row else None def list_users(self) -> list[dict[str, Any]]: with self.connect() as connection: rows = connection.execute( "SELECT id, is_admin, is_active, created_at, updated_at " "FROM users ORDER BY id" ).fetchall() return [dict(row) for row in rows] def create_user( self, user_id: str, password_hash: str, *, is_admin: bool = False ) -> None: now = _now() with self._lock, self.connect() as connection: connection.execute( "INSERT INTO users(id, password_hash, is_admin, is_active, " "token_version, created_at, updated_at) VALUES(?, ?, ?, 1, 0, ?, ?)", (user_id, password_hash, int(is_admin), now, now), ) def update_user( self, user_id: str, *, password_hash: str | None = None, is_admin: bool | None = None, is_active: bool | None = None, ) -> bool: assignments: list[str] = [] values: list[Any] = [] invalidate_tokens = False if password_hash is not None: assignments.append("password_hash = ?") values.append(password_hash) invalidate_tokens = True if is_admin is not None: assignments.append("is_admin = ?") values.append(int(is_admin)) if is_active is not None: assignments.append("is_active = ?") values.append(int(is_active)) if not is_active: invalidate_tokens = True if invalidate_tokens: assignments.append("token_version = token_version + 1") if not assignments: return self.get_user(user_id) is not None assignments.append("updated_at = ?") values.extend([_now(), user_id]) with self._lock, self.connect() as connection: connection.execute("BEGIN IMMEDIATE") try: current = connection.execute( "SELECT is_admin, is_active FROM users WHERE id = ?", (user_id,) ).fetchone() if not current: connection.execute("ROLLBACK") return False removes_admin = current["is_admin"] and current["is_active"] and ( is_admin is False or is_active is False ) if removes_admin: admin_count = connection.execute( "SELECT COUNT(*) FROM users WHERE is_admin = 1 AND is_active = 1" ).fetchone()[0] if admin_count <= 1: connection.execute("ROLLBACK") raise LastAdminError cursor = connection.execute( f"UPDATE users SET {', '.join(assignments)} WHERE id = ?", values ) connection.execute("COMMIT") return cursor.rowcount == 1 except Exception: if connection.in_transaction: connection.execute("ROLLBACK") raise # Sessions def create_session( self, session: TranslationSession, owner_id: str, title: str | None = None ) -> str: session_id = uuid.uuid4().hex[:16] now = _now() data = session.model_copy(update={"progress": None}).model_dump(mode="json") phase = self._phase_from_data(data) status = "completed" if phase >= 4 else "draft" with self._lock, self.connect() as connection: connection.execute( "INSERT INTO translation_sessions(id, owner_id, title, status, current_phase, " "data_json, created_at, updated_at, completed_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)", ( session_id, owner_id, title or _default_title(session.source_text), status, phase, json.dumps(data, ensure_ascii=False), now, now, now if phase >= 4 else None, ), ) return session_id def get_session( self, session_id: str, owner_id: str ) -> tuple[TranslationSession, int, dict[str, Any]] | None: with self.connect() as connection: row = connection.execute( "SELECT * FROM translation_sessions WHERE id = ? AND owner_id = ?", (session_id, owner_id), ).fetchone() if not row: return None metadata = dict(row) return TranslationSession(**json.loads(row["data_json"])), row["version"], metadata def update_session( self, session_id: str, owner_id: str, patch: dict[str, Any], *, expected_version: int | None = None, snapshot_reason: str | None = None, current_phase: int | None = None, ) -> tuple[TranslationSession, int] | None: with self._lock, self.connect() as connection: connection.execute("BEGIN IMMEDIATE") try: row = connection.execute( "SELECT * FROM translation_sessions WHERE id = ? AND owner_id = ?", (session_id, owner_id), ).fetchone() if not row: connection.execute("ROLLBACK") return None if row["archived_at"] is not None: connection.execute("ROLLBACK") raise SessionArchivedError if expected_version is not None and row["version"] != expected_version: connection.execute("ROLLBACK") raise SessionConflictError old_data = json.loads(row["data_json"]) if snapshot_reason and any( old_data.get(key) for key in ("phase1_result", "phase3_result", "phase4_result") ): connection.execute( "INSERT INTO session_revisions(session_id, snapshot_json, reason, created_at) " "VALUES(?, ?, ?, ?) ON CONFLICT(session_id) DO UPDATE SET " "snapshot_json = excluded.snapshot_json, reason = excluded.reason, " "created_at = excluded.created_at", (session_id, row["data_json"], snapshot_reason, _now()), ) merged = {**old_data, **patch, "progress": None} session = TranslationSession(**merged) version = row["version"] + 1 phase = current_phase if current_phase is not None else row["current_phase"] status = "completed" if phase >= 4 else "draft" completed_at = _now() if phase >= 4 else None title = row["title"] if patch.get("source_text") is not None and title in {"새 번역", _default_title(old_data.get("source_text", ""))}: title = _default_title(session.source_text) cursor = connection.execute( "UPDATE translation_sessions SET title = ?, status = ?, current_phase = ?, " "version = ?, data_json = ?, updated_at = ?, completed_at = ? " "WHERE id = ? AND owner_id = ? AND version = ?", ( title, status, phase, version, json.dumps(session.model_dump(mode="json"), ensure_ascii=False), _now(), completed_at, session_id, owner_id, row["version"], ), ) if cursor.rowcount != 1: connection.execute("ROLLBACK") raise SessionConflictError connection.execute("COMMIT") return session, version except Exception: if connection.in_transaction: connection.execute("ROLLBACK") raise def list_sessions( self, owner_id: str, *, include_archived: bool = False, limit: int = 50 ) -> list[dict[str, Any]]: archived_clause = "" if include_archived else "AND archived_at IS NULL" with self.connect() as connection: rows = connection.execute( f"SELECT id, title, status, current_phase, version, data_json, " f"created_at, updated_at, completed_at, archived_at " f"FROM translation_sessions WHERE owner_id = ? {archived_clause} " f"ORDER BY updated_at DESC LIMIT ?", (owner_id, min(max(limit, 1), 100)), ).fetchall() summaries = [] for row in rows: data = json.loads(row["data_json"]) summaries.append( { "session_id": row["id"], "title": row["title"], "source_preview": data.get("source_text", "")[:120], "target_language": data.get("target_language", "한국어"), "status": row["status"], "current_phase": row["current_phase"], "version": row["version"], "created_at": row["created_at"], "updated_at": row["updated_at"], "archived_at": row["archived_at"], } ) return summaries def set_archived(self, session_id: str, owner_id: str, archived: bool) -> bool: with self._lock, self.connect() as connection: row = connection.execute( "SELECT current_phase FROM translation_sessions WHERE id = ? AND owner_id = ?", (session_id, owner_id), ).fetchone() if not row: return False cursor = connection.execute( "UPDATE translation_sessions SET archived_at = ?, status = ?, updated_at = ?, " "version = version + 1 " "WHERE id = ? AND owner_id = ?", ( _now() if archived else None, "archived" if archived else ("completed" if row["current_phase"] >= 4 else "draft"), _now(), session_id, owner_id, ), ) return cursor.rowcount == 1 @staticmethod def _phase_from_data(data: dict[str, Any]) -> int: if data.get("phase4_result"): return 4 if data.get("phase3_result"): return 3 if data.get("phase2_confirmed"): return 2 if data.get("phase1_result"): return 1 return 0 def rename_session(self, session_id: str, owner_id: str, title: str) -> bool: with self._lock, self.connect() as connection: cursor = connection.execute( "UPDATE translation_sessions SET title = ?, updated_at = ? " "WHERE id = ? AND owner_id = ?", (title.strip()[:120] or "새 번역", _now(), session_id, owner_id), ) return cursor.rowcount == 1 def delete_session(self, session_id: str, owner_id: str) -> bool: with self._lock, self.connect() as connection: cursor = connection.execute( "DELETE FROM translation_sessions WHERE id = ? AND owner_id = ?", (session_id, owner_id), ) return cursor.rowcount == 1 def clone_session(self, session_id: str, owner_id: str) -> str | None: result = self.get_session(session_id, owner_id) if not result: return None session, _, metadata = result clone = session.model_copy(update={"progress": None}) return self.create_session(clone, owner_id, f"{metadata['title']} (복사본)") def get_revision(self, session_id: str, owner_id: str) -> dict[str, Any] | None: with self.connect() as connection: row = connection.execute( "SELECT r.snapshot_json, r.reason, r.created_at FROM session_revisions r " "JOIN translation_sessions s ON s.id = r.session_id " "WHERE r.session_id = ? AND s.owner_id = ?", (session_id, owner_id), ).fetchone() if not row: return None return { "data": json.loads(row["snapshot_json"]), "reason": row["reason"], "created_at": row["created_at"], } def restore_revision(self, session_id: str, owner_id: str) -> bool: revision = self.get_revision(session_id, owner_id) if not revision: return False result = self.get_session(session_id, owner_id) if not result: return False _, version, _ = result data = revision["data"] phase = self._phase_from_data(data) restored = self.update_session( session_id, owner_id, data, expected_version=version, snapshot_reason="revision_restore", current_phase=phase, ) return restored is not None def healthcheck(self) -> bool: with self.connect() as connection: return connection.execute("SELECT 1").fetchone()[0] == 1 database = Database()