459 lines
18 KiB
Python
459 lines
18 KiB
Python
"""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()
|