from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
from collections import defaultdict
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from sqlite3 import Connection, IntegrityError
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.household_classification import (
    CLASSIFICATION_VERSION,
    classify_household_row,
)
from jarvis_finance.services.transfer_pairing import confirm_transfer_pair

CONTRACT_VERSION = "household_import_v1"
PAIRING_VERSION = "transfer_pairing_v3"
PAIRING_CLASSES = ("safe", "review", "ambiguous", "unmatched")
MAX_IMPORT_BYTES = 5 * 1024 * 1024


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 _norm(value: Any) -> str:
    return " ".join(str(value or "").strip().casefold().split())


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]) -> 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")
        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
        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,
) -> 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())
        },
        "classification_version": CLASSIFICATION_VERSION,
    }))


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
    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": description[:120] if profile == "visa_credit_card" else None,
            "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 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]:
    validated_files = _validated_files(payload)
    existing_files, existing_source, existing_logical = _existing_sets(conn)
    legacy_business = _legacy_business_identities(conn)
    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()
    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
        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
        ]
        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":
                row["mapping"] = _mapping_for_file(conn, profile, row["source_reference"], item)
                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"
    _apply_reversal_matches(conn, normalized)
    pairs = _pair_rows(normalized)

    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"]): str(item["name"])
        for item in conn.execute(
            "SELECT category_id,name FROM budget_categories WHERE is_active=1 ORDER BY sort_order,name,category_id"
        ).fetchall()
    }
    for index, row in enumerate(normalized, 1):
        classification_v2 = classify_household_row(conn, row)
        override = category_overrides.get(f"row_{index}")
        if override is not None:
            category_id = str(override)
            if category_id not in valid_categories:
                raise HTTPException(status_code=422, detail="category override is not active")
            classification_v2.update(
                category_id=category_id,
                category_name=valid_categories[category_id],
                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"

    money_rows = [{"kind": "row", "id": r["source_row_fingerprint"],
                   "transaction_date": r["transaction_date"], "signed_amount": r["signed_amount"], "row": r}
                  for r in normalized if r["source_type"] != "migros_receipts"
                  and not r["disposition"].startswith("duplicate_")
                  and "migros" in _norm(r["description"])]
    existing_money = _existing_migros_money(conn)
    receipt_links = []
    for receipt in [r for r in normalized if r["source_type"] == "migros_receipts"]:
        receipt_total = Decimal(receipt["amount"])
        eligible = receipt_total >= Decimal("50.00")
        existing_same_day = [r for r in existing_money if r["transaction_date"] == receipt["transaction_date"]]
        current_same_day = [r for r in money_rows if r["transaction_date"] == receipt["transaction_date"]]
        existing_exact = [r for r in existing_same_day if abs(abs(Decimal(r["signed_amount"])) - receipt_total) <= Decimal("0.01")]
        existing_transactions = [r for r in existing_exact if r["kind"] == "transaction"]
        existing_candidates = [r for r in existing_exact if r["kind"] == "candidate"]
        current_exact = [r for r in current_same_day if abs(abs(Decimal(r["signed_amount"])) - receipt_total) <= Decimal("0.01")]
        chosen = None
        current_source_mode = bool(money_rows)
        if current_source_mode and eligible and len(current_exact) == 1:
            chosen = existing_transactions[0] if len(existing_transactions) == 1 else current_exact[0]
        elif not current_source_mode and len(existing_transactions) == 1:
            chosen = existing_transactions[0]
        elif not current_source_mode and not existing_transactions and len(existing_candidates) == 1:
            chosen = existing_candidates[0]
        linked_choice = chosen
        if linked_choice is not None:
            status = "linked"
        elif current_source_mode and not eligible:
            status = "unmatched"
        elif current_source_mode and (len(current_exact) > 1 or current_same_day):
            status = "review"
        elif not current_source_mode and (len(existing_transactions) > 1 or len(existing_candidates) > 1 or existing_same_day):
            status = "review"
        else:
            status = "unmatched"
        if chosen is None and status == "review":
            review_matches = current_same_day if current_source_mode else existing_same_day
            if len(review_matches) == 1:
                chosen = review_matches[0]
        difference = abs(abs(Decimal(chosen["signed_amount"])) - receipt_total) if chosen else None
        if eligible and status != "linked":
            receipt["requires_review"] = True
            receipt["disposition"] = "review"
        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})
    baseline = _baseline(conn)
    input_fp = _input_fingerprint(validated_files, payload.get("category_overrides"))
    safe_rows = [{k: v for k, v in row.items() if k not in {"source_reference", "mapping", "line_items", "provider_id"}}
                 | {"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)}
    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": len(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,
                    "counts": counts, "expected_budget_effect": expected_budget_effect,
                    "expected_writes": expected_writes}
    preview_fp = _sha(_canonical(preview_core))
    return preview_core | {"preview_fingerprint": preview_fp,
                           "confirmable": not errors, "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."""
    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"]: f"row_{index}"
        for index, item in enumerate(result["rows"], 1)
    }
    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",
                "account_id", "account_name",
            }
        }
        | {"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 item["description"],
            "description": item["description"],
            "amount": item["signed_amount"],
            "currency": item["currency"],
            "category_id": classification_v2.get("category_id"),
            "category_name": classification_v2.get("category_name"),
            "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",
        })
    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"]
    ]
    return {
        "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"])}
            for item in conn.execute(
                "SELECT category_id,name FROM budget_categories WHERE is_active=1 ORDER BY sort_order,name,category_id"
            ).fetchall()
        ],
        "transfer_pairs": pairs,
        "receipt_links": links,
        "counts": result["counts"],
        "expected_budget_effect": result["expected_budget_effect"],
        "expected_writes": result["expected_writes"],
        "confirmable": result["confirmable"],
        "errors": [
            {key: value for key, value in item.items() if key != "row_token"}
            for item in result["errors"]
        ],
    }


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"),
             "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")
        )
    )
    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,
            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,
            _canonical({
                "contract_version": CONTRACT_VERSION,
                "classification_version": CLASSIFICATION_VERSION,
                "classification_origin": row.get("classification_v2", {}).get("origin"),
                "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"))
    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")
        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 confirmable")
    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["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 {}),
                "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"]:
            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 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 {alias}.currency_original='CHF' THEN CAST({alias}.amount_original AS REAL) ELSE 0 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")
    amount = _amount_chf_expression()
    totals = conn.execute(
        f"""SELECT
              COALESCE(SUM(CASE WHEN transaction_type='income' THEN abs({amount}) ELSE 0 END),0) AS income,
              COALESCE(SUM(CASE WHEN transaction_type='expense' THEN abs({amount}) ELSE 0 END),0) AS expense,
              COALESCE(SUM(CASE WHEN transaction_type='refund' THEN abs({amount}) ELSE 0 END),0) AS refund,
              COALESCE(SUM(CASE WHEN transaction_type='transfer' THEN abs({amount}) ELSE 0 END),0) AS transfer_volume
            FROM budget_transactions t
            WHERE status='confirmed' AND substr(transaction_date,1,7)=?""",
        (selected_month,),
    ).fetchone()
    review_count = int(conn.execute(
        """SELECT COUNT(*) FROM budget_transaction_candidates
           WHERE household_batch_id IS NOT NULL AND requires_review=1
             AND status IN ('needs_review','pending','transfer_candidate','covered_by_source')"""
    ).fetchone()[0])
    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(str(totals["income"])) + Decimal(str(totals["refund"]))
    expense = Decimal(str(totals["expense"]))
    return {
        "month": selected_month,
        "income_chf": format(income, ".2f"),
        "expense_chf": format(expense, ".2f"),
        "balance_chf": format(income - expense, ".2f"),
        # Two balanced legs represent one budget-neutral transfer; expose its volume only.
        "neutral_transfers_chf": format(Decimal(str(totals["transfer_volume"])) / Decimal("2"), ".2f"),
        "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 review_count else "current",
            "label": "Prüfung offen" if review_count else "Aktuell",
            "message": (
                f"{review_count} unklare Buchung(en) sind nicht in Einnahmen oder Ausgaben enthalten."
                if review_count else "Nur bestätigte Haushaltsbuchungen sind enthalten."
            ),
        },
    }


def list_household_transactions(
    conn: Connection,
    *,
    period: 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 = 100,
) -> dict[str, Any]:
    limit = max(1, min(int(limit), 500))
    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)
    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)
    if category:
        where.append("lower(coalesce(bc.name,'')) LIKE ?"); params.append("%" + category.casefold() + "%")
    source_label_sql = "CASE t.source_type WHEN 'import_candidate' THEN 'Import' WHEN 'manual' THEN 'Manuell' ELSE t.source_type END"
    if source:
        where.append(f"lower({source_label_sql}) LIKE ?"); params.append("%" + source.casefold() + "%")
    if merchant:
        where.append("lower(coalesce(t.description,'')) LIKE ?"); params.append("%" + merchant.casefold() + "%")
    if review in {"open", "ignored"}:
        # Productive rows are never open review rows; keep this explicit instead of mixing staging candidates.
        where.append("1=0")
    amount = _amount_chf_expression()
    rows = conn.execute(
        f"""SELECT t.transaction_date,ba.name AS account_name,t.transaction_type,
                   bc.name AS category_name,t.description AS merchant_name,t.description,
                   {source_label_sql} AS source_label,'reviewed' AS review_status,
                   {amount} AS amount_chf,t.currency_original AS currency
            FROM budget_transactions t
            LEFT JOIN budget_accounts ba ON ba.budget_account_id=t.account_id
            LEFT JOIN budget_categories bc ON bc.category_id=t.category_id
            WHERE {' AND '.join(where)}
            ORDER BY t.transaction_date DESC,t.created_at DESC,t.budget_transaction_id
            LIMIT ?""",
        (*params, limit),
    ).fetchall()
    items = [dict(row) | {"amount_chf": format(Decimal(str(row["amount_chf"])), ".2f")} for row in rows]
    return {"items": items, "total": len(items)}


def _open_household_review_rows(conn: Connection, limit: int = 500) -> list[Any]:
    return conn.execute(
        """SELECT * FROM budget_transaction_candidates
           WHERE household_batch_id IS NOT NULL AND requires_review=1
             AND status IN ('needs_review','pending','transfer_candidate','covered_by_source')
           ORDER BY transaction_date DESC,transaction_candidate_id LIMIT ?""",
        (limit,),
    ).fetchall()


def _review_item_token(candidate_id: str) -> str:
    return "item_" + _sha(candidate_id)[:16]


def _human_review_items(conn: Connection, limit: int = 500) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    for candidate in _open_household_review_rows(conn, limit):
        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,
        })
        if candidate["proposed_category_id"]:
            category_id, category_name = _category_for_review(conn, str(candidate["proposed_category_id"]))
            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 {"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"
        items.append({
            "item_token": _review_item_token(str(candidate["transaction_candidate_id"])),
            "date": candidate["transaction_date"],
            "merchant": candidate["merchant"] or candidate["description"],
            "amount": signed,
            "currency": candidate["currency_original"],
            "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_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": classification["user_state"] == "proposal_ready",
            "can_confirm": classification["user_state"] == "proposal_ready" and bool(classification.get("category_id")),
            "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) -> tuple[str | None, str | None]:
    row = conn.execute(
        "SELECT category_id,name FROM budget_categories WHERE category_id=? AND is_active=1",
        (category_id,),
    ).fetchone()
    return (str(row["category_id"]), str(row["name"])) if row else (None, None)


def get_household_review(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 requires_review=1
             AND status IN ('needs_review','pending','transfer_candidate','covered_by_source')
           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, 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"])}
            for item in conn.execute(
                "SELECT category_id,name 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 _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")
        category_id = str(raw.get("category_id") or candidate["proposed_category_id"] or "")
        category_id, category_name = _category_for_review(conn, category_id)
        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"]),
        })
    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 = "income" if amount > 0 else "expense"
            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 requires_review=1
             AND status IN ('needs_review','pending','transfer_candidate','covered_by_source')"""
    ).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 requires_review=1
             AND status IN ('needs_review','pending','transfer_candidate','covered_by_source')
             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