"""Thin persistence for the run snapshot (CONTRACT §2). The Python service **reads and writes** the shared tables but **never creates or migrates** them (CI4 owns the schema). Two backends share one interface: * :class:`MySQLStore` — PyMySQL against the shared MySQL 8 database. * :class:`InMemoryStore` — a dict-backed fallback used by the offline runner and whenever ``MYSQL_*`` is unset, so the service/CLI runs with no DB. Writes cover: ``runs``, ``planned_lines``, ``invoices``, ``results``, ``traces``. Reads cover the memory tables ``name_map`` and ``memory_rules`` plus run read-back for the API. Review + memory *writes* are owned by PHP (CONTRACT §3) — not here. """ from __future__ import annotations import json import logging from datetime import datetime, timezone from typing import Any, Optional, Sequence from app.matching.models import Invoice, MatchResult, PlannedLine logger = logging.getLogger(__name__) def _now() -> datetime: return datetime.now(timezone.utc) class Store: """Persistence interface shared by the MySQL and in-memory backends.""" # --- runs ------------------------------------------------------------- def create_run( self, month: str, provider: str, model: str, source_filename: str ) -> int: raise NotImplementedError def set_run_status(self, run_id: int, status: str) -> None: raise NotImplementedError def finish_run( self, run_id: int, status: str, totals: Optional[dict[str, Any]] = None, error: Optional[str] = None, ) -> None: raise NotImplementedError # --- snapshot writes -------------------------------------------------- def save_planned_lines( self, run_id: int, lines: Sequence[PlannedLine] ) -> dict[int, int]: """Persist planned lines; return ``source_row -> planned_lines.id``.""" raise NotImplementedError def save_invoices(self, run_id: int, invoices: Sequence[Invoice]) -> dict[int, int]: """Persist invoices; return ``odoo_id -> invoices.id``.""" raise NotImplementedError def save_results( self, run_id: int, results: Sequence[MatchResult], planned_id_map: dict[int, int], invoice_id_map: dict[int, int], ) -> None: raise NotImplementedError def save_traces(self, run_id: int, rows: Sequence[dict[str, Any]]) -> None: raise NotImplementedError # --- memory reads ----------------------------------------------------- def load_name_map(self) -> list[dict[str, Any]]: raise NotImplementedError def load_memory_rules(self) -> list[dict[str, Any]]: raise NotImplementedError # --- API reads -------------------------------------------------------- def get_run(self, run_id: int) -> Optional[dict[str, Any]]: raise NotImplementedError def list_runs(self, limit: int = 50) -> list[dict[str, Any]]: raise NotImplementedError def close(self) -> None: # pragma: no cover - trivial pass class InMemoryStore(Store): """Dict-backed store for offline runs and no-DB environments.""" def __init__(self) -> None: self._runs: dict[int, dict[str, Any]] = {} self._planned: dict[int, list[dict[str, Any]]] = {} self._invoices: dict[int, list[dict[str, Any]]] = {} self._results: dict[int, list[dict[str, Any]]] = {} self._traces: dict[int, list[dict[str, Any]]] = {} self._name_map: list[dict[str, Any]] = [] self._memory_rules: list[dict[str, Any]] = [] self._seq = 0 def _next_id(self) -> int: self._seq += 1 return self._seq # Seeders so the offline runner can inject memory without a DB. def seed_name_map(self, entries: list[dict[str, Any]]) -> None: self._name_map = list(entries) def seed_memory_rules(self, rules: list[dict[str, Any]]) -> None: self._memory_rules = list(rules) def create_run(self, month, provider, model, source_filename) -> int: run_id = self._next_id() self._runs[run_id] = { "id": run_id, "month": month, "provider": provider, "model": model, "status": "pending", "source_filename": source_filename, "totals_json": None, "error": None, "created_at": _now().isoformat(sep=" ", timespec="seconds"), "finished_at": None, } return run_id def set_run_status(self, run_id, status) -> None: self._runs[run_id]["status"] = status def finish_run(self, run_id, status, totals=None, error=None) -> None: run = self._runs[run_id] run["status"] = status run["totals_json"] = totals run["error"] = error run["finished_at"] = _now().isoformat(sep=" ", timespec="seconds") def save_planned_lines(self, run_id, lines) -> dict[int, int]: mapping: dict[int, int] = {} rows = [] for line in lines: pid = self._next_id() mapping[line.source_row] = pid row = line.to_dict() row.update({"id": pid, "run_id": run_id}) rows.append(row) self._planned[run_id] = rows return mapping def save_invoices(self, run_id, invoices) -> dict[int, int]: mapping: dict[int, int] = {} rows = [] for inv in invoices: iid = self._next_id() mapping[inv.odoo_id] = iid row = inv.to_dict() row.update({"id": iid, "run_id": run_id}) rows.append(row) self._invoices[run_id] = rows return mapping def save_results(self, run_id, results, planned_id_map, invoice_id_map) -> None: rows = [] for res in results: row = res.to_dict() rid = self._next_id() row["id"] = rid row["run_id"] = run_id row["planned_line_id"] = ( planned_id_map.get(res.source_row) if res.source_row is not None else None ) row["matched_invoice_id"] = ( invoice_id_map.get(res.matched_invoice_id) if res.matched_invoice_id is not None else None ) row["review_status"] = "pending" rows.append(row) self._results[run_id] = rows def save_traces(self, run_id, rows) -> None: self._traces[run_id] = [dict(r, run_id=run_id) for r in rows] def load_name_map(self) -> list[dict[str, Any]]: return list(self._name_map) def load_memory_rules(self) -> list[dict[str, Any]]: return list(self._memory_rules) def get_run(self, run_id) -> Optional[dict[str, Any]]: run = self._runs.get(run_id) if run is None: return None return { **run, "planned_lines": self._planned.get(run_id, []), "invoices": self._invoices.get(run_id, []), "results": self._results.get(run_id, []), } def list_runs(self, limit=50) -> list[dict[str, Any]]: runs = sorted(self._runs.values(), key=lambda r: r["id"], reverse=True) return runs[:limit] class MySQLStore(Store): """PyMySQL-backed store against the shared database (never migrates).""" def __init__( self, host: str, port: int, database: str, user: str, password: str ) -> None: try: import pymysql from pymysql.cursors import DictCursor except ImportError as exc: # pragma: no cover - needs pymysql absent raise RuntimeError( "PyMySQL is required for MySQLStore; install it or leave MYSQL_* " "unset to use the in-memory fallback." ) from exc self._pymysql = pymysql self._conn = pymysql.connect( host=host, port=port, db=database, user=user, password=password, charset="utf8mb4", cursorclass=DictCursor, autocommit=True, ) def _cursor(self): self._conn.ping(reconnect=True) return self._conn.cursor() def create_run(self, month, provider, model, source_filename) -> int: with self._cursor() as cur: cur.execute( "INSERT INTO runs (month, provider, model, status, source_filename, " "created_at) VALUES (%s, %s, %s, %s, %s, %s)", (month, provider, model, "pending", source_filename, _now()), ) return int(cur.lastrowid) def set_run_status(self, run_id, status) -> None: with self._cursor() as cur: cur.execute("UPDATE runs SET status=%s WHERE id=%s", (status, run_id)) def finish_run(self, run_id, status, totals=None, error=None) -> None: with self._cursor() as cur: cur.execute( "UPDATE runs SET status=%s, totals_json=%s, error=%s, finished_at=%s " "WHERE id=%s", (status, json.dumps(totals) if totals is not None else None, error, _now(), run_id), ) def save_planned_lines(self, run_id, lines) -> dict[int, int]: mapping: dict[int, int] = {} with self._cursor() as cur: for line in lines: cur.execute( "INSERT INTO planned_lines (run_id, source_row, client_name, " "canonical_client, odoo_customer_id, planned_amount, project_value, " "match_source) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", ( run_id, line.source_row, line.client_name, line.canonical_client, line.odoo_customer_id, line.planned_amount, line.project_value, line.match_source, ), ) mapping[line.source_row] = int(cur.lastrowid) return mapping def save_invoices(self, run_id, invoices) -> dict[int, int]: mapping: dict[int, int] = {} with self._cursor() as cur: for inv in invoices: cur.execute( "INSERT INTO invoices (run_id, odoo_id, number, customer, " "odoo_customer_id, invoice_date, amount_untaxed, amount_total, sent) " "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", ( run_id, inv.odoo_id, inv.number, inv.customer, inv.odoo_customer_id, inv.invoice_date, inv.amount_untaxed, inv.amount_total, 1 if inv.sent else 0, ), ) mapping[inv.odoo_id] = int(cur.lastrowid) return mapping def save_results(self, run_id, results, planned_id_map, invoice_id_map) -> None: with self._cursor() as cur: for res in results: planned_line_id = ( planned_id_map.get(res.source_row) if res.source_row is not None else None ) matched_invoice_id = ( invoice_id_map.get(res.matched_invoice_id) if res.matched_invoice_id is not None else None ) cur.execute( "INSERT INTO results (run_id, planned_line_id, matched_invoice_id, " "status, variance, confidence, tier, reasoning, candidates_json, " "review_status) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", ( run_id, planned_line_id, matched_invoice_id, res.status, res.variance, res.confidence, res.tier, res.reasoning, json.dumps([c.to_dict() for c in res.candidates]), "pending", ), ) def save_traces(self, run_id, rows) -> None: with self._cursor() as cur: for row in rows: cur.execute( "INSERT INTO traces (run_id, span_id, parent_span_id, name, " "attributes_json, started_at, duration_ms) VALUES (%s,%s,%s,%s,%s,%s,%s)", ( run_id, row["span_id"], row.get("parent_span_id"), row["name"], json.dumps(row.get("attributes_json", {})), row["started_at"], row["duration_ms"], ), ) def load_name_map(self) -> list[dict[str, Any]]: with self._cursor() as cur: cur.execute( "SELECT planning_name, odoo_customer_id, odoo_customer_name, " "confidence, source FROM name_map" ) return list(cur.fetchall()) def load_memory_rules(self) -> list[dict[str, Any]]: with self._cursor() as cur: cur.execute( "SELECT scope, rule_type, params_json, note FROM memory_rules" ) rows = list(cur.fetchall()) for row in rows: if isinstance(row.get("params_json"), str): try: row["params_json"] = json.loads(row["params_json"]) except (json.JSONDecodeError, TypeError): row["params_json"] = {} return rows def get_run(self, run_id) -> Optional[dict[str, Any]]: with self._cursor() as cur: cur.execute("SELECT * FROM runs WHERE id=%s", (run_id,)) run = cur.fetchone() if run is None: return None cur.execute("SELECT * FROM planned_lines WHERE run_id=%s", (run_id,)) planned = list(cur.fetchall()) cur.execute("SELECT * FROM invoices WHERE run_id=%s", (run_id,)) invoices = list(cur.fetchall()) cur.execute("SELECT * FROM results WHERE run_id=%s", (run_id,)) results = list(cur.fetchall()) run["planned_lines"] = planned run["invoices"] = invoices run["results"] = results return run def list_runs(self, limit=50) -> list[dict[str, Any]]: with self._cursor() as cur: cur.execute( "SELECT * FROM runs ORDER BY id DESC LIMIT %s", (int(limit),) ) return list(cur.fetchall()) def close(self) -> None: # pragma: no cover - trivial try: self._conn.close() except Exception: pass def build_store(settings: "Any") -> Store: """Return a MySQL store when configured, else the in-memory fallback.""" if settings.mysql_configured: try: return MySQLStore( host=settings.mysql_host, port=settings.mysql_port, database=settings.mysql_database, user=settings.mysql_user, password=settings.mysql_password or "", ) except Exception as exc: logger.warning("MySQL unavailable (%s); using in-memory store.", exc) return InMemoryStore()