from __future__ import annotations import base64 import hashlib import hmac import json import os import re import secrets import time from collections import defaultdict from datetime import date, datetime from decimal import Decimal, InvalidOperation from sqlite3 import Connection from typing import Any from fastapi import HTTPException from jarvis_finance.audit.log import record_audit_event from jarvis_finance.services.budget_common import now from jarvis_finance.services.budget_csv_imports import IMPORT_PROFILES, parse_budget_csv_text from jarvis_finance.services.budget_transactions import refund_link_is_valid from jarvis_finance.services.household_classification import ( CLASSIFICATION_VERSION, USER_CONFIRMED_UNMATCHED_TRANSFER, USER_DECISION_VERSION, ClassificationLookupCache, ambiguous_review_family, classify_household_row, extract_counterparty, merchant_family, normalize_merchant, ) from jarvis_finance.services.household_financials import get_household_financial_summary from jarvis_finance.services.transfer_pairing import confirm_transfer_pair CONTRACT_VERSION = "household_import_v1" PAIRING_VERSION = "transfer_pairing_v3" CLUSTER_DECISION_VERSION = "cluster_decision_v2" PAIRING_CLASSES = ("safe", "review", "ambiguous", "unmatched") MAX_IMPORT_BYTES = 5 * 1024 * 1024 CONCEPT_V2_MINIMUM_COVERAGE_RATIO = Decimal("0.9000") CONCEPT_V2_MAXIMUM_MANUAL_DECISION_RATIO = Decimal("0.1000") STRICT_MINIMUM_COVERAGE_RATIO = Decimal("0.9500") STRICT_MAXIMUM_INDIVIDUAL_REVIEW_COUNT = 49 STRICT_MAXIMUM_INDIVIDUAL_REVIEW_RATIO = Decimal("0.0300") STRICT_MAXIMUM_MERCHANT_REVIEW_CLUSTER_COUNT = 25 OWNER_ATTESTED_TRANSFER_EVIDENCE = "owner_attested_known_household_counterparty_v1" HOUSEHOLD_OPEN_REVIEW_SQL = "household_batch_id IS NOT NULL AND status='needs_review'" def _sha(value: str) -> str: key = os.environ.get("JARVIS_FINANCE_FINGERPRINT_KEY") if not key: raise HTTPException(status_code=503, detail="household fingerprint key is not configured") return hmac.new(key.encode("utf-8"), value.encode("utf-8"), hashlib.sha256).hexdigest() def _canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) def get_household_open_review_count(conn: Connection) -> int: return int( conn.execute( f"SELECT COUNT(*) FROM budget_transaction_candidates WHERE {HOUSEHOLD_OPEN_REVIEW_SQL}" ).fetchone()[0] ) def _owner_neutral_cluster_approval( cluster_token: str, row_tokens: list[str], baseline_fingerprint: str, issued_at: int, expires_at: int, nonce: str, approval_evidence: str = OWNER_ATTESTED_TRANSFER_EVIDENCE, ) -> str: """Mint a transient server-secret-bound token for one approved neutral cluster.""" configured_key = os.environ.get("JARVIS_FINANCE_OPERATOR_APPROVAL_KEY") if not configured_key: raise HTTPException(status_code=503, detail="household operator approval key is not configured") approval_payload = _canonical({ "scope": "owner_confirmed_neutral_household_transfer_v2", "approval_evidence": approval_evidence, "cluster_token": cluster_token, "row_tokens": sorted(row_tokens), "baseline_fingerprint": baseline_fingerprint, "issued_at": issued_at, "expires_at": expires_at, "nonce": nonce, }) return hmac.new( configured_key.encode("utf-8"), approval_payload.encode("utf-8"), hashlib.sha256, ).hexdigest() def authorize_owner_neutral_cluster( conn: Connection, payload: dict[str, Any], operator_key: str | None, ) -> dict[str, Any]: """Authorize one transient exact-row transfer decision at the trusted operator boundary.""" configured_key = os.environ.get("JARVIS_FINANCE_OPERATOR_APPROVAL_KEY") if not configured_key: raise HTTPException(status_code=503, detail="household operator approval key is not configured") if not operator_key or not hmac.compare_digest(operator_key, configured_key): raise HTTPException(status_code=403, detail="operator approval is not authorized") cluster_token = str(payload.get("cluster_token") or "") row_tokens_raw = payload.get("approved_row_tokens") or [] preview_request = payload.get("preview_request") evidence = str(payload.get("approval_evidence") or "") if ( payload.get("confirm_owner_attestation") is not True or evidence != OWNER_ATTESTED_TRANSFER_EVIDENCE or not cluster_token or not isinstance(row_tokens_raw, list) or not row_tokens_raw or not isinstance(preview_request, dict) or bool(preview_request.get("cluster_decisions")) ): raise HTTPException(status_code=422, detail="exact owner attestation is required") row_tokens = [str(value) for value in row_tokens_raw] if len(row_tokens) != len(set(row_tokens)) or any(not value for value in row_tokens): raise HTTPException(status_code=422, detail="approved row tokens must be unique and non-empty") reconstructed = preview_household_import(conn, preview_request) matching_clusters = [ cluster for cluster in reconstructed["merchant_clusters"] if cluster["cluster_token"] == cluster_token ] if len(matching_clusters) != 1: raise HTTPException(status_code=409, detail="owner-attested merchant cluster is stale") complete_row_tokens = [str(value) for value in matching_clusters[0]["row_tokens"]] if sorted(row_tokens) != sorted(complete_row_tokens): raise HTTPException( status_code=422, detail="owner attestation must cover the complete current merchant cluster", ) issued_at = int(time.time()) expires_at = issued_at + 15 * 60 baseline_fingerprint = _baseline(conn) nonce = secrets.token_urlsafe(24) return { "cluster_token": cluster_token, "decision_type": USER_CONFIRMED_UNMATCHED_TRANSFER, "owner_confirmed": True, "approval_evidence": evidence, "approved_row_tokens": row_tokens, "approval_baseline_fingerprint": baseline_fingerprint, "approval_issued_at": issued_at, "approval_expires_at": expires_at, "approval_nonce": nonce, "approval_token": _owner_neutral_cluster_approval( cluster_token, row_tokens, baseline_fingerprint, issued_at, expires_at, nonce, evidence, ), "excluded_row_tokens": [], } def _norm(value: Any) -> str: return " ".join(str(value or "").strip().casefold().split()) _UNMATCHED_TRANSFER_MARKERS = ( "uebertrag eigenes konto", "ubertrag eigenes konto", "übertrag eigenes konto", "transfer eigenes konto", "internal transfer", "own account transfer", "kontoübertrag", "kontouebertrag", "wallet top up", "wallet aufladung", ) _UNMATCHED_TRANSFER_DISQUALIFIERS = ( "fee", "charge", "commission", "kommission", "kosten", "entgelt", "gebuhr", "gebühr", ) def _has_unmatched_transfer_evidence( row: dict[str, Any], preliminary: dict[str, Any] ) -> bool: if ( row.get("source_type") not in {"akb_bank", "raiffeisen_bank"} or str(row.get("currency") or "") != "CHF" or not (row.get("mapping") or {}).get("budget_account_id") or preliminary.get("user_state") != "decision_needed" ): return False if row.get("pairing_class") == "ambiguous": return True text = _norm(f"{row.get('description') or ''} {row.get('merchant') or ''}") return ( not any(marker in text for marker in _UNMATCHED_TRANSFER_DISQUALIFIERS) and any(marker in text for marker in _UNMATCHED_TRANSFER_MARKERS) ) def _money(value: Any) -> Decimal | None: text = str(value or "").strip().replace("'", "").replace(" ", "") if not text: return None if "," in text and "." in text: text = text.replace(".", "").replace(",", ".") if text.rfind(",") > text.rfind(".") else text.replace(",", "") elif "," in text: text = text.replace(",", ".") try: return Decimal(text) except InvalidOperation: return None def _date(value: Any) -> str | None: text = str(value or "").strip()[:10] for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d/%m/%Y"): try: return datetime.strptime(text, fmt).date().isoformat() except ValueError: continue return None def _low(row: dict[str, Any]) -> dict[str, Any]: return {_norm(key): value for key, value in row.items()} def _source_reference(profile: str, row: dict[str, Any], fallback: Any = None) -> str: spec = IMPORT_PROFILES[profile] low = _low(row) for column in spec.account_columns: value = str(low.get(_norm(column)) or "").strip() if value: return value return str(fallback or "").strip() def _mapping_account_is_compatible(profile: str, row: Any) -> bool: budget_type = str(row["budget_account_type"] or "") canonical_type = str(row["canonical_account_type"] or "") currency_ok = str(row["budget_currency"] or "").upper() == "CHF" and str( row["canonical_currency"] or "" ).upper() == "CHF" performance_ok = int(row["performance_included"] or 0) == 0 if profile == "visa_credit_card": role_ok = budget_type in {"credit_card", "credit_card_liability"} and canonical_type in { "credit_card", "credit_card_liability", "liability", } portfolio_ok = str(row["portfolio_bucket"] or "") == "liability" else: role_ok = budget_type in {"cash", "checking", "savings"} and canonical_type == "cash" portfolio_ok = str(row["portfolio_bucket"] or "") == "cash" return bool(currency_ok and performance_ok and role_ok and portfolio_ok) def _mapping( conn: Connection, profile: str, source_reference: str, mapping_id: str | None = None, ) -> dict[str, Any] | None: if not source_reference and not mapping_id: return None row = conn.execute( """SELECT m.mapping_id,m.budget_account_id,m.canonical_account_id,m.reference_hint, m.source_reference_hash,ba.name AS account_name, ba.account_type AS budget_account_type,ba.currency AS budget_currency, a.account_type AS canonical_account_type,a.currency AS canonical_currency, a.performance_included,a.portfolio_bucket FROM household_account_source_mappings m JOIN budget_accounts ba ON ba.budget_account_id=m.budget_account_id AND ba.is_active=1 AND ba.linked_account_id=m.canonical_account_id JOIN accounts a ON a.account_id=m.canonical_account_id AND a.is_active=1 WHERE m.contract_version=? AND m.source_type=? AND (? IS NULL OR m.mapping_id=?) AND (?='' OR m.source_reference_hash=?) AND m.is_active=1""", ( CONTRACT_VERSION, profile, mapping_id, mapping_id, source_reference, _sha(_norm(source_reference)) if source_reference else "", ), ).fetchone() return dict(row) if row and _mapping_account_is_compatible(profile, row) else None def _mapping_for_file( conn: Connection, profile: str, source_reference: str, file_item: dict[str, Any], ) -> dict[str, Any] | None: """Resolve one row against the mappings explicitly allowed for its file. A provider file may contain multiple stable account references. The legacy ``mapping_id`` remains supported for single-account files; ``mapping_ids`` binds a multi-account file without persisting or returning raw references. """ raw_mapping_ids = file_item.get("mapping_ids") if raw_mapping_ids is None: return _mapping( conn, profile, source_reference, str(file_item.get("mapping_id") or "") or None, ) if not isinstance(raw_mapping_ids, list) or not raw_mapping_ids: raise HTTPException(status_code=422, detail="mapping_ids must be a non-empty list") mapping_ids = [str(value or "") for value in raw_mapping_ids] if any(not value for value in mapping_ids) or len(set(mapping_ids)) != len(mapping_ids): raise HTTPException(status_code=422, detail="mapping_ids must contain unique mapping identifiers") matches = [ match for mapping_id in mapping_ids if (match := _mapping(conn, profile, source_reference, mapping_id)) is not None ] return matches[0] if len(matches) == 1 else None def configure_source_mapping(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: profile = str(payload.get("source_type") or "") if profile not in IMPORT_PROFILES or profile == "migros_receipts": raise HTTPException(status_code=422, detail="source_type does not support an account mapping") source_reference = str(payload.get("source_reference") or "").strip() budget_account_id = str(payload.get("budget_account_id") or "") if not source_reference or not budget_account_id: raise HTTPException(status_code=422, detail="source_reference and budget_account_id are required") account = conn.execute( """SELECT ba.budget_account_id,ba.linked_account_id,ba.name, ba.account_type AS budget_account_type,ba.currency AS budget_currency, a.account_type AS canonical_account_type,a.currency AS canonical_currency, a.performance_included,a.portfolio_bucket FROM budget_accounts ba JOIN accounts a ON a.account_id=ba.linked_account_id WHERE ba.budget_account_id=? AND ba.is_active=1 AND a.is_active=1""", (budget_account_id,), ).fetchone() if not account or not account["linked_account_id"]: raise HTTPException(status_code=422, detail="active budget account must be linked to an active canonical account") if not _mapping_account_is_compatible(profile, account): raise HTTPException(status_code=422, detail="source type is incompatible with the selected account semantics") reference_hash = _sha(_norm(source_reference)) mapping_id = "hhmap_" + _sha(f"{profile}|{reference_hash}")[:24] existing = conn.execute( """SELECT contract_version,budget_account_id,canonical_account_id,is_active FROM household_account_source_mappings WHERE source_type=? AND source_reference_hash=?""", (profile, reference_hash), ).fetchone() if existing and ( str(existing["contract_version"]) == CONTRACT_VERSION and str(existing["budget_account_id"]) == budget_account_id and str(existing["canonical_account_id"]) == str(account["linked_account_id"]) and bool(existing["is_active"]) ): return {"status": "unchanged", "idempotent": True, "mapping_id": mapping_id, "source_type": profile, "account_id": budget_account_id, "account_name": account["name"], "reference_hint": "configured"} timestamp = now() had_outer_transaction = conn.in_transaction if had_outer_transaction: conn.execute("SAVEPOINT household_mapping") else: conn.execute("BEGIN IMMEDIATE") try: conn.execute( """INSERT INTO household_account_source_mappings( mapping_id,contract_version,source_type,source_reference_hash,budget_account_id, canonical_account_id,reference_hint,is_active,created_at,updated_at) VALUES (?,?,?,?,?,?,?,1,?,?) ON CONFLICT(source_type,source_reference_hash) DO UPDATE SET budget_account_id=excluded.budget_account_id, canonical_account_id=excluded.canonical_account_id, reference_hint=excluded.reference_hint,is_active=1,updated_at=excluded.updated_at""", (mapping_id, CONTRACT_VERSION, profile, reference_hash, budget_account_id, account["linked_account_id"], "ref_" + reference_hash[:12], timestamp, timestamp), ) record_audit_event( conn, source="household_import", action="account_source_mapping_confirmed", entity_type="household_account_source_mapping", entity_id=mapping_id, new_values={"contract_version": CONTRACT_VERSION, "source_type": profile, "budget_account_id": budget_account_id}, created_by="user", ) if had_outer_transaction: conn.execute("RELEASE SAVEPOINT household_mapping") else: conn.commit() except Exception: if had_outer_transaction: conn.execute("ROLLBACK TO SAVEPOINT household_mapping") conn.execute("RELEASE SAVEPOINT household_mapping") else: conn.rollback() raise return {"status": "confirmed", "mapping_id": mapping_id, "source_type": profile, "account_id": budget_account_id, "account_name": account["name"], "reference_hint": "configured"} def list_source_mappings(conn: Connection) -> list[dict[str, Any]]: rows = conn.execute( """SELECT m.mapping_id,m.source_type,m.reference_hint,m.budget_account_id, m.canonical_account_id,ba.name AS account_name,m.is_active FROM household_account_source_mappings m JOIN budget_accounts ba ON ba.budget_account_id=m.budget_account_id ORDER BY m.source_type,ba.name,m.mapping_id""" ).fetchall() return [dict(row) | {"is_active": bool(row["is_active"])} for row in rows] def _baseline(conn: Connection) -> str: """Hash every relation which can change preview classification or confirm writes.""" tables = ( "household_account_source_mappings", "household_import_batches", "household_import_files", "household_import_items", "household_migros_links", "budget_transaction_candidates", "budget_transactions", "budget_transfers", "budget_transfer_pairs", "budget_accounts", "accounts", "budget_categories", "budget_merchants", "budget_merchant_aliases", "budget_review_rules", "budget_rule_suggestions", "budget_recurring_payments", ) serial: dict[str, list[list[Any]]] = {} for table in tables: columns = [str(row[1]) for row in conn.execute(f"PRAGMA table_info({table})")] rows = [list(row) for row in conn.execute(f"SELECT * FROM {table}").fetchall()] serial[table] = [columns] + sorted(rows, key=lambda row: _canonical(row)) return _sha(_canonical(serial)) def _validated_files( payload: dict[str, Any], performance: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: files = payload.get("files") if not isinstance(files, list) or not files: raise HTTPException(status_code=422, detail="files must be a non-empty list") total = 0 validated: list[dict[str, Any]] = [] for item in files: if not isinstance(item, dict): raise HTTPException(status_code=422, detail="each file must be an object") csv_value = item.get("csv_text") if not isinstance(csv_value, str) or not csv_value.strip(): raise HTTPException(status_code=422, detail="each file must contain a non-empty CSV") size = len(csv_value.encode("utf-8")) total += size if size > MAX_IMPORT_BYTES or total > MAX_IMPORT_BYTES: raise HTTPException(status_code=413, detail="import exceeds the 5 MB request limit") hint = item.get("profile") if hint not in (None, "", "auto") and str(hint) not in IMPORT_PROFILES: raise HTTPException(status_code=422, detail="unsupported CSV profile") parse_started = time.perf_counter() try: parsed = parse_budget_csv_text(csv_value, None if hint in (None, "", "auto") else str(hint)) except (ValueError, TypeError): raise HTTPException(status_code=422, detail="unsupported CSV format") from None finally: if performance is not None: performance.setdefault("parsing_by_file", []).append({ "file_index": len(validated) + 1, "profile": str(hint or "auto"), "seconds": round(time.perf_counter() - parse_started, 6), }) if not parsed["rows"]: raise HTTPException(status_code=422, detail="CSV contains no data rows") validated.append({"item": item, "parsed": parsed, "file_fingerprint": _sha(f"{parsed['profile']}|{csv_value}")}) return validated def _input_fingerprint( validated: list[dict[str, Any]], category_overrides: dict[str, Any] | None = None, user_decisions: dict[str, Any] | None = None, cluster_decisions: list[dict[str, Any]] | None = None, ) -> str: return _sha(_canonical({ "files": [{ "profile": entry["parsed"]["profile"], "file_fingerprint": entry["file_fingerprint"], "mapping_id": str(entry["item"].get("mapping_id") or ""), "mapping_ids": sorted( str(value) for value in (entry["item"].get("mapping_ids") or []) ), "source_reference_hash": _sha(_norm(entry["item"].get("source_reference"))) if entry["item"].get("source_reference") else "", } for entry in validated], "category_overrides": { str(key): str(value) for key, value in sorted((category_overrides or {}).items()) }, "user_decisions": { str(key): str(value) for key, value in sorted((user_decisions or {}).items()) }, "user_decision_version": USER_DECISION_VERSION, "cluster_decisions": sorted( [ { "cluster_token": str(item.get("cluster_token") or ""), "category_id": str(item.get("category_id") or ""), "decision_type": str(item.get("decision_type") or "category"), "owner_confirmed": item.get("owner_confirmed") is True, "approval_evidence": str(item.get("approval_evidence") or ""), "approved_row_tokens": sorted( str(value) for value in (item.get("approved_row_tokens") or []) ), "approval_baseline_fingerprint": str( item.get("approval_baseline_fingerprint") or "" ), "approval_issued_at": item.get("approval_issued_at"), "approval_expires_at": item.get("approval_expires_at"), "approval_nonce": str(item.get("approval_nonce") or ""), "approval_token": str(item.get("approval_token") or ""), "excluded_row_tokens": sorted( str(value) for value in (item.get("excluded_row_tokens") or []) ), } for item in (cluster_decisions or []) ], key=lambda item: str(item["cluster_token"]), ), "classification_version": CLASSIFICATION_VERSION, "cluster_decision_version": CLUSTER_DECISION_VERSION, })) def _validated_user_decisions(payload: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, str]: raw = payload.get("user_decisions") or {} if not isinstance(raw, dict): raise HTTPException(status_code=422, detail="user_decisions must be an object") allowed_tokens = {str(row["_row_token"]) for row in rows} result: dict[str, str] = {} for token, decision in raw.items(): token = str(token) decision = str(decision) if token not in allowed_tokens: raise HTTPException(status_code=409, detail="user decision row is stale") if decision != USER_CONFIRMED_UNMATCHED_TRANSFER: raise HTTPException(status_code=422, detail="unsupported household user decision") result[token] = decision return result def _merchant_cluster_key(row: dict[str, Any]) -> tuple[str, str, str, bool]: classification = row["classification_v2"] value = row.get("merchant") or row.get("description") family = merchant_family(value) ambiguous = ambiguous_review_family(value) if not family else None if family: identity = "family:" + family["family_id"] label = family["label"] decision_allowed = True elif ambiguous: normalized = normalize_merchant(value) identity = "ambiguous:" + ambiguous["family_id"] + ":merchant:" + normalized label = str(value or ambiguous["label"]) # A user may classify an exact marketplace/cash family as one bounded # decision unit. Generic TWINT without a counterparty remains row-level. decision_allowed = ambiguous["family_id"] != "unknown_twint_counterparties" else: normalized = normalize_merchant(value) identity = "merchant:" + normalized label = str(value or "Unbekannter Händler") decision_allowed = True return identity, str(classification["transaction_semantics"]), label, decision_allowed def _apply_cluster_decisions( conn: Connection, rows: list[dict[str, Any]], payload: dict[str, Any], valid_categories: dict[str, dict[str, str]], ) -> list[dict[str, Any]]: grouped: dict[tuple[str, str, str, bool], list[tuple[int, dict[str, Any]]]] = defaultdict(list) for index, row in enumerate(rows, 1): classification = row["classification_v2"] if ( row["source_type"] != "migros_receipts" and row["currency"] == "CHF" and row.get("mapping") and classification["user_state"] == "decision_needed" and row["disposition"] not in { "pending", "superseded_pending", "duplicate_file", "duplicate_source_row", } ): grouped[_merchant_cluster_key(row)].append((index, row)) clusters: list[dict[str, Any]] = [] by_token: dict[str, dict[str, Any]] = {} ordered = sorted(grouped.items(), key=lambda item: (item[0][0], item[0][1])) for (identity, semantics, label, decision_allowed), members in ordered: member_tokens = [str(row["_row_token"]) for _, row in members] token = "cluster_" + _sha(_canonical({ "identity": identity, "semantics": semantics, "row_tokens": sorted(member_tokens), "classification_version": CLASSIFICATION_VERSION, }))[:24] public_members = [ { "row_token": str(row["_row_token"]), "date": row["transaction_date"], "merchant": row.get("merchant") or row["description"], "amount": row["signed_amount"], "currency": row["currency"], } for _, row in members ] record = { "cluster_token": token, "merchant_family": label, "transaction_semantics": semantics, "category_decision_allowed": decision_allowed, "count": len(members), "row_tokens": member_tokens, "examples": public_members[:3], "members": public_members, "category_id": None, "category_name": None, "decision_type": None, "selected_count": 0, "excluded_count": 0, "cluster_identity": identity, } clusters.append(record) by_token[token] = record | {"_members": members} raw_decisions = payload.get("cluster_decisions") or [] if not isinstance(raw_decisions, list) or not all(isinstance(item, dict) for item in raw_decisions): raise HTTPException(status_code=422, detail="cluster_decisions must be a list of objects") seen: set[str] = set() for raw in raw_decisions: token = str(raw.get("cluster_token") or "") if token in seen or token not in by_token: raise HTTPException(status_code=409, detail="merchant cluster is stale or duplicated") seen.add(token) excluded_raw = raw.get("excluded_row_tokens") or [] if not isinstance(excluded_raw, list): raise HTTPException(status_code=422, detail="excluded_row_tokens must be a list") excluded = [str(value) for value in excluded_raw] if len(set(excluded)) != len(excluded): raise HTTPException(status_code=409, detail="excluded merchant rows are duplicated") cluster = by_token[token] if not cluster["category_decision_allowed"]: raise HTTPException(status_code=422, detail="ambiguous merchant family requires row-level review") allowed = set(cluster["row_tokens"]) if not set(excluded).issubset(allowed): raise HTTPException(status_code=409, detail="excluded merchant row is outside the displayed cluster") included = [item for item in cluster["_members"] if str(item[1]["_row_token"]) not in set(excluded)] if not included: raise HTTPException(status_code=422, detail="merchant cluster decision selects no rows") public = next(item for item in clusters if item["cluster_token"] == token) decision_type = str(raw.get("decision_type") or "category") if decision_type == "category": category_id = str(raw.get("category_id") or "") category_meta = valid_categories.get(category_id) if not category_meta: raise HTTPException(status_code=409, detail="merchant cluster category is not active") if category_meta["category_type"] != cluster["transaction_semantics"]: raise HTTPException( status_code=422, detail="merchant cluster category does not match transaction semantic", ) for _, row in included: classification = row["classification_v2"] classification.update( category_id=category_id, category_name=category_meta["name"], user_state="proposal_ready", origin="user_cluster_decision", internal_reason="bounded_user_cluster_decision", learned_rule=False, user_message="Kategorie für die exakt angezeigte Händlergruppe festgelegt.", ) row["proposed_category_id"] = category_id row["proposed_category_name"] = category_meta["name"] row["user_state"] = "proposal_ready" if row["disposition"] == "candidate": row["requires_review"] = False public.update( category_id=category_id, category_name=category_meta["name"], decision_type="category", selected_count=len(included), excluded_count=len(excluded), ) elif decision_type == USER_CONFIRMED_UNMATCHED_TRANSFER: complete_cluster_tokens = [str(value) for value in cluster["row_tokens"]] included_tokens = [str(member[1]["_row_token"]) for member in included] approved_tokens_raw = raw.get("approved_row_tokens") or [] evidence = str(raw.get("approval_evidence") or "") approval_baseline = str(raw.get("approval_baseline_fingerprint") or "") approval_nonce = str(raw.get("approval_nonce") or "") try: approval_issued_at = int(raw.get("approval_issued_at")) approval_expires_at = int(raw.get("approval_expires_at")) except (TypeError, ValueError) as exc: raise HTTPException(status_code=422, detail="bounded neutral cluster approval timestamps are invalid") from exc expected_approval = _owner_neutral_cluster_approval( token, complete_cluster_tokens, approval_baseline, approval_issued_at, approval_expires_at, approval_nonce, evidence, ) now = int(time.time()) if approval_baseline != _baseline(conn): raise HTTPException(status_code=409, detail="bounded neutral cluster approval is stale") if ( approval_expires_at - approval_issued_at != 15 * 60 or approval_issued_at > now + 30 or now > approval_expires_at ): raise HTTPException(status_code=409, detail="bounded neutral cluster approval has expired") if ( raw.get("owner_confirmed") is not True or raw.get("category_id") or evidence != OWNER_ATTESTED_TRANSFER_EVIDENCE or bool(excluded) or sorted(included_tokens) != sorted(complete_cluster_tokens) or not approval_nonce or not isinstance(approved_tokens_raw, list) or sorted(str(value) for value in approved_tokens_raw) != sorted(complete_cluster_tokens) or not hmac.compare_digest(str(raw.get("approval_token") or ""), expected_approval) ): raise HTTPException( status_code=422, detail="bounded neutral cluster requires server-bound owner approval and no category", ) for _, row in included: text = _norm(f"{row.get('description') or ''} {row.get('merchant') or ''}") if ( row.get("source_type") not in {"akb_bank", "raiffeisen_bank"} or str(row.get("currency") or "") != "CHF" or not row.get("mapping") or any(marker in text for marker in _UNMATCHED_TRANSFER_DISQUALIFIERS) ): raise HTTPException( status_code=422, detail="bounded neutral cluster contains an ineligible row", ) for _, row in included: classification = row["classification_v2"] classification.update( category_id=None, category_name=None, user_state="special_case", transaction_semantics=USER_CONFIRMED_UNMATCHED_TRANSFER, budget_effect_chf="0.00", origin="owner_confirmed_cluster_decision", internal_reason="bounded_owner_confirmed_household_transfer", decision_version=USER_DECISION_VERSION, learned_rule=False, user_message="Exakt gebundene Händlergruppe als neutralen Haushaltstransfer bestätigt.", ) row["classification"] = USER_CONFIRMED_UNMATCHED_TRANSFER row["user_decision"] = USER_CONFIRMED_UNMATCHED_TRANSFER row["unmatched_transfer_candidate"] = True row["proposed_category_id"] = None row["proposed_category_name"] = None row["user_state"] = "special_case" row["transaction_semantics"] = USER_CONFIRMED_UNMATCHED_TRANSFER row["requires_review"] = False row["disposition"] = "candidate" public.update( decision_type=USER_CONFIRMED_UNMATCHED_TRANSFER, selected_count=len(included), excluded_count=len(excluded), ) else: raise HTTPException(status_code=422, detail="unsupported merchant cluster decision type") for cluster in clusters: cluster.pop("cluster_identity", None) return clusters def _normal_row(profile: str, row: dict[str, Any], index: int, fallback_ref: Any) -> dict[str, Any] | None: low = _low(row) reference = _source_reference(profile, row, fallback_ref) if profile in {"raiffeisen_bank", "akb_bank"}: description = str(low.get("text") or low.get("buchungstext") or "").strip() tx_date = _date(low.get("booked at") or low.get("buchung") or low.get("valuta date") or low.get("valuta")) amount = _money(low.get("credit/debit amount")) if amount is None: credit, debit = _money(low.get("gutschrift")), _money(low.get("belastung")) amount = credit if credit is not None else (-abs(debit) if debit is not None else None) currency = "CHF" provider_id = "" pending = False elif profile == "visa_credit_card": description = str(low.get("merchantname") or low.get("details") or "").strip() tx_date = _date(low.get("date") or low.get("valutadate")) source_amount = _money(low.get("amount")) amount = source_amount currency = str(low.get("currency") or "CHF").upper() provider_id = str(low.get("transactionid") or "").strip() status_text = _norm(low.get("status") or low.get("transactionstatus") or low.get("bookingstatus") or low.get("statetype")) pending = status_text in {"pending", "pendent", "authorised", "authorized", "vorgemerkt"} or _norm(low.get("ispending")) in {"1", "true", "yes"} else: return None if amount is None or not tx_date or not description: return None source_key = provider_id or _canonical({k: _norm(v) for k, v in low.items() if k not in {"status", "transactionstatus", "bookingstatus", "ispending"}}) source_fp = _sha(f"{profile}|{_sha(_norm(reference))}|{source_key}") # The current VISA liability export exposes provider-specific amount/state # columns and reports purchases as positive liability increases. Older # generic CardId CSVs already use economic signs and must not be inverted. liability_view = profile == "visa_credit_card" and any( key in low for key in ("originalamount", "originalcurrency", "statetype") ) signed = -amount if liability_view else amount logical_fp = _sha(_canonical(["household", profile, _sha(_norm(reference)), tx_date, format(signed, "f"), currency, _norm(description)])) text = _norm(description) card_statement_payment = profile == "visa_credit_card" and "ihre zahlung" in text payment_text = any(token in text for token in ( "visa", "viseca", "kreditkarte", "credit card", "kartenabrechnung", "card payment", "ihre zahlung", )) if pending: classification, disposition, review = "credit_card_pending", "pending", True elif card_statement_payment or (profile == "visa_credit_card" and signed > 0 and payment_text): classification, disposition, review = "credit_card_payment_counterpost", "review", True elif profile == "visa_credit_card" and signed > 0: classification, disposition, review = "credit_card_refund", "candidate", False elif payment_text: classification, disposition, review = "credit_card_payment", "review", True elif signed > 0: classification, disposition, review = "income_candidate", "review", True else: classification, disposition, review = "expense_candidate", "candidate", True if currency != "CHF" and not review: # Missing/changed FX data must never turn a clear source row into an implicit CHF posting. disposition, review = "review", True counterparty = extract_counterparty(description, profile) return {"row_index": index, "source_type": profile, "source_reference": reference, "source_row_fingerprint": source_fp, "logical_fingerprint": logical_fp, "transaction_date": tx_date, "description": description[:300], "merchant": (counterparty or description)[:120], "counterparty_hash": _sha("counterparty|" + (counterparty or _norm(description))), "signed_amount": format(signed, "f"), "amount": format(abs(signed), "f"), "currency": currency, "classification": classification, "disposition": disposition, "requires_review": review, "pending": pending, "provider_id": provider_id} def _migros_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: groups: dict[str, list[tuple[int, dict[str, Any]]]] = defaultdict(list) for index, row in enumerate(rows, 1): low = _low(row) key = "|".join(_norm(low.get(k)) for k in ("datum", "zeit", "filiale", "kassennummer", "transaktionsnummer")) groups[key].append((index, low)) result = [] for key, values in sorted(groups.items()): first = values[0][1] tx_date = _date(first.get("datum")) amounts = [_money(item.get("umsatz")) for _, item in values] total = sum((amount for amount in amounts if amount is not None), Decimal("0")) if not tx_date or total == 0: continue source_fp = _sha(f"migros_receipts|{key}") logical_fp = _sha(_canonical(["migros_receipt", tx_date, format(abs(total), "f"), _norm(first.get("filiale"))])) result.append({"row_index": values[0][0], "source_type": "migros_receipts", "source_reference": "", "source_row_fingerprint": source_fp, "logical_fingerprint": logical_fp, "transaction_date": tx_date, "description": "Migros receipt", "merchant": "Migros", "signed_amount": format(-abs(total), "f"), "amount": format(abs(total), "f"), "currency": "CHF", "classification": "migros_receipt_detail", "disposition": "receipt_detail", "requires_review": False, "pending": False, "receipt_key": source_fp[:24], "line_items": [{"row": idx, "name": str(item.get("artikel") or "")[:200], "amount": format(abs(amount), "f") if amount is not None else None} for (idx, item), amount in zip(values, amounts)]}) return result def _existing_sets(conn: Connection) -> tuple[set[str], set[str], set[str]]: files = {row[0] for row in conn.execute("SELECT file_fingerprint FROM household_import_files")} source = {row[0] for row in conn.execute("SELECT source_row_fingerprint FROM household_import_items")} logical = {row[0] for row in conn.execute("SELECT logical_fingerprint FROM household_import_items")} return files, source, logical def _business_identity(account_id: str, tx_date: str, amount: Any, currency: str, description: str) -> str: return _sha(_canonical(["legacy_business", account_id, tx_date, format(Decimal(str(amount)), "f"), currency, _norm(description)])) def _legacy_business_identities(conn: Connection) -> set[str]: identities: set[str] = set() for row in conn.execute( """SELECT account_source,transaction_date, COALESCE(signed_amount_original,amount_original) AS amount,currency_original,description FROM budget_transaction_candidates WHERE account_source IS NOT NULL AND status NOT IN ('ignored','duplicate')""" ): identities.add(_business_identity(row["account_source"], row["transaction_date"], row["amount"], row["currency_original"], row["description"])) for row in conn.execute( """SELECT account_id,transaction_date,amount_original,currency_original,description FROM budget_transactions WHERE status='confirmed'""" ): identities.add(_business_identity(row["account_id"], row["transaction_date"], row["amount_original"], row["currency_original"], row["description"])) return identities def _existing_migros_money(conn: Connection) -> list[dict[str, Any]]: """Return canonical transactions and unconsumed non-receipt money candidates.""" candidates = conn.execute( """SELECT c.transaction_candidate_id,c.transaction_date,c.signed_amount_original,c.amount_original FROM budget_transaction_candidates c LEFT JOIN budget_transactions t ON t.source_candidate_id=c.transaction_candidate_id AND t.status='confirmed' WHERE c.source_type NOT IN ('migros_receipt','migros_receipts') AND c.requires_review=0 AND c.status NOT IN ('pending','ignored','duplicate') AND CAST(COALESCE(c.signed_amount_original,c.amount_original) AS REAL) < 0 AND c.classification NOT IN ( 'credit_card_payment','credit_card_payment_counterpost', 'user_confirmed_unmatched_transfer','transfer' ) AND t.budget_transaction_id IS NULL AND NOT EXISTS ( SELECT 1 FROM household_migros_links ml WHERE ml.status='linked' AND ml.money_candidate_id=c.transaction_candidate_id ) AND (lower(c.description) LIKE '%migros%' OR lower(coalesce(c.merchant,'')) LIKE '%migros%') ORDER BY c.transaction_candidate_id""" ).fetchall() transactions = conn.execute( """SELECT t.budget_transaction_id,t.transaction_date,t.amount_original FROM budget_transactions t WHERE t.status='confirmed' AND t.transaction_type<>'transfer' AND lower(t.description) LIKE '%migros%' AND NOT EXISTS ( SELECT 1 FROM household_migros_links ml WHERE ml.status='linked' AND ml.money_transaction_id=t.budget_transaction_id ) ORDER BY t.budget_transaction_id""" ).fetchall() result = [ {"kind": "candidate", "id": row["transaction_candidate_id"], "transaction_date": row["transaction_date"], "signed_amount": str(row["signed_amount_original"] or row["amount_original"])} for row in candidates ] result.extend( {"kind": "transaction", "id": row["budget_transaction_id"], "transaction_date": row["transaction_date"], "signed_amount": str(row["amount_original"])} for row in transactions ) return result def _reversal_key(value: Any) -> str: text = _norm(value) for token in ("storno", "reversal", "reversed", "cancelled", "canceled", "annulation"): text = text.replace(token, " ") return " ".join(text.split()) def _apply_reversal_matches(conn: Connection, rows: list[dict[str, Any]]) -> None: for row in rows: if ( row["disposition"] in {"pending", "superseded_pending"} or row["disposition"].startswith("duplicate_") or row["classification"] in {"possible_logical_duplicate", "possible_legacy_duplicate"} ): continue text = _norm(row.get("description")) if ( row.get("source_type") != "visa_credit_card" or Decimal(str(row.get("signed_amount") or "0")) <= 0 or not any(token in text for token in ("storno", "reversal", "reversed", "cancelled", "canceled", "annulation")) ): continue mapping = row.get("mapping") if not mapping: row.update(classification="credit_card_reversal_unmatched", disposition="review", requires_review=True) continue tx_date = date.fromisoformat(str(row["transaction_date"])) wanted = abs(Decimal(str(row["signed_amount"]))) key = _reversal_key(row["description"]) candidates = conn.execute( """SELECT budget_transaction_id,transaction_date,description,amount_original FROM budget_transactions WHERE account_id=? AND status='confirmed' AND transaction_type='expense' ORDER BY transaction_date,budget_transaction_id""", (mapping["budget_account_id"],), ).fetchall() matches = [] for candidate in candidates: candidate_date = date.fromisoformat(str(candidate["transaction_date"])) if candidate_date > tx_date or (tx_date - candidate_date).days > 90: continue if abs(Decimal(str(candidate["amount_original"]))) != wanted: continue candidate_key = _reversal_key(candidate["description"]) if key == candidate_key or (candidate_key and (candidate_key in key or key in candidate_key)): matches.append(str(candidate["budget_transaction_id"])) if len(matches) == 1: row.update( classification="credit_card_reversal", disposition="candidate", requires_review=False, reversal_of_transaction_id=matches[0], ) else: row.update(classification="credit_card_reversal_unmatched", disposition="review", requires_review=True) def _pair_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: blocked_classifications = { "possible_logical_duplicate", "possible_legacy_duplicate", "credit_card_reversal_unmatched", } candidates = [ row for row in rows if row.get("mapping") and row["classification"] not in blocked_classifications and row["disposition"] not in {"pending", "superseded_pending", "duplicate_file", "duplicate_source_row", "duplicate_logical", "receipt_detail"} ] possibilities: dict[int, list[int]] = defaultdict(list) for i, left in enumerate(candidates): for j, right in enumerate(candidates): if i >= j or left["mapping"]["budget_account_id"] == right["mapping"]["budget_account_id"]: continue if left["currency"] != right["currency"] or Decimal(left["signed_amount"]) * Decimal(right["signed_amount"]) >= 0: continue if abs(Decimal(left["signed_amount"])) != abs(Decimal(right["signed_amount"])): continue if abs((date.fromisoformat(left["transaction_date"]) - date.fromisoformat(right["transaction_date"])).days) > 3: continue # Generic "payment" occurs on ordinary purchases and is not own-account evidence. # Card settlements are handled by the explicit credit-card classifications below. transfer_words = ("transfer", "uebertrag", "übertrag", "umbuchung", "own account", "kontoausgleich") text_support = any(word in _norm(left["description"] + " " + right["description"]) for word in transfer_words) card_payment = {left["classification"], right["classification"]} == { "credit_card_payment", "credit_card_payment_counterpost" } if not text_support and not card_payment: continue possibilities[i].append(j) possibilities[j].append(i) pairs = [] used: set[int] = set() for i, row in enumerate(candidates): if i in used or Decimal(row["signed_amount"]) >= 0: continue matches = possibilities.get(i, []) if len(matches) == 1 and len(possibilities.get(matches[0], [])) == 1: j = matches[0] target = candidates[j] card_settlement = {row["classification"], target["classification"]} == { "credit_card_payment", "credit_card_payment_counterpost" } date_gap = abs( (date.fromisoformat(row["transaction_date"]) - date.fromisoformat(target["transaction_date"])).days ) # The outer three-day boundary is useful for finding a possible # card settlement, but is not strong enough for automatic pairing. if card_settlement and date_gap == 3: used.update({i, j}) row["pairing_class"] = target["pairing_class"] = "ambiguous" pairs.append({"pairing_class": "ambiguous", "source_row_fingerprint": row["source_row_fingerprint"], "target_row_fingerprint": None, "amount": row["amount"], "currency": row["currency"], "reason_codes": ["card_settlement_edge_window", "manual_review_required"]}) continue used.update({i, j}) pairs.append({"pairing_class": "safe", "source_row_fingerprint": row["source_row_fingerprint"], "target_row_fingerprint": target["source_row_fingerprint"], "amount": row["amount"], "currency": row["currency"], "reason_codes": ["stable_own_accounts", "opposite_signs", "exact_amount", "same_currency", "unique_bidirectional_match", "date_within_3_days"]}) row["pairing_class"] = target["pairing_class"] = "safe" row["disposition"] = target["disposition"] = "transfer_confirmed" elif len(matches) > 1: row["pairing_class"] = "ambiguous" pairs.append({"pairing_class": "ambiguous", "source_row_fingerprint": row["source_row_fingerprint"], "target_row_fingerprint": None, "amount": row["amount"], "currency": row["currency"], "reason_codes": ["multiple_equal_counterbookings", "manual_review_required"]}) elif row["classification"] in {"credit_card_payment", "income_candidate"}: row["pairing_class"] = "unmatched" pairs.append({"pairing_class": "unmatched", "source_row_fingerprint": row["source_row_fingerprint"], "target_row_fingerprint": None, "amount": row["amount"], "currency": row["currency"], "reason_codes": ["counterbooking_missing", "manual_review_required"]}) else: row["pairing_class"] = "review" return pairs def _preview_household_import_internal(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: performance: dict[str, Any] = { "parsing_by_file": [], "normalization_seconds": 0.0, "classification_seconds": 0.0, "merchant_history_queries_seconds": 0.0, "transfer_pairing_seconds": 0.0, "migros_linking_seconds": 0.0, "fingerprint_readiness_seconds": 0.0, } validated_files = _validated_files(payload, performance) merchant_started = time.perf_counter() existing_files, existing_source, existing_logical = _existing_sets(conn) legacy_business = _legacy_business_identities(conn) performance["merchant_history_queries_seconds"] += time.perf_counter() - merchant_started normalized: list[dict[str, Any]] = [] files: list[dict[str, Any]] = [] errors: list[dict[str, str]] = [] seen_source: set[str] = set() seen_logical: set[str] = set() mapping_lookup_cache: dict[tuple[str, str, str, tuple[str, ...]], dict[str, Any] | None] = {} for file_index, entry in enumerate(validated_files, 1): item, parsed = entry["item"], entry["parsed"] profile = parsed["profile"] file_fp = entry["file_fingerprint"] file_duplicate = file_fp in existing_files normalization_started = time.perf_counter() file_rows = _migros_rows(parsed["rows"]) if profile == "migros_receipts" else [ row for idx, raw in enumerate(parsed["rows"], 1) if (row := _normal_row(profile, raw, idx, item.get("source_reference"))) is not None ] performance["normalization_seconds"] += time.perf_counter() - normalization_started if not file_rows: errors.append({"code": "source_file_contains_no_importable_rows", "source_type": profile}) periods = sorted(row["transaction_date"] for row in file_rows if row.get("transaction_date")) files.append({ "file_token": "file_" + file_fp[:12], "profile": profile, "file_fingerprint": file_fp, "physical_row_count": int(parsed.get("physical_row_count", len(parsed["rows"]))), "logical_row_count": len(file_rows), "row_count": len(file_rows), "period_start": periods[0] if periods else None, "period_end": periods[-1] if periods else None, "duplicate": file_duplicate, }) for row in file_rows: row["file_index"] = file_index row["file_fingerprint"] = file_fp if profile != "migros_receipts": mapping_key = ( profile, str(row["source_reference"] or ""), str(item.get("mapping_id") or ""), tuple(sorted(str(value) for value in (item.get("mapping_ids") or []))), ) if mapping_key not in mapping_lookup_cache: mapping_started = time.perf_counter() mapping_lookup_cache[mapping_key] = _mapping_for_file( conn, profile, row["source_reference"], item ) performance["merchant_history_queries_seconds"] += ( time.perf_counter() - mapping_started ) row["mapping"] = mapping_lookup_cache[mapping_key] if not row["mapping"]: errors.append({"code": "account_source_mapping_missing_or_mismatch", "source_type": profile, "row_token": "row_" + row["source_row_fingerprint"][:12]}) else: row["mapping"] = None if file_duplicate: row["disposition"] = "duplicate_file" elif row["source_row_fingerprint"] in existing_source or row["source_row_fingerprint"] in seen_source: row["disposition"] = "duplicate_source_row" elif row["logical_fingerprint"] in existing_logical or row["logical_fingerprint"] in seen_logical: # A business-level near match without the same source row is not safe to suppress: # two genuine purchases can share date, amount and merchant. Keep it in review with # a source-qualified storage identity so the immutable audit trail remains unique. row["classification"] = "possible_logical_duplicate" row["disposition"] = "review" row["requires_review"] = True row["logical_fingerprint"] = _sha( row["logical_fingerprint"] + "|" + row["source_row_fingerprint"] ) elif row.get("mapping") and _business_identity( row["mapping"]["budget_account_id"], row["transaction_date"], row["signed_amount"], row["currency"], row["description"], ) in legacy_business: row["classification"] = "possible_legacy_duplicate" row["disposition"] = "review" row["requires_review"] = True seen_source.add(row["source_row_fingerprint"]) seen_logical.add(row["logical_fingerprint"]) normalized.append(row) # A final card row supersedes its pending representation in the same request. by_source: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in normalized: by_source[row["source_row_fingerprint"]].append(row) for versions in by_source.values(): finals = [row for row in versions if not row.get("pending")] pending_versions = [row for row in versions if row.get("pending")] if finals and pending_versions: for row in pending_versions: row["disposition"] = "superseded_pending" selected = finals[0] for duplicate_final in finals[1:]: duplicate_final["disposition"] = "duplicate_source_row" if selected["source_row_fingerprint"] not in existing_source and selected["logical_fingerprint"] not in existing_logical: selected["disposition"] = "candidate" merchant_started = time.perf_counter() _apply_reversal_matches(conn, normalized) performance["merchant_history_queries_seconds"] += time.perf_counter() - merchant_started pairing_started = time.perf_counter() pairs = _pair_rows(normalized) performance["transfer_pairing_seconds"] = time.perf_counter() - pairing_started category_overrides = payload.get("category_overrides") or {} if not isinstance(category_overrides, dict): raise HTTPException(status_code=422, detail="category_overrides must be an object") valid_categories = { str(item["category_id"]): { "name": str(item["name"]), "category_type": str(item["category_type"]), } for item in conn.execute( """SELECT category_id,name,category_type FROM budget_categories WHERE is_active=1 ORDER BY sort_order,name,category_id""" ).fetchall() } for row in normalized: row["_row_token"] = "row_" + _sha( f"{row['source_type']}|{row['source_reference']}|{row['source_row_fingerprint']}", )[:24] user_decisions = _validated_user_decisions(payload, normalized) lookup_cache = ClassificationLookupCache(conn) lookup_started = time.perf_counter() lookup_cache.category_rows() lookup_cache.histories() lookup_cache.merchant_defaults() lookup_cache.recurring() for source_type in sorted({str(row.get("source_type") or "") for row in normalized}): lookup_cache.aliases(source_type) lookup_cache.review_rules(source_type) performance["merchant_history_queries_seconds"] += time.perf_counter() - lookup_started classification_started = time.perf_counter() for index, row in enumerate(normalized, 1): row_token = str(row["_row_token"]) selected_decision = user_decisions.get(row_token) row["user_decision"] = None preliminary = classify_household_row(conn, row, lookup_cache) row["unmatched_transfer_candidate"] = _has_unmatched_transfer_evidence( row, preliminary ) if selected_decision and not row["unmatched_transfer_candidate"]: raise HTTPException( status_code=422, detail="user-confirmed unmatched transfer requires an unresolved unmatched transfer candidate", ) row["user_decision"] = selected_decision classification_v2 = ( classify_household_row(conn, row, lookup_cache) if selected_decision else preliminary ) classification_v2["counterparty_hash"] = row.get("counterparty_hash") override = category_overrides.get(row_token) if override is None: override = category_overrides.get(f"row_{index}") if override is not None: category_id = str(override) category_meta = valid_categories.get(category_id) expected_type = str(classification_v2["transaction_semantics"]) if ( not category_meta or expected_type not in {"expense", "income"} or category_meta["category_type"] != expected_type ): raise HTTPException( status_code=422, detail="category override is not active for this transaction semantic", ) classification_v2.update( category_id=category_id, category_name=category_meta["name"], user_state="proposal_ready", origin="user_override", learned_rule=False, internal_reason="inline_category_override", user_message="Kategorie von Ihnen angepasst.", ) row["classification_v2"] = classification_v2 row["proposed_category_id"] = classification_v2.get("category_id") row["proposed_category_name"] = classification_v2.get("category_name") row["user_state"] = classification_v2["user_state"] row["transaction_semantics"] = classification_v2["transaction_semantics"] if ( classification_v2["user_state"] == "proposal_ready" and classification_v2.get("category_id") and row["disposition"] == "candidate" ): row["requires_review"] = False if classification_v2["transaction_semantics"] == "user_confirmed_unmatched_transfer": row["classification"] = "user_confirmed_unmatched_transfer" row["requires_review"] = False row["disposition"] = "candidate" performance["classification_seconds"] = time.perf_counter() - classification_started merchant_clusters = _apply_cluster_decisions(conn, normalized, payload, valid_categories) migros_started = time.perf_counter() money_rows = [ { "kind": "row", "id": row["source_row_fingerprint"], "transaction_date": row["transaction_date"], "signed_amount": row["signed_amount"], "row": row, "linkable": row["disposition"] == "candidate" and not row["requires_review"], } for row in normalized if row["source_type"] != "migros_receipts" and row["disposition"] in {"candidate", "review"} and not row.get("pending") and row.get("mapping") and row["currency"] == "CHF" and row.get("transaction_semantics") == "expense" and "migros" in _norm(row["description"]) ] existing_money = _existing_migros_money(conn) receipt_rows = [row for row in normalized if row["source_type"] == "migros_receipts"] money_pool = [*existing_money, *money_rows] exact_by_receipt: dict[str, list[dict[str, Any]]] = {} exact_receipts_by_money: dict[tuple[str, str], int] = defaultdict(int) for receipt in receipt_rows: receipt_total = Decimal(receipt["amount"]) matches = [ money for money in money_pool if money["transaction_date"] == receipt["transaction_date"] and abs(abs(Decimal(money["signed_amount"])) - receipt_total) <= Decimal("0.01") ] exact_by_receipt[receipt["source_row_fingerprint"]] = matches for match in matches: exact_receipts_by_money[(str(match["kind"]), str(match["id"]))] += 1 receipt_links = [] for receipt in receipt_rows: receipt_total = Decimal(receipt["amount"]) exact = exact_by_receipt[receipt["source_row_fingerprint"]] bidirectional = [ match for match in exact if exact_receipts_by_money[(str(match["kind"]), str(match["id"]))] == 1 ] chosen = bidirectional[0] if len(exact) == 1 and len(bidirectional) == 1 else None same_day = [ money for money in money_pool if money["transaction_date"] == receipt["transaction_date"] ] if chosen is not None and chosen.get("linkable", True): status = "linked" elif chosen is not None: status = "review" elif exact or same_day: status = "review" if len(same_day) == 1: chosen = same_day[0] else: status = "unmatched" difference = abs(abs(Decimal(chosen["signed_amount"])) - receipt_total) if chosen else None receipt["requires_review"] = status == "review" receipt["disposition"] = "review" if status == "review" else "receipt_detail" receipt_links.append({"receipt_row_fingerprint": receipt["source_row_fingerprint"], "money_row_fingerprint": chosen["id"] if chosen and chosen["kind"] == "row" else None, "money_candidate_id": chosen["id"] if chosen and chosen["kind"] == "candidate" else None, "money_transaction_id": chosen["id"] if chosen and chosen["kind"] == "transaction" else None, "receipt_total": receipt["amount"], "money_total": format(abs(Decimal(chosen["signed_amount"])), "f") if chosen else None, "difference": format(difference, "f") if difference is not None else None, "status": status}) performance["migros_linking_seconds"] = time.perf_counter() - migros_started fingerprint_started = time.perf_counter() baseline = _baseline(conn) input_fp = _input_fingerprint( validated_files, payload.get("category_overrides"), user_decisions, payload.get("cluster_decisions"), ) safe_rows = [{k: v for k, v in row.items() if k not in {"source_reference", "mapping", "line_items", "provider_id", "user_decision", "counterparty_hash"}} | {"row_token": "row_" + row["source_row_fingerprint"][:12], "account_id": row["mapping"]["budget_account_id"] if row.get("mapping") else None, "account_name": row["mapping"]["account_name"] if row.get("mapping") else None} for row in normalized] duplicate_count = sum(row["disposition"].startswith("duplicate_") or row["disposition"] == "superseded_pending" for row in normalized) counts = {"files": len(files), "rows": len(normalized), "candidates": sum(row["disposition"] in {"candidate", "review", "receipt_detail", "transfer_confirmed"} for row in normalized), "safe_transfer_pairs": sum(pair["pairing_class"] == "safe" for pair in pairs), "duplicates": duplicate_count, "pending": sum(row["disposition"] in {"pending", "superseded_pending"} for row in normalized), "review": sum(bool(row["requires_review"]) for row in normalized), "receipt_links": len(receipt_links), "proposals": sum(row.get("user_state") == "proposal_ready" and not row["disposition"].startswith("duplicate_") for row in normalized), "decisions": sum(row.get("user_state") == "decision_needed" and not row["disposition"].startswith("duplicate_") for row in normalized), "special_cases": sum(row.get("user_state") == "special_case" and not row["disposition"].startswith("duplicate_") for row in normalized), "unmatched_neutral_transfers": sum(row.get("transaction_semantics") in {"credit_card_payment", "user_confirmed_unmatched_transfer"} and row.get("pairing_class") != "safe" for row in normalized), "credit_card_payments": sum(row.get("transaction_semantics") == "credit_card_payment" for row in normalized), "linked_migros_receipts": sum(link["status"] == "linked" for link in receipt_links), "unlinked_migros_receipts": sum(link["status"] != "linked" for link in receipt_links)} ordinary_rows = [ row for row in normalized if row["source_type"] != "migros_receipts" and row["currency"] == "CHF" and row.get("mapping") and row["disposition"] not in { "pending", "superseded_pending", "duplicate_file", "duplicate_source_row", } and row.get("pairing_class") != "safe" and row.get("transaction_semantics") not in { "transfer", "credit_card_payment", USER_CONFIRMED_UNMATCHED_TRANSFER, } ] unresolved_rows = [ row for row in ordinary_rows if row["classification_v2"]["user_state"] == "decision_needed" ] unresolved_token_set = { str(row["_row_token"]) for row in unresolved_rows } unresolved_clusters = [ cluster for cluster in merchant_clusters if set(cluster["row_tokens"]) & unresolved_token_set ] grouped_review_clusters = [ cluster for cluster in unresolved_clusters if cluster["category_decision_allowed"] and len(set(cluster["row_tokens"]) & unresolved_token_set) >= 2 ] actionable_group_tokens = { token for cluster in grouped_review_clusters for token in cluster["row_tokens"] if token in unresolved_token_set } individual_review_count = sum( str(row["_row_token"]) not in actionable_group_tokens for row in unresolved_rows ) covered_count = len(ordinary_rows) - len(unresolved_rows) coverage_ratio = ( Decimal(covered_count) / Decimal(len(ordinary_rows)) if ordinary_rows else Decimal("1") ) individual_ratio = ( Decimal(individual_review_count) / Decimal(len(ordinary_rows)) if ordinary_rows else Decimal("0") ) manual_decision_count = len(unresolved_rows) manual_decision_ratio = ( Decimal(manual_decision_count) / Decimal(len(ordinary_rows)) if ordinary_rows else Decimal("0") ) review_threshold = { "ordinary_monetary_count": len(ordinary_rows), "covered_count": covered_count, "coverage_ratio": format(coverage_ratio.quantize(Decimal("0.0001")), "f"), "minimum_coverage_ratio": format(STRICT_MINIMUM_COVERAGE_RATIO, "f"), "individual_review_count": individual_review_count, "maximum_individual_review_count": STRICT_MAXIMUM_INDIVIDUAL_REVIEW_COUNT, "individual_review_ratio": format(individual_ratio.quantize(Decimal("0.0001")), "f"), "maximum_individual_review_ratio": format(STRICT_MAXIMUM_INDIVIDUAL_REVIEW_RATIO, "f"), "merchant_review_cluster_count": len(grouped_review_clusters), "maximum_merchant_review_cluster_count": STRICT_MAXIMUM_MERCHANT_REVIEW_CLUSTER_COUNT, } concept_v2_readiness = { "ordinary_monetary_count": len(ordinary_rows), "plausible_proposal_count": covered_count, "plausible_proposal_ratio": format(coverage_ratio.quantize(Decimal("0.0001")), "f"), "minimum_plausible_proposal_ratio": format(CONCEPT_V2_MINIMUM_COVERAGE_RATIO, "f"), "manual_decision_count": manual_decision_count, "manual_decision_ratio": format(manual_decision_ratio.quantize(Decimal("0.0001")), "f"), "maximum_manual_decision_ratio": format(CONCEPT_V2_MAXIMUM_MANUAL_DECISION_RATIO, "f"), "coverage_target_met": coverage_ratio >= CONCEPT_V2_MINIMUM_COVERAGE_RATIO, "manual_decision_target_met": manual_decision_ratio <= CONCEPT_V2_MAXIMUM_MANUAL_DECISION_RATIO, } concept_v2_readiness["gate_met"] = bool( concept_v2_readiness["coverage_target_met"] and concept_v2_readiness["manual_decision_target_met"] ) strict_quality_indicator = { "coverage_target_met": coverage_ratio >= STRICT_MINIMUM_COVERAGE_RATIO, "individual_review_count_met": individual_review_count <= STRICT_MAXIMUM_INDIVIDUAL_REVIEW_COUNT, "individual_review_ratio_met": individual_ratio <= STRICT_MAXIMUM_INDIVIDUAL_REVIEW_RATIO, "merchant_cluster_target_met": ( len(grouped_review_clusters) <= STRICT_MAXIMUM_MERCHANT_REVIEW_CLUSTER_COUNT ), } strict_quality_indicator["indicator_met"] = all(strict_quality_indicator.values()) migros_money_identities = [ (link.get("money_row_fingerprint"), link.get("money_candidate_id"), link.get("money_transaction_id")) for link in receipt_links if link["status"] == "linked" ] no_migros_double_count = len(migros_money_identities) == len(set(migros_money_identities)) no_card_payment_as_income = not any( row.get("transaction_semantics") == "credit_card_payment" and row["classification_v2"].get("budget_effect_chf") != "0.00" for row in ordinary_rows ) technical = not errors readiness_checks = { "no_unresolved_account_or_sign_semantics": technical and all(row.get("mapping") for row in ordinary_rows), "no_card_payment_as_income": no_card_payment_as_income, "no_migros_double_count": no_migros_double_count, "no_critical_mapping_or_balance_deviation": technical, "concept_v2_coverage_target_met": bool(concept_v2_readiness["coverage_target_met"]), "concept_v2_manual_decision_target_met": bool(concept_v2_readiness["manual_decision_target_met"]), } business_ready = technical and all(readiness_checks.values()) writable = [row for row in normalized if row["disposition"] in {"candidate", "review", "receipt_detail", "transfer_confirmed"}] materialized = [row for row in writable if not row["requires_review"] and row["source_type"] != "migros_receipts" and row["disposition"] != "transfer_confirmed" and row["currency"] == "CHF" and row.get("mapping")] income = sum((Decimal(row["signed_amount"]) for row in materialized if row.get("transaction_semantics") == "income"), Decimal("0")) expenses = sum((abs(Decimal(row["signed_amount"])) for row in materialized if row.get("transaction_semantics") == "expense"), Decimal("0")) expected_budget_effect = { "income_chf": format(income, ".2f"), "expense_chf": format(expenses, ".2f"), "neutral_transfer_chf": format(sum((abs(Decimal(row["signed_amount"])) for row in materialized if row.get("transaction_semantics") in {"transfer", "user_confirmed_unmatched_transfer", "credit_card_payment"}), Decimal("0")), ".2f"), } expected_writes = { "import_batches": 1, "import_files": sum(not item["duplicate"] for item in files), "candidates": len(writable), "transactions": len(materialized) + 2 * counts["safe_transfer_pairs"], "transfer_pairs": counts["safe_transfer_pairs"], "migros_links": sum(link["status"] == "linked" for link in receipt_links), } preview_core = {"contract_version": CONTRACT_VERSION, "pairing_version": PAIRING_VERSION, "classification_version": CLASSIFICATION_VERSION, "input_fingerprint": input_fp, "baseline_fingerprint": baseline, "files": files, "rows": safe_rows, "transfer_pairs": pairs, "receipt_links": receipt_links, "merchant_clusters": merchant_clusters, "counts": counts, "review_threshold": review_threshold, "concept_v2_readiness": concept_v2_readiness, "strict_quality_indicator": strict_quality_indicator, "readiness_checks": readiness_checks, "technically_confirmable": technical, "business_ready_for_confirm": business_ready, "expected_budget_effect": expected_budget_effect, "expected_writes": expected_writes} preview_fp = _sha(_canonical(preview_core)) performance["fingerprint_readiness_seconds"] = time.perf_counter() - fingerprint_started for key, value in list(performance.items()): if isinstance(value, float): performance[key] = round(value, 6) return preview_core | {"preview_fingerprint": preview_fp, "performance": performance, "confirmable": technical, "errors": errors} def preview_household_import(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: """Return the complete human preview without exposing internal row/file identities.""" api_started = time.perf_counter() result = _preview_household_import_internal(conn, payload) files = [ {key: value for key, value in item.items() if key not in {"file_fingerprint", "file_token"}} | {"file_token": f"file_{index}"} for index, item in enumerate(result["files"], 1) ] row_tokens = { item["source_row_fingerprint"]: str(item["_row_token"]) for item in result["rows"] } rows = [ { key: value for key, value in item.items() if key not in { "file_fingerprint", "source_row_fingerprint", "logical_fingerprint", "reversal_of_transaction_id", "row_token", "classification_v2", "_row_token", "account_id", "account_name", "description", } } | {"row_token": row_tokens[item["source_row_fingerprint"]]} for item in result["rows"] ] receipt_status = { link["receipt_row_fingerprint"]: link["status"] for link in result["receipt_links"] } items = [] for item in result["rows"]: classification_v2 = item["classification_v2"] special_hint = None semantics = classification_v2["transaction_semantics"] if semantics == "user_confirmed_unmatched_transfer": special_hint = classification_v2["user_message"] elif semantics in {"transfer", "credit_card_payment"}: special_hint = "Gegenbuchung fehlt" if item.get("pairing_class") != "safe" else "Möglicher interner Transfer" elif item["classification"] in {"possible_logical_duplicate", "possible_legacy_duplicate"}: special_hint = "Mögliche Dublette" elif item["source_type"] == "migros_receipts" and receipt_status.get(item["source_row_fingerprint"]) == "linked": special_hint = "Passender Migros-Beleg gefunden" elif item["source_type"] == "migros_receipts" and receipt_status.get(item["source_row_fingerprint"]) != "linked": special_hint = "Zahlungsgegenposten fehlt" account_id = str(item.get("account_id") or "") items.append({ "row_token": row_tokens[item["source_row_fingerprint"]], "date": item["transaction_date"], "merchant": item.get("merchant") or "Unbekannter Händler", "description": item.get("merchant") or "Unbekannter Händler", "amount": item["signed_amount"], "currency": item["currency"], "category_id": classification_v2.get("category_id"), "category_name": classification_v2.get("category_name"), "transaction_semantics": semantics, "user_state": classification_v2["user_state"], "state_label": { "proposal_ready": "Vorschlag bereit", "decision_needed": "Entscheidung nötig", "special_case": "Sonderfall erkannt", }[classification_v2["user_state"]], "explanation": classification_v2["user_message"], "special_hint": special_hint, "source_label": { "akb_bank": "AKB", "raiffeisen_bank": "Raiffeisen", "visa_credit_card": "VISA", "migros_receipts": "Migros-Beleg", }.get(item["source_type"], "Import"), "account_hint": ("•••• " + _sha(account_id)[-4:]) if account_id else None, "can_confirm": ( classification_v2["user_state"] == "proposal_ready" or semantics == "user_confirmed_unmatched_transfer" ), "selected": classification_v2["user_state"] == "proposal_ready", "decision_actions": ( [USER_CONFIRMED_UNMATCHED_TRANSFER] if item.get("unmatched_transfer_candidate") is True else [] ), "selected_user_decision": ( USER_CONFIRMED_UNMATCHED_TRANSFER if semantics == USER_CONFIRMED_UNMATCHED_TRANSFER else None ), }) pairs = [ { key: value for key, value in item.items() if key not in {"source_row_fingerprint", "target_row_fingerprint"} } | { "source_row_token": row_tokens.get(item["source_row_fingerprint"]), "target_row_token": ( row_tokens.get(item["target_row_fingerprint"]) if item.get("target_row_fingerprint") else None ), } for item in result["transfer_pairs"] ] links = [ { key: value for key, value in item.items() if key not in { "receipt_row_fingerprint", "money_row_fingerprint", "money_candidate_id", "money_transaction_id", } } | { "receipt_row_token": row_tokens.get(item["receipt_row_fingerprint"]), "money_row_token": ( row_tokens.get(item["money_row_fingerprint"]) if item.get("money_row_fingerprint") else None ), } for item in result["receipt_links"] ] response = { "contract_version": result["contract_version"], "pairing_version": result["pairing_version"], "classification_version": result["classification_version"], "preview_fingerprint": result["preview_fingerprint"], "baseline_fingerprint": result["baseline_fingerprint"], "files": files, "rows": rows, "items": items, "categories": [ { "category_id": str(item["category_id"]), "name": str(item["name"]), "category_type": str(item["category_type"]), } for item in conn.execute( """SELECT category_id,name,category_type FROM budget_categories WHERE is_active=1 ORDER BY sort_order,name,category_id""" ).fetchall() ], "transfer_pairs": pairs, "receipt_links": links, "merchant_clusters": result["merchant_clusters"], "counts": result["counts"], "review_threshold": result["review_threshold"], "concept_v2_readiness": result["concept_v2_readiness"], "strict_quality_indicator": result["strict_quality_indicator"], "readiness_checks": result["readiness_checks"], "technically_confirmable": result["technically_confirmable"], "business_ready_for_confirm": result["business_ready_for_confirm"], "expected_budget_effect": result["expected_budget_effect"], "expected_writes": result["expected_writes"], "performance": result["performance"], "confirmable": result["confirmable"], "errors": [ {key: value for key, value in item.items() if key != "row_token"} for item in result["errors"] ], } response["performance"]["api_total_seconds"] = round( time.perf_counter() - api_started, 6 ) return response def _candidate_id(row: dict[str, Any]) -> str: return "btxcand_hh_" + row["source_row_fingerprint"][:24] def _insert_candidate(conn: Connection, row: dict[str, Any], batch_id: str, timestamp: str) -> str: candidate_id = _candidate_id(row) status = "transfer_candidate" if row["disposition"] == "transfer_confirmed" else ("covered_by_source" if row["source_type"] == "migros_receipts" else "needs_review") tx_type = "refund" if row["classification"] in {"credit_card_refund", "credit_card_reversal"} else ("income" if Decimal(row["signed_amount"]) > 0 else "expense") if row["disposition"] == "transfer_confirmed" or row.get("transaction_semantics") == "user_confirmed_unmatched_transfer": tx_type = "transfer" conn.execute( """INSERT INTO budget_transaction_candidates( transaction_candidate_id,source_file_label,source_row_or_range,source_type, transaction_date,value_date,description,merchant,amount_original,signed_amount_original, currency_original,confidence,requires_review,status,notes,created_at,updated_at, classification,review_reason,rule_id,rule_name,source_priority,receipt_key, account_source,raw_fingerprint,household_batch_id,source_row_fingerprint,logical_fingerprint) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (candidate_id, "household:" + row["file_fingerprint"][:12], f"R{row['row_index']}", row["source_type"], row["transaction_date"], row["transaction_date"], row["description"], row.get("merchant"), row["amount"], row["signed_amount"], row["currency"], "1.00" if row["disposition"] == "transfer_confirmed" else "0.70", 1 if row["requires_review"] else 0, status, _canonical({ "contract_version": CONTRACT_VERSION, "classification_version": CLASSIFICATION_VERSION, "transaction_type": tx_type, "classification_origin": row.get("classification_v2", {}).get("origin"), "counterparty_hash": row.get("classification_v2", {}).get("counterparty_hash"), "evidence_transaction_ids": row.get("classification_v2", {}).get("evidence_transaction_ids", []), "user_state": row.get("user_state"), }), timestamp, timestamp, row["classification"], "household_import_review" if row["requires_review"] else None, CONTRACT_VERSION, CONTRACT_VERSION, 90, row.get("receipt_key"), row["mapping"]["budget_account_id"] if row.get("mapping") else None, row["source_row_fingerprint"], batch_id, row["source_row_fingerprint"], row["logical_fingerprint"]), ) conn.execute( """UPDATE budget_transaction_candidates SET proposed_category_id=?,proposed_category_name=? WHERE transaction_candidate_id=?""", (row.get("proposed_category_id"), row.get("proposed_category_name"), candidate_id), ) return candidate_id def _confirm_safe_nontransfer_candidate( conn: Connection, row: dict[str, Any], candidate_id: str, timestamp: str, ) -> str | None: if ( row["requires_review"] or row["source_type"] == "migros_receipts" or row["disposition"] == "transfer_confirmed" or row["currency"] != "CHF" or not row.get("mapping") ): return None transaction_id = "btx_hh_" + row["source_row_fingerprint"][:24] transaction_type = ( "transfer" if row.get("transaction_semantics") == "user_confirmed_unmatched_transfer" else ( "refund" if row["classification"] in {"credit_card_refund", "credit_card_reversal"} else ("income" if Decimal(row["signed_amount"]) > 0 else "expense") ) ) reversal_id = row.get("reversal_of_transaction_id") if transaction_type == "refund" and reversal_id and not refund_link_is_valid( conn, str(reversal_id), str(row["signed_amount"]), exclude_transaction_id=transaction_id, ): return None conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,booking_date, description,payee,amount_original,currency_original,fx_rate_to_chf,amount_chf, fx_status,category_id,status,source_type,source_candidate_id,reversal_of_transaction_id, notes,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,'1',?,'not_needed',?,'confirmed','import_candidate',?,?,?,?,?)""", ( transaction_id, row["mapping"]["budget_account_id"], transaction_type, row["transaction_date"], row["transaction_date"], row["description"], row.get("merchant"), row["signed_amount"], row["currency"], row["signed_amount"], row.get("proposed_category_id"), candidate_id, reversal_id, _canonical({ "contract_version": CONTRACT_VERSION, "classification_version": CLASSIFICATION_VERSION, "classification_origin": row.get("classification_v2", {}).get("origin"), "counterparty_hash": row.get("classification_v2", {}).get("counterparty_hash"), "user_state": row.get("user_state"), **({"reversal_of_transaction_id": row["reversal_of_transaction_id"]} if row.get("reversal_of_transaction_id") else {}), }), timestamp, timestamp, ), ) conn.execute( """UPDATE budget_transaction_candidates SET status='confirmed',requires_review=0,confirmed_transaction_id=?,updated_at=? WHERE transaction_candidate_id=?""", (transaction_id, timestamp, candidate_id), ) return transaction_id def confirm_household_import(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: expected_preview = str(payload.get("preview_fingerprint") or "") expected_baseline = str(payload.get("baseline_fingerprint") or "") if not expected_preview or not expected_baseline or payload.get("confirm") is not True: raise HTTPException(status_code=422, detail="confirm=true and both fingerprints are required") validated_files = _validated_files(payload) current_input = _input_fingerprint( validated_files, payload.get("category_overrides"), payload.get("user_decisions"), payload.get("cluster_decisions"), ) existing = conn.execute("SELECT * FROM household_import_batches WHERE preview_fingerprint=?", (expected_preview,)).fetchone() if existing: if current_input != existing["input_fingerprint"] or expected_baseline != existing["baseline_fingerprint"]: raise HTTPException(status_code=409, detail="confirmed preview does not match this input") current_time = int(time.time()) for decision in payload.get("cluster_decisions") or []: if decision.get("decision_type") != USER_CONFIRMED_UNMATCHED_TRANSFER: continue try: approval_expires_at = int(decision.get("approval_expires_at")) except (TypeError, ValueError) as exc: raise HTTPException(status_code=422, detail="bounded neutral cluster approval timestamps are invalid") from exc if current_time > approval_expires_at: raise HTTPException(status_code=409, detail="bounded neutral cluster approval has expired") return {"status": "confirmed", "batch_id": existing["batch_id"], "idempotent": True, "counts": {key: existing[key] for key in ("file_count", "row_count", "candidate_count", "transfer_pair_count", "duplicate_count", "review_count", "receipt_link_count")}} reconstructed = _preview_household_import_internal(conn, payload) if reconstructed["preview_fingerprint"] != expected_preview: raise HTTPException(status_code=409, detail="preview fingerprint mismatch; preview again") if reconstructed["baseline_fingerprint"] != expected_baseline: raise HTTPException(status_code=409, detail="database baseline changed; preview again") if not reconstructed["confirmable"]: raise HTTPException(status_code=409, detail="preview is not technically confirmable") if not reconstructed["business_ready_for_confirm"]: raise HTTPException(status_code=409, detail="preview is not business-ready for confirm") timestamp = now() batch_id = "hhbatch_" + expected_preview[:24] reconstructed_rows = reconstructed["rows"] # map fields needed by writes from the current stable mapping table for entry in validated_files: file_item, parsed = entry["item"], entry["parsed"] originals = _migros_rows(parsed["rows"]) if parsed["profile"] == "migros_receipts" else [r for i, raw in enumerate(parsed["rows"], 1) if (r := _normal_row(parsed["profile"], raw, i, file_item.get("source_reference")))] for original in originals: targets = [row for row in reconstructed_rows if row["source_row_fingerprint"] == original["source_row_fingerprint"]] for target in targets: target["mapping"] = _mapping_for_file( conn, original["source_type"], original["source_reference"], file_item, ) if original["source_type"] != "migros_receipts" else None target["line_items"] = original.get("line_items", []) writable = [row for row in reconstructed_rows if row["disposition"] in {"candidate", "review", "receipt_detail", "transfer_confirmed"}] rows_by_fp = {row["source_row_fingerprint"]: row for row in writable} had_outer_transaction = conn.in_transaction if had_outer_transaction: conn.execute("SAVEPOINT household_confirm") else: conn.execute("BEGIN IMMEDIATE") try: # Close the preview-to-write TOCTOU window after acquiring the write lock/savepoint. if _baseline(conn) != expected_baseline: raise HTTPException(status_code=409, detail="database baseline changed; preview again") counts = reconstructed["counts"] expected_pair_count = sum(pair["pairing_class"] == "safe" for pair in reconstructed["transfer_pairs"]) expected_link_count = sum( link["status"] == "linked" and link["receipt_row_fingerprint"] in rows_by_fp for link in reconstructed["receipt_links"] ) audit_id = record_audit_event( conn, source="household_import", action="household_import_confirmed", entity_type="household_import_batch", entity_id=batch_id, new_values={ "contract_version": CONTRACT_VERSION, "classification_version": CLASSIFICATION_VERSION, "pairing_version": PAIRING_VERSION, "preview_fingerprint": expected_preview, "sources": sorted({str(row["source_type"]) for row in reconstructed_rows}), "masked_accounts": sorted({ "••••" + _sha(str(row["mapping"]["budget_account_id"]))[-4:] for row in reconstructed_rows if row.get("mapping") }), "classification_origins": sorted({ str(row.get("classification_v2", {}).get("origin") or "unresolved") for row in reconstructed_rows }), "category_override_count": len(payload.get("category_overrides") or {}), "user_decision_count": len(payload.get("user_decisions") or {}), "cluster_decision_count": len(payload.get("cluster_decisions") or []), "owner_attested_neutral_cluster_count": sum( cluster.get("decision_type") == USER_CONFIRMED_UNMATCHED_TRANSFER for cluster in reconstructed.get("merchant_clusters", []) ), "owner_attestation_evidence_version": OWNER_ATTESTED_TRANSFER_EVIDENCE, "cluster_decision_version": CLUSTER_DECISION_VERSION, "user_decision_version": USER_DECISION_VERSION, "business_ready_for_confirm": reconstructed["business_ready_for_confirm"], "counts": counts, "expected_writes": reconstructed["expected_writes"], "actual_writes": reconstructed["expected_writes"], "status": "confirmed", }, created_by="user", ) # Insert the FK parent before candidates, files, items, and receipt links. conn.execute( """INSERT INTO household_import_batches VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (batch_id, CONTRACT_VERSION, expected_preview, expected_baseline, reconstructed["input_fingerprint"], counts["files"], counts["rows"], len(writable), expected_pair_count, counts["duplicates"], counts["review"], expected_link_count, "confirmed", audit_id, timestamp, "user"), ) candidate_ids: dict[str, str] = {} for row in writable: candidate_ids[row["source_row_fingerprint"]] = _insert_candidate(conn, row, batch_id, timestamp) _confirm_safe_nontransfer_candidate( conn, row, candidate_ids[row["source_row_fingerprint"]], timestamp, ) if row["source_type"] == "migros_receipts": for item in row.get("line_items", []): item_fp = _sha(_canonical([row["source_row_fingerprint"], item["row"], item["name"], item["amount"]])) conn.execute( """INSERT INTO budget_import_line_items( line_item_id,transaction_candidate_id,source_file_label,receipt_key, source_row_or_range,item_name,quantity,is_promotion,amount_original, currency_original,raw_fingerprint,created_at) VALUES (?,?,?,?,?,?,NULL,0,?,'CHF',?,?)""", ("bhhli_" + item_fp[:24], candidate_ids[row["source_row_fingerprint"]], "household:" + row["file_fingerprint"][:12], row.get("receipt_key") or row["source_row_fingerprint"][:24], f"R{item['row']}", item["name"], item["amount"], item_fp, timestamp), ) pair_ids: list[str] = [] row_pair_ids: dict[str, str] = {} for pair in reconstructed["transfer_pairs"]: if pair["pairing_class"] != "safe": continue source, target = rows_by_fp[pair["source_row_fingerprint"]], rows_by_fp[pair["target_row_fingerprint"]] if Decimal(source["signed_amount"]) > 0: source, target = target, source pair_id = "btpair_hh_" + _sha(source["source_row_fingerprint"] + target["source_row_fingerprint"])[:24] evidence = {"matcher_version": PAIRING_VERSION, "pairing_class": "safe", "batch_id": batch_id, "merchant_text_decisive": False} conn.execute( """INSERT INTO budget_transfer_pairs(transfer_pair_id,source_candidate_id,target_candidate_id, source_account_id,target_account_id,source_signed_amount,target_signed_amount,currency, source_booking_date,target_booking_date,source_value_date,target_value_date,status,quality_status, evidence_json,reason_codes_json,budget_effect_chf,created_at,created_by,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,'safe',?,?,'0',?,'household_import',?)""", (pair_id, candidate_ids[source["source_row_fingerprint"]], candidate_ids[target["source_row_fingerprint"]], source["mapping"]["budget_account_id"], target["mapping"]["budget_account_id"], source["signed_amount"], target["signed_amount"], source["currency"], source["transaction_date"], target["transaction_date"], source["transaction_date"], target["transaction_date"], "proposed", _canonical(evidence), _canonical(pair["reason_codes"]), timestamp, timestamp), ) confirm_transfer_pair(conn, pair_id, decision_by="household_import") pair_ids.append(pair_id) row_pair_ids[source["source_row_fingerprint"]] = pair_id row_pair_ids[target["source_row_fingerprint"]] = pair_id link_count = 0 for link in reconstructed["receipt_links"]: if link["status"] != "linked": continue receipt_id = candidate_ids.get(link["receipt_row_fingerprint"]) if not receipt_id: continue money_id = candidate_ids.get(link.get("money_row_fingerprint") or "") or link.get("money_candidate_id") money_transaction_id = link.get("money_transaction_id") if link["status"] == "linked" and bool(money_id) == bool(money_transaction_id): raise RuntimeError("linked Migros receipt must resolve to exactly one money movement") if money_id: conn.execute("UPDATE budget_transaction_candidates SET linked_candidate_id=? WHERE transaction_candidate_id=?", (money_id, receipt_id)) conn.execute( """INSERT INTO household_migros_links(receipt_link_id,batch_id,receipt_candidate_id, money_candidate_id,money_transaction_id,receipt_total,money_total,difference,status,created_at) VALUES (?,?,?,?,?,?,?,?,?,?)""", ("hhmig_" + link["receipt_row_fingerprint"][:24], batch_id, receipt_id, money_id, money_transaction_id, link["receipt_total"], link["money_total"], link["difference"], link["status"], timestamp), ) link_count += 1 for file in reconstructed["files"]: if not file["duplicate"]: conn.execute("INSERT INTO household_import_files VALUES (?,?,?,?,?,?)", ("hhfile_" + file["file_fingerprint"][:24], batch_id, file["profile"], file["file_fingerprint"], file["row_count"], timestamp)) for row in writable: conn.execute( """INSERT INTO household_import_items(household_item_id,batch_id,source_type, source_row_fingerprint,logical_fingerprint,disposition,candidate_id,transfer_pair_id,created_at) VALUES (?,?,?,?,?,?,?,?,?)""", ("hhitem_" + row["source_row_fingerprint"][:24], batch_id, row["source_type"], row["source_row_fingerprint"], row["logical_fingerprint"], row["disposition"], candidate_ids[row["source_row_fingerprint"]], row_pair_ids.get(row["source_row_fingerprint"]), timestamp), ) if len(pair_ids) != expected_pair_count or link_count != expected_link_count: raise RuntimeError("household import write counts diverged from preview") if had_outer_transaction: conn.execute("RELEASE SAVEPOINT household_confirm") else: conn.commit() except Exception: if had_outer_transaction: conn.execute("ROLLBACK TO SAVEPOINT household_confirm") conn.execute("RELEASE SAVEPOINT household_confirm") else: conn.rollback() raise return {"status": "confirmed", "batch_id": batch_id, "idempotent": False, "counts": {"file_count": counts["files"], "row_count": counts["rows"], "candidate_count": len(writable), "transfer_pair_count": len(pair_ids), "duplicate_count": counts["duplicates"], "review_count": counts["review"], "receipt_link_count": link_count}} def list_household_batches(conn: Connection, limit: int = 30) -> list[dict[str, Any]]: limit = max(1, min(int(limit), 100)) rows = conn.execute( """SELECT batch_id,contract_version,file_count,row_count,candidate_count, transfer_pair_count,duplicate_count,review_count,receipt_link_count,status,confirmed_at FROM household_import_batches ORDER BY confirmed_at DESC,batch_id LIMIT ?""", (limit,) ).fetchall() return [dict(row) for row in rows] def get_household_import_options(conn: Connection) -> dict[str, Any]: return { "contract_version": CONTRACT_VERSION, "max_file_bytes": MAX_IMPORT_BYTES, "max_request_bytes": MAX_IMPORT_BYTES, "profiles": [{"source_type": key, "requires_mapping": key != "migros_receipts"} for key in sorted(IMPORT_PROFILES)], "source_mappings": list_source_mappings(conn), } def _amount_chf_expression(alias: str = "t") -> str: return ( f"CASE WHEN {alias}.amount_chf IS NOT NULL THEN CAST({alias}.amount_chf AS REAL) " f"WHEN upper({alias}.currency_original)='CHF' THEN CAST({alias}.amount_original AS REAL) ELSE NULL END" ) def get_household_overview(conn: Connection, month: str | None = None) -> dict[str, Any]: selected_month = month or date.today().strftime("%Y-%m") if not re.fullmatch(r"\d{4}-(0[1-9]|1[0-2])", selected_month): raise HTTPException(status_code=422, detail="month must use YYYY-MM") financials = get_household_financial_summary(conn, period=selected_month) review_count = get_household_open_review_count(conn) last = conn.execute( """SELECT confirmed_at,status,file_count,row_count FROM household_import_batches ORDER BY confirmed_at DESC,batch_id DESC LIMIT 1""" ).fetchone() income = Decimal(financials["income_chf"]) expense = Decimal(financials["expense_chf"]) data_issues: list[str] = [] if review_count: data_issues.append(f"{review_count} unklare Buchung(en) sind nicht in Einnahmen oder Ausgaben enthalten.") if financials["unavailable_chf_count"]: data_issues.append( f"{financials['unavailable_chf_count']} Buchung(en) ohne bestätigten CHF-Wert sind aus den Summen ausgeschlossen." ) if financials["unlinked_refund_count"]: data_issues.append( f"{financials['unlinked_refund_count']} nicht verknüpfte Rückerstattung(en) haben bis zur Klärung keine Finanzwirkung." ) if financials["transfer_membership_conflict_count"]: data_issues.append( f"{financials['transfer_membership_conflict_count']} widersprüchliche Transferverknüpfung(en) sperren das Transfer-Volumen." ) return { "month": selected_month, "income_chf": format(income, ".2f"), "expense_chf": format(expense, ".2f"), "balance_chf": format(income - expense, ".2f"), "neutral_transfers_chf": "0.00", "neutral_transfer_volume_chf": financials["neutral_transfer_volume_chf"], "neutral_transfer_count": financials["neutral_transfer_count"], "unavailable_transfer_count": financials["unavailable_transfer_count"], "unlinked_refund_count": financials["unlinked_refund_count"], "unavailable_chf_count": financials["unavailable_chf_count"], "transfer_membership_conflict_count": financials["transfer_membership_conflict_count"], "semantics_version": financials["semantics_version"], "categories": financials["categories"], "review_count": review_count, "last_import": ({ "source_label": f"{int(last['file_count'])} Datei(en), {int(last['row_count'])} Zeile(n)", "imported_at": last["confirmed_at"], "status": last["status"], } if last else None), "data_status": { "status": "partial" if data_issues else "current", "label": "Prüfung offen" if review_count or financials["unlinked_refund_count"] else ( "Teilweise verfügbar" if data_issues else "Aktuell" ), "message": " ".join(data_issues) if data_issues else "Nur bestätigte Haushaltsbuchungen sind enthalten.", }, } def _encode_page_cursor(parts: list[str]) -> str: return base64.urlsafe_b64encode(_canonical(parts).encode()).decode().rstrip("=") def _decode_page_cursor(value: str | None, expected_scope: str) -> tuple[str, str, str] | None: if not value: return None try: raw = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)).decode() parts = json.loads(raw) except (ValueError, UnicodeDecodeError, json.JSONDecodeError): raise HTTPException(status_code=422, detail="invalid pagination cursor") from None if not isinstance(parts, list) or len(parts) != 4 or str(parts[0]) != expected_scope: raise HTTPException(status_code=409, detail="pagination cursor does not match current filters") return str(parts[1]), str(parts[2]), str(parts[3]) def _household_transaction_data_version(conn: Connection) -> str: """Digest every value that can change transaction paging or rendered rows.""" digest = hashlib.sha256() queries = ( "SELECT * FROM budget_transactions WHERE status='confirmed' ORDER BY budget_transaction_id", "SELECT * FROM budget_categories ORDER BY category_id", "SELECT * FROM budget_accounts ORDER BY budget_account_id", ) for query in queries: cursor = conn.execute(query) digest.update(_canonical([description[0] for description in cursor.description or ()]).encode()) for row in cursor: digest.update(_canonical([row[index] for index in range(len(row))]).encode()) return digest.hexdigest()[:20] def _household_review_data_version(conn: Connection) -> str: """Digest all review rows, classifier lookups, categories, and mapping inputs.""" digest = hashlib.sha256() queries = ( """SELECT * FROM budget_transaction_candidates WHERE household_batch_id IS NOT NULL ORDER BY transaction_candidate_id""", """SELECT c.* FROM budget_transaction_candidates c JOIN budget_transactions t ON t.source_candidate_id=c.transaction_candidate_id WHERE t.status='confirmed' AND t.category_id IS NOT NULL ORDER BY c.transaction_candidate_id""", "SELECT * FROM budget_transactions WHERE status='confirmed' ORDER BY budget_transaction_id", "SELECT * FROM budget_categories ORDER BY category_id", "SELECT * FROM budget_merchants ORDER BY merchant_id", "SELECT * FROM budget_merchant_aliases ORDER BY alias_id", "SELECT * FROM budget_review_rules ORDER BY rule_id", "SELECT * FROM budget_recurring_payments ORDER BY recurring_id", "SELECT * FROM household_account_source_mappings ORDER BY mapping_id", "SELECT * FROM budget_accounts ORDER BY budget_account_id", "SELECT * FROM accounts ORDER BY account_id", ) for query in queries: cursor = conn.execute(query) digest.update(_canonical([description[0] for description in cursor.description or ()]).encode()) for row in cursor: digest.update(_canonical([row[index] for index in range(len(row))]).encode()) return digest.hexdigest()[:20] def list_household_transactions( conn: Connection, *, period: str | None = None, date_from: str | None = None, date_to: str | None = None, account: str | None = None, transaction_type: str | None = None, category: str | None = None, merchant: str | None = None, source: str | None = None, review: str | None = None, limit: int = 48, cursor: str | None = None, ) -> dict[str, Any]: limit = max(1, min(int(limit), 100)) where = ["t.status='confirmed'"] params: list[Any] = [] if period: if not re.fullmatch(r"\d{4}-(0[1-9]|1[0-2])", period): raise HTTPException(status_code=422, detail="period must use YYYY-MM") where.append("substr(t.transaction_date,1,7)=?") params.append(period) for label, value, operator in (("date_from", date_from, ">="), ("date_to", date_to, "<=")): if value: try: date.fromisoformat(value) except ValueError: raise HTTPException(status_code=422, detail=f"{label} must use YYYY-MM-DD") from None where.append(f"t.transaction_date {operator} ?") params.append(value) if date_from and date_to and date_from > date_to: raise HTTPException(status_code=422, detail="date_from must not be after date_to") if account: where.append("lower(coalesce(ba.name,'')) LIKE ?") params.append("%" + account.casefold() + "%") if transaction_type: where.append("t.transaction_type=?") params.append(transaction_type) reversal_target_sql = ( "COALESCE(t.reversal_of_transaction_id, " "CASE WHEN json_valid(t.notes) " "THEN json_extract(t.notes,'$.reversal_of_transaction_id') END)" ) original_amount_sql = ( "CASE WHEN original.amount_chf IS NOT NULL THEN CAST(original.amount_chf AS REAL) " "WHEN upper(original.currency_original)='CHF' THEN CAST(original.amount_original AS REAL) ELSE NULL END" ) refund_amount_sql = ( "CASE WHEN r.amount_chf IS NOT NULL THEN CAST(r.amount_chf AS REAL) " "WHEN upper(r.currency_original)='CHF' THEN CAST(r.amount_original AS REAL) ELSE NULL END" ) refund_target_r_sql = ( "COALESCE(r.reversal_of_transaction_id,CASE WHEN json_valid(r.notes) " "THEN json_extract(r.notes,'$.reversal_of_transaction_id') END)" ) current_amount_sql = _amount_chf_expression() original_join_sql = ( f"original.budget_transaction_id={reversal_target_sql} AND original.status='confirmed' " "AND t.transaction_type='refund' AND original.transaction_type IN ('expense','fee') " f"AND {current_amount_sql} IS NOT NULL AND {original_amount_sql} IS NOT NULL AND " f"(SELECT COALESCE(SUM(abs({refund_amount_sql})),0) FROM budget_transactions r " "WHERE r.status='confirmed' AND r.transaction_type='refund' " f"AND {refund_target_r_sql}=original.budget_transaction_id AND (" "r.transaction_date limit page_rows = rows[:limit] items = [] for row in page_rows: item = dict(row) item["transaction_token"] = "tx_" + _sha(str(item["budget_transaction_id"]))[:20] item["amount_chf"] = ( format(Decimal(str(item["amount_chf"])), ".2f") if item["amount_chf"] is not None else None ) item["amount_original"] = format(Decimal(str(item["amount_original"])), ".2f") item.pop("budget_transaction_id", None) item.pop("created_at", None) items.append(item) next_cursor = None if has_more and page_rows: last = page_rows[-1] next_cursor = _encode_page_cursor([ scope, str(last["transaction_date"]), str(last["created_at"]), str(last["budget_transaction_id"]), ]) if _household_transaction_data_version(conn) != data_version: raise HTTPException(status_code=409, detail="transaction dataset changed; restart pagination") return { "items": items, "total": filtered_total, "total_all": total_all, "filtered_total": filtered_total, "page_count": len(items), "page_size": limit, "has_more": has_more, "next_cursor": next_cursor, "as_of": now(), "data_version": data_version, } def _open_household_review_rows(conn: Connection, limit: int | None = None) -> list[Any]: sql = """SELECT * FROM budget_transaction_candidates WHERE household_batch_id IS NOT NULL AND status='needs_review' ORDER BY transaction_date DESC,transaction_candidate_id DESC""" if limit is None: return conn.execute(sql).fetchall() return conn.execute(sql + " LIMIT ?", (max(1, int(limit)),)).fetchall() def _review_item_token(candidate_id: str) -> str: return "item_" + _sha(candidate_id)[:16] def _duplicate_semantics(candidate: Any) -> str: signed = Decimal(str(candidate["signed_amount_original"] or candidate["amount_original"] or "0")) return "income" if signed > 0 else "expense" def duplicate_original_is_compatible(candidate: Any, original: Any) -> bool: """Return the exact, fail-closed duplicate relation used by reads and writes.""" candidate_signed = Decimal( str(candidate["signed_amount_original"] or candidate["amount_original"] or "0") ) original_signed = Decimal(str(original["amount_original"] or "0")) candidate_merchant = re.sub( r"[^a-z0-9]+", "", str(candidate["merchant"] or candidate["description"] or "").casefold(), ) original_merchant = re.sub( r"[^a-z0-9]+", "", str(original["payee"] or original["description"] or "").casefold(), ) try: within_window = abs( ( date.fromisoformat(str(candidate["transaction_date"])) - date.fromisoformat(str(original["transaction_date"])) ).days ) <= 14 except ValueError: return False return bool( candidate_signed != 0 and original_signed == candidate_signed and str(original["currency_original"]).upper() == str(candidate["currency_original"]).upper() and str(original["transaction_type"]) == _duplicate_semantics(candidate) and within_window and candidate_merchant and candidate_merchant == original_merchant ) def _has_eligible_duplicate_original(conn: Connection, candidate: Any) -> bool: rows = conn.execute( """SELECT * FROM budget_transactions WHERE status='confirmed' AND currency_original=? AND transaction_type=? AND transaction_date BETWEEN date(?,'-14 day') AND date(?,'+14 day') ORDER BY transaction_date DESC,budget_transaction_id DESC""", ( candidate["currency_original"], _duplicate_semantics(candidate), candidate["transaction_date"], candidate["transaction_date"], ), ).fetchall() return any(duplicate_original_is_compatible(candidate, row) for row in rows) def _review_account_token(account_id: str) -> str: return "account_" + _sha(account_id)[:20] def _review_source_account_is_eligible(conn: Connection, candidate: Any) -> bool: account_id = str(candidate["account_source"] or "") currency = str(candidate["currency_original"] or "") if not account_id or not currency: return False return bool( conn.execute( """SELECT 1 FROM budget_accounts WHERE budget_account_id=? AND is_active=1 AND upper(currency)=upper(?)""", (account_id, currency), ).fetchone() ) def _review_transfer_accounts(conn: Connection, candidate: Any) -> list[dict[str, str]]: if ( str(candidate["status"]) != "needs_review" or str(candidate["classification"] or "") not in {"expense_candidate", "income_candidate"} or str(candidate["source_type"]) not in {"akb_bank", "raiffeisen_bank"} or str(candidate["currency_original"]).upper() != "CHF" or not _review_source_account_is_eligible(conn, candidate) ): return [] rows = conn.execute( """SELECT budget_account_id,name,account_type,currency FROM budget_accounts WHERE is_active=1 AND budget_account_id<>? AND upper(currency)=upper(?) ORDER BY name,budget_account_id""", (candidate["account_source"], candidate["currency_original"]), ).fetchall() return [ { "account_token": _review_account_token(str(row["budget_account_id"])), "account_name": str(row["name"]), "account_hint": "•••• " + _sha(str(row["budget_account_id"]))[-4:], "account_type": str(row["account_type"]), "currency": str(row["currency"]), } for row in rows ] def _review_item_available_actions( conn: Connection, candidate: Any, transfer_accounts: list[dict[str, str]] | None = None, ) -> list[str]: classification = str(candidate["classification"] or "") if str(candidate["status"]) == "duplicate" and candidate["duplicate_of_transaction_id"]: return ["reopen_duplicate"] if str(candidate["status"]) != "needs_review": return [] actions: list[str] = [] if classification in {"possible_logical_duplicate", "possible_legacy_duplicate"}: if ( str(candidate["currency_original"]).upper() == "CHF" and candidate["account_source"] ): actions.append("keep_not_duplicate") if _has_eligible_duplicate_original(conn, candidate): actions.append("exclude_duplicate") if transfer_accounts is None: transfer_accounts = _review_transfer_accounts(conn, candidate) if transfer_accounts: actions.append("internal_transfer") if ( classification in {"credit_card_payment", "credit_card_payment_counterpost"} and str(candidate["source_type"]) in {"akb_bank", "raiffeisen_bank"} and str(candidate["currency_original"]).upper() == "CHF" and _review_source_account_is_eligible(conn, candidate) and conn.execute( """SELECT COUNT(DISTINCT m.budget_account_id) FROM household_account_source_mappings m JOIN budget_accounts ba ON ba.budget_account_id=m.budget_account_id AND ba.is_active=1 WHERE m.contract_version=? AND m.source_type='visa_credit_card' AND m.is_active=1 AND ba.account_type IN ('credit_card','credit_card_liability') AND upper(ba.currency)=upper(?)""", (CONTRACT_VERSION, candidate["currency_original"]), ).fetchone()[0] == 1 ): actions.append("credit_card_settlement") return actions def _human_review_items(conn: Connection, candidates: list[Any]) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] lookup_cache = ClassificationLookupCache(conn) mapping_compatibility: dict[tuple[str, str], bool] = {} for candidate in candidates: signed = str(candidate["signed_amount_original"] or candidate["amount_original"] or "0") classification = classify_household_row(conn, { "source_type": candidate["source_type"], "description": candidate["description"], "merchant": candidate["merchant"], "signed_amount": signed, "currency": candidate["currency_original"], "transaction_date": candidate["transaction_date"], "classification": candidate["classification"], "disposition": "review", "requires_review": True, "mapping": {"budget_account_id": candidate["account_source"]} if candidate["account_source"] else None, }, cache=lookup_cache) if candidate["proposed_category_id"]: category_type = "income" if Decimal(signed) > 0 else "expense" category_id, category_name = _category_for_review( conn, str(candidate["proposed_category_id"]), category_type ) if category_id: classification.update( category_id=category_id, category_name=category_name, user_state="proposal_ready", user_message="Kategorie aus der bestätigten Importvorschau.", ) special_hint = None raw_class = str(candidate["classification"] or "") if raw_class in { "credit_card_payment", "credit_card_payment_counterpost", "credit_card_pending", "migros_receipt_detail", "possible_logical_duplicate", "possible_legacy_duplicate", "user_confirmed_unmatched_transfer", }: classification["user_state"] = "special_case" elif candidate["proposed_category_id"] and classification.get("category_id"): classification["user_state"] = "proposal_ready" else: classification["user_state"] = "decision_needed" if raw_class in {"possible_logical_duplicate", "possible_legacy_duplicate"}: special_hint = "Mögliche Dublette" elif raw_class in {"credit_card_payment", "credit_card_payment_counterpost"}: special_hint = "Gegenbuchung fehlt" elif candidate["source_type"] == "migros_receipts": special_hint = "Zahlungsgegenposten fehlt" transaction_type = "income" if Decimal(signed) > 0 else "expense" mapping_key = (str(candidate["source_type"]), str(candidate["account_source"] or "")) if mapping_key not in mapping_compatibility: mapping_compatibility[mapping_key] = bool( candidate["account_source"] and _review_mapping_is_compatible(conn, mapping_key[0], mapping_key[1]) ) can_confirm = bool( classification["user_state"] != "special_case" and candidate["currency_original"] == "CHF" and mapping_compatibility[mapping_key] ) transfer_accounts = _review_transfer_accounts(conn, candidate) available_actions = _review_item_available_actions( conn, candidate, transfer_accounts ) items.append({ "item_token": _review_item_token(str(candidate["transaction_candidate_id"])), "candidate_version": int(candidate["review_version"] or 1), "candidate_baseline": _sha(_canonical({ key: candidate[key] for key in ( "transaction_candidate_id", "status", "review_version", "updated_at", "classification", "account_source", "transaction_date", "signed_amount_original", "amount_original", "currency_original", "duplicate_of_transaction_id", "confirmed_transaction_id", ) })), "date": candidate["transaction_date"], "merchant": candidate["merchant"] or candidate["description"], "amount": signed, "currency": candidate["currency_original"], "transaction_type": transaction_type, "category_id": classification.get("category_id"), "category_name": classification.get("category_name"), "user_state": classification["user_state"], "state_label": {"proposal_ready": "Vorschlag bereit", "decision_needed": "Entscheidung nötig", "special_case": "Sonderfall erkannt"}[classification["user_state"]], "explanation": classification["user_message"], "similar_confirmed_count": int(classification.get("evidence_count") or 0), "special_hint": special_hint, "source_type": str(candidate["source_type"]), "source_label": {"akb_bank": "AKB", "raiffeisen_bank": "Raiffeisen", "visa_credit_card": "VISA", "migros_receipts": "Migros-Beleg"}.get(str(candidate["source_type"]), "Import"), "account_hint": ("•••• " + _sha(str(candidate["account_source"]))[-4:]) if candidate["account_source"] else None, "selected": can_confirm and classification["user_state"] == "proposal_ready" and bool(classification.get("category_id")), "can_confirm": can_confirm, "available_actions": available_actions, "eligible_transfer_accounts": transfer_accounts, "capabilities": { action: action in available_actions for action in ( "keep_not_duplicate", "exclude_duplicate", "reopen_duplicate", "credit_card_settlement", "internal_transfer", ) }, "menu_actions": [ action for action, enabled in ( ("transfer", raw_class not in {"credit_card_pending", "migros_receipt_detail"}), ("split", raw_class not in {"credit_card_pending", "migros_receipt_detail"}), ("duplicate", raw_class in {"possible_logical_duplicate", "possible_legacy_duplicate"}), ("ignore", True), ("details", True), ) if enabled ], }) return items def _category_for_review( conn: Connection, category_id: str, category_type: str ) -> tuple[str | None, str | None]: row = conn.execute( """SELECT category_id,name FROM budget_categories WHERE category_id=? AND is_active=1 AND category_type=?""", (category_id, category_type), ).fetchone() return (str(row["category_id"]), str(row["name"])) if row else (None, None) def _get_household_review_legacy(conn: Connection, limit: int = 100) -> dict[str, Any]: limit = max(1, min(int(limit), 500)) rows = conn.execute( """SELECT classification,COUNT(*) AS item_count, COALESCE(SUM(abs(CAST(signed_amount_original AS REAL))),0) AS amount, MIN(transaction_date) AS first_date,MAX(transaction_date) AS last_date, MIN(currency_original) AS min_currency,MAX(currency_original) AS max_currency, SUM(CASE WHEN account_source IS NULL OR account_source='' THEN 1 ELSE 0 END) AS missing_accounts FROM budget_transaction_candidates WHERE household_batch_id IS NOT NULL AND status='needs_review' GROUP BY classification ORDER BY classification LIMIT ?""", (limit,), ).fetchall() groups = [] for row in rows: classification = str(row["classification"] or "unclear") supported: list[str] = ["ignore"] if classification in {"possible_logical_duplicate", "possible_legacy_duplicate"}: supported.insert(0, "duplicate") if ( row["min_currency"] == "CHF" and row["max_currency"] == "CHF" and int(row["missing_accounts"] or 0) == 0 ): if classification in {"income_candidate", "credit_card_refund"}: supported.insert(0, "income") elif classification not in {"credit_card_pending", "migros_receipt_detail"}: supported.insert(0, "expense") groups.append({ "group_key": "review_" + _sha(classification)[:12], "title": classification.replace("_", " ").title(), "subtitle": f"{row['first_date']} bis {row['last_date']}", "count": int(row["item_count"]), "amount_chf": format(Decimal(str(row["amount"])), ".2f"), "proposed_action": supported[0], "source_labels": ["Haushaltsimport"], "can_confirm": bool(supported), "supported_actions": supported, "warning": "Die gewählte Aktion wird vor dem Schreiben vollständig als Auswirkung angezeigt.", }) unlinked = int(conn.execute( "SELECT COUNT(*) FROM household_migros_links WHERE status<>'linked'" ).fetchone()[0]) if unlinked: groups.append({ "group_key": "review_migros_link", "title": "Migros-Zuordnung prüfen", "subtitle": "Bon-Details ohne eindeutige Gesamtbelastung", "count": unlinked, "amount_chf": "0.00", "proposed_action": "Migros-Bon zuordnen", "source_labels": ["Migros"], "can_confirm": False, "supported_actions": [], "warning": "Keine automatische Zuordnung bei Differenz, Mehrdeutigkeit oder fehlender Belastung.", }) items = _human_review_items(conn, _open_household_review_rows(conn, limit)) return { "review_count": len(items), "proposal_count": sum(item["user_state"] == "proposal_ready" for item in items), "decision_count": sum(item["user_state"] != "proposal_ready" for item in items), "items": items, "categories": [ { "category_id": str(item["category_id"]), "name": str(item["name"]), "category_type": str(item["category_type"]), } for item in conn.execute( """SELECT category_id,name,category_type FROM budget_categories WHERE is_active=1 ORDER BY sort_order,name,category_id""" ).fetchall() ], "groups": groups, "unsafe_groups_can_confirm": False, "classification_version": CLASSIFICATION_VERSION, } def get_household_review( conn: Connection, limit: int = 25, *, cursor: str | None = None, state: str | None = None, source: str | None = None, special_only: bool = False, ) -> dict[str, Any]: """Return one bounded page plus canonical global review counts. Rows already covered by another monetary source (for example receipt detail rows) and confirmed transfer legs are deliberately not open work. """ limit = max(1, min(int(limit), 100)) special_classes = ( "credit_card_payment", "credit_card_payment_counterpost", "credit_card_pending", "migros_receipt_detail", "possible_logical_duplicate", "possible_legacy_duplicate", "user_confirmed_unmatched_transfer", ) placeholders = ",".join("?" for _ in special_classes) special_sql = f"classification IN ({placeholders})" valid_proposal_sql = ( "proposed_category_id IS NOT NULL AND EXISTS (" "SELECT 1 FROM budget_categories proposed " "WHERE proposed.category_id=budget_transaction_candidates.proposed_category_id " "AND proposed.is_active=1 AND proposed.category_type=(CASE " "WHEN CAST(COALESCE(signed_amount_original,amount_original,'0') AS REAL)>0 " "THEN 'income' ELSE 'expense' END))" ) open_sql = HOUSEHOLD_OPEN_REVIEW_SQL data_version = _household_review_data_version(conn) summary = conn.execute( f"""SELECT COUNT(*) AS total_open, SUM(CASE WHEN {valid_proposal_sql} AND NOT ({special_sql}) THEN 1 ELSE 0 END) AS proposal_ready, SUM(CASE WHEN NOT ({valid_proposal_sql}) AND NOT ({special_sql}) THEN 1 ELSE 0 END) AS decision_needed, SUM(CASE WHEN {special_sql} THEN 1 ELSE 0 END) AS special_cases FROM budget_transaction_candidates WHERE {open_sql}""", (*special_classes, *special_classes, *special_classes), ).fetchone() total_open = int(summary["total_open"] or 0) proposal_ready = int(summary["proposal_ready"] or 0) decision_needed = int(summary["decision_needed"] or 0) special_cases = int(summary["special_cases"] or 0) source_labels = { "akb_bank": "AKB", "raiffeisen_bank": "Raiffeisen", "visa_credit_card": "VISA", "migros_receipts": "Migros-Beleg", } source_options = [ {"value": str(row["source_type"]), "label": source_labels.get(str(row["source_type"]), "Import")} for row in conn.execute( f"SELECT DISTINCT source_type FROM budget_transaction_candidates WHERE {open_sql} ORDER BY source_type" ).fetchall() ] where = [open_sql] params: list[Any] = [] if state: if state not in {"proposal_ready", "decision_needed", "special_case", "needs_decision"}: raise HTTPException(status_code=422, detail="invalid review state filter") if state == "special_case": where.append(special_sql) params.extend(special_classes) elif state == "proposal_ready": where.append(f"{valid_proposal_sql} AND NOT ({special_sql})") params.extend(special_classes) elif state == "needs_decision": where.append(f"(NOT ({valid_proposal_sql}) OR ({special_sql}))") params.extend(special_classes) else: where.append(f"NOT ({valid_proposal_sql}) AND NOT ({special_sql})") params.extend(special_classes) if special_only and state != "special_case": where.append(special_sql) params.extend(special_classes) if source: if source not in IMPORT_PROFILES: raise HTTPException(status_code=422, detail="invalid review source filter") where.append("source_type=?") params.append(source) scope = _sha(_canonical([state, source, bool(special_only), data_version]))[:20] cursor_parts = _decode_page_cursor(cursor, scope) filtered_total = int(conn.execute( f"SELECT COUNT(*) FROM budget_transaction_candidates WHERE {' AND '.join(where)}", params, ).fetchone()[0]) page_where = list(where) page_params = list(params) if cursor_parts: cursor_date, _unused, cursor_id = cursor_parts page_where.append("(transaction_date limit page_rows = list(candidates[:limit]) items = _human_review_items(conn, page_rows) next_cursor = None if has_more and page_rows: last = page_rows[-1] next_cursor = _encode_page_cursor([ scope, str(last["transaction_date"]), "", str(last["transaction_candidate_id"]), ]) group_rows = conn.execute( f"""SELECT classification,COUNT(*) AS item_count, COALESCE(SUM(abs(CAST(signed_amount_original AS REAL))),0) AS amount, MIN(transaction_date) AS first_date,MAX(transaction_date) AS last_date, MIN(currency_original) AS min_currency,MAX(currency_original) AS max_currency, SUM(CASE WHEN account_source IS NULL OR account_source='' THEN 1 ELSE 0 END) AS missing_accounts FROM budget_transaction_candidates WHERE {open_sql} GROUP BY classification ORDER BY classification""" ).fetchall() groups = [] for row in group_rows: classification = str(row["classification"] or "unclear") supported: list[str] = ["ignore"] if classification in {"possible_logical_duplicate", "possible_legacy_duplicate"}: supported.insert(0, "duplicate") if row["min_currency"] == "CHF" and row["max_currency"] == "CHF" and int(row["missing_accounts"] or 0) == 0: if classification in {"income_candidate", "credit_card_refund"}: supported.insert(0, "income") elif classification not in {"credit_card_pending", "migros_receipt_detail"}: supported.insert(0, "expense") groups.append({ "group_key": "review_" + _sha(classification)[:12], "title": classification.replace("_", " ").title(), "subtitle": f"{row['first_date']} bis {row['last_date']}", "count": int(row["item_count"]), "amount_chf": format(Decimal(str(row["amount"])), ".2f"), "proposed_action": supported[0], "source_labels": ["Haushaltsimport"], "can_confirm": bool(supported), "supported_actions": supported, "warning": "Die gewählte Aktion wird vor dem Schreiben vollständig als Auswirkung angezeigt.", }) category_options = [ { "category_id": str(item["category_id"]), "name": str(item["name"]), "category_type": str(item["category_type"]), } for item in conn.execute( """SELECT category_id,name,category_type FROM budget_categories WHERE is_active=1 ORDER BY sort_order,name,category_id""" ).fetchall() ] if _household_review_data_version(conn) != data_version: raise HTTPException(status_code=409, detail="review dataset changed; restart pagination") return { "total_open": total_open, "proposal_ready": proposal_ready, "decision_needed": decision_needed, "special_cases": special_cases, "filtered_total": filtered_total, "items": items, "has_more": has_more, "next_cursor": next_cursor, "as_of": now(), "data_version": data_version, "source_options": source_options, # Compatibility aliases are canonical global counts, never page length. "review_count": total_open, "proposal_count": proposal_ready, "decision_count": decision_needed + special_cases, "categories": category_options, "groups": groups, "unsafe_groups_can_confirm": False, "classification_version": CLASSIFICATION_VERSION, } def _review_mapping_is_compatible(conn: Connection, profile: str, budget_account_id: str) -> bool: if profile not in IMPORT_PROFILES or profile == "migros_receipts": return False row = conn.execute( """SELECT ba.account_type AS budget_account_type,ba.currency AS budget_currency, a.account_type AS canonical_account_type,a.currency AS canonical_currency, a.performance_included,a.portfolio_bucket FROM household_account_source_mappings m JOIN budget_accounts ba ON ba.budget_account_id=m.budget_account_id AND ba.is_active=1 AND ba.linked_account_id=m.canonical_account_id JOIN accounts a ON a.account_id=m.canonical_account_id AND a.is_active=1 WHERE m.contract_version=? AND m.source_type=? AND m.budget_account_id=? AND m.is_active=1 LIMIT 1""", (CONTRACT_VERSION, profile, budget_account_id), ).fetchone() return bool(row and _mapping_account_is_compatible(profile, row)) def _review_batch_selection(conn: Connection, payload: dict[str, Any]) -> list[dict[str, Any]]: raw_items = payload.get("items") if not isinstance(raw_items, list) or not raw_items: raise HTTPException(status_code=422, detail="at least one review item is required") open_rows = _open_household_review_rows(conn) by_token = {_review_item_token(str(row["transaction_candidate_id"])): row for row in open_rows} selected: list[dict[str, Any]] = [] seen: set[str] = set() for raw in raw_items: if not isinstance(raw, dict): raise HTTPException(status_code=422, detail="review items must be objects") token = str(raw.get("item_token") or "") if token in seen or token not in by_token: raise HTTPException(status_code=409, detail="review item is stale or duplicated") seen.add(token) candidate = by_token[token] if str(candidate["classification"] or "") in { "credit_card_payment", "credit_card_payment_counterpost", "credit_card_pending", "migros_receipt_detail", "possible_logical_duplicate", "possible_legacy_duplicate", "user_confirmed_unmatched_transfer", }: raise HTTPException(status_code=409, detail="review item requires its dedicated action") amount = Decimal(str(candidate["signed_amount_original"] or candidate["amount_original"])) transaction_type = "income" if amount > 0 else "expense" category_id = str(raw.get("category_id") or candidate["proposed_category_id"] or "") category_id, category_name = _category_for_review(conn, category_id, transaction_type) if not category_id: raise HTTPException(status_code=409, detail="a current category is required") if candidate["currency_original"] != "CHF" or not candidate["account_source"]: raise HTTPException(status_code=409, detail="review item cannot be safely confirmed") if not _review_mapping_is_compatible( conn, str(candidate["source_type"]), str(candidate["account_source"]) ): raise HTTPException(status_code=409, detail="review item account mapping changed; import again") selected.append({ "item_token": token, "candidate_id": str(candidate["transaction_candidate_id"]), "updated_at": str(candidate["updated_at"]), "category_id": category_id, "category_name": category_name, "source_type": str(candidate["source_type"]), "account_id": str(candidate["account_source"]), "transaction_date": str(candidate["transaction_date"]), "description": str(candidate["description"]), "merchant": str(candidate["merchant"] or candidate["description"]), "signed_amount": str(candidate["signed_amount_original"] or candidate["amount_original"]), "transaction_type": transaction_type, }) return selected def _review_batch_request_fingerprint(payload: dict[str, Any]) -> str: raw_items = payload.get("items") if not isinstance(raw_items, list) or not raw_items or not all(isinstance(item, dict) for item in raw_items): raise HTTPException(status_code=422, detail="at least one review item is required") identity = sorted( ( str(item.get("item_token") or ""), str(item.get("category_id") or ""), ) for item in raw_items ) return _sha(_canonical([ CLASSIFICATION_VERSION, "household_review_batch_request_v2", str(payload.get("baseline_fingerprint") or ""), identity, ])) def preview_household_review_batch(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: selected = _review_batch_selection(conn, payload) baseline = _baseline(conn) identity = [ {key: item[key] for key in ("candidate_id", "updated_at", "category_id", "signed_amount")} for item in selected ] preview_fingerprint = _sha(_canonical([ CLASSIFICATION_VERSION, "household_review_batch_v2", baseline, identity, ])) return { "classification_version": CLASSIFICATION_VERSION, "preview_fingerprint": preview_fingerprint, "baseline_fingerprint": baseline, "proposal_count": len(selected), "decision_count": 0, "expected_writes": {"transactions": len(selected), "candidate_updates": len(selected), "audit_events": 1}, "items": [ {"item_token": item["item_token"], "category_id": item["category_id"], "category_name": item["category_name"]} for item in selected ], "summary": f"{len(selected)} Vorschlag/Vorschläge werden exakt wie angezeigt bestätigt.", } def confirm_household_review_batch(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: expected_preview = str(payload.get("preview_fingerprint") or "") expected_baseline = str(payload.get("baseline_fingerprint") or "") if not expected_preview or not expected_baseline or payload.get("confirm") is not True: raise HTTPException(status_code=422, detail="confirm=true and both review fingerprints are required") audit_entity_id = "hhreviewv2_" + expected_preview[:24] request_fingerprint = _review_batch_request_fingerprint(payload) prior = conn.execute( """SELECT new_values_json FROM audit_log WHERE entity_type='household_review_batch' AND entity_id=? AND action='household_review_batch_confirmed'""", (audit_entity_id,), ).fetchone() if prior: try: prior_values = json.loads(str(prior["new_values_json"] or "{}")) except json.JSONDecodeError: prior_values = {} if prior_values.get("selection_fingerprint") != request_fingerprint: raise HTTPException(status_code=409, detail="review retry payload differs from the confirmed selection") return {"status": "confirmed", "confirmed_count": 0, "idempotent": True, "message": "Vorschläge waren bereits bestätigt."} preview = preview_household_review_batch(conn, payload) if preview["preview_fingerprint"] != expected_preview or preview["baseline_fingerprint"] != expected_baseline: raise HTTPException(status_code=409, detail="review preview changed; preview again") selected = _review_batch_selection(conn, payload) had_outer_transaction = conn.in_transaction if had_outer_transaction: conn.execute("SAVEPOINT household_review_batch_v2") else: conn.execute("BEGIN IMMEDIATE") try: if _baseline(conn) != expected_baseline: raise HTTPException(status_code=409, detail="review baseline changed; preview again") timestamp = now() for item in selected: amount = Decimal(item["signed_amount"]) transaction_type = item["transaction_type"] transaction_id = "btx_hhv2_" + _sha(item["candidate_id"] + "|" + item["category_id"])[:24] conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,booking_date, description,payee,amount_original,currency_original,fx_rate_to_chf,amount_chf, fx_status,category_id,status,source_type,source_candidate_id,notes,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,'1',?,'not_needed',?,'confirmed','import_candidate',?,?,?,?)""", (transaction_id, item["account_id"], transaction_type, item["transaction_date"], item["transaction_date"], item["description"], item["merchant"], format(amount, "f"), "CHF", format(amount, "f"), item["category_id"], item["candidate_id"], _canonical({"classification_version": CLASSIFICATION_VERSION, "review_contract": "household_review_batch_v2"}), timestamp, timestamp), ) conn.execute( """UPDATE budget_transaction_candidates SET status='confirmed',requires_review=0,proposed_category_id=?,proposed_category_name=?, confirmed_transaction_id=?,confirmed_at=?,confirmed_by='user',updated_at=? WHERE transaction_candidate_id=?""", (item["category_id"], item["category_name"], transaction_id, timestamp, timestamp, item["candidate_id"]), ) record_audit_event( conn, source="household_review", action="household_review_batch_confirmed", entity_type="household_review_batch", entity_id=audit_entity_id, new_values={ "classification_version": CLASSIFICATION_VERSION, "preview_fingerprint": expected_preview, "selection_fingerprint": request_fingerprint, "proposal_count": len(selected), "expected_writes": preview["expected_writes"], "actual_writes": {"transactions": len(selected), "candidate_updates": len(selected), "audit_events": 1}, "category_changes": [{"candidate_id": item["candidate_id"], "category_id": item["category_id"]} for item in selected], "status": "confirmed", }, created_by="user", ) if had_outer_transaction: conn.execute("RELEASE SAVEPOINT household_review_batch_v2") else: conn.commit() except Exception: if had_outer_transaction: conn.execute("ROLLBACK TO SAVEPOINT household_review_batch_v2") conn.execute("RELEASE SAVEPOINT household_review_batch_v2") else: conn.rollback() raise return {"status": "confirmed", "confirmed_count": len(selected), "idempotent": False, "message": f"{len(selected)} Vorschlag/Vorschläge bestätigt."} def preview_household_review_action(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: group_key = str(payload.get("group_key") or "") action = str(payload.get("action") or "") classification = _classification_for_group(conn, group_key) rows = _review_rows(conn, classification) supported = _supported_review_actions(rows, classification) if action not in supported: raise HTTPException(status_code=409, detail="review action is not safe for this group") baseline = _baseline(conn) identity = [{"candidate": str(row["transaction_candidate_id"]), "updated_at": row["updated_at"]} for row in rows] preview_fp = _sha(_canonical([CONTRACT_VERSION, group_key, action, baseline, identity])) total = sum((abs(Decimal(str(row["signed_amount_original"] or row["amount_original"] or "0"))) for row in rows), Decimal("0")) return { "preview_fingerprint": preview_fp, "baseline_fingerprint": baseline, "summary": f"{len(rows)} Buchung(en): {action}", "impact": { "confirmed_count": len(rows) if action in {"income", "expense"} else 0, "skipped_count": len(rows) if action in {"duplicate", "ignore"} else 0, "expense_chf": format(total if action == "expense" else Decimal("0"), ".2f"), "neutral_transfer_count": 0, }, "warnings": (["Die Buchungen bleiben unkategorisiert."] if action in {"income", "expense"} else []), } def confirm_household_review_action(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: expected_preview = str(payload.get("preview_fingerprint") or "") expected_baseline = str(payload.get("baseline_fingerprint") or "") if not expected_preview or not expected_baseline: raise HTTPException(status_code=422, detail="both review fingerprints are required") audit_entity_id = "hhreview_" + expected_preview[:24] if conn.execute( "SELECT 1 FROM audit_log WHERE entity_type='household_review_group' AND entity_id=? AND action='household_review_confirmed'", (audit_entity_id,), ).fetchone(): return {"status": "confirmed", "confirmed_count": 0, "idempotent": True, "message": "Prüfentscheidung war bereits bestätigt"} preview = preview_household_review_action(conn, payload) if preview["preview_fingerprint"] != expected_preview or preview["baseline_fingerprint"] != expected_baseline: raise HTTPException(status_code=409, detail="review preview changed; preview again") group_key = str(payload.get("group_key") or "") action = str(payload.get("action") or "") classification = _classification_for_group(conn, group_key) had_outer_transaction = conn.in_transaction if had_outer_transaction: conn.execute("SAVEPOINT household_review_confirm") else: conn.execute("BEGIN IMMEDIATE") try: if _baseline(conn) != expected_baseline: raise HTTPException(status_code=409, detail="database baseline changed; preview again") rows = _review_rows(conn, classification) if action not in _supported_review_actions(rows, classification): raise HTTPException(status_code=409, detail="review action is no longer safe") timestamp = now() confirmed_count = 0 for row in rows: candidate_id = str(row["transaction_candidate_id"]) if action in {"duplicate", "ignore"}: conn.execute( "UPDATE budget_transaction_candidates SET status=?,requires_review=0,updated_at=? WHERE transaction_candidate_id=?", ("duplicate" if action == "duplicate" else "ignored", timestamp, candidate_id), ) continue signed = abs(Decimal(str(row["signed_amount_original"] or row["amount_original"]))) amount = signed if action == "income" else -signed transaction_id = "btx_hhr_" + _sha(candidate_id + "|" + action)[:24] conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,transaction_type,transaction_date,booking_date, description,payee,amount_original,currency_original,fx_rate_to_chf,amount_chf, fx_status,category_id,status,source_type,source_candidate_id,notes,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,'1',?,'not_needed',NULL,'confirmed','import_candidate',?,?,?,?)""", ( transaction_id, row["account_source"], action, row["transaction_date"], row["transaction_date"], row["description"], row["merchant"], format(amount, "f"), "CHF", format(amount, "f"), candidate_id, _canonical({"contract_version": CONTRACT_VERSION, "review_action": action}), timestamp, timestamp, ), ) conn.execute( """UPDATE budget_transaction_candidates SET status='confirmed',requires_review=0, confirmed_transaction_id=?,updated_at=? WHERE transaction_candidate_id=?""", (transaction_id, timestamp, candidate_id), ) confirmed_count += 1 record_audit_event( conn, source="household_review", action="household_review_confirmed", entity_type="household_review_group", entity_id=audit_entity_id, new_values={"contract_version": CONTRACT_VERSION, "action": action, "count": len(rows)}, created_by="user", ) if had_outer_transaction: conn.execute("RELEASE SAVEPOINT household_review_confirm") else: conn.commit() except Exception: if had_outer_transaction: conn.execute("ROLLBACK TO SAVEPOINT household_review_confirm") conn.execute("RELEASE SAVEPOINT household_review_confirm") else: conn.rollback() raise return { "status": "confirmed", "confirmed_count": confirmed_count, "idempotent": False, "message": f"{len(rows)} Prüfentscheidung(en) gespeichert", } def _classification_for_group(conn: Connection, group_key: str) -> str: values = conn.execute( """SELECT DISTINCT classification FROM budget_transaction_candidates WHERE household_batch_id IS NOT NULL AND status='needs_review'""" ).fetchall() for row in values: classification = str(row[0] or "unclear") if "review_" + _sha(classification)[:12] == group_key: return classification raise HTTPException(status_code=404, detail="household review group not found") def _review_rows(conn: Connection, classification: str) -> list[Any]: rows = conn.execute( """SELECT transaction_candidate_id,transaction_date,description,merchant,amount_original, signed_amount_original,currency_original,account_source,status,updated_at FROM budget_transaction_candidates WHERE household_batch_id IS NOT NULL AND status='needs_review' AND COALESCE(classification,'unclear')=? ORDER BY transaction_candidate_id""", (classification,), ).fetchall() if not rows: raise HTTPException(status_code=404, detail="household review group not found") return list(rows) def _supported_review_actions(rows: list[Any], classification: str) -> list[str]: supported = ["ignore"] if classification in {"possible_logical_duplicate", "possible_legacy_duplicate"}: supported.insert(0, "duplicate") all_chf_mapped = all(row["currency_original"] == "CHF" and row["account_source"] for row in rows) if all_chf_mapped: if classification in {"income_candidate", "credit_card_refund"}: supported.insert(0, "income") elif classification not in {"credit_card_pending", "migros_receipt_detail"}: supported.insert(0, "expense") return supported