# ruff: noqa: E701, E702, E731 from __future__ import annotations import csv import hashlib import json import re from collections import defaultdict from dataclasses import dataclass from datetime import datetime from decimal import Decimal, InvalidOperation from sqlite3 import Connection from typing import Any from fastapi import HTTPException from jarvis_finance.audit.log import record_audit_event from jarvis_finance.services.budget_common import new_id, now, row_to_dict from jarvis_finance.services.budget_transactions import confirm_budget_transaction SUBSCRIPTION_PATTERNS = ["APPLE", "APPLE.COM", "GOOGLE", "AMAZON PRIME", "NETFLIX", "DISNEY", "DISNEYPLUS"] MIGROS_PATTERNS = ["MIGROS", "MIGROL", "CUMULUS", "MMM "] GALAXUS_PATTERNS = ["GALAXUS"] INCOME_PATTERNS = [ "LOHN", "SALARY", "ARBEITGEBER", "PAYROLL", "GEHALT", "ERNE", "MUSIKSCHULE", "GASSER", "GASSER BAUUNTERNEHMEN", "BONUS", "GRATIFIKATION", "SALAER", "RUECKVERGUETUNG", "RÜCKVERGÜTUNG", "RUECKERSTATTUNG", "RÜCKERSTATTUNG", "ERSTATTUNG", "SOLARSTROM", "VERGUETUNG", "VERGÜTUNG", ] STRONG_INCOME_PATTERNS = [ "LOHN", "SALARY", "GEHALT", "PAYROLL", "ERNE", "MUSIKSCHULE", "GASSER", "GASSER BAUUNTERNEHMEN", "BONUS", "GRATIFIKATION", "SOLARSTROM", "RUECKVERGUETUNG", "RÜCKVERGÜTUNG", ] POSSIBLE_INCOME_PATTERNS = ["RUECKERSTATTUNG", "RÜCKERSTATTUNG", "ERSTATTUNG", "VERGUETUNG", "VERGÜTUNG", "GUTSCHRIFT"] TRANSFER_PATTERNS = ["RAIFFEISEN", "AKB", "AARGAUISCHE KANTONALBANK", "ÜBERTRAG", "UEBERTRAG", "TRANSFER", "EIGENES KONTO", "KONTOAUSGLEICH", "UMBUCHUNG"] SWISSLOS_TRANSFER_PATTERNS = ["TWINT SWISSLOS", "SWISSLOS E-COMMERCE"] NEGATIVE_INCOME_PATTERNS = [ "KREDITKARTE", "VISA", "CREDIT CARD", "KARTENABRECHNUNG", "TRUE WEALTH", "TRUEWEALTH", "DEPOT", "INVESTMENT", "WERTSCHRIFT", "SPAREN", "KONTOAUSGLEICH", "UMBUCHUNG", ] def _decimal_text(value: object) -> str | None: if value in (None, ""): return None text = str(value).strip().replace("'", "").replace("CHF", "").replace("−", "-") text = text.replace("+", "") if text.count(",") == 1 and text.count(".") == 0: text = text.replace(",", ".") text = text.replace(" ", "") try: return format(abs(Decimal(text)), "f") except (InvalidOperation, ValueError): return None def _signed_decimal(value: object) -> Decimal | None: if value in (None, ""): return None text = str(value).strip().replace("'", "").replace("CHF", "").replace("−", "-").replace("+", "") if text.count(",") == 1 and text.count(".") == 0: text = text.replace(",", ".") text = text.replace(" ", "") try: return Decimal(text) except (InvalidOperation, ValueError): return None def _normalise(value: object) -> str: return str(value or "").strip().lower().replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("\ufeff", "") def _upper_ascii(value: object) -> str: return _normalise(value).upper() def _date_text(value: object) -> str | None: text = str(value or "").strip() if not text: return None if re.match(r"^\d{4}-\d{2}-\d{2}", text): return text[:10] m = re.match(r"^(\d{1,2})\.(\d{1,2})\.(\d{4})", text) if m: d, mo, y = m.groups() return f"{y}-{int(mo):02d}-{int(d):02d}" return text[:10] if len(text) >= 10 else None def _category_name(conn: Connection, category_id: str | None) -> str | None: if not category_id: return None row = conn.execute("SELECT name FROM budget_categories WHERE category_id=?", (category_id,)).fetchone() return str(row["name"]) if row else None def _category_by_name(conn: Connection, needles: list[str]) -> tuple[str | None, str | None]: cats = conn.execute("SELECT category_id, name FROM budget_categories WHERE is_active=1").fetchall() for needle in needles: n = _normalise(needle) for cat in cats: if n and (n in _normalise(cat["name"]) or _normalise(cat["name"]) in n): return str(cat["category_id"]), str(cat["name"]) return None, None def suggest_category(conn: Connection, description: str, source_label: str = "") -> tuple[str | None, str | None, str, bool]: text = _normalise(f"{source_label} {description}") if "migros" in text or "cumulus" in text: cid, cname = _category_by_name(conn, ["Essen & Haushalt", "Grosseinkauf", "Essen", "Haushalt", "Leben"]) return cid, cname or "Essen & Haushalt / Grosseinkauf", "0.80", cid is None keyword_map = [ (["krankenkasse", "versicherung", "axa", "zurich", "css", "helsana"], ["Versicherungen", "Krankenkasse"]), (["coop", "denner", "aldi", "lidl"], ["Essen", "Haushalt", "Leben"]), (["sbb", "zug", "bahn", "parking", "tankstelle", "mazda"], ["Mobilität", "Auto"]), (["lohn", "salary", "einkommen", "gutschrift"], ["Einnahmen", "Lohn"]), ] for words, cats in keyword_map: if any(w in text for w in words): cid, cname = _category_by_name(conn, cats) return cid, cname or cats[0], "0.65", cid is None return None, None, "0.20", True def _duplicate_of(conn: Connection, tx_date: str | None, amount: str | None, description: str) -> str | None: if not tx_date or not amount: return None row = conn.execute( """ SELECT budget_transaction_id FROM budget_transactions WHERE transaction_date=? AND status IN ('confirmed','reversed') AND COALESCE(amount_chf, amount_original)=? AND lower(description)=lower(?) ORDER BY created_at LIMIT 1 """, (tx_date, amount, description), ).fetchone() return str(row["budget_transaction_id"]) if row else None def _candidate_key(source_label: str, row_ref: str, tx_date: str | None, amount: str | None, description: str) -> str: raw = "|".join([source_label, row_ref, tx_date or "", amount or "", description]) return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] USER_OPEN_STATUSES = {"pending", "auto_categorized", "needs_review", "transfer_candidate"} USER_DUPLICATE_STATUSES = {"duplicate", "duplicate_candidate", "possible_duplicate", "duplicate_blocked"} USER_EXCLUDED_STATUSES = {"confirmed", "ignored", "superseded", "auto_ignored_duplicate", "covered_by_source", "covered_by_migros", "reference_2025", "archived_reference", "already_processed"} | USER_DUPLICATE_STATUSES USER_VISIBLE_SPECIAL_STATUSES = USER_DUPLICATE_STATUSES | {"covered_by_source", "covered_by_migros", "already_processed"} NORMAL_REVIEW_EXCLUDED_STATUSES = USER_EXCLUDED_STATUSES | {"covered_by_migros", "covered_by_source"} DUPLICATE_CONFIRM_BLOCKED_STATUSES = USER_DUPLICATE_STATUSES | {"covered_by_migros", "covered_by_source", "ignored", "superseded", "auto_ignored_duplicate", "reference_2025", "archived_reference", "already_processed"} DUPLICATE_CONFIRM_BLOCK_MESSAGE = "Kandidat ist als Duplikat/covered/ignored markiert und kann nicht normal bestätigt werden." STATUS_LABELS = { "pending": "Offen", "auto_categorized": "Sicher vorgeschlagen", "needs_review": "Review nötig", "confirmed": "Bestätigt", "ignored": "Ignoriert", "covered_by_source": "Durch Quelle abgedeckt", "covered_by_migros": "Durch Migros abgedeckt", "duplicate_candidate": "Mögliches Duplikat", "possible_duplicate": "Mögliches Duplikat", "duplicate_blocked": "Blockiertes Duplikat", "duplicate": "Mögliches Duplikat", "transfer_candidate": "Möglicher Transfer", "already_processed": "Bereits verarbeitet", "superseded": "Ersetzt", } def _candidate_view(row) -> dict[str, Any]: data = row_to_dict(row) | {"requires_review": bool(row["requires_review"])} data["status_label"] = data.get("status_label") or STATUS_LABELS.get(str(data.get("status") or ""), str(data.get("status") or "")) try: meta = json.loads(str(data.get("notes") or "{}")) if str(data.get("notes") or "").strip().startswith("{") else {} except json.JSONDecodeError: meta = {} transfer_meta = meta.get("transfer") if isinstance(meta.get("transfer"), dict) else {} if transfer_meta: data["transfer_pair_status"] = transfer_meta.get("pair_status") or "unmatched" data["transfer_target_account_name"] = transfer_meta.get("to_account_name") data["transfer_from_account_name"] = transfer_meta.get("from_account_name") data["counterparty_candidate_id"] = transfer_meta.get("counterparty_candidate_id") data["budget_impact"] = transfer_meta.get("budget_impact") or "neutral" return data def _normalized_merchant_name(value: object) -> str: return re.sub(r"[^a-z0-9]+", " ", _normalise(value)).strip() def _merchant_text(row_or_description: Any, source_type: str | None = None) -> str: if isinstance(row_or_description, dict): return _normalise(" ".join(str(row_or_description.get(k) or "") for k in ("merchant", "description", "source_file_label"))) return _normalise(row_or_description) def _merchant_alias_matches(alias, candidate) -> bool: if alias["source_type"] and alias["source_type"] != candidate["source_type"]: return False text = _merchant_text(row_to_dict(candidate)) pattern = str(alias["pattern"] or "") norm_pattern = _normalise(pattern) if alias["match_type"] == "exact": return text == norm_pattern or _normalise(candidate["merchant"] or "") == norm_pattern if alias["match_type"] == "regex": try: return re.search(pattern, text, re.IGNORECASE) is not None except re.error: return False return norm_pattern in text def _fingerprint(parts: list[object]) -> str: raw = "|".join(str(p or "").strip() for p in parts) return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:32] def _insert_candidate(conn: Connection, row: dict[str, Any]) -> bool: if row.get("receipt_key") and row.get("source_type") in {"migros_receipt", "migros_receipts"}: existing_receipt = conn.execute( """ SELECT 1 FROM budget_transaction_candidates WHERE source_type IN ('migros_receipt','migros_receipts') AND receipt_key=? AND status NOT IN ('superseded','ignored','reference_2025','archived_reference') """, (row.get("receipt_key"),), ).fetchone() if existing_receipt: return False existing = conn.execute( """ SELECT 1 FROM budget_transaction_candidates WHERE source_file_label=? AND source_row_or_range=? AND transaction_date IS ? AND description=? AND COALESCE(amount_original,'')=COALESCE(?, '') """, (row["source_file_label"], row.get("source_row_or_range"), row.get("transaction_date"), row["description"], row.get("amount_original")), ).fetchone() if existing: return False 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, proposed_category_id, proposed_category_name, duplicate_of_transaction_id, confidence, requires_review, status, notes, created_at, updated_at, classification, review_reason, rule_id, rule_name, source_priority, covered_by_source, receipt_key, linked_candidate_id, account_source, raw_fingerprint ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( row["transaction_candidate_id"], row["source_file_label"], row.get("source_row_or_range"), row.get("source_type") or "csv_seed", row.get("transaction_date"), row.get("value_date"), row["description"], row.get("merchant"), row.get("amount_original"), row.get("signed_amount_original"), row.get("currency_original") or "CHF", row.get("proposed_category_id"), row.get("proposed_category_name"), row.get("duplicate_of_transaction_id"), row.get("confidence") or "0", 1 if row.get("requires_review", True) else 0, row.get("status") or "pending", row.get("notes"), row["created_at"], row["updated_at"], row.get("classification"), row.get("review_reason"), row.get("rule_id"), row.get("rule_name"), row.get("source_priority") or 50, row.get("covered_by_source"), row.get("receipt_key"), row.get("linked_candidate_id"), row.get("account_source"), row.get("raw_fingerprint"), ), ) return True def seed_transaction_candidates_from_rows(conn: Connection, rows: list[dict[str, Any]] | list[list[Any]], *, source_file_label: str, source_type: str = "csv_seed") -> dict[str, int]: created = duplicates = review = suggested = ignored = confirmed = 0 for idx, raw in enumerate(rows, start=1): if isinstance(raw, dict): low = {_normalise(k): v for k, v in raw.items()} description = str(low.get("beschreibung") or low.get("description") or low.get("text") or low.get("artikel") or low.get("merchant") or low.get("merchantname") or low.get("haendler") or "").strip() merchant = str(low.get("merchant") or low.get("merchantname") or low.get("filiale") or low.get("haendler") or "").strip() or None tx_date = _date_text(low.get("datum") or low.get("date") or low.get("buchungsdatum") or low.get("valutadate")) amount = _decimal_text(low.get("betrag") or low.get("amount") or low.get("umsatz") or low.get("wert") or low.get("originalamount")) else: cells = [str(c).strip() for c in raw] description = next((c for c in cells if c and _decimal_text(c) is None and not c[:4].isdigit()), "") merchant = None tx_date = next((_date_text(c) for c in cells if _date_text(c)), None) amount = next((_decimal_text(c) for c in cells if _decimal_text(c) is not None), None) if not description or not amount: ignored += 1 continue cid, cname, conf, needs_review = suggest_category(conn, description, source_file_label) dup = _duplicate_of(conn, tx_date, amount, description) status = "duplicate" if dup else ("needs_review" if needs_review else "pending") if dup: duplicates += 1 if cid: suggested += 1 if needs_review: review += 1 ts = now() row_ref = f"R{idx}" inserted = _insert_candidate(conn, { "transaction_candidate_id": "btxcand_" + _candidate_key(source_file_label, row_ref, tx_date, amount, description), "source_file_label": source_file_label, "source_row_or_range": row_ref, "source_type": source_type, "transaction_date": tx_date, "description": description[:300], "merchant": merchant, "amount_original": amount, "currency_original": "CHF", "proposed_category_id": cid, "proposed_category_name": cname, "duplicate_of_transaction_id": dup, "confidence": conf, "requires_review": needs_review or bool(dup), "status": status, "notes": "mvp_light_category_suggestion", "classification": "legacy_csv_seed", "review_reason": "legacy_seed_review" if needs_review else "legacy_seed_suggested", "raw_fingerprint": _fingerprint([source_file_label, row_ref, tx_date, amount, description]), "created_at": ts, "updated_at": ts, }) if inserted: created += 1 conn.commit() return {"candidate_count": created, "suggested_count": suggested, "review_required_count": review, "possible_duplicate_count": duplicates, "ignored_count": ignored, "confirmed_count": confirmed} def seed_transaction_candidates_from_csv_text(conn: Connection, text: str, *, source_file_label: str, source_type: str = "csv_seed") -> dict[str, int]: sample = text[:4096] dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") if sample.strip() else csv.excel reader = csv.DictReader(text.splitlines(), dialect=dialect) if reader.fieldnames: return seed_transaction_candidates_from_rows(conn, list(reader), source_file_label=source_file_label, source_type=source_type) return seed_transaction_candidates_from_rows(conn, list(csv.reader(text.splitlines(), dialect=dialect)), source_file_label=source_file_label, source_type=source_type) def seed_migros_candidates_from_rows(conn: Connection, rows: list[dict[str, Any]], *, source_file_label: str, auto_category_id: str | None = None, threshold_chf: str = "50.00", candidate_source_type: str = "migros_receipt") -> dict[str, int]: _threshold = Decimal(threshold_chf) if not auto_category_id: auto_category_id, auto_category_name = _category_id_name(conn, "Essen & Haushalt") else: auto_category_name = _category_name(conn, auto_category_id) grouped: dict[str, list[tuple[int, dict[str, Any]]]] = defaultdict(list) ignored = 0 for idx, raw in enumerate(rows, start=1): low = {_normalise(k): v for k, v in raw.items()} tx_date = _date_text(low.get("datum") or low.get("date")) time = str(low.get("zeit") or low.get("time") or "").strip() store = str(low.get("filiale") or low.get("store") or "Migros").strip() register = str(low.get("kasse") or low.get("kassennummer") or low.get("register") or "").strip() txn = str(low.get("transaktion") or low.get("transaktionsnummer") or low.get("bon") or low.get("bon-id") or "").strip() amount = _decimal_text(low.get("umsatz") or low.get("betrag") or low.get("amount")) if not tx_date or not amount: ignored += 1 continue key = _fingerprint([tx_date, time, store, register, txn]) grouped[key].append((idx, raw)) created = auto = review = line_count = 0 for key, items in grouped.items(): first_low = {_normalise(k): v for k, v in items[0][1].items()} tx_date = _date_text(first_low.get("datum") or first_low.get("date")) time = str(first_low.get("zeit") or first_low.get("time") or "").strip() store = str(first_low.get("filiale") or first_low.get("store") or "Migros").strip() or "Migros" register = str(first_low.get("kasse") or first_low.get("kassennummer") or first_low.get("register") or "").strip() txn = str(first_low.get("transaktion") or first_low.get("transaktionsnummer") or first_low.get("bon") or first_low.get("bon-id") or "").strip() total = sum(Decimal(_decimal_text(({_normalise(k): v for k, v in raw.items()}).get("umsatz") or ({_normalise(k): v for k, v in raw.items()}).get("betrag") or ({_normalise(k): v for k, v in raw.items()}).get("amount")) or "0") for _, raw in items) amount = format(total, "f") needs_review = auto_category_id is None status = "needs_review" if needs_review else "auto_categorized" classification = "migros_review_missing_category" if needs_review else "migros_receipt_food_household" review_reason = "migros_receipt_missing_food_household_category" if needs_review else "migros_receipt_auto_food_household" row_ref = f"BON:{tx_date}:{time}:{store}:{register}:{txn}"[:180] ts = now() candidate_id = "btxcand_" + _candidate_key(source_file_label, row_ref, tx_date, amount, "Migros Bon") inserted = _insert_candidate(conn, { "transaction_candidate_id": candidate_id, "source_file_label": source_file_label, "source_row_or_range": row_ref, "source_type": candidate_source_type, "transaction_date": tx_date, "description": f"Migros Bon {store}"[:300], "merchant": store, "amount_original": amount, "currency_original": "CHF", "proposed_category_id": None if needs_review else auto_category_id, "proposed_category_name": None if needs_review else auto_category_name, "confidence": "0.90" if not needs_review else "0.60", "requires_review": needs_review, "status": status, "notes": json.dumps({"rule": "migros_receipt_food_household", "legacy_threshold_chf": threshold_chf, "line_items": len(items), "article_rows_are_detail_only": True}, ensure_ascii=False), "classification": classification, "review_reason": review_reason, "rule_id": "migros_receipt_food_household" if not needs_review else "migros_receipt_missing_category", "rule_name": "Migros Bon = Essen & Haushalt" if not needs_review else "Migros Bon Kategorie fehlt", "source_priority": 90, "receipt_key": key, "raw_fingerprint": key, "created_at": ts, "updated_at": ts, }) if inserted: created += 1 auto += 0 if needs_review else 1 review += 1 if needs_review else 0 for idx, raw in items: low = {_normalise(k): v for k, v in raw.items()} item_amount = _decimal_text(low.get("umsatz") or low.get("betrag") or low.get("amount")) conn.execute( """ INSERT OR IGNORE 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, 'CHF', ?, ?) """, ("bline_" + _fingerprint([source_file_label, key, idx]), candidate_id, source_file_label, key, f"R{idx}", str(low.get("artikel") or low.get("article") or "").strip(), str(low.get("menge") or low.get("quantity") or "").strip() or None, 1 if _normalise(low.get("aktion") or "") in {"1", "true", "ja", "yes"} else 0, item_amount, _fingerprint([source_file_label, idx, raw]), ts), ) line_count += 1 conn.commit() return {"receipt_candidate_count": created, "candidate_count": created, "line_item_count": line_count, "auto_assigned_count": auto, "review_required_count": review, "ignored_count": ignored} def preview_migros_receipts_2026(conn: Connection, *, year: str = "2026") -> dict[str, Any]: rows = conn.execute( """ SELECT status, COUNT(*) AS count FROM budget_transaction_candidates WHERE source_type='migros_receipt' AND substr(COALESCE(transaction_date,''),1,4)=? GROUP BY status """, (year,), ).fetchall() by_status = {str(r["status"]): int(r["count"] or 0) for r in rows} receipt_total = sum(by_status.values()) confirmed = by_status.get("confirmed", 0) open_count = sum(count for status, count in by_status.items() if status not in {"confirmed", "ignored", "superseded", "reference_2025", "archived_reference"}) covered = conn.execute( """ SELECT COUNT(*) FROM budget_transaction_candidates WHERE status='covered_by_migros' AND substr(COALESCE(transaction_date,''),1,4)=? """, (year,), ).fetchone()[0] line_items = conn.execute( """ SELECT COUNT(*) FROM budget_import_line_items li JOIN budget_transaction_candidates c ON c.transaction_candidate_id=li.transaction_candidate_id WHERE c.source_type='migros_receipt' AND substr(COALESCE(c.transaction_date,''),1,4)=? """, (year,), ).fetchone()[0] article_candidates = conn.execute( """ SELECT COUNT(*) FROM budget_transaction_candidates WHERE substr(COALESCE(transaction_date,''),1,4)=? AND (lower(COALESCE(source_file_label,'')) LIKE '%migros%' OR lower(COALESCE(source_file_label,'')) LIKE '%cumulus%') AND COALESCE(source_type,'') NOT IN ('migros_receipt') AND status NOT IN ('superseded','ignored','reference_2025','archived_reference') """, (year,), ).fetchone()[0] return { "year": year, "receipt_candidate_count": receipt_total, "confirmed_receipt_count": confirmed, "open_receipt_count": open_count, "covered_by_migros_count": int(covered or 0), "line_item_count": int(line_items or 0), "article_candidate_count": int(article_candidates or 0), "status_counts": by_status, "dry_run": True, } def confirm_migros_receipts_2026(conn: Connection, *, account_id: str, year: str = "2026") -> dict[str, Any]: if not account_id: raise HTTPException(status_code=422, detail="account_id required") category_id, category_name = _category_id_name(conn, "Essen & Haushalt") if not category_id: raise HTTPException(status_code=422, detail="Essen & Haushalt category required") rows = conn.execute( """ SELECT * FROM budget_transaction_candidates WHERE source_type='migros_receipt' AND substr(COALESCE(transaction_date,''),1,4)=? AND status NOT IN ('confirmed','ignored','covered_by_migros','superseded','reference_2025','archived_reference') ORDER BY transaction_date, created_at """, (year,), ).fetchall() ts = now() confirmed = skipped = 0 errors: list[dict[str, str]] = [] for row in rows: cid = str(row["transaction_candidate_id"]) try: conn.execute( """ UPDATE budget_transaction_candidates SET proposed_category_id=?, proposed_category_name=?, confidence='0.98', requires_review=0, status='auto_categorized', classification='migros_receipt_food_household', review_reason='migros_receipt_auto_food_household', rule_id='migros_receipt_food_household', rule_name='Migros Bon = Essen & Haushalt', updated_at=? WHERE transaction_candidate_id=? """, (category_id, category_name, ts, cid), ) confirm_transaction_candidate(conn, cid, {"account_id": account_id, "category_id": category_id, "budget_year": year}) confirmed += 1 except Exception as exc: # keep per-candidate auditability and aggregate reporting skipped += 1 errors.append({"candidate_id": cid, "error": str(getattr(exc, "detail", str(exc)))}) audit_id = record_audit_event( conn, source="budget_phase111_migros", action="migros_receipts_2026_confirmed", entity_type="budget_transaction_candidate", entity_id=f"migros:{year}", new_values={"confirmed_count": confirmed, "skipped_count": skipped, "error_count": len(errors), "line_item_transaction_count": 0}, created_by="agent", ) conn.commit() return {"year": year, "confirmed_count": confirmed, "skipped_count": skipped, "errors": errors, "line_item_transaction_count": 0, "audit_id": audit_id} def _matches_any(text: str, patterns: list[str]) -> bool: hay = _normalise(text) return any(_normalise(p) in hay for p in patterns) PHASE16_RULES: list[dict[str, Any]] = [ {"id": "p16_food_supermarket", "name": "Essen & Haushalt · Supermarkt/Lebensmittel", "category": "Essen & Haushalt", "classification": "food_household", "confidence": "0.88", "patterns": ["Migros", "Coop", "ALDI SUISSE", "Lidl", "Denner", "Volg", "Schmidt's Märkte", "SPAR", "Yavuz Market", "Supermercato Dpiu"]}, {"id": "p16_food_convenience", "name": "Essen & Haushalt · Convenience/Kiosk", "category": "Essen & Haushalt", "classification": "food_household", "confidence": "0.84", "patterns": ["Migrolino", "Coop Pronto", "Coop Vitality", "Avec", "K Kiosk", "Sonneland Kiosk"]}, {"id": "p16_food_restaurant", "name": "Essen & Haushalt · Restaurant/Takeaway", "category": "Essen & Haushalt", "classification": "food_household", "confidence": "0.82", "patterns": ["McDonald's", "Burger King", "Holy Cow", "Subway", "Brezelkönig", "Dunkin", "Restaurant Smiling Fish", "Restaurant Aquarena", "FELFEL AG", "Just Eat", "PostFinance Restaurant W", "Erne Chuchi", "DORY & DU", "Marche Burger", "Ristorante Sale e Pepe", "Marché Restaurants", "Herrlisberg Gran Bar", "Restaurant Pantanal", "Ristorante Pizzeria La Pa", "Mammami", "Restaurant Zoologischer G", "Restaurant Big Sterne", "Artigiano Pizzeria Nap", "Restaurant Waldkantine", "Restaurant Tannegg", "Gasthaus Felsgarten", "Lenzo Palace", "kunz AG art of sweets", "Spezialitäten-Metzgerei Bechinger", "Müller Drogeriemarkt", "Sonneland AG Shop", "Sportcenter Bustelbach"]}, {"id": "p16_home_garden", "name": "Hausrat / Möbel & Garten", "category": "Hausrat / Möbel & Garten", "classification": "home_garden", "confidence": "0.86", "patterns": ["IKEA", "Möbelmarkt Dogern", "JUMBO", "Boni-Shop", "Landi", "Zulauf Gartencenter", "Grillfurst Schweiz GmbH"]}, {"id": "p16_leisure_subscriptions", "name": "Freizeit / Ausflüge / Abos", "category": "Freizeit / Ausflüge / Abos", "classification": "subscription_auto", "confidence": "0.90", "patterns": ["Apple", "Apple.com", "Google", "Amazon Prime", "Netflix", "Disney", "Disney Plus", "Disney Streaming", "DisneyPlus", "Paramount+", "Prime Video CH", "Babbel", "OpenAI", "ChatGPT", "Europa-Park", "Knie's Kinderzoo", "Zoo Zürich", "Zoo Basel", "Zoo Hasel", "Technorama", "Legionarspfad", "Bad Schinznach", "aquabasilea", "Thermalbad Zurzach", "Alpamare", "sole uno", "Thermalquelle", "Kurhotel im Park", "Kraftreaktor", "Seilbahn Weissenstein", "Grand Casino Baden", "Pathé", "Cinema Excelsior", "Koi-Breeder", "OTT-Feuerwerk"]}, {"id": "p16_health_medical", "name": "Gesundheit / Medizin", "category": "Gesundheit / Medizin", "classification": "health_medical", "confidence": "0.88", "patterns": ["Kantonsspital Baden", "KSB", "Universitätsspital Zürich", "Spital Bulach", "Spital Wil", "Apotheke Suessbach", "Vindonissa Rotpunkt", "Loewen Apotheke Frick", "Rotpunkt Apotheken", "Storchen Apotheke", "Apotheke Tag", "Analytica Med", "Laboratori", "Stiftung Gesundheit", "iHerb"]}, {"id": "p16_auto_transport", "name": "Auto / Transport", "category": "Auto / Transport", "classification": "auto_transport", "confidence": "0.87", "patterns": ["Agrola", "Migrol", "Tamoil", "AVIA", "Esso", "Royal Dutch Shell", "Shell", "Voegtlin-Meyer", "Tankstelle Gebr. Knecht", "Tank-Rastanlage Neckarbur", "Spital Bulach - Parking", "ParkplatzAnnerstrasse", "Parkplatz Rosengarten", "City Parkhaus Zürich", "Parkhaus Trafo Baden", "EasyPark", "Eisi Parkhaus", "Parkhaus Messe Zürich", "Parking Gais-Center", "Mobility Hub Parkservice", "Parkingpay", "SBB CFF FFS", "SBB P+Rail Turgi", "Flughafen Zürich", "Palma de Mallorca Airport", "Volvo Car Corporation"]}, {"id": "p16_shopping", "name": "Shopping / Kleidung & Elektronik", "category": "Shopping / Kleidung & Elektronik", "classification": "shopping", "confidence": "0.84", "patterns": ["Apfelkiste", "Apple", "MediaMarkt", "H&M", "C&A", "VAN GRAAF", "Claire's", "Dosenbach", "Vero Moda", "OLYMP Digital", "Decathlon", "Smyths Toys", "AliExpress"]}, {"id": "p16_pets", "name": "Haustiere", "category": "Haustiere", "classification": "pets", "confidence": "0.89", "patterns": ["QUALIPET", "Fressnapf", "Kleintierpraxis Tim"]}, {"id": "p16_admin_other", "name": "Sonstiges / Administration", "category": "Sonstiges / Administration", "classification": "admin_other", "confidence": "0.76", "patterns": ["Finanzverw.des Kt Schaffh", "Gemeinde Windisch", "Gemeindeverw. Spreitenbac", "Regionalpolizei Brugg", "Dropbox", "Norton", "2CO.COM", "BITDEFENDER", "Bitdefender", "Alpha Progression", "Solar Manager AG", "TradingView", "Ricardo", "Beck-Mobile", "Eventlogistik-Armyliq", "Payment Solution & Servic", "SegPay.com", "Sami International"]}, ] GALAXUS_DIGITEC_RULE = {"id": "p16_galaxus_digitec_review", "name": "Galaxus/Digitec manuell prüfen oder splitten", "category": "Shopping / Kleidung & Elektronik", "classification": "galaxus_review", "confidence": "0.72", "patterns": ["Digitec Galaxus AG", "Galaxus", "Digitec"]} def _is_primary_migros_card_text(text: str) -> bool: hay = _normalise(text) if "migrol" in hay: return False return any(p in hay for p in ["migros", "cumulus", "mmm "]) def _phase16_match(text: str) -> dict[str, Any] | None: if _matches_any(text, GALAXUS_DIGITEC_RULE["patterns"]): return GALAXUS_DIGITEC_RULE for rule in PHASE16_RULES: if _matches_any(text, rule["patterns"]): return rule return None def _category_id_name(conn: Connection, name: str) -> tuple[str | None, str]: row = conn.execute("SELECT category_id, name FROM budget_categories WHERE is_active=1 AND lower(name)=lower(?) ORDER BY sort_order LIMIT 1", (name,)).fetchone() if row: return str(row["category_id"]), str(row["name"]) cid, cname = _category_by_name(conn, [name]) return cid, cname or name def apply_user_budget_rules_to_candidates(conn: Connection, *, year: str = "2026") -> dict[str, int]: rows = conn.execute( """ SELECT * FROM budget_transaction_candidates WHERE substr(COALESCE(transaction_date,''),1,4)=? AND source_type IN ('credit_card_csv','migros_receipt') AND status NOT IN ('confirmed','ignored','superseded') ORDER BY created_at """, (year,), ).fetchall() counts = { "candidate_total": 0, "auto_categorized_count": 0, "review_required_count": 0, "galaxus_digitec_review_count": 0, "covered_by_migros_count": 0, "migros_under_50_count": 0, "migros_over_50_count": 0, "subscription_count": 0, "health_medical_count": 0, "auto_transport_count": 0, "shopping_count": 0, "pets_count": 0, "admin_other_count": 0, "unclear_count": 0, "productive_confirmed_count": 0, } changed = 0 ts = now() for row in rows: counts["candidate_total"] += 1 desc = f"{row['merchant'] or ''} {row['description'] or ''}" source_type = row["source_type"] status = "needs_review"; requires_review = 1; category_id = None; category_name = None confidence = "0.35"; classification = "unclear"; review_reason = "Keine sichere Regel gefunden"; rule_id = "p16_unclear"; rule_name = "Unklar · manuelle Prüfung"; covered_by = None if source_type == "credit_card_csv" and _is_primary_migros_card_text(desc): status = "covered_by_migros"; requires_review = 0; confidence = "0.96"; classification = "covered_by_migros" review_reason = "Migros-CSV ist primäre Quelle für Migros-Ausgaben"; rule_id = "p16_card_migros_covered"; rule_name = "Kreditkarten-Migros durch Migros-CSV abgedeckt"; covered_by = "migros_csv" counts["covered_by_migros_count"] += 1 elif source_type == "migros_receipt": category_id, category_name = _category_id_name(conn, "Essen & Haushalt") classification = "migros_receipt_food_household" confidence = "0.98" rule_id = "migros_receipt_food_household" rule_name = "Migros Bon = Essen & Haushalt" review_reason = "Migros Bon automatisch Essen & Haushalt; Artikelzeilen bleiben Detaildaten" status = "auto_categorized" requires_review = 0 counts["migros_under_50_count"] += 1 else: rule = _phase16_match(desc) if rule: category_id, category_name = _category_id_name(conn, str(rule["category"])) classification = str(rule["classification"]); confidence = str(rule["confidence"]); rule_id = str(rule["id"]); rule_name = str(rule["name"]) if rule is GALAXUS_DIGITEC_RULE: status = "needs_review"; requires_review = 1; review_reason = "Galaxus/Digitec bitte manuell prüfen oder splitten"; counts["galaxus_digitec_review_count"] += 1 elif classification == "admin_other" and Decimal(confidence) < Decimal("0.80"): status = "needs_review"; requires_review = 1; review_reason = "Sonstiges/Administration eher manuell prüfen" else: status = "auto_categorized"; requires_review = 0; review_reason = f"Automatischer Vorschlag nach Nutzerregel: {rule_name}" if classification == "subscription_auto": counts["subscription_count"] += 1 if classification == "health_medical": counts["health_medical_count"] += 1 if classification == "auto_transport": counts["auto_transport_count"] += 1 if classification == "shopping": counts["shopping_count"] += 1 if classification == "pets": counts["pets_count"] += 1 if classification == "admin_other": counts["admin_other_count"] += 1 else: counts["unclear_count"] += 1 if status == "auto_categorized": counts["auto_categorized_count"] += 1 if requires_review: counts["review_required_count"] += 1 conn.execute( """ UPDATE budget_transaction_candidates SET proposed_category_id=?, proposed_category_name=?, confidence=?, requires_review=?, status=?, classification=?, review_reason=?, rule_id=?, rule_name=?, covered_by_source=?, updated_at=? WHERE transaction_candidate_id=? """, (category_id, category_name, confidence, requires_review, status, classification, review_reason, rule_id, rule_name, covered_by, ts, row["transaction_candidate_id"]), ) changed += 1 audit_id = record_audit_event(conn, source="budget_phase16_rules", action="apply_user_budget_rules", entity_type="budget_transaction_candidate", entity_id=f"year:{year}", new_values=counts | {"changed_count": changed}, created_by="agent") conn.commit() counts["audit_id"] = audit_id # type: ignore[assignment] return counts def seed_credit_card_candidates_from_rows(conn: Connection, rows: list[dict[str, Any]], *, source_file_label: str, subscription_category_id: str | None = None, migros_covered: bool = True, candidate_source_type: str = "credit_card_csv") -> dict[str, int]: sub_name = _category_name(conn, subscription_category_id) created = subs = galaxus = covered = dup_count = review = ignored = 0 for idx, raw in enumerate(rows, start=1): low = {_normalise(k): v for k, v in raw.items()} desc = str(low.get("beschreibung") or low.get("description") or low.get("text") or low.get("merchant") or low.get("merchantname") or low.get("details") or "").strip() tx_date = _date_text(low.get("datum") or low.get("date") or low.get("buchungsdatum")) amount = _decimal_text(low.get("betrag") or low.get("amount") or low.get("umsatz")) if not desc or not amount: ignored += 1; continue is_sub = _matches_any(desc, SUBSCRIPTION_PATTERNS) is_galaxus = _matches_any(desc, GALAXUS_PATTERNS) is_migros = _matches_any(desc, MIGROS_PATTERNS) dup = _duplicate_of(conn, tx_date, amount, desc) status = "pending"; needs_review = False; category_id = None; category_name = None; classification = "card_uncategorized"; reason = "card_review"; conf = "0.40"; covered_by = None if is_sub: category_id = subscription_category_id; category_name = sub_name; classification = "subscription_auto"; reason = "subscription_merchant_rule"; conf = "0.90"; subs += 1 if is_migros and migros_covered: status = "covered_by_migros"; needs_review = False; classification = "covered_by_migros"; reason = "credit_card_migros_covered_by_migros"; conf = "0.95"; covered_by = "migros_csv" covered += 1 elif dup: status = "duplicate"; needs_review = True; reason = "matched_existing_transaction"; dup_count += 1 elif is_galaxus: status = "needs_review"; needs_review = True; classification = "galaxus_review"; reason = "galaxus_requires_user_category_or_split"; conf = "0.70"; galaxus += 1; review += 1 else: status = "needs_review"; needs_review = True; review += 1 ts = now(); row_ref = f"R{idx}" if _insert_candidate(conn, { "transaction_candidate_id": "btxcand_" + _candidate_key(source_file_label, row_ref, tx_date, amount, desc), "source_file_label": source_file_label, "source_row_or_range": row_ref, "source_type": candidate_source_type, "transaction_date": tx_date, "description": desc[:300], "merchant": desc[:120], "amount_original": amount, "currency_original": "CHF", "proposed_category_id": category_id, "proposed_category_name": category_name, "duplicate_of_transaction_id": dup, "confidence": conf, "requires_review": needs_review, "status": status, "notes": reason, "classification": classification, "review_reason": reason, "rule_id": classification, "source_priority": 70, "covered_by_source": covered_by, "raw_fingerprint": _fingerprint([source_file_label, row_ref, tx_date, amount, desc]), "created_at": ts, "updated_at": ts, }): created += 1 conn.commit() return {"candidate_count": created, "subscription_auto_count": subs, "galaxus_review_count": galaxus, "covered_by_migros_count": covered, "possible_duplicate_count": dup_count, "review_required_count": review, "ignored_count": ignored} @dataclass class _BankRow: idx: int date: str | None value_date: str | None desc: str amount_abs: str amount_signed: Decimal account: str | None def _bank_rows(rows: list[dict[str, Any]]) -> list[_BankRow]: parsed = [] for idx, raw in enumerate(rows, start=1): low = {_normalise(k): v for k, v in raw.items()} desc = str(low.get("beschreibung") or low.get("description") or low.get("text") or low.get("buchungstext") or low.get("name") or "").strip() date = _date_text(low.get("datum") or low.get("date") or low.get("buchungsdatum") or low.get("booked at") or low.get("buchung")) value_date = _date_text(low.get("valuta date") or low.get("valuta") or date) signed = _signed_decimal(low.get("betrag") or low.get("amount") or low.get("umsatz") or low.get("credit/debit amount")) if signed is None: debit = _signed_decimal(low.get("belastung")) credit = _signed_decimal(low.get("gutschrift")) if credit is not None: signed = credit elif debit is not None: signed = -abs(debit) if not desc or signed is None: continue parsed.append(_BankRow(idx, date or value_date, value_date, desc, format(abs(signed), "f"), signed, str(low.get("konto") or low.get("account") or low.get("iban") or "").strip() or None)) return parsed def _find_transfer_indices(rows: list[_BankRow]) -> set[int]: indices: set[int] = set() for i, left in enumerate(rows): for right in rows[i + 1:]: if left.amount_abs != right.amount_abs or left.amount_signed * right.amount_signed >= 0: continue if left.date and right.date: d1 = datetime.fromisoformat(left.date); d2 = datetime.fromisoformat(right.date) if abs((d1 - d2).days) > 3: continue joined = f"{left.desc} {right.desc} {left.account or ''} {right.account or ''}" if _matches_any(joined, TRANSFER_PATTERNS): indices.add(left.idx); indices.add(right.idx) return indices def _bank_candidate_decision(description: str, *, is_positive: bool, matched_transfer_pair: bool = False, source_type: str | None = None, account_source: str | None = None) -> dict[str, Any]: """Classify a bank row without confirming it. Positive salary/employer patterns win over generic bank names, while explicit own-account / card / investment transfer patterns must never become income. """ if _matches_any(description, ["TRUE WEALTH", "TRUEWEALTH"]): return {"status": "transfer_candidate", "requires_review": True, "classification": "investment_transfer", "review_reason": "true_wealth_investment_transfer", "transaction_type": "transfer", "confidence": "0.95"} if _matches_any(description, ["VISA", "KREDITKARTE", "KARTENABRECHNUNG", "CREDIT CARD"]): return {"status": "transfer_candidate", "requires_review": True, "classification": "credit_card_payment", "review_reason": "credit_card_payment_avoids_double_counting", "transaction_type": "transfer", "confidence": "0.95"} if matched_transfer_pair or _matches_any(description, TRANSFER_PATTERNS): return {"status": "transfer_candidate", "requires_review": True, "classification": "transfer_candidate", "review_reason": "possible_internal_transfer", "transaction_type": "transfer", "confidence": "0.88"} if is_positive and _matches_any(description, STRONG_INCOME_PATTERNS): return {"status": "pending", "requires_review": False, "classification": "income_candidate", "review_reason": "strong_income_pattern", "transaction_type": "income", "confidence": "0.90"} if is_positive and _matches_any(description, POSSIBLE_INCOME_PATTERNS) and not _matches_any(description, NEGATIVE_INCOME_PATTERNS): return {"status": "needs_review", "requires_review": True, "classification": "possible_income", "review_reason": "possible_income_pattern", "transaction_type": "income", "confidence": "0.65"} return {"status": "needs_review", "requires_review": True, "classification": "bank_review", "review_reason": "bank_payment_unclear", "transaction_type": "expense" if not is_positive else "income", "confidence": "0.50"} def seed_bank_transfer_candidates_from_rows(conn: Connection, rows: list[dict[str, Any]], *, source_file_label: str, income_category_id: str | None = None, candidate_source_type: str = "bank_csv") -> dict[str, int]: parsed = _bank_rows(rows) transfer_indices = _find_transfer_indices(parsed) income_name = _category_name(conn, income_category_id) created = income = transfer = review = dup_count = investment_transfer = credit_card_payment = 0 for br in parsed: dup = _duplicate_of(conn, br.date, br.amount_abs, br.desc) category_id = None; category_name = None decision = _bank_candidate_decision(br.desc, is_positive=br.amount_signed > 0, matched_transfer_pair=br.idx in transfer_indices, source_type=candidate_source_type, account_source=br.account) status = str(decision["status"]); needs_review = bool(decision["requires_review"]); classification = str(decision["classification"]); reason = str(decision["review_reason"]); tx_type = str(decision["transaction_type"]); conf = str(decision["confidence"]) if classification == "income_candidate": category_id = income_category_id; category_name = income_name; income += 1 elif classification == "possible_income": category_id = income_category_id; category_name = income_name; review += 1 elif classification in {"transfer_candidate", "investment_transfer", "credit_card_payment"}: transfer += 1 investment_transfer += 1 if classification == "investment_transfer" else 0 credit_card_payment += 1 if classification == "credit_card_payment" else 0 elif dup: status = "duplicate_candidate"; classification = "possible_duplicate"; reason = "matched_existing_transaction"; dup_count += 1 else: review += 1 ts = now(); row_ref = f"R{br.idx}" if _insert_candidate(conn, { "transaction_candidate_id": "btxcand_" + _candidate_key(source_file_label, row_ref, br.date, br.amount_abs, br.desc), "source_file_label": source_file_label, "source_row_or_range": row_ref, "source_type": candidate_source_type, "transaction_date": br.date, "value_date": br.value_date, "description": br.desc[:300], "merchant": None, "amount_original": br.amount_abs, "signed_amount_original": format(br.amount_signed, "f"), "currency_original": "CHF", "proposed_category_id": category_id, "proposed_category_name": category_name, "duplicate_of_transaction_id": dup, "confidence": conf, "requires_review": needs_review, "status": status, "notes": json.dumps({"transaction_type": tx_type, "reason": reason, "target_account_hint": decision.get("target_account_hint"), "transfer": {"budget_impact": "neutral", "pair_status": "unmatched", "from_account_name": "Raiffeisen" if candidate_source_type == "raiffeisen_bank" else None, "to_account_name": decision.get("target_account_hint")}}, ensure_ascii=False), "classification": classification, "review_reason": reason, "rule_id": classification, "rule_name": decision.get("rule_name"), "source_priority": 80, "account_source": br.account, "raw_fingerprint": _fingerprint([source_file_label, row_ref, br.date, br.amount_abs, br.desc, br.account]), "created_at": ts, "updated_at": ts, }): created += 1 conn.commit() from jarvis_finance.services.transfer_pairing import propose_transfer_pairs propose_transfer_pairs(conn) return {"candidate_count": created, "income_candidate_count": income, "transfer_candidate_count": transfer, "investment_transfer_count": investment_transfer, "credit_card_payment_count": credit_card_payment, "review_required_count": review, "possible_duplicate_count": dup_count} def list_transaction_candidates(conn: Connection, *, status: str | None = None, tab: str | None = None, source_type: str | None = None, category_id: str | None = None, merchant: str | None = None, rule: str | None = None, min_confidence: str | None = None, date_from: str | None = None, date_to: str | None = None, amount_min: str | None = None, amount_max: str | None = None, sort_by: str = "created_at", sort_dir: str = "desc", budget_year: str = "2026", include_reference: bool = False, import_session_id: str | None = None) -> list[dict[str, Any]]: clauses=[]; params=[] if budget_year: clauses.append("substr(transaction_date,1,4)=?"); params.append(budget_year) normal_excluded_sql = "'confirmed','ignored','covered_by_migros','covered_by_source','auto_ignored_duplicate','duplicate','duplicate_candidate','possible_duplicate','duplicate_blocked','superseded','reference_2025','archived_reference','already_processed','transfer_candidate'" if status == "open": clauses.append("status IN ('pending','auto_categorized','needs_review')") elif status: clauses.append("status=?"); params.append(status) elif not include_reference and not tab: clauses.append(f"status NOT IN ({normal_excluded_sql})") if tab: mapping = { "auto": "requires_review=0 AND status IN ('pending','auto_categorized')", "review": "requires_review=1 AND status IN ('needs_review','pending','transfer_candidate')", "subscriptions": "classification='subscription_auto'", "galaxus": "classification='galaxus_review'", "migros_over_50": "classification='migros_review_over_threshold'", "health": "classification='health_medical'", "auto_transport": "classification='auto_transport'", "shopping": "classification IN ('shopping','galaxus_review')", "pets": "classification='pets'", "unclear": "classification IN ('admin_other','unclear','legacy_csv_seed','card_uncategorized')", "covered_by_migros": "status='covered_by_migros'", "duplicates": "status IN ('duplicate','duplicate_candidate','possible_duplicate','duplicate_blocked')", "auto_ignored_duplicates": "status='auto_ignored_duplicate'", "covered_by_source": "status IN ('covered_by_migros','covered_by_source')", "transfers": "classification='transfer_candidate' OR status='transfer_candidate'", "bank": "source_type IN ('bank_csv','raiffeisen_bank','akb_bank')", "visa": "source_type='visa_credit_card' OR source_type='credit_card_csv'", "migros": "(source_type IN ('migros_receipts','migros_receipt') OR status='covered_by_migros') AND status NOT IN ('confirmed','duplicate','duplicate_candidate','possible_duplicate','duplicate_blocked','superseded','ignored','reference_2025','archived_reference')", "raiffeisen": "source_type='raiffeisen_bank'", "akb": "source_type='akb_bank'", "income": "classification='income_candidate'", "possible_income": "classification='possible_income'", "money_in": "classification IN ('income_candidate','possible_income')", "not_income_transfer": "classification IN ('transfer_candidate','investment_transfer','credit_card_payment') OR status='transfer_candidate'", "credit_card_payment": "classification='credit_card_payment'", "investment_transfers": "classification='investment_transfer'", "reference_2025": "status IN ('reference_2025','archived_reference')", "ignored": "status IN ('ignored','covered_by_migros','covered_by_source','superseded','reference_2025','archived_reference')", } if tab in mapping: clauses.append(mapping[tab]) if source_type: clauses.append("source_type=?"); params.append(source_type) if import_session_id: clauses.append("import_session_id=?"); params.append(import_session_id) if category_id: clauses.append("proposed_category_id=?"); params.append(category_id) if merchant: clauses.append("lower(COALESCE(merchant,'') || ' ' || COALESCE(description,'')) LIKE ?"); params.append(f"%{merchant.lower()}%") if rule: clauses.append("lower(COALESCE(rule_id,'') || ' ' || COALESCE(rule_name,'') || ' ' || COALESCE(review_reason,'')) LIKE ?"); params.append(f"%{rule.lower()}%") if min_confidence: clauses.append("CAST(COALESCE(confidence,'0') AS REAL) >= CAST(? AS REAL)"); params.append(min_confidence) if date_from: clauses.append("transaction_date>=?"); params.append(date_from) if date_to: clauses.append("transaction_date<=?"); params.append(date_to) if amount_min: clauses.append("CAST(COALESCE(amount_original,'0') AS REAL) >= CAST(? AS REAL)"); params.append(amount_min) if amount_max: clauses.append("CAST(COALESCE(amount_original,'0') AS REAL) <= CAST(? AS REAL)"); params.append(amount_max) where = "WHERE " + " AND ".join(clauses) if clauses else "" sort_map = {"date": "transaction_date", "merchant": "lower(COALESCE(merchant, description))", "category": "proposed_category_name", "confidence": "CAST(COALESCE(confidence,'0') AS REAL)", "source": "source_type", "status": "status", "created_at": "created_at"} order_col = sort_map.get(sort_by, "created_at") direction = "ASC" if sort_dir.lower() == "asc" else "DESC" rows = conn.execute(f"SELECT * FROM budget_transaction_candidates {where} ORDER BY {order_col} {direction}, created_at DESC LIMIT 500", params).fetchall() return [_candidate_view(r) for r in rows] def reclassify_bank_income_candidates(conn: Connection, *, budget_year: str = "2026") -> dict[str, int]: """Re-score existing AKB/Raiffeisen candidates without confirming bookings.""" cid_income, cname_income = _category_by_name(conn, ["Lohn Marcel", "Einnahmen", "Sonstige Einnahmen", "Rückerstattungen"]) rows = conn.execute( """ SELECT * FROM budget_transaction_candidates WHERE source_type IN ('akb_bank','raiffeisen_bank','bank_csv') AND substr(COALESCE(transaction_date,''),1,4)=? AND status NOT IN ('confirmed','ignored','covered_by_migros','covered_by_source','auto_ignored_duplicate','superseded','reference_2025','archived_reference') """, (budget_year,), ).fetchall() counts = {"scanned_count": len(rows), "income_candidate_count": 0, "possible_income_count": 0, "transfer_candidate_count": 0, "investment_transfer_count": 0, "credit_card_payment_count": 0, "unclear_money_in_count": 0, "updated_count": 0} ts = now() for row in rows: meta = {} try: meta = json.loads(row["notes"] or "{}") if str(row["notes"] or "").startswith("{") else {} except json.JSONDecodeError: meta = {} is_positive = str(meta.get("transaction_type") or "").lower() == "income" decision = _bank_candidate_decision(str(row["description"] or ""), is_positive=is_positive, source_type=str(row["source_type"] or ""), account_source=str(row["account_source"] or "")) classification = str(decision["classification"]) if is_positive and classification == "bank_review": counts["unclear_money_in_count"] += 1 if classification == "income_candidate": counts["income_candidate_count"] += 1 if classification == "possible_income": counts["possible_income_count"] += 1 if classification in {"transfer_candidate", "investment_transfer", "credit_card_payment"}: counts["transfer_candidate_count"] += 1 if classification == "investment_transfer": counts["investment_transfer_count"] += 1 if classification == "credit_card_payment": counts["credit_card_payment_count"] += 1 proposed_category_id = cid_income if classification in {"income_candidate", "possible_income"} else row["proposed_category_id"] proposed_category_name = cname_income if classification in {"income_candidate", "possible_income"} else row["proposed_category_name"] notes = json.dumps({"transaction_type": decision["transaction_type"], "reason": decision["review_reason"], "target_account_hint": decision.get("target_account_hint"), "transfer": {"budget_impact": "neutral", "pair_status": "unmatched", "from_account_name": "Raiffeisen" if row["source_type"] == "raiffeisen_bank" else None, "to_account_name": decision.get("target_account_hint")}, "reclassified_by": "income_csv_recognition_v1"}, ensure_ascii=False) before = (row["status"], row["classification"], row["review_reason"], row["proposed_category_id"]) after = (decision["status"], classification, decision["review_reason"], proposed_category_id) if before != after: conn.execute( """ UPDATE budget_transaction_candidates SET status=?, classification=?, requires_review=?, review_reason=?, confidence=?, notes=?, proposed_category_id=?, proposed_category_name=?, rule_name=?, updated_at=? WHERE transaction_candidate_id=? """, (decision["status"], classification, 1 if decision["requires_review"] else 0, decision["review_reason"], decision["confidence"], notes, proposed_category_id, proposed_category_name, decision.get("rule_name"), ts, row["transaction_candidate_id"]), ) counts["updated_count"] += 1 audit_id = record_audit_event(conn, source="vue_dashboard", action="bank_income_candidates_reclassified", entity_type="budget_transaction_candidate", entity_id=f"bank_income_reclassify_{budget_year}", new_values=counts, created_by="system") conn.commit() counts["audit_written"] = 1 if audit_id else 0 return counts def classify_swisslos_internal_transfers(conn: Connection, *, budget_year: str = "2026") -> dict[str, Any]: """Compatibility entry point; all text-specific behavior is intentionally removed.""" from jarvis_finance.services.transfer_pairing import propose_transfer_pairs result = propose_transfer_pairs(conn) return { "updated_count": result["proposed_count"] + result["unmatched_count"] + result["ambiguous_count"], "matched_count": result["proposed_count"], "unmatched_count": result["unmatched_count"] + result["ambiguous_count"], "matcher_version": "transfer_pairing_v2", "budget_year": budget_year, } def preview_transaction_candidate(conn: Connection, candidate_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") payload = payload or {} status = str(row["status"] or "") if status in DUPLICATE_CONFIRM_BLOCKED_STATUSES and not payload.get("duplicate_override"): raise HTTPException(status_code=409, detail=DUPLICATE_CONFIRM_BLOCK_MESSAGE) cat_id = payload.get("category_id") or row["proposed_category_id"] notes = row["notes"] or "" tx_type = payload.get("transaction_type") or "expense" try: meta = json.loads(notes) if notes and notes.startswith("{") else {} tx_type = payload.get("transaction_type") or meta.get("transaction_type") or tx_type except json.JSONDecodeError: pass if not cat_id and tx_type in {"expense", "income"}: raise HTTPException(status_code=422, detail="category required before confirm") if cat_id and not conn.execute("SELECT 1 FROM budget_categories WHERE category_id=? AND is_active=1", (cat_id,)).fetchone(): raise HTTPException(status_code=422, detail="active category required before confirm") if not payload.get("account_id") and tx_type != "transfer": raise HTTPException(status_code=422, detail="account_id required before confirm") if tx_type == "transfer" and not (payload.get("from_account_id") or payload.get("account_id")): raise HTTPException(status_code=422, detail="from_account_id required before transfer confirm") if tx_type == "transfer" and not payload.get("to_account_id"): raise HTTPException(status_code=422, detail="to_account_id required before transfer confirm") if not row["transaction_date"]: raise HTTPException(status_code=422, detail="transaction date required before confirm") if not row["amount_original"]: raise HTTPException(status_code=422, detail="amount required before confirm") if not row["currency_original"]: raise HTTPException(status_code=422, detail="currency required before confirm") return { "preview_id": new_id("preview"), "summary": "Buchungskandidat übernehmen", "warnings": ["Möglicher Duplikat-Treffer"] if row["duplicate_of_transaction_id"] else ([] if cat_id or tx_type == "transfer" else ["Kategorie fehlt"]), "payload": { "account_id": payload.get("account_id") or payload.get("from_account_id"), "from_account_id": payload.get("from_account_id") or payload.get("account_id"), "to_account_id": payload.get("to_account_id"), "transaction_type": tx_type, "transaction_date": payload.get("transaction_date") or row["transaction_date"], "description": payload.get("description") or row["description"], "payee": payload.get("payee") or row["merchant"], "amount_original": row["amount_original"], "currency_original": row["currency_original"], "category_id": cat_id, "source_type": "import_candidate", "source_candidate_id": candidate_id, "notes": f"Confirmed from transaction candidate {candidate_id}", }, } def confirm_transaction_candidate(conn: Connection, candidate_id: str, payload: dict[str, Any]) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") if row["status"] == "confirmed": raise HTTPException(status_code=409, detail="candidate already confirmed") if row["status"] in DUPLICATE_CONFIRM_BLOCKED_STATUSES: raise HTTPException(status_code=409, detail=DUPLICATE_CONFIRM_BLOCK_MESSAGE) if str(row["transaction_date"] or "")[:4] != str(payload.get("budget_year") or "2026"): raise HTTPException(status_code=409, detail="candidate outside active budget year") preview = preview_transaction_candidate(conn, candidate_id, payload) if preview["payload"].get("transaction_type") == "transfer" or row["status"] == "transfer_candidate" or "transfer" in str(row["classification"] or ""): from jarvis_finance.services.transfer_pairing import confirm_transfer_pair pair = conn.execute( """ SELECT transfer_pair_id FROM budget_transfer_pairs WHERE status='proposed' AND (source_candidate_id=? OR target_candidate_id=?) ORDER BY created_at DESC LIMIT 1 """, (candidate_id, candidate_id), ).fetchone() if not pair: raise HTTPException( status_code=409, detail="transfer counterbooking required before confirm", ) return confirm_transfer_pair( conn, str(pair["transfer_pair_id"]), decision_by="user", note=str(payload.get("note") or "") or None, ) result = confirm_budget_transaction(conn, preview["payload"]) ts = now() conn.execute("UPDATE budget_transaction_candidates SET status='confirmed', confirmed_transaction_id=?, confirmed_at=?, confirmed_by='user', updated_at=? WHERE transaction_candidate_id=?", (result["entity_id"], ts, ts, candidate_id)) record_audit_event(conn, source="vue_dashboard", action="budget_transaction_created", entity_type="budget_transaction", entity_id=result["entity_id"], new_values={"source_candidate_id": candidate_id}, created_by="user") audit_id = record_audit_event(conn, source="vue_dashboard", action="candidate_confirmed", entity_type="budget_transaction_candidate", entity_id=candidate_id, old_values={"status": row["status"]}, new_values={"status": "confirmed", "confirmed_transaction_id": result["entity_id"], "category_id": preview["payload"].get("category_id")}, created_by="user") conn.commit() return {"status": "confirmed", "entity_id": result["entity_id"], "audit_id": audit_id, "message": "Buchungskandidat übernommen"} def confirm_duplicate_override(conn: Connection, candidate_id: str, payload: dict[str, Any]) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") if row["status"] == "confirmed": raise HTTPException(status_code=409, detail="candidate already confirmed") if row["status"] not in USER_DUPLICATE_STATUSES: raise HTTPException(status_code=409, detail="candidate is not a duplicate override candidate") reason = str(payload.get("reason") or payload.get("override_reason") or "").strip() if len(reason) < 8: raise HTTPException(status_code=422, detail="duplicate override reason required") if str(row["transaction_date"] or "")[:4] != str(payload.get("budget_year") or "2026"): raise HTTPException(status_code=409, detail="candidate outside active budget year") preview_payload = dict(payload) | {"duplicate_override": True, "notes": f"Duplicate override: {reason}"} preview = preview_transaction_candidate(conn, candidate_id, preview_payload) result = confirm_budget_transaction(conn, preview["payload"] | {"notes": f"Duplicate override from transaction candidate {candidate_id}: {reason}"}) ts = now() conn.execute("UPDATE budget_transaction_candidates SET status='confirmed', confirmed_transaction_id=?, confirmed_at=?, confirmed_by='user', updated_at=? WHERE transaction_candidate_id=?", (result["entity_id"], ts, ts, candidate_id)) record_audit_event(conn, source="vue_dashboard", action="budget_transaction_created_duplicate_override", entity_type="budget_transaction", entity_id=result["entity_id"], new_values={"source_candidate_id": candidate_id, "override_reason": reason}, created_by="user") audit_id = record_audit_event(conn, source="vue_dashboard", action="candidate_duplicate_override_confirmed", entity_type="budget_transaction_candidate", entity_id=candidate_id, old_values={"status": row["status"]}, new_values={"status": "confirmed", "confirmed_transaction_id": result["entity_id"], "override_reason": reason}, user_text_note=reason, created_by="user") conn.commit() return {"status": "confirmed", "entity_id": result["entity_id"], "audit_id": audit_id, "message": "Duplikat-Override bestätigt"} def ignore_transaction_candidate(conn: Connection, candidate_id: str, note: str | None = None) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") ts = now() conn.execute("UPDATE budget_transaction_candidates SET status='ignored', notes=COALESCE(?, notes), updated_at=? WHERE transaction_candidate_id=?", (note, ts, candidate_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="ignore", entity_type="budget_transaction_candidate", entity_id=candidate_id, old_values={"status": row["status"]}, new_values={"status": "ignored"}, created_by="user") conn.commit() return {"status": "ignored", "entity_id": candidate_id, "audit_id": audit_id, "message": "Buchungskandidat ignoriert"} def reopen_transaction_candidate(conn: Connection, candidate_id: str, note: str | None = None) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") if row["status"] == "confirmed": raise HTTPException(status_code=409, detail="confirmed candidates cannot be reopened") new_status = "needs_review" if row["requires_review"] else "pending" ts = now() conn.execute("UPDATE budget_transaction_candidates SET status=?, review_reason=COALESCE(review_reason, 'user_reopened'), notes=COALESCE(?, notes), updated_at=? WHERE transaction_candidate_id=?", (new_status, note, ts, candidate_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="reopen", entity_type="budget_transaction_candidate", entity_id=candidate_id, old_values={"status": row["status"]}, new_values={"status": new_status}, user_text_note=note, created_by="user") conn.commit() return {"status": new_status, "entity_id": candidate_id, "audit_id": audit_id, "message": "Buchungskandidat wieder geöffnet"} def update_transaction_candidate_category(conn: Connection, candidate_id: str, category_id: str, note: str | None = None) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") if row["status"] in USER_DUPLICATE_STATUSES: raise HTTPException(status_code=409, detail="duplicate candidates must be reopened before category editing") if not conn.execute("SELECT 1 FROM budget_categories WHERE category_id=? AND is_active=1", (category_id,)).fetchone(): raise HTTPException(status_code=422, detail="active category not found") cname = _category_name(conn, category_id) ts = now() conn.execute("UPDATE budget_transaction_candidates SET proposed_category_id=?, proposed_category_name=?, requires_review=0, status='pending', review_reason='user_category_override', updated_at=? WHERE transaction_candidate_id=?", (category_id, cname, ts, candidate_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="update_category", entity_type="budget_transaction_candidate", entity_id=candidate_id, old_values=row_to_dict(row), new_values={"category_id": category_id, "category_name": cname}, user_text_note=note, created_by="user") conn.commit() return {"status": "updated", "entity_id": candidate_id, "audit_id": audit_id} def create_candidate_split(conn: Connection, candidate_id: str, splits: list[dict[str, Any]]) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") if not splits: raise HTTPException(status_code=422, detail="splits required") ts = now() conn.execute("DELETE FROM budget_candidate_splits WHERE transaction_candidate_id=?", (candidate_id,)) for split in splits: amount = _decimal_text(split.get("amount_original")) if not amount: raise HTTPException(status_code=422, detail="split amount required") conn.execute("INSERT INTO budget_candidate_splits(split_id, transaction_candidate_id, category_id, amount_original, notes, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)", (new_id("bcsplit"), candidate_id, split.get("category_id"), amount, split.get("notes"), ts, ts)) conn.execute("UPDATE budget_transaction_candidates SET requires_review=1, status='needs_review', review_reason='split_pending_confirm', updated_at=? WHERE transaction_candidate_id=?", (ts, candidate_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="split", entity_type="budget_transaction_candidate_split", entity_id=candidate_id, new_values={"split_count": len(splits)}, created_by="user") conn.commit() return {"status": "split_created", "entity_id": candidate_id, "split_count": len(splits), "audit_id": audit_id} def mark_candidate_type(conn: Connection, candidate_id: str, candidate_type: str, note: str | None = None) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") allowed = { "transfer": ("transfer_candidate", "transfer_candidate", "user_marked_transfer", "mark_transfer"), "investment_transfer": ("transfer_candidate", "investment_transfer", "user_marked_investment_transfer", "mark_investment_transfer"), "credit_card_payment": ("transfer_candidate", "credit_card_payment", "user_marked_credit_card_payment", "mark_credit_card_payment"), } if candidate_type not in allowed: raise HTTPException(status_code=422, detail="unsupported candidate type") status, classification, reason, action = allowed[candidate_type] ts = now() conn.execute("UPDATE budget_transaction_candidates SET status=?, classification=?, requires_review=1, review_reason=?, notes=COALESCE(?, notes), updated_at=? WHERE transaction_candidate_id=?", (status, classification, reason, note, ts, candidate_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action=action, entity_type="budget_transaction_candidate", entity_id=candidate_id, old_values={"status": row["status"], "classification": row["classification"]}, new_values={"status": status, "classification": classification}, user_text_note=note, created_by="user") conn.commit() return {"status": status, "classification": classification, "entity_id": candidate_id, "audit_id": audit_id} def mark_candidate_transfer(conn: Connection, candidate_id: str, note: str | None = None) -> dict[str, Any]: return mark_candidate_type(conn, candidate_id, "transfer", note) def bulk_update_transaction_candidate_category(conn: Connection, *, candidate_ids: list[str], category_id: str, note: str | None = None) -> dict[str, int]: updated = skipped = 0 for cid in candidate_ids: row = conn.execute("SELECT status FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (cid,)).fetchone() if not row or row["status"] == "confirmed": skipped += 1 continue update_transaction_candidate_category(conn, cid, category_id, note or "Bulk category assignment") updated += 1 return {"updated_count": updated, "skipped_count": skipped} def _safe_auto_candidate_where() -> str: return """ status IN ('pending','auto_categorized') AND COALESCE(requires_review, 0)=0 AND duplicate_of_transaction_id IS NULL AND COALESCE(covered_by_source, '')='' AND ( classification='subscription_auto' OR classification IN ('migros_auto_under_threshold','migros_receipt_food_household') OR classification='auto_transport' ) AND lower(COALESCE(description,'') || ' ' || COALESCE(merchant,'')) NOT LIKE '%galaxus%' AND lower(COALESCE(description,'') || ' ' || COALESCE(merchant,'')) NOT LIKE '%digitec%' """ def preview_safe_auto_candidates(conn: Connection) -> dict[str, Any]: rows = conn.execute( f""" SELECT classification, COUNT(*) AS count FROM budget_transaction_candidates WHERE {_safe_auto_candidate_where()} GROUP BY classification ORDER BY classification """ ).fetchall() total = sum(int(r["count"] or 0) for r in rows) return { "preview_id": new_id("preview"), "summary": f"{total} sichere Auto-Kandidaten bereit für Confirm", "count": total, "groups": [row_to_dict(r) for r in rows], "warnings": ["Preview only: Confirm benötigt Zielkonto und bucht nur sichere Kandidaten."], } def batch_confirm_safe_candidates(conn: Connection, *, account_id: str, candidate_ids: list[str] | None = None) -> dict[str, int]: if not account_id: raise HTTPException(status_code=422, detail="account_id required") allowed_rows = conn.execute(f"SELECT transaction_candidate_id FROM budget_transaction_candidates WHERE {_safe_auto_candidate_where()}").fetchall() allowed_ids = {str(r["transaction_candidate_id"]) for r in allowed_rows} requested_ids = set(candidate_ids or allowed_ids) confirmed = skipped = 0 for cid in sorted(requested_ids): if cid not in allowed_ids: skipped += 1 continue confirm_transaction_candidate(conn, cid, {"account_id": account_id}) confirmed += 1 record_audit_event(conn, source="vue_dashboard", action="batch_confirm_safe_candidates", entity_type="budget_transaction_candidate", entity_id="batch", new_values={"confirmed_count": confirmed, "skipped_count": skipped}, created_by="user") conn.commit() return {"confirmed_count": confirmed, "skipped_count": skipped} def candidate_user_count_summary(conn: Connection, *, budget_year: str = "2026") -> dict[str, int]: rows = conn.execute( """ SELECT status, classification, requires_review, duplicate_of_transaction_id FROM budget_transaction_candidates WHERE substr(COALESCE(transaction_date,''),1,4)=? """, (budget_year,), ).fetchall() def count(pred): return sum(1 for r in rows if pred(r)) normal_open = lambda r: r["status"] in USER_OPEN_STATUSES duplicates = lambda r: r["status"] in USER_DUPLICATE_STATUSES or bool(r["duplicate_of_transaction_id"]) covered = lambda r: r["status"] in {"covered_by_source", "covered_by_migros"} already = lambda r: r["status"] in {"already_processed", "auto_ignored_duplicate"} return { "open_total": count(normal_open), "auto_categorized": count(lambda r: r["status"] in {"pending", "auto_categorized"} and not bool(r["requires_review"])), "review_needed": count(lambda r: normal_open(r) and bool(r["requires_review"])), "galaxus_digitec": count(lambda r: normal_open(r) and r["classification"] == "galaxus_review"), "migros_over_50": count(lambda r: normal_open(r) and r["classification"] == "migros_review_over_threshold"), "covered_by_source": count(covered), "covered_by_migros": count(lambda r: r["status"] == "covered_by_migros"), "unclear": count(lambda r: normal_open(r) and r["classification"] in {"admin_other", "unclear", "legacy_csv_seed", "card_uncategorized"}), "possible_duplicates": count(duplicates), "transfer_candidates": count(lambda r: normal_open(r) and (r["status"] == "transfer_candidate" or r["classification"] == "transfer_candidate")), "income_candidates": count(lambda r: normal_open(r) and r["classification"] == "income_candidate"), "possible_income": count(lambda r: normal_open(r) and r["classification"] == "possible_income"), "already_processed": count(already), } def review_candidate_summary(conn: Connection, *, budget_year: str = "2026") -> dict[str, int]: return candidate_user_count_summary(conn, budget_year=budget_year) BLOCKED_BULK_EDIT_STATUSES = {"confirmed", "covered_by_migros", "covered_by_source", "auto_ignored_duplicate", "superseded", "reference_2025", "archived_reference"} | USER_DUPLICATE_STATUSES BLOCKED_BULK_CONFIRM_STATUSES = BLOCKED_BULK_EDIT_STATUSES | {"ignored", "transfer_candidate"} def _batch_confirm_block_reason(row: Any, *, budget_year: str) -> str | None: status = str(row["status"] or "") classification = str(row["classification"] or "") if status in USER_DUPLICATE_STATUSES: return "duplicate_candidate_blocked" if status in {"covered_by_source", "covered_by_migros"}: return status if status in {"auto_ignored_duplicate", "superseded", "reference_2025", "archived_reference", "confirmed", "ignored"}: return f"not_confirmable:{status}" if status == "transfer_candidate" or "transfer" in classification: return "transfer_candidate_blocked" if str(row["transaction_date"] or "")[:4] != str(budget_year): return "outside_active_budget_year" if not row["proposed_category_id"] and classification not in {"income_candidate", "possible_income"}: return "missing_category" if not row["transaction_date"]: return "missing_transaction_date" if not row["amount_original"]: return "missing_amount" if not row["currency_original"]: return "missing_currency" return None def _blocker_summary(conflicts: list[dict[str, Any]]) -> dict[str, int]: def count(needle: str) -> int: return sum(1 for c in conflicts if needle in str(c.get("reason") or "")) return { "duplicate_candidates": count("duplicate"), "covered_by_source": count("covered_by_source") + count("covered_by_migros"), "transfer_candidates": count("transfer"), "without_category": count("missing_category"), "missing_required_fields": count("missing_transaction_date") + count("missing_amount") + count("missing_currency"), "not_confirmable": count("not_confirmable"), } def preview_batch_confirm_candidates(conn: Connection, *, candidate_ids: list[str], account_id: str, budget_year: str = "2026") -> dict[str, Any]: rows = [] conflicts: list[dict[str, Any]] = [] categories: dict[str, int] = {} sources: dict[str, int] = {} possible_duplicates = 0 for cid in candidate_ids: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (cid,)).fetchone() if not row: conflicts.append({"candidate_id": cid, "reason": "missing"}); continue reason = _batch_confirm_block_reason(row, budget_year=budget_year) if reason: conflicts.append({"candidate_id": cid, "reason": reason}); continue try: preview_transaction_candidate(conn, cid, {"account_id": account_id, "category_id": row["proposed_category_id"], "budget_year": budget_year}) except Exception as exc: conflicts.append({"candidate_id": cid, "reason": str(getattr(exc, "detail", str(exc)))}); continue rows.append(row) categories[str(row["proposed_category_name"] or "Review nötig")] = categories.get(str(row["proposed_category_name"] or "Review nötig"), 0) + 1 sources[str(row["source_type"] or "Quelle unbekannt")] = sources.get(str(row["source_type"] or "Quelle unbekannt"), 0) + 1 if row["duplicate_of_transaction_id"]: possible_duplicates += 1 if not rows and conflicts: raise HTTPException(status_code=409, detail={"message": "no confirmable candidates", "conflicts": conflicts, "blockers": _blocker_summary(conflicts)}) preview_id = new_id("batchprev") blockers = _blocker_summary(conflicts) record_audit_event(conn, source="vue_dashboard", action="batch_confirm_candidates_preview", entity_type="budget_transaction_candidate", entity_id=preview_id, new_values={"count": len(rows), "conflicts": len(conflicts), "categories": categories, "sources": sources, "blockers": blockers}, created_by="user") conn.commit() return {"preview_id": preview_id, "count": len(rows), "categories": categories, "sources": sources, "possible_duplicates": possible_duplicates, "conflicts": conflicts, "blockers": blockers, "can_confirm": len(conflicts) == 0 and len(rows) > 0, "requires_explicit_confirm": True} def batch_mark_candidates(conn: Connection, *, candidate_ids: list[str], action: str, account_id: str | None = None, category_id: str | None = None, preview_id: str | None = None, note: str | None = None, budget_year: str = "2026") -> dict[str, Any]: counts: dict[str, Any] = {"updated_count": 0, "confirmed_count": 0, "ignored_count": 0, "reopened_count": 0, "transfer_count": 0, "covered_by_migros_count": 0, "skipped_count": 0, "errors": []} if action == "confirm" and not preview_id: raise HTTPException(status_code=422, detail="batch confirm requires preview_id") for cid in candidate_ids: try: row_for_year = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (cid,)).fetchone() if not row_for_year: raise HTTPException(status_code=404, detail="candidate not found") if str(row_for_year["transaction_date"] or "")[:4] != str(budget_year): raise HTTPException(status_code=409, detail="candidate outside active budget year") if action == "category": if row_for_year["status"] in BLOCKED_BULK_EDIT_STATUSES: raise HTTPException(status_code=409, detail=f"candidate not editable:{row_for_year['status']}") if not category_id: raise HTTPException(status_code=422, detail="category_id required") update_transaction_candidate_category(conn, cid, category_id, note or "Batch category assignment"); counts["updated_count"] += 1 elif action == "confirm": reason = _batch_confirm_block_reason(row_for_year, budget_year=budget_year) if reason: raise HTTPException(status_code=409, detail=reason) confirm_transaction_candidate(conn, cid, {"account_id": account_id, "budget_year": budget_year}); counts["confirmed_count"] += 1 elif action == "ignore": ignore_transaction_candidate(conn, cid, note or "Batch ignore"); counts["ignored_count"] += 1 elif action == "reopen": reopen_transaction_candidate(conn, cid, note or "Batch reopen"); counts["reopened_count"] += 1 elif action == "transfer": mark_candidate_transfer(conn, cid, note or "Batch transfer"); counts["transfer_count"] += 1 elif action == "covered_by_migros": ts = now(); row = conn.execute("SELECT status FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (cid,)).fetchone() if not row: raise HTTPException(status_code=404, detail="candidate not found") conn.execute("UPDATE budget_transaction_candidates SET status='covered_by_migros', classification='covered_by_migros', covered_by_source='migros_csv', requires_review=0, review_reason='user_batch_marked_covered_by_migros', updated_at=? WHERE transaction_candidate_id=?", (ts, cid)) record_audit_event(conn, source="vue_dashboard", action="mark_covered_by_migros", entity_type="budget_transaction_candidate", entity_id=cid, old_values={"status": row["status"]}, new_values={"status": "covered_by_migros"}, created_by="user") counts["covered_by_migros_count"] += 1 else: raise HTTPException(status_code=422, detail="invalid batch action") except Exception as exc: counts["skipped_count"] += 1 detail = getattr(exc, "detail", str(exc)) counts["errors"].append({"candidate_id": cid, "error": str(detail)}) audit_id = record_audit_event(conn, source="vue_dashboard", action="batch_confirm_candidates" if action == "confirm" else f"batch_{action}_candidates", entity_type="budget_transaction_candidate", entity_id=preview_id or "batch", new_values=counts | {"candidate_count": len(candidate_ids)}, created_by="user") conn.commit() return counts | {"audit_id": audit_id} def confirm_candidate_split(conn: Connection, candidate_id: str, payload: dict[str, Any]) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction candidate not found") if row["status"] in {"confirmed", "ignored", "covered_by_migros", "superseded"} | USER_DUPLICATE_STATUSES: raise HTTPException(status_code=409, detail="candidate is not splittable") splits = payload.get("splits") or [] if not payload.get("account_id"): raise HTTPException(status_code=422, detail="account_id required") if not splits: raise HTTPException(status_code=422, detail="splits required") total = sum(Decimal(_decimal_text(s.get("amount_original")) or "0") for s in splits) candidate_amount = Decimal(str(row["amount_original"] or "0")) if total != candidate_amount: raise HTTPException(status_code=422, detail="split sum must match candidate amount") group_id = new_id("splitgrp") ts = now() created_ids = [] conn.execute("DELETE FROM budget_candidate_splits WHERE transaction_candidate_id=?", (candidate_id,)) for idx, split in enumerate(splits, start=1): amount = _decimal_text(split.get("amount_original")) cat = split.get("category_id") if not cat or not conn.execute("SELECT 1 FROM budget_categories WHERE category_id=? AND is_active=1", (cat,)).fetchone(): raise HTTPException(status_code=422, detail="active split category required") result = confirm_budget_transaction(conn, { "account_id": payload["account_id"], "transaction_type": payload.get("transaction_type") or "expense", "transaction_date": row["transaction_date"], "description": f"{row['description']} · Split {idx}", "amount_original": amount, "currency_original": row["currency_original"], "category_id": cat, "source_type": "import_candidate", "source_candidate_id": candidate_id, "notes": f"Split group {group_id}: {split.get('note') or ''}", "tag_names": [split.get("tag")] if split.get("tag") else [], }) created_ids.append(result["entity_id"]) conn.execute("INSERT INTO budget_candidate_splits(split_id, transaction_candidate_id, category_id, amount_original, notes, tag_name, confirmed_transaction_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", (new_id("bcsplit"), candidate_id, cat, amount, split.get("note"), split.get("tag"), result["entity_id"], ts, ts)) primary_confirmed_id = created_ids[0] if created_ids else group_id conn.execute("UPDATE budget_transaction_candidates SET status='confirmed', confirmed_transaction_id=?, confirmed_at=?, confirmed_by='user', updated_at=? WHERE transaction_candidate_id=?", (primary_confirmed_id, ts, ts, candidate_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="candidate_split_confirmed", entity_type="budget_transaction_candidate", entity_id=candidate_id, new_values={"group_id": group_id, "primary_transaction_id": primary_confirmed_id, "transaction_ids": created_ids, "split_count": len(created_ids)}, created_by="user") conn.commit() return {"status": "confirmed", "group_id": primary_confirmed_id, "split_group_id": group_id, "created_transactions": len(created_ids), "transaction_ids": created_ids, "audit_id": audit_id} def create_review_category_for_candidate(conn: Connection, candidate_id: str, payload: dict[str, Any]) -> dict[str, Any]: if not conn.execute("SELECT 1 FROM budget_transaction_candidates WHERE transaction_candidate_id=?", (candidate_id,)).fetchone(): raise HTTPException(status_code=404, detail="transaction candidate not found") name = str(payload.get("name") or "").strip() if not name: raise HTTPException(status_code=422, detail="category name required") ts = now(); category_id = new_id("bcat") conn.execute("INSERT INTO budget_categories(category_id, parent_category_id, name, category_type, color, icon, is_active, sort_order, created_at, updated_at) VALUES (?, ?, ?, ?, NULL, NULL, 1, 999, ?, ?)", (category_id, payload.get("parent_category_id") or None, name, payload.get("category_type") or "expense", ts, ts)) update_transaction_candidate_category(conn, candidate_id, category_id, "Created from review") audit_id = record_audit_event(conn, source="vue_dashboard", action="review_category_created", entity_type="budget_category", entity_id=category_id, new_values={"name": name, "candidate_id": candidate_id}, created_by="user") conn.commit() return {"status": "created", "category_id": category_id, "audit_id": audit_id} def create_budget_rule(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: text = str(payload.get("merchant_contains") or "").strip() if not text: raise HTTPException(status_code=422, detail="merchant_contains required") target_status = payload.get("target_status") or "review" if target_status not in {"auto", "review"}: raise HTTPException(status_code=422, detail="invalid target_status") priority = int(payload.get("priority") or 100) confidence = str(payload.get("confidence") or "0.82") rule_id = new_id("brule"); ts = now() conn.execute("INSERT INTO budget_review_rules(rule_id, merchant_contains, source_type, category_id, target_status, is_active, created_at, updated_at, priority, confidence, notes) VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?)", (rule_id, text, payload.get("source_type") or None, payload.get("category_id") or None, target_status, ts, ts, priority, confidence, payload.get("notes"))) audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_rule_created", entity_type="budget_review_rule", entity_id=rule_id, new_values={"merchant_contains": text, "target_status": target_status, "priority": priority, "confidence": confidence}, created_by="user") conn.commit() return {"status": "created", "rule_id": rule_id, "audit_id": audit_id} def list_budget_rules(conn: Connection, *, include_inactive: bool = False) -> list[dict[str, Any]]: where = "" if include_inactive else "WHERE r.is_active=1" rows = conn.execute(f"SELECT r.*, c.name AS category_name FROM budget_review_rules r LEFT JOIN budget_categories c ON c.category_id=r.category_id {where} ORDER BY r.priority ASC, r.created_at DESC").fetchall() return [row_to_dict(r) | {"is_active": bool(r["is_active"])} for r in rows] def _rule_match_rows(conn: Connection, rule_id: str): rule = conn.execute("SELECT * FROM budget_review_rules WHERE rule_id=?", (rule_id,)).fetchone() if not rule: raise HTTPException(status_code=404, detail="rule not found") clauses = ["status NOT IN ('confirmed','ignored','covered_by_migros','covered_by_source','auto_ignored_duplicate','superseded','reference_2025','archived_reference')", "lower(COALESCE(merchant,'') || ' ' || COALESCE(description,'')) LIKE ?"] params: list[Any] = [f"%{str(rule['merchant_contains']).lower()}%"] if rule["source_type"]: clauses.append("source_type=?"); params.append(rule["source_type"]) return rule, conn.execute("SELECT * FROM budget_transaction_candidates WHERE " + " AND ".join(clauses), params).fetchall() def test_budget_rule(conn: Connection, rule_id: str) -> dict[str, Any]: rule, rows = _rule_match_rows(conn, rule_id) return {"rule_id": rule_id, "match_count": len(rows), "would_set_category": rule["category_id"], "would_set_status": rule["target_status"], "confidence": rule["confidence"], "candidate_ids": [r["transaction_candidate_id"] for r in rows]} def apply_budget_rules_to_open_candidates(conn: Connection, rule_id: str) -> dict[str, Any]: rule, rows = _rule_match_rows(conn, rule_id) if not rule["is_active"]: raise HTTPException(status_code=409, detail="rule inactive") status = "auto_categorized" if rule["target_status"] == "auto" else "needs_review" requires_review = 0 if rule["target_status"] == "auto" else 1 cname = _category_name(conn, rule["category_id"]) ts = now(); updated = 0 for row in rows: conn.execute("UPDATE budget_transaction_candidates SET proposed_category_id=?, proposed_category_name=?, status=?, requires_review=?, rule_id=?, rule_name=?, review_reason='rule_mvp_suggestion_only', confidence=?, updated_at=? WHERE transaction_candidate_id=?", (rule["category_id"], cname, status, requires_review, rule_id, f"Regel: enthält {rule['merchant_contains']}", str(rule["confidence"] or "0.82"), ts, row["transaction_candidate_id"])) updated += 1 audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_rule_applied_to_candidates", entity_type="budget_review_rule", entity_id=rule_id, new_values={"updated_count": updated, "productive_transaction_count": 0}, created_by="user") conn.commit() return {"updated_count": updated, "audit_id": audit_id, "productive_transaction_count": 0} def update_budget_rule(conn: Connection, rule_id: str, payload: dict[str, Any]) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_review_rules WHERE rule_id=?", (rule_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="rule not found") allowed = {"merchant_contains", "source_type", "category_id", "target_status", "priority", "confidence", "notes", "is_active"} updates: dict[str, Any] = {k: payload[k] for k in allowed if k in payload} if "target_status" in updates and updates["target_status"] not in {"auto", "review"}: raise HTTPException(status_code=422, detail="invalid target_status") if "merchant_contains" in updates and not str(updates["merchant_contains"] or "").strip(): raise HTTPException(status_code=422, detail="merchant_contains required") if not updates: raise HTTPException(status_code=422, detail="no editable fields supplied") updates["updated_at"] = now() set_clause = ", ".join([f"{k}=?" for k in updates]) conn.execute(f"UPDATE budget_review_rules SET {set_clause} WHERE rule_id=?", [updates[k] for k in updates] + [rule_id]) audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_rule_updated", entity_type="budget_review_rule", entity_id=rule_id, old_values=row_to_dict(row), new_values=updates, created_by="user") conn.commit() return {"status": "updated", "rule_id": rule_id, "audit_id": audit_id} def prioritize_budget_rule(conn: Connection, rule_id: str, priority: int) -> dict[str, Any]: return update_budget_rule(conn, rule_id, {"priority": int(priority)}) | {"priority": int(priority)} def deactivate_budget_rule(conn: Connection, rule_id: str) -> dict[str, Any]: row = conn.execute("SELECT is_active FROM budget_review_rules WHERE rule_id=?", (rule_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="rule not found") ts = now(); conn.execute("UPDATE budget_review_rules SET is_active=0, updated_at=? WHERE rule_id=?", (ts, rule_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_rule_deactivated", entity_type="budget_review_rule", entity_id=rule_id, old_values={"is_active": row["is_active"]}, new_values={"is_active": 0}, created_by="user") conn.commit(); return {"status": "deactivated", "rule_id": rule_id, "audit_id": audit_id} def create_merchant(conn: Connection, payload: dict[str, Any]) -> dict[str, Any]: display_name = str(payload.get("display_name") or "").strip() if not display_name: raise HTTPException(status_code=422, detail="display_name required") default_category_id = payload.get("default_category_id") or None if default_category_id and not conn.execute("SELECT 1 FROM budget_categories WHERE category_id=? AND is_active=1", (default_category_id,)).fetchone(): raise HTTPException(status_code=422, detail="active default category required") merchant_id = new_id("bmerch") ts = now() normalized = _normalized_merchant_name(display_name) conn.execute("INSERT INTO budget_merchants(merchant_id, display_name, normalized_name, default_category_id, is_active, notes, created_at, updated_at) VALUES (?, ?, ?, ?, 1, ?, ?, ?)", (merchant_id, display_name, normalized, default_category_id, payload.get("notes"), ts, ts)) audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_merchant_created", entity_type="budget_merchant", entity_id=merchant_id, new_values={"display_name": display_name, "default_category_id": default_category_id}, created_by="user") conn.commit() return {"status": "created", "merchant_id": merchant_id, "audit_id": audit_id} def list_merchants(conn: Connection, *, include_inactive: bool = False) -> list[dict[str, Any]]: where = "" if include_inactive else "WHERE m.is_active=1" rows = conn.execute(f""" SELECT m.*, c.name AS default_category_name, (SELECT COUNT(*) FROM budget_merchant_aliases a WHERE a.merchant_id=m.merchant_id AND a.is_active=1) AS alias_count, (SELECT COUNT(*) FROM budget_transaction_candidates t WHERE t.merchant_id=m.merchant_id AND t.status NOT IN ('confirmed','ignored','covered_by_source','covered_by_migros','auto_ignored_duplicate','superseded')) AS open_candidate_count FROM budget_merchants m LEFT JOIN budget_categories c ON c.category_id=m.default_category_id {where} ORDER BY m.is_active DESC, lower(m.display_name) """).fetchall() return [row_to_dict(r) | {"is_active": bool(r["is_active"])} for r in rows] def create_merchant_alias(conn: Connection, merchant_id: str, payload: dict[str, Any]) -> dict[str, Any]: if not conn.execute("SELECT 1 FROM budget_merchants WHERE merchant_id=? AND is_active=1", (merchant_id,)).fetchone(): raise HTTPException(status_code=404, detail="merchant not found") pattern = str(payload.get("pattern") or "").strip() match_type = payload.get("match_type") or "contains" if not pattern: raise HTTPException(status_code=422, detail="pattern required") if match_type not in {"contains", "exact", "regex"}: raise HTTPException(status_code=422, detail="invalid match_type") alias_id = new_id("bmalias") ts = now() conn.execute("INSERT INTO budget_merchant_aliases(alias_id, merchant_id, pattern, match_type, source_type, priority, is_active, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)", (alias_id, merchant_id, pattern, match_type, payload.get("source_type") or None, int(payload.get("priority") or 100), ts, ts)) audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_merchant_alias_created", entity_type="budget_merchant_alias", entity_id=alias_id, new_values={"merchant_id": merchant_id, "pattern": pattern, "match_type": match_type}, created_by="user") conn.commit() return {"status": "created", "alias_id": alias_id, "audit_id": audit_id} def update_merchant_category(conn: Connection, merchant_id: str, category_id: str | None) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_merchants WHERE merchant_id=?", (merchant_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="merchant not found") if category_id and not conn.execute("SELECT 1 FROM budget_categories WHERE category_id=? AND is_active=1", (category_id,)).fetchone(): raise HTTPException(status_code=422, detail="active category required") ts = now() conn.execute("UPDATE budget_merchants SET default_category_id=?, updated_at=? WHERE merchant_id=?", (category_id, ts, merchant_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_merchant_category_updated", entity_type="budget_merchant", entity_id=merchant_id, old_values={"default_category_id": row["default_category_id"]}, new_values={"default_category_id": category_id}, created_by="user") conn.commit() return {"status": "updated", "merchant_id": merchant_id, "audit_id": audit_id} def _open_candidates_for_merchants(conn: Connection): return conn.execute("SELECT * FROM budget_transaction_candidates WHERE status NOT IN ('confirmed','ignored','covered_by_source','covered_by_migros','auto_ignored_duplicate','superseded','reference_2025','archived_reference')").fetchall() def apply_merchant_to_open_candidates(conn: Connection, merchant_id: str | None = None) -> dict[str, Any]: merchant_clause = "AND m.merchant_id=?" if merchant_id else "" params: list[Any] = [merchant_id] if merchant_id else [] aliases = conn.execute(f""" SELECT a.*, m.display_name, m.default_category_id, c.name AS default_category_name FROM budget_merchant_aliases a JOIN budget_merchants m ON m.merchant_id=a.merchant_id AND m.is_active=1 LEFT JOIN budget_categories c ON c.category_id=m.default_category_id WHERE a.is_active=1 {merchant_clause} ORDER BY a.priority ASC, length(a.pattern) DESC """, params).fetchall() updated = 0 ts = now() for cand in _open_candidates_for_merchants(conn): for alias in aliases: if not _merchant_alias_matches(alias, cand): continue status = "auto_categorized" if alias["default_category_id"] else "needs_review" requires_review = 0 if alias["default_category_id"] else 1 conn.execute(""" UPDATE budget_transaction_candidates SET merchant_id=?, merchant_display_name=?, merchant=?, proposed_category_id=COALESCE(?, proposed_category_id), proposed_category_name=COALESCE(?, proposed_category_name), status=?, status_label=?, requires_review=?, review_reason='merchant_alias_match', confidence='0.90', updated_at=? WHERE transaction_candidate_id=? """, (alias["merchant_id"], alias["display_name"], alias["display_name"], alias["default_category_id"], alias["default_category_name"], status, STATUS_LABELS[status], requires_review, ts, cand["transaction_candidate_id"])) updated += 1 break audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_merchant_applied_to_candidates", entity_type="budget_merchant", entity_id=merchant_id or "all", new_values={"updated_count": updated}, created_by="user") conn.commit() return {"updated_count": updated, "audit_id": audit_id} def merge_merchants(conn: Connection, source_merchant_id: str, target_merchant_id: str) -> dict[str, Any]: if source_merchant_id == target_merchant_id: raise HTTPException(status_code=422, detail="source and target must differ") source = conn.execute("SELECT * FROM budget_merchants WHERE merchant_id=?", (source_merchant_id,)).fetchone() target = conn.execute("SELECT * FROM budget_merchants WHERE merchant_id=?", (target_merchant_id,)).fetchone() if not source or not target: raise HTTPException(status_code=404, detail="merchant not found") ts = now() conn.execute("UPDATE budget_merchant_aliases SET merchant_id=?, updated_at=? WHERE merchant_id=?", (target_merchant_id, ts, source_merchant_id)) conn.execute("UPDATE budget_transaction_candidates SET merchant_id=?, merchant_display_name=?, merchant=? WHERE merchant_id=?", (target_merchant_id, target["display_name"], target["display_name"], source_merchant_id)) conn.execute("UPDATE budget_merchants SET is_active=0, updated_at=?, notes=COALESCE(notes,'') || ' merged_into:' || ? WHERE merchant_id=?", (ts, target_merchant_id, source_merchant_id)) audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_merchant_merged", entity_type="budget_merchant", entity_id=target_merchant_id, old_values={"source_merchant_id": source_merchant_id}, new_values={"target_merchant_id": target_merchant_id}, created_by="user") conn.commit() return {"status": "merged", "source_merchant_id": source_merchant_id, "target_merchant_id": target_merchant_id, "audit_id": audit_id} def _similar_merchant(a: str, b: str) -> bool: aa = set(_normalized_merchant_name(a).split()) bb = set(_normalized_merchant_name(b).split()) if not aa or not bb: return False return bool(aa & bb) or _normalized_merchant_name(a) in _normalized_merchant_name(b) or _normalized_merchant_name(b) in _normalized_merchant_name(a) def detect_duplicate_candidates(conn: Connection, *, day_window: int = 2) -> dict[str, Any]: candidates = conn.execute(""" SELECT * FROM budget_transaction_candidates WHERE status NOT IN ('confirmed','ignored','covered_by_source','covered_by_migros','auto_ignored_duplicate','superseded','reference_2025','archived_reference') AND transaction_date IS NOT NULL AND amount_original IS NOT NULL """).fetchall() transactions = conn.execute("SELECT * FROM budget_transactions WHERE status='confirmed' AND transaction_date IS NOT NULL").fetchall() safe = possible = 0 ts = now() for cand in candidates: try: cdate = datetime.fromisoformat(str(cand["transaction_date"])[:10]) camount = Decimal(str(cand["amount_original"])) except Exception: continue cmerchant = str(cand["merchant_display_name"] or cand["merchant"] or cand["description"] or "") for tx in transactions: try: tdate = datetime.fromisoformat(str(tx["transaction_date"])[:10]) tamount = Decimal(str(tx["amount_original"])) except Exception: continue day_diff = abs((cdate - tdate).days) if day_diff > day_window or camount != tamount or str(cand["currency_original"] or "CHF") != str(tx["currency_original"] or "CHF"): continue if not _similar_merchant(cmerchant, str(tx["payee"] or tx["description"] or "")): continue exact_text = _normalized_merchant_name(cmerchant) == _normalized_merchant_name(str(tx["payee"] or tx["description"] or "")) high_conf = Decimal(str(cand["confidence"] or "0")) >= Decimal("0.95") if day_diff == 0 and exact_text and high_conf: conn.execute("UPDATE budget_transaction_candidates SET status='auto_ignored_duplicate', status_label='Automatisch ignoriertes Duplikat', duplicate_of_transaction_id=?, requires_review=0, review_reason='safe_duplicate_existing_transaction', updated_at=? WHERE transaction_candidate_id=?", (tx["budget_transaction_id"], ts, cand["transaction_candidate_id"])) record_audit_event(conn, source="vue_dashboard", action="auto_ignore_duplicate", entity_type="budget_transaction_candidate", entity_id=cand["transaction_candidate_id"], old_values={"status": cand["status"]}, new_values={"status": "auto_ignored_duplicate", "duplicate_of_transaction_id": tx["budget_transaction_id"]}, created_by="system") safe += 1 else: conn.execute("UPDATE budget_transaction_candidates SET status='duplicate_candidate', status_label=?, duplicate_of_transaction_id=?, requires_review=1, review_reason='duplicate_detection_v1', updated_at=? WHERE transaction_candidate_id=?", (STATUS_LABELS["duplicate_candidate"], tx["budget_transaction_id"], ts, cand["transaction_candidate_id"])) possible += 1 break audit_id = record_audit_event(conn, source="vue_dashboard", action="budget_duplicate_detection_v2", entity_type="budget_transaction_candidate", entity_id="duplicate_detection_v2", new_values={"safe_duplicate_count": safe, "possible_duplicate_count": possible}, created_by="system") conn.commit() return {"duplicate_count": safe + possible, "safe_duplicate_count": safe, "possible_duplicate_count": possible, "audit_id": audit_id} def archive_reference_year_candidates(conn: Connection, *, active_year: str = "2026", reference_year: str = "2025") -> dict[str, int | str]: ts = now() rows = conn.execute("SELECT * FROM budget_transaction_candidates WHERE substr(transaction_date,1,4)=? AND status NOT IN ('confirmed','reference_2025','archived_reference','superseded')", (reference_year,)).fetchall() updated = 0 for row in rows: status = f"reference_{reference_year}" if reference_year == "2025" else "archived_reference" conn.execute("UPDATE budget_transaction_candidates SET status=?, requires_review=0, review_reason=?, updated_at=? WHERE transaction_candidate_id=?", (status, f"reference_year_not_active_budget_{active_year}", ts, row["transaction_candidate_id"])) record_audit_event(conn, source="phase110_cleanup", action="archive_reference_year_candidate", entity_type="budget_transaction_candidate", entity_id=row["transaction_candidate_id"], old_values=row_to_dict(row), new_values={"status": status, "active_year": active_year}, created_by="system") updated += 1 conn.commit() return {"updated_count": updated, "active_year": active_year, "reference_year": reference_year} def cleanup_migros_article_candidates(conn: Connection, *, dry_run: bool = True) -> dict[str, int]: where = """ status IN ('pending','needs_review','duplicate') AND (lower(COALESCE(source_file_label,'')) LIKE '%migros%' OR lower(COALESCE(source_file_label,'')) LIKE '%cumulus%') AND COALESCE(source_type,'') NOT IN ('migros_receipt') """ rows = conn.execute(f"SELECT transaction_candidate_id FROM budget_transaction_candidates WHERE {where}").fetchall() count = len(rows) if dry_run: return {"article_candidate_count": count, "superseded_count": 0, "dry_run": 1} ts = now() conn.execute(f"UPDATE budget_transaction_candidates SET status='superseded', review_reason='superseded_by_migros_receipt_aggregation', updated_at=? WHERE {where}", (ts,)) audit_id = record_audit_event(conn, source="budget_phase15_cleanup", action="cleanup_migros_article_candidates", entity_type="budget_transaction_candidate", entity_id="migros_article_candidates", new_values={"article_candidate_count": count, "status": "superseded"}, created_by="agent") conn.commit() return {"article_candidate_count": count, "superseded_count": count, "audit_id": audit_id, "dry_run": 0}