from __future__ import annotations from .shared import ( Any, db, fail, json, new_id, normalize_date, now_iso, ok, row_to_dict, ) VALID_REVIEW_STATUSES = {"pending", "decided", "held", "needs_recheck", "ignored"} VALID_DECISION_TYPES = { "account_transfer", "real_expense", "real_income", "investment_transfer", "excluded_from_analysis", "duplicate_or_cancel", "unknown", } VALID_RECONCILIATION_STATUSES = {"open", "closed", "needs_recheck", "ignored"} SENSITIVE_PAYLOAD_KEYS = {"raw_payload", "password", "account_password", "secret", "token", "file_path"} def _normalize_review_status(value: str | None) -> str: status = (value or "pending").strip() or "pending" if status not in VALID_REVIEW_STATUSES: raise ValueError(f"status must be one of {sorted(VALID_REVIEW_STATUSES)}") return status def _normalize_decision_type(value: str | None) -> str: decision_type = (value or "unknown").strip() or "unknown" if decision_type not in VALID_DECISION_TYPES: raise ValueError(f"decision_type must be one of {sorted(VALID_DECISION_TYPES)}") return decision_type def _normalize_reconciliation_status(value: str | None) -> str: status = (value or "open").strip() or "open" if status not in VALID_RECONCILIATION_STATUSES: raise ValueError(f"status must be one of {sorted(VALID_RECONCILIATION_STATUSES)}") return status def _safe_payload(target_payload: Any) -> str: if target_payload in (None, ""): return "{}" if isinstance(target_payload, str): try: parsed = json.loads(target_payload) except json.JSONDecodeError: return json.dumps({"summary": target_payload[:500]}, ensure_ascii=False) elif isinstance(target_payload, dict): parsed = target_payload else: parsed = {"value": str(target_payload)[:500]} safe: dict[str, Any] = {} for key, value in parsed.items(): key_text = str(key) if key_text.lower() in SENSITIVE_PAYLOAD_KEYS: continue if isinstance(value, (str, int, float, bool)) or value is None: safe[key_text] = str(value)[:500] if isinstance(value, str) else value elif isinstance(value, (list, tuple)): safe[key_text] = [str(item)[:200] for item in value[:20]] else: safe[key_text] = str(value)[:500] return json.dumps(safe, ensure_ascii=False, sort_keys=True) def _decode_payload(row: dict[str, Any]) -> dict[str, Any]: raw = row.get("target_payload") or "{}" try: row["target_payload"] = json.loads(raw) except json.JSONDecodeError: row["target_payload"] = {} return row def _fetch_latest_decisions(conn: Any, targets: list[tuple[str, str]]) -> dict[tuple[str, str], dict[str, Any]]: if not targets: return {} by_target: dict[tuple[str, str], dict[str, Any]] = {} for target_type, target_id in targets: row = conn.execute( """ SELECT * FROM review_decisions WHERE target_type=? AND target_id=? ORDER BY decided_at DESC, updated_at DESC, created_at DESC LIMIT 1 """, (target_type, target_id), ).fetchone() if row: by_target[(target_type, target_id)] = _decode_payload(row_to_dict(row)) return by_target def attach_review_decisions(db_conn: Any, items: list[dict[str, Any]], target_type: str, id_field: str = "id") -> list[dict[str, Any]]: targets = [(target_type, str(item.get(id_field) or "")) for item in items if item.get(id_field)] decisions = _fetch_latest_decisions(db_conn, targets) enriched: list[dict[str, Any]] = [] for item in items: target_id = str(item.get(id_field) or "") decision = decisions.get((target_type, target_id)) state = decision.get("status") if decision else "pending" enriched.append({**item, "review_status": state, "review_decision": decision}) return enriched def list_review_decisions(target_type: str = "", target_id: str = "", status: str = "", limit: int = 100, db_path: str | None = None) -> dict[str, Any]: clauses: list[str] = [] params: list[Any] = [] if target_type: clauses.append("target_type=?") params.append(target_type) if target_id: clauses.append("target_id=?") params.append(target_id) if status: try: status = _normalize_review_status(status) except ValueError as exc: return fail("validation_error", str(exc)) clauses.append("status=?") params.append(status) where = " WHERE " + " AND ".join(clauses) if clauses else "" max_limit = max(1, min(int(limit), 1000)) with db(db_path) as conn: rows = conn.execute( f"SELECT * FROM review_decisions{where} ORDER BY decided_at DESC, updated_at DESC LIMIT ?", (*params, max_limit), ).fetchall() decisions = [_decode_payload(row_to_dict(row)) for row in rows] return ok(review_decisions=decisions, decisions=decisions, count=len(decisions)) def upsert_review_decision( target_type: str, target_id: str, status: str = "decided", decision_type: str = "unknown", reason_code: str = "", memo: str = "", operator: str = "operator", review_after: str = "", target_payload: Any = None, decision_id: str = "", db_path: str | None = None, ) -> dict[str, Any]: target_type = (target_type or "").strip() target_id = (target_id or "").strip() if not target_type or not target_id: return fail("validation_error", "target_type and target_id are required") try: normalized_status = _normalize_review_status(status) normalized_decision_type = _normalize_decision_type(decision_type) normalized_review_after = normalize_date(review_after, default_today=False) if review_after else "" except Exception as exc: return fail("validation_error", str(exc)) ts = now_iso() payload_text = _safe_payload(target_payload) with db(db_path) as conn: existing = None if decision_id: existing = conn.execute("SELECT * FROM review_decisions WHERE id=?", (decision_id,)).fetchone() if not existing: return fail("not_found", f"review decision not found: {decision_id}") else: existing = conn.execute( "SELECT * FROM review_decisions WHERE target_type=? AND target_id=? ORDER BY decided_at DESC, updated_at DESC LIMIT 1", (target_type, target_id), ).fetchone() row_id = decision_id or (existing["id"] if existing else new_id("rd")) created_at = existing["created_at"] if existing else ts decided_at = ts if normalized_status in {"decided", "held", "needs_recheck", "ignored"} else "" conn.execute( """ INSERT INTO review_decisions(id, target_type, target_id, status, decision_type, reason_code, memo, operator, decided_at, review_after, target_payload, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET target_type=excluded.target_type, target_id=excluded.target_id, status=excluded.status, decision_type=excluded.decision_type, reason_code=excluded.reason_code, memo=excluded.memo, operator=excluded.operator, decided_at=excluded.decided_at, review_after=excluded.review_after, target_payload=excluded.target_payload, updated_at=excluded.updated_at """, (row_id, target_type, target_id, normalized_status, normalized_decision_type, reason_code or "", memo or "", operator or "operator", decided_at, normalized_review_after, payload_text, created_at, ts), ) row = conn.execute("SELECT * FROM review_decisions WHERE id=?", (row_id,)).fetchone() return ok(review_decision=_decode_payload(row_to_dict(row)), decision_id=row_id) def hold_review_decision(target_type: str, target_id: str, reason_code: str = "held", memo: str = "", operator: str = "operator", review_after: str = "", db_path: str | None = None) -> dict[str, Any]: return upsert_review_decision(target_type, target_id, status="held", decision_type="unknown", reason_code=reason_code, memo=memo, operator=operator, review_after=review_after, db_path=db_path) def mark_review_recheck(target_type: str, target_id: str, reason_code: str = "needs_recheck", memo: str = "", operator: str = "operator", review_after: str = "", db_path: str | None = None) -> dict[str, Any]: return upsert_review_decision(target_type, target_id, status="needs_recheck", decision_type="unknown", reason_code=reason_code, memo=memo, operator=operator, review_after=review_after, db_path=db_path) def release_review_decision(target_type: str, target_id: str, memo: str = "", operator: str = "operator", db_path: str | None = None) -> dict[str, Any]: return upsert_review_decision(target_type, target_id, status="pending", decision_type="unknown", reason_code="released", memo=memo, operator=operator, review_after="", db_path=db_path) def list_reconciliation_sessions(account_id: str = "", status: str = "", basis_date: str = "", limit: int = 100, db_path: str | None = None) -> dict[str, Any]: clauses: list[str] = [] params: list[Any] = [] if account_id: clauses.append("account_id=?") params.append(account_id) if status: try: status = _normalize_reconciliation_status(status) except ValueError as exc: return fail("validation_error", str(exc)) clauses.append("status=?") params.append(status) if basis_date: try: basis_date = normalize_date(basis_date) except Exception as exc: return fail("validation_error", str(exc)) clauses.append("basis_date=?") params.append(basis_date) where = " WHERE " + " AND ".join(clauses) if clauses else "" with db(db_path) as conn: rows = conn.execute( f"SELECT * FROM reconciliation_sessions{where} ORDER BY basis_date DESC, updated_at DESC LIMIT ?", (*params, max(1, min(int(limit), 1000))), ).fetchall() sessions = [row_to_dict(row) for row in rows] return ok(reconciliation_sessions=sessions, sessions=sessions, count=len(sessions)) def upsert_reconciliation_session( account_id: str, basis_date: str, expected_balance: int | None = None, actual_balance: int | None = None, difference: int | None = None, status: str = "open", source_note: str = "", memo: str = "", operator: str = "operator", session_id: str = "", db_path: str | None = None, ) -> dict[str, Any]: account_id = (account_id or "").strip() if not account_id: return fail("validation_error", "account_id is required") try: basis = normalize_date(basis_date, default_today=False) normalized_status = _normalize_reconciliation_status(status) expected = int(expected_balance) if expected_balance is not None else None actual = int(actual_balance) if actual_balance is not None else None diff = int(difference) if difference is not None else ((expected - actual) if expected is not None and actual is not None else None) except Exception as exc: return fail("validation_error", str(exc)) ts = now_iso() closed_at = ts if normalized_status == "closed" else "" with db(db_path) as conn: account = conn.execute("SELECT id FROM accounts WHERE id=?", (account_id,)).fetchone() if not account: return fail("not_found", f"account not found: {account_id}") existing = None if session_id: existing = conn.execute("SELECT * FROM reconciliation_sessions WHERE id=?", (session_id,)).fetchone() if not existing: return fail("not_found", f"reconciliation session not found: {session_id}") else: existing = conn.execute("SELECT * FROM reconciliation_sessions WHERE account_id=? AND basis_date=?", (account_id, basis)).fetchone() row_id = session_id or (existing["id"] if existing else new_id("rs")) created_at = existing["created_at"] if existing else ts if normalized_status != "closed" and existing: closed_at = existing["closed_at"] if existing["status"] == "closed" and normalized_status == "closed" else "" conn.execute( """ INSERT INTO reconciliation_sessions(id, account_id, basis_date, expected_balance, actual_balance, difference, status, closed_at, source_note, memo, operator, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET account_id=excluded.account_id, basis_date=excluded.basis_date, expected_balance=excluded.expected_balance, actual_balance=excluded.actual_balance, difference=excluded.difference, status=excluded.status, closed_at=excluded.closed_at, source_note=excluded.source_note, memo=excluded.memo, operator=excluded.operator, updated_at=excluded.updated_at """, (row_id, account_id, basis, expected, actual, diff, normalized_status, closed_at, source_note or "", memo or "", operator or "operator", created_at, ts), ) row = conn.execute("SELECT * FROM reconciliation_sessions WHERE id=?", (row_id,)).fetchone() return ok(reconciliation_session=row_to_dict(row), session_id=row_id) def close_reconciliation_session(session_id: str = "", account_id: str = "", basis_date: str = "", memo: str = "", operator: str = "operator", db_path: str | None = None) -> dict[str, Any]: with db(db_path) as conn: if session_id: row = conn.execute("SELECT * FROM reconciliation_sessions WHERE id=?", (session_id,)).fetchone() elif account_id and basis_date: try: basis = normalize_date(basis_date, default_today=False) except Exception as exc: return fail("validation_error", str(exc)) row = conn.execute("SELECT * FROM reconciliation_sessions WHERE account_id=? AND basis_date=?", (account_id, basis)).fetchone() else: return fail("validation_error", "session_id or account_id+basis_date is required") if not row: return fail("not_found", "reconciliation session not found") existing = row_to_dict(row) return upsert_reconciliation_session( account_id=existing["account_id"], basis_date=existing["basis_date"], expected_balance=existing.get("expected_balance"), actual_balance=existing.get("actual_balance"), difference=existing.get("difference"), status="closed", source_note=existing.get("source_note", ""), memo=memo or existing.get("memo", ""), operator=operator, session_id=existing["id"], db_path=db_path, ) def get_review_operating_indicators(db_path: str | None = None) -> dict[str, Any]: with db(db_path) as conn: review_rows = conn.execute("SELECT status, COUNT(*) AS count FROM review_decisions GROUP BY status").fetchall() pending_count = int(conn.execute("SELECT COUNT(*) FROM review_decisions WHERE status='pending'").fetchone()[0]) held_recheck_count = int(conn.execute("SELECT COUNT(*) FROM review_decisions WHERE status IN ('held','needs_recheck')").fetchone()[0]) unreconciled_accounts = int(conn.execute("SELECT COUNT(DISTINCT account_id) FROM reconciliation_sessions WHERE status<>'closed'").fetchone()[0]) last_closed = conn.execute("SELECT MAX(closed_at) AS closed_at FROM reconciliation_sessions WHERE status='closed' AND closed_at<>''").fetchone()["closed_at"] uncategorized_pending = int(conn.execute("SELECT COUNT(*) FROM transactions WHERE status='active' AND COALESCE(category,'')='' ").fetchone()[0]) by_status = {row["status"]: int(row["count"] or 0) for row in review_rows} return ok( indicators={ "pending_review_count": pending_count, "held_recheck_count": held_recheck_count, "unreconciled_account_count": unreconciled_accounts, "last_reconciliation_closed_at": last_closed or "", "uncategorized_recommendation_backlog": uncategorized_pending, "review_decisions_by_status": by_status, } )