"""Application database (SQLite): documents, audit log, and managed prompts. This is the real persistence layer for the retail IT deployment — every processed document, every audit event, and every prompt version is stored here. (Metrics/KPIs live in app/metrics.py; the vector index lives in app/rag_store.py.) Production swap: point APP_DB_PATH at Postgres-compatible storage, or replace the thin DAO with SQLAlchemy — the call sites only use the methods below. """ from __future__ import annotations import json import sqlite3 import threading import time from pathlib import Path class Database: def __init__(self, db_path: str | Path) -> None: self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) self._lock = threading.Lock() self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False) self._conn.row_factory = sqlite3.Row self._init() self._seed_prompts() def _init(self): with self._lock: self._conn.executescript( """ CREATE TABLE IF NOT EXISTS documents ( run_id TEXT PRIMARY KEY, doc_id TEXT, doc_type TEXT, category TEXT, channel TEXT, ocr_backend TEXT, confidence REAL, posted INTEGER, requires_review INTEGER, total_cost_usd REAL, total_tokens INTEGER, latency_ms REAL, flags TEXT, extracted TEXT, created_at REAL); CREATE INDEX IF NOT EXISTS idx_docs_cat ON documents(category); CREATE TABLE IF NOT EXISTS audit_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, action TEXT, run_id TEXT, actor TEXT, detail TEXT); CREATE TABLE IF NOT EXISTS prompts ( name TEXT PRIMARY KEY, version INTEGER, content TEXT, updated_at REAL, updated_by TEXT); CREATE TABLE IF NOT EXISTS prompt_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, version INTEGER, content TEXT, ts REAL, actor TEXT); """ ) self._conn.commit() # --- documents ----------------------------------------------------------- def save_document(self, run: dict) -> None: res = run.get("result", {}) or {} with self._lock: self._conn.execute( """INSERT OR REPLACE INTO documents (run_id, doc_id, doc_type, category, channel, ocr_backend, confidence, posted, requires_review, total_cost_usd, total_tokens, latency_ms, flags, extracted, created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (run.get("run_id"), run.get("doc_id"), run.get("doc_type"), run.get("category"), run.get("channel"), run.get("ocr_backend"), run.get("confidence"), int(bool(res.get("posted"))), int(bool(run.get("hitl"))), run.get("total_cost_usd", 0.0), (run.get("total_input_tokens", 0) + run.get("total_output_tokens", 0)), run.get("latency_ms", 0.0), json.dumps(run.get("flags", [])), json.dumps(res.get("extracted")), time.time()), ) self._conn.commit() def list_documents(self, category: str | None = None, limit: int = 100) -> list[dict]: with self._lock: if category and category != "all": rows = self._conn.execute( "SELECT * FROM documents WHERE category=? ORDER BY created_at DESC LIMIT ?", (category, limit)).fetchall() else: rows = self._conn.execute( "SELECT * FROM documents ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall() return [self._doc_row(r) for r in rows] def get_document(self, run_id: str) -> dict | None: with self._lock: r = self._conn.execute("SELECT * FROM documents WHERE run_id=?", (run_id,)).fetchone() return self._doc_row(r) if r else None def category_counts(self) -> dict: with self._lock: rows = self._conn.execute( "SELECT category, COUNT(*) c FROM documents GROUP BY category").fetchall() return {r["category"]: r["c"] for r in rows} @staticmethod def _doc_row(r: sqlite3.Row) -> dict: d = dict(r) for k in ("flags", "extracted"): if d.get(k): try: d[k] = json.loads(d[k]) except (json.JSONDecodeError, TypeError): pass d["posted"] = bool(d.get("posted")) d["requires_review"] = bool(d.get("requires_review")) return d # --- audit --------------------------------------------------------------- def audit(self, action: str, run_id: str | None = None, actor: str = "system", detail: dict | None = None) -> None: with self._lock: self._conn.execute( "INSERT INTO audit_log (ts, action, run_id, actor, detail) VALUES (?,?,?,?,?)", (time.time(), action, run_id, actor, json.dumps(detail or {}))) self._conn.commit() def recent_audit(self, limit: int = 100) -> list[dict]: with self._lock: rows = self._conn.execute( "SELECT * FROM audit_log ORDER BY ts DESC LIMIT ?", (limit,)).fetchall() out = [] for r in rows: d = dict(r) try: d["detail"] = json.loads(d["detail"] or "{}") except (json.JSONDecodeError, TypeError): pass out.append(d) return out # --- prompt management --------------------------------------------------- def _seed_prompts(self): from .prompts import ( BROWSER_AGENT_SYSTEM, CLASSIFY_SYSTEM, COMPLEX_REFINE_SYSTEM, EXTRACT_SYSTEM, NORMALIZE_SYSTEM, ) defaults = { "classify": CLASSIFY_SYSTEM, "extract": EXTRACT_SYSTEM, "normalize": NORMALIZE_SYSTEM, "browser_agent": BROWSER_AGENT_SYSTEM, "complex_refine": COMPLEX_REFINE_SYSTEM, } with self._lock: for name, content in defaults.items(): exists = self._conn.execute( "SELECT 1 FROM prompts WHERE name=?", (name,)).fetchone() if not exists: self._conn.execute( "INSERT INTO prompts (name, version, content, updated_at, updated_by) " "VALUES (?,?,?,?,?)", (name, 1, content, time.time(), "seed")) self._conn.commit() def list_prompts(self) -> list[dict]: with self._lock: rows = self._conn.execute( "SELECT name, version, length(content) AS chars, updated_at, updated_by " "FROM prompts ORDER BY name").fetchall() return [dict(r) for r in rows] def get_prompt(self, name: str) -> dict | None: with self._lock: r = self._conn.execute("SELECT * FROM prompts WHERE name=?", (name,)).fetchone() return dict(r) if r else None def set_prompt(self, name: str, content: str, actor: str = "admin") -> dict: with self._lock: cur = self._conn.execute("SELECT version FROM prompts WHERE name=?", (name,)).fetchone() version = (cur["version"] + 1) if cur else 1 self._conn.execute( "INSERT OR REPLACE INTO prompts (name, version, content, updated_at, updated_by) " "VALUES (?,?,?,?,?)", (name, version, content, time.time(), actor)) self._conn.execute( "INSERT INTO prompt_history (name, version, content, ts, actor) VALUES (?,?,?,?,?)", (name, version, content, time.time(), actor)) self._conn.commit() return {"name": name, "version": version} def prompt_history(self, name: str, limit: int = 20) -> list[dict]: with self._lock: rows = self._conn.execute( "SELECT version, ts, actor, length(content) chars FROM prompt_history " "WHERE name=? ORDER BY version DESC LIMIT ?", (name, limit)).fetchall() return [dict(r) for r in rows]