from __future__ import annotations import hashlib import json from datetime import date 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, row_to_dict PAIR_STATUSES = {"proposed", "confirmed", "rejected", "superseded", "unmatched"} OPEN_CANDIDATE_STATUSES = {"pending", "auto_categorized", "needs_review", "transfer_candidate"} ACCOUNT_SOURCE_HINTS = { "raiffeisen_bank": ("raiffeisen",), "akb_bank": ("akb", "aargauische kantonalbank"), } TRANSFER_TEXT_HINTS = ( "transfer", "uebertrag", "übertrag", "eigenes konto", "kontoausgleich", "umbuchung", "twint", ) MATCHER_VERSION = "transfer_pairing_v2" CONFIRMATION_NOTICE = ( "Als interner Transfer erkannt. Die endgültige Verbuchung erfolgt erst nach Bestätigung." ) def _decimal(value: object) -> Decimal | None: if value in (None, ""): return None try: return Decimal(str(value)) except (InvalidOperation, ValueError): return None def _signed_amount(row: Any) -> Decimal | None: explicit = _decimal(row["signed_amount_original"]) if explicit is not None and explicit != 0: return explicit amount = _decimal(row["amount_original"]) if amount is None or amount == 0: return None try: meta = json.loads(str(row["notes"] or "{}")) except json.JSONDecodeError: meta = {} transaction_type = str(meta.get("transaction_type") or "").lower() if transaction_type == "income": return abs(amount) if transaction_type in {"expense", "fee"}: return -abs(amount) return None def _candidate_date(row: Any) -> date | None: text = str(row["value_date"] or row["transaction_date"] or "")[:10] try: return date.fromisoformat(text) except ValueError: return None def _own_account(conn: Connection, row: Any) -> Any | None: accounts = conn.execute( """ SELECT budget_account_id, name, linked_account_id FROM budget_accounts WHERE is_active=1 ORDER BY name, budget_account_id """ ).fetchall() account_source = str(row["account_source"] or "").strip().casefold() if account_source: exact = [ item for item in accounts if account_source in { str(item["budget_account_id"] or "").casefold(), str(item["linked_account_id"] or "").casefold(), str(item["name"] or "").casefold(), } ] if len(exact) == 1: return exact[0] hints = ACCOUNT_SOURCE_HINTS.get(str(row["source_type"] or ""), ()) hinted = [ item for item in accounts if any(hint in str(item["name"] or "").casefold() for hint in hints) ] return hinted[0] if len(hinted) == 1 else None def _pair_id(source_candidate_id: str, target_candidate_id: str | None) -> str: raw = f"{source_candidate_id}|{target_candidate_id or 'unmatched'}" return "btpair_" + hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] def _transfer_id(pair_id: str) -> str: return "btrf_" + hashlib.sha256(pair_id.encode("utf-8")).hexdigest()[:24] def _transaction_id(pair_id: str, side: str) -> str: raw = f"{pair_id}|{side}" return "btx_" + hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] def _text_support(row: Any) -> bool: text = f"{row['description'] or ''} {row['merchant'] or ''}".casefold() return any(hint in text for hint in TRANSFER_TEXT_HINTS) def _eligible_rows(conn: Connection) -> list[Any]: placeholders = ",".join("?" for _ in OPEN_CANDIDATE_STATUSES) return conn.execute( f""" SELECT * FROM budget_transaction_candidates WHERE status IN ({placeholders}) AND confirmed_transaction_id IS NULL AND amount_original IS NOT NULL AND currency_original IS NOT NULL ORDER BY transaction_candidate_id """, tuple(sorted(OPEN_CANDIDATE_STATUSES)), ).fetchall() def _is_rejected_relation(conn: Connection, source_id: str, target_id: str) -> bool: return bool( conn.execute( """ SELECT 1 FROM budget_transfer_pairs WHERE source_candidate_id=? AND target_candidate_id=? AND status='rejected' """, (source_id, target_id), ).fetchone() ) def _match_candidates( conn: Connection, outflow: dict[str, Any], inflows: list[dict[str, Any]], *, date_window_days: int, ) -> list[dict[str, Any]]: matches = [] for inflow in inflows: if outflow["account_id"] == inflow["account_id"]: continue if outflow["currency"] != inflow["currency"]: continue if abs(outflow["amount"]) != abs(inflow["amount"]): continue if abs((outflow["date"] - inflow["date"]).days) > date_window_days: continue if _is_rejected_relation(conn, outflow["candidate_id"], inflow["candidate_id"]): continue matches.append(inflow) return matches def _redacted_pair_audit(pair: dict[str, Any]) -> dict[str, Any]: return { "transfer_pair_id": pair["transfer_pair_id"], "status": pair["status"], "source_account_name": pair["source_account_name"], "target_account_name": pair.get("target_account_name"), "amount": pair["amount"], "currency": pair["currency"], "budget_effect_chf": "0", "matcher_version": MATCHER_VERSION, } def _upsert_pair( conn: Connection, *, source: dict[str, Any], target: dict[str, Any] | None, status: str, quality_status: str, reason_codes: list[str], alternative_count: int, date_window_days: int, ) -> tuple[str, bool]: pair_id = _pair_id(source["candidate_id"], target["candidate_id"] if target else None) existing = conn.execute( "SELECT status FROM budget_transfer_pairs WHERE transfer_pair_id=?", (pair_id,), ).fetchone() if existing and existing["status"] in {"confirmed", "rejected"}: return pair_id, False timestamp = now() evidence = { "matcher_version": MATCHER_VERSION, "date_window_days": date_window_days, "alternative_count": alternative_count, "different_sources": bool(target and source["source_type"] != target["source_type"]), "different_import_runs": bool( target and source["source_file_label"] != target["source_file_label"] ), "text_support": bool(target and (_text_support(source["row"]) or _text_support(target["row"]))), "merchant_text_decisive": False, } values = ( pair_id, source["candidate_id"], target["candidate_id"] if target else None, source["account_id"], target["account_id"] if target else None, format(source["amount"], "f"), format(target["amount"], "f") if target else None, source["currency"], source["row"]["transaction_date"], target["row"]["transaction_date"] if target else None, source["row"]["value_date"], target["row"]["value_date"] if target else None, status, quality_status, json.dumps(evidence, sort_keys=True), json.dumps(reason_codes, sort_keys=True), timestamp, timestamp, ) 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '0', ?, 'system', ?) ON CONFLICT(transfer_pair_id) DO UPDATE SET target_candidate_id=excluded.target_candidate_id, target_account_id=excluded.target_account_id, target_signed_amount=excluded.target_signed_amount, target_booking_date=excluded.target_booking_date, target_value_date=excluded.target_value_date, status=excluded.status, quality_status=excluded.quality_status, evidence_json=excluded.evidence_json, reason_codes_json=excluded.reason_codes_json, updated_at=excluded.updated_at """, values, ) return pair_id, existing is None def _set_candidate_pair_metadata( conn: Connection, source: dict[str, Any], target: dict[str, Any], pair_id: str, ) -> None: timestamp = now() for candidate, counterpart, direction in ( (source, target, "outflow"), (target, source, "inflow"), ): row = candidate["row"] try: metadata = json.loads(str(row["notes"] or "{}")) except json.JSONDecodeError: metadata = {} if not isinstance(metadata, dict): metadata = {} metadata["transaction_type"] = "transfer" metadata["transfer"] = { "pair_id": pair_id, "pair_status": "proposed", "direction": direction, "budget_impact": "neutral", } conn.execute( """ UPDATE budget_transaction_candidates SET status='transfer_candidate', classification='transfer_candidate', requires_review=1, review_reason='generic_internal_transfer_match', proposed_category_id=NULL, proposed_category_name=NULL, confidence='0.95', rule_id=?, rule_name='Generischer interner Transfer', linked_candidate_id=?, notes=?, updated_at=? WHERE transaction_candidate_id=? """, ( MATCHER_VERSION, counterpart["candidate_id"], json.dumps(metadata, ensure_ascii=False, sort_keys=True), timestamp, candidate["candidate_id"], ), ) def _supersede_other_open_pairs( conn: Connection, *, source_candidate_id: str, current_pair_id: str, ) -> int: rows = conn.execute( """ SELECT transfer_pair_id, status FROM budget_transfer_pairs WHERE source_candidate_id=? AND transfer_pair_id<>? AND status IN ('proposed','unmatched') """, (source_candidate_id, current_pair_id), ).fetchall() timestamp = now() for row in rows: conn.execute( """ UPDATE budget_transfer_pairs SET status='superseded', superseded_by_pair_id=?, updated_at=? WHERE transfer_pair_id=? """, (current_pair_id, timestamp, row["transfer_pair_id"]), ) record_audit_event( conn, source="transfer_matcher", action="transfer_pair_superseded", entity_type="budget_transfer_pair", entity_id=row["transfer_pair_id"], old_values={"status": row["status"]}, new_values={"status": "superseded", "superseded_by_pair_id": current_pair_id}, created_by="system", ) return len(rows) def propose_transfer_pairs(conn: Connection, *, date_window_days: int = 3) -> dict[str, int]: if not 0 <= date_window_days <= 7: raise HTTPException(status_code=422, detail="date_window_days must be between 0 and 7") normalized: list[dict[str, Any]] = [] for row in _eligible_rows(conn): amount = _signed_amount(row) booking_date = _candidate_date(row) account = _own_account(conn, row) if amount is None or booking_date is None or account is None: continue normalized.append( { "row": row, "candidate_id": str(row["transaction_candidate_id"]), "amount": amount, "currency": str(row["currency_original"]), "date": booking_date, "account_id": str(account["budget_account_id"]), "account_name": str(account["name"]), "source_type": str(row["source_type"]), "source_file_label": str(row["source_file_label"]), } ) outflows = [item for item in normalized if item["amount"] < 0] inflows = [item for item in normalized if item["amount"] > 0] forward_matches = { item["candidate_id"]: _match_candidates( conn, item, inflows, date_window_days=date_window_days, ) for item in outflows } reverse_counts: dict[str, int] = {} for matches in forward_matches.values(): for target in matches: reverse_counts[target["candidate_id"]] = reverse_counts.get(target["candidate_id"], 0) + 1 counts = { "scanned_count": len(normalized), "proposed_count": 0, "unmatched_count": 0, "ambiguous_count": 0, "superseded_count": 0, } for source in outflows: matches = forward_matches[source["candidate_id"]] unique_target = ( matches[0] if len(matches) == 1 and reverse_counts.get(matches[0]["candidate_id"]) == 1 else None ) if unique_target: reason_codes = [ "known_own_accounts", "different_accounts", "opposite_signs", "exact_amount", "same_currency", f"date_within_{date_window_days}_days", "merchant_text_not_decisive", "manual_confirm_required", ] pair_id, created = _upsert_pair( conn, source=source, target=unique_target, status="proposed", quality_status="strong", reason_codes=reason_codes, alternative_count=1, date_window_days=date_window_days, ) _set_candidate_pair_metadata(conn, source, unique_target, pair_id) counts["proposed_count"] += 1 counts["superseded_count"] += _supersede_other_open_pairs( conn, source_candidate_id=source["candidate_id"], current_pair_id=pair_id, ) if created: record_audit_event( conn, source="transfer_matcher", action="transfer_pair_proposed", entity_type="budget_transfer_pair", entity_id=pair_id, new_values={ "status": "proposed", "source_account_name": source["account_name"], "target_account_name": unique_target["account_name"], "amount": format(abs(source["amount"]), "f"), "currency": source["currency"], "budget_effect_chf": "0", "matcher_version": MATCHER_VERSION, }, confirmed=False, created_by="system", ) continue ambiguous = bool(matches) reason_codes = [ "known_own_source_account", "counterbooking_not_selected", "manual_confirm_required", ] if ambiguous: reason_codes.append("ambiguous_counterbookings") counts["ambiguous_count"] += 1 else: reason_codes.append("counterbooking_missing") counts["unmatched_count"] += 1 pair_id, _created = _upsert_pair( conn, source=source, target=None, status="unmatched", quality_status="ambiguous" if ambiguous else "unmatched", reason_codes=reason_codes, alternative_count=len(matches), date_window_days=date_window_days, ) counts["superseded_count"] += _supersede_other_open_pairs( conn, source_candidate_id=source["candidate_id"], current_pair_id=pair_id, ) conn.commit() return counts def list_transfer_pairs( conn: Connection, *, status: str | None = None, ) -> list[dict[str, Any]]: if status and status not in PAIR_STATUSES: raise HTTPException(status_code=422, detail="invalid transfer pair status") where = "WHERE pair.status=?" if status else "" params: tuple[str, ...] = (status,) if status else () rows = conn.execute( f""" SELECT pair.*, source_account.name AS source_account_name, target_account.name AS target_account_name FROM budget_transfer_pairs pair JOIN budget_accounts source_account ON source_account.budget_account_id=pair.source_account_id LEFT JOIN budget_accounts target_account ON target_account.budget_account_id=pair.target_account_id {where} ORDER BY CASE pair.status WHEN 'proposed' THEN 0 WHEN 'unmatched' THEN 1 ELSE 2 END, pair.created_at DESC, pair.transfer_pair_id """, params, ).fetchall() result = [] for row in rows: item = row_to_dict(row) try: evidence = json.loads(str(item.pop("evidence_json") or "{}")) except json.JSONDecodeError: evidence = {} try: reason_codes = json.loads(str(item.pop("reason_codes_json") or "[]")) except json.JSONDecodeError: reason_codes = [] amount = format(abs(Decimal(str(item["source_signed_amount"]))), "f") target_name = item.get("target_account_name") if item["quality_status"] == "ambiguous": explanation = f"{evidence.get('alternative_count', 0)} gleich plausible Gegenbuchungen; keine automatische Auswahl." elif item["status"] == "unmatched": explanation = "Gegenbuchung fehlt; der Vorgang bleibt ungeklärt und wird nicht gebucht." else: explanation = "Gegenbuchung mit gleichem Betrag, gleicher Währung und passendem Datum gefunden." item.update( { "source_signed_amount": str(item.pop("source_signed_amount")), "target_signed_amount": item.pop("target_signed_amount"), "amount": amount, "reason_codes": reason_codes, "alternative_count": int(evidence.get("alternative_count") or 0), "explanation": explanation, "confirmation_notice": CONFIRMATION_NOTICE, "summary": ( f"Interner Transfer {item['source_account_name']} → {target_name} · " f"{item['currency']} {amount} · Budgeteffekt CHF 0" if target_name else f"Möglicher interner Transfer {item['source_account_name']} · {item['currency']} {amount} · ungeklärt" ), } ) result.append(item) return result def _load_pair_and_candidates(conn: Connection, pair_id: str) -> tuple[Any, Any, Any]: pair = conn.execute( "SELECT * FROM budget_transfer_pairs WHERE transfer_pair_id=?", (pair_id,), ).fetchone() if not pair: raise HTTPException(status_code=404, detail="transfer pair not found") if not pair["target_candidate_id"]: raise HTTPException(status_code=409, detail="transfer pair has no selected counterbooking") source = conn.execute( "SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (pair["source_candidate_id"],), ).fetchone() target = conn.execute( "SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (pair["target_candidate_id"],), ).fetchone() if not source or not target: raise HTTPException(status_code=409, detail="transfer pair candidates missing") return pair, source, target def _validate_confirmable(conn: Connection, pair: Any, source: Any, target: Any) -> None: if source["status"] not in OPEN_CANDIDATE_STATUSES or target["status"] not in OPEN_CANDIDATE_STATUSES: raise HTTPException(status_code=409, detail="transfer pair candidates are no longer open") source_amount = _signed_amount(source) target_amount = _signed_amount(target) if source_amount is None or target_amount is None or source_amount >= 0 or target_amount <= 0: raise HTTPException(status_code=409, detail="transfer pair signs changed") if abs(source_amount) != abs(target_amount): raise HTTPException(status_code=409, detail="transfer pair amounts changed") if source["currency_original"] != target["currency_original"]: raise HTTPException(status_code=409, detail="transfer pair currency changed") source_account = _own_account(conn, source) target_account = _own_account(conn, target) if not source_account or not target_account or source_account["budget_account_id"] == target_account["budget_account_id"]: raise HTTPException(status_code=409, detail="transfer pair account relationship changed") if source_account["budget_account_id"] != pair["source_account_id"] or target_account["budget_account_id"] != pair["target_account_id"]: raise HTTPException(status_code=409, detail="transfer pair account mapping changed") competing = conn.execute( """ SELECT transfer_pair_id FROM budget_transfer_pairs WHERE status='confirmed' AND transfer_pair_id<>? AND (source_candidate_id IN (?, ?) OR target_candidate_id IN (?, ?)) LIMIT 1 """, ( pair["transfer_pair_id"], pair["source_candidate_id"], pair["target_candidate_id"], pair["source_candidate_id"], pair["target_candidate_id"], ), ).fetchone() if competing: raise HTTPException(status_code=409, detail="candidate already belongs to a confirmed transfer pair") def _start_atomic(conn: Connection, name: str) -> bool: nested = conn.in_transaction if nested: conn.execute(f"SAVEPOINT {name}") else: conn.execute("BEGIN IMMEDIATE") return nested def _finish_atomic(conn: Connection, name: str, nested: bool) -> None: if nested: conn.execute(f"RELEASE SAVEPOINT {name}") else: conn.commit() def _rollback_atomic(conn: Connection, name: str, nested: bool) -> None: if nested: conn.execute(f"ROLLBACK TO SAVEPOINT {name}") conn.execute(f"RELEASE SAVEPOINT {name}") else: conn.rollback() def confirm_transfer_pair( conn: Connection, pair_id: str, *, decision_by: str = "user", note: str | None = None, ) -> dict[str, Any]: existing = conn.execute( "SELECT status, confirmed_transfer_id, budget_effect_chf FROM budget_transfer_pairs WHERE transfer_pair_id=?", (pair_id,), ).fetchone() if not existing: raise HTTPException(status_code=404, detail="transfer pair not found") if existing["status"] == "confirmed": return { "status": "confirmed", "entity_id": existing["confirmed_transfer_id"], "audit_id": None, "budget_effect_chf": existing["budget_effect_chf"], "idempotent": True, "message": "Transferpaar war bereits bestätigt", } if existing["status"] != "proposed": raise HTTPException(status_code=409, detail="only proposed transfer pairs can be confirmed") savepoint = "transfer_pair_confirm" nested = _start_atomic(conn, savepoint) try: pair, source, target = _load_pair_and_candidates(conn, pair_id) if pair["status"] == "confirmed": _finish_atomic(conn, savepoint, nested) return { "status": "confirmed", "entity_id": pair["confirmed_transfer_id"], "audit_id": None, "budget_effect_chf": pair["budget_effect_chf"], "idempotent": True, "message": "Transferpaar war bereits bestätigt", } if pair["status"] != "proposed": raise HTTPException(status_code=409, detail="only proposed transfer pairs can be confirmed") _validate_confirmable(conn, pair, source, target) transfer_id = _transfer_id(pair_id) source_tx_id = _transaction_id(pair_id, "source") target_tx_id = _transaction_id(pair_id, "target") timestamp = now() description = "Interner Transfer" for tx_id, candidate, account_id, amount in ( (source_tx_id, source, pair["source_account_id"], pair["source_signed_amount"]), (target_tx_id, target, pair["target_account_id"], pair["target_signed_amount"]), ): conn.execute( """ INSERT INTO budget_transactions( budget_transaction_id, account_id, transaction_type, transaction_date, booking_date, description, 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 (?, ?, 'transfer', ?, ?, ?, ?, ?, ?, ?, ?, NULL, 'confirmed', 'import_candidate', ?, NULL, ?, ?) """, ( tx_id, account_id, candidate["transaction_date"], candidate["transaction_date"], description, amount, pair["currency"], "1" if pair["currency"] == "CHF" else None, amount if pair["currency"] == "CHF" else None, "not_needed" if pair["currency"] == "CHF" else "missing", candidate["transaction_candidate_id"], timestamp, timestamp, ), ) conn.execute( """ INSERT INTO budget_transfers( transfer_id, from_transaction_id, to_transaction_id, from_account_id, to_account_id, amount_original, currency_original, fx_rate_to_chf, notes, created_at, transfer_type ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, 'internal_transfer') """, ( transfer_id, source_tx_id, target_tx_id, pair["source_account_id"], pair["target_account_id"], format(abs(Decimal(str(pair["source_signed_amount"]))), "f"), pair["currency"], "1" if pair["currency"] == "CHF" else None, timestamp, ), ) for candidate_id, tx_id, counterpart_id in ( (pair["source_candidate_id"], source_tx_id, pair["target_candidate_id"]), (pair["target_candidate_id"], target_tx_id, pair["source_candidate_id"]), ): conn.execute( """ UPDATE budget_transaction_candidates SET status='confirmed', classification='transfer', linked_candidate_id=?, confirmed_transaction_id=?, confirmed_at=?, confirmed_by=?, updated_at=? WHERE transaction_candidate_id=? AND status IN ('pending','auto_categorized','needs_review','transfer_candidate') """, (counterpart_id, tx_id, timestamp, decision_by, timestamp, candidate_id), ) if conn.execute("SELECT changes()").fetchone()[0] != 1: raise HTTPException(status_code=409, detail="transfer pair candidate was consumed concurrently") conn.execute( """ UPDATE budget_transfer_pairs SET status='confirmed', confirmed_transfer_id=?, budget_effect_chf='0', decided_at=?, decided_by=?, decision_note=?, updated_at=? WHERE transfer_pair_id=? AND status='proposed' """, (transfer_id, timestamp, decision_by, note, timestamp, pair_id), ) if conn.execute("SELECT changes()").fetchone()[0] != 1: raise HTTPException(status_code=409, detail="transfer pair was decided concurrently") pair_view = next(item for item in list_transfer_pairs(conn) if item["transfer_pair_id"] == pair_id) audit_id = record_audit_event( conn, source="vue_dashboard", action="transfer_pair_confirmed", entity_type="budget_transfer_pair", entity_id=pair_id, old_values={"status": "proposed"}, new_values=_redacted_pair_audit(pair_view), user_text_note=note, created_by=decision_by, ) _finish_atomic(conn, savepoint, nested) return { "status": "confirmed", "entity_id": transfer_id, "audit_id": audit_id, "budget_effect_chf": "0", "idempotent": False, "message": "Transferpaar bestätigt", } except (HTTPException, IntegrityError): _rollback_atomic(conn, savepoint, nested) raise except Exception: _rollback_atomic(conn, savepoint, nested) raise def reject_transfer_pair( conn: Connection, pair_id: str, *, decision_by: str = "user", note: str | None = None, ) -> dict[str, Any]: savepoint = "transfer_pair_reject" nested = _start_atomic(conn, savepoint) try: pair = conn.execute( "SELECT * FROM budget_transfer_pairs WHERE transfer_pair_id=?", (pair_id,), ).fetchone() if not pair: raise HTTPException(status_code=404, detail="transfer pair not found") if pair["status"] == "rejected": _finish_atomic(conn, savepoint, nested) return { "status": "rejected", "entity_id": pair_id, "audit_id": None, "idempotent": True, "message": "Transferpaar war bereits abgelehnt", } if pair["status"] == "confirmed": raise HTTPException( status_code=409, detail="confirmed transfer pairs cannot be rejected", ) if pair["status"] not in {"proposed", "unmatched"}: raise HTTPException( status_code=409, detail="transfer pair cannot be rejected in current status", ) timestamp = now() conn.execute( """ UPDATE budget_transfer_pairs SET status='rejected', decided_at=?, decided_by=?, decision_note=?, updated_at=? WHERE transfer_pair_id=? AND status IN ('proposed','unmatched') """, (timestamp, decision_by, note, timestamp, pair_id), ) if conn.execute("SELECT changes()").fetchone()[0] != 1: raise HTTPException( status_code=409, detail="transfer pair was decided concurrently", ) for candidate_id in ( pair["source_candidate_id"], pair["target_candidate_id"], ): if not candidate_id: continue candidate = conn.execute( "SELECT notes, signed_amount_original FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,), ).fetchone() if not candidate: continue try: metadata = json.loads(str(candidate["notes"] or "{}")) except json.JSONDecodeError: metadata = {} if not isinstance(metadata, dict): metadata = {} signed = _decimal(candidate["signed_amount_original"]) metadata["transaction_type"] = "income" if signed is not None and signed > 0 else "expense" metadata["transfer"] = { "pair_id": pair_id, "pair_status": "rejected", "budget_impact": "neutral", } conn.execute( """ UPDATE budget_transaction_candidates SET status='needs_review', classification='bank_review', requires_review=1, review_reason='transfer_pair_rejected', linked_candidate_id=NULL, rule_id=NULL, rule_name=NULL, confidence='0.50', notes=?, updated_at=? WHERE transaction_candidate_id=? AND status='transfer_candidate' """, ( json.dumps(metadata, ensure_ascii=False, sort_keys=True), timestamp, candidate_id, ), ) audit_id = record_audit_event( conn, source="vue_dashboard", action="transfer_pair_rejected", entity_type="budget_transfer_pair", entity_id=pair_id, old_values={"status": pair["status"]}, new_values={"status": "rejected", "budget_effect_chf": "0"}, user_text_note=note, created_by=decision_by, ) _finish_atomic(conn, savepoint, nested) return { "status": "rejected", "entity_id": pair_id, "audit_id": audit_id, "idempotent": False, "message": "Transferpaar abgelehnt; Kandidaten bleiben offen", } except (HTTPException, IntegrityError): _rollback_atomic(conn, savepoint, nested) raise except Exception: _rollback_atomic(conn, savepoint, nested) raise