from __future__ import annotations from decimal import Decimal from sqlite3 import Connection from fastapi import HTTPException from jarvis_finance.audit.log import record_audit_event from jarvis_finance.services.budget_common import ( new_id, now, require_decimal_text, row_to_dict, validate_currency, ) TX_TYPES = {"income", "expense", "transfer", "refund", "fee", "adjustment", "reversal"} def refund_link_is_valid( conn: Connection, origin_id: str | None, refund_amount_chf: str | None, *, exclude_transaction_id: str | None = None, ) -> bool: if not origin_id or refund_amount_chf is None: return False origin = conn.execute( """SELECT transaction_type,amount_chf,amount_original,currency_original FROM budget_transactions WHERE budget_transaction_id=? AND status='confirmed'""", (origin_id,), ).fetchone() if not origin or origin["transaction_type"] not in {"expense", "fee"}: return False origin_chf = origin["amount_chf"] if origin_chf is None and str(origin["currency_original"] or "").upper() == "CHF": origin_chf = origin["amount_original"] if origin_chf is None: return False rows = conn.execute( """SELECT budget_transaction_id,amount_chf,amount_original,currency_original FROM budget_transactions WHERE status='confirmed' AND transaction_type='refund' AND COALESCE(reversal_of_transaction_id, CASE WHEN json_valid(notes) THEN json_extract(notes,'$.reversal_of_transaction_id') END)=?""", (origin_id,), ).fetchall() used = Decimal("0") for row in rows: if exclude_transaction_id and row["budget_transaction_id"] == exclude_transaction_id: continue amount = row["amount_chf"] if amount is None and str(row["currency_original"] or "").upper() == "CHF": amount = row["amount_original"] if amount is not None: used += abs(Decimal(str(amount))) return used + abs(Decimal(str(refund_amount_chf))) <= abs(Decimal(str(origin_chf))) def _validate_transaction_link(conn: Connection, tx: dict) -> None: origin_id = tx.get("reversal_of_transaction_id") if origin_id and tx["transaction_type"] not in {"refund", "reversal"}: raise HTTPException(status_code=422, detail="reversal link is only valid for refund or reversal") if tx["transaction_type"] == "refund" and origin_id and not refund_link_is_valid( conn, str(origin_id), tx.get("amount_chf"), exclude_transaction_id=str(tx["budget_transaction_id"]), ): raise HTTPException(status_code=422, detail="refund link must reference an eligible expense within its remaining CHF amount") def _fx_fields(conn: Connection, currency: str, amount_text: str) -> tuple[str, str | None, str | None]: cur = validate_currency(currency) if cur == "CHF": return "not_needed", "1", amount_text row = conn.execute( "SELECT rate FROM fx_rates WHERE base_currency=? AND quote_currency='CHF' ORDER BY rate_date DESC LIMIT 1", (cur,), ).fetchone() if not row or row["rate"] in (None, ""): return "missing", None, None rate = Decimal(str(row["rate"])) return "ok", format(rate, "f"), format(Decimal(amount_text) * rate, "f") def _normalise_tx_payload(conn: Connection, payload: dict, *, force_type: str | None = None) -> dict: account_id = str(payload.get("account_id") or "").strip() if not conn.execute("SELECT 1 FROM budget_accounts WHERE budget_account_id=? AND is_active=1", (account_id,)).fetchone(): raise HTTPException(status_code=422, detail="active budget account required") tx_type = force_type or str(payload.get("transaction_type") or "expense").strip().lower() if tx_type not in TX_TYPES: raise HTTPException(status_code=422, detail="invalid transaction_type") description = str(payload.get("description") or "").strip() if not description: raise HTTPException(status_code=422, detail="description is required") amount = require_decimal_text(str(payload.get("amount_original") or ""), "amount_original") currency = validate_currency(payload.get("currency_original") or "CHF") fx_status, fx_rate, amount_chf = _fx_fields(conn, currency, amount) if payload.get("fx_rate_to_chf"): fx_rate = require_decimal_text(str(payload["fx_rate_to_chf"]), "fx_rate_to_chf") amount_chf = format(Decimal(amount) * Decimal(fx_rate), "f") fx_status = "not_needed" if currency == "CHF" else "manual_override" return { "budget_transaction_id": payload.get("budget_transaction_id") or new_id("btx"), "account_id": account_id, "transaction_type": tx_type, "transaction_date": str(payload.get("transaction_date") or now()[:10]), "booking_date": payload.get("booking_date") or None, "description": description, "payee": payload.get("payee") or None, "merchant_id": payload.get("merchant_id") or None, "amount_original": amount, "currency_original": currency, "fx_rate_to_chf": fx_rate, "amount_chf": amount_chf, "fx_status": fx_status, "category_id": payload.get("category_id") or None, "source_type": payload.get("source_type") or "manual", "source_candidate_id": payload.get("source_candidate_id") or None, "notes": payload.get("notes") or None, "tag_names": payload.get("tag_names") or [], "reversal_of_transaction_id": payload.get("reversal_of_transaction_id") or None, } def create_budget_transaction_preview(conn: Connection, payload: dict) -> dict: tx = _normalise_tx_payload(conn, payload) _validate_transaction_link(conn, tx) warnings = [] if tx["fx_status"] == "missing": warnings.append("FX fehlt: CHF-Wert bleibt leer, Buchung kann bewusst unvollständig gespeichert werden.") if not tx["category_id"] and tx["transaction_type"] in {"income", "expense", "refund", "fee"}: warnings.append("Kategorie fehlt: Buchung bleibt unkategorisiert.") return {"preview_id": new_id("preview"), "summary": f"{tx['transaction_type']} · {tx['currency_original']} {tx['amount_original']} · {tx['description']}", "fx_status": tx["fx_status"], "warnings": warnings, "payload": tx} def _insert_tags(conn: Connection, tx_id: str, tag_names: list[str]) -> None: for name in tag_names: row = conn.execute("SELECT tag_id FROM budget_tags WHERE name=?", (name,)).fetchone() if row: conn.execute("INSERT OR IGNORE INTO budget_transaction_tags(budget_transaction_id, tag_id) VALUES (?, ?)", (tx_id, row["tag_id"])) def _confirm_budget_transaction(conn: Connection, payload: dict) -> dict: tx = _normalise_tx_payload(conn, payload) _validate_transaction_link(conn, tx) if tx["category_id"] and not conn.execute("SELECT 1 FROM budget_categories WHERE category_id=?", (tx["category_id"],)).fetchone(): raise HTTPException(status_code=422, detail="category not found") ts = now() conn.execute( """ INSERT INTO budget_transactions( budget_transaction_id, account_id, transaction_type, transaction_date, booking_date, description, payee, merchant_id, amount_original, currency_original, fx_rate_to_chf, amount_chf, fx_status, category_id, status, source_type, notes, created_at, updated_at, reversal_of_transaction_id, source_candidate_id ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'confirmed', ?, ?, ?, ?, ?, ?) """, (tx["budget_transaction_id"], tx["account_id"], tx["transaction_type"], tx["transaction_date"], tx["booking_date"], tx["description"], tx["payee"], tx["merchant_id"], tx["amount_original"], tx["currency_original"], tx["fx_rate_to_chf"], tx["amount_chf"], tx["fx_status"], tx["category_id"], tx["source_type"], tx["notes"], ts, ts, tx["reversal_of_transaction_id"], tx["source_candidate_id"]), ) _insert_tags(conn, tx["budget_transaction_id"], tx["tag_names"]) audit_id = record_audit_event(conn, source="vue_dashboard", action="confirm", entity_type="budget_transaction", entity_id=tx["budget_transaction_id"], new_values=tx, user_text_note=tx.get("notes"), created_by="user") conn.commit() return {"status": "confirmed", "entity_id": tx["budget_transaction_id"], "audit_id": audit_id, "message": "Budgetbuchung gespeichert"} def confirm_budget_transaction(conn: Connection, payload: dict) -> dict: started_transaction = not conn.in_transaction if started_transaction: conn.execute("BEGIN IMMEDIATE") try: return _confirm_budget_transaction(conn, payload) except Exception: if started_transaction and conn.in_transaction: conn.rollback() raise def reverse_budget_transaction_preview(conn: Connection, transaction_id: str, payload: dict | None = None) -> dict: row = conn.execute("SELECT * FROM budget_transactions WHERE budget_transaction_id=? AND status='confirmed'", (transaction_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="confirmed transaction not found") reason = (payload or {}).get("reason") or "Reversal" reversal = row_to_dict(row) | {"budget_transaction_id": new_id("btx"), "transaction_type": "reversal", "description": f"Reversal: {row['description']}", "notes": reason, "reversal_of_transaction_id": transaction_id, "tag_names": []} return {"preview_id": new_id("preview"), "summary": f"Reversal · {row['description']}", "warnings": [], "payload": reversal} def confirm_reverse_budget_transaction(conn: Connection, transaction_id: str, payload: dict) -> dict: result = confirm_budget_transaction(conn, payload | {"reversal_of_transaction_id": transaction_id, "transaction_type": "reversal"}) ts = now() conn.execute("UPDATE budget_transactions SET status='reversed', updated_at=? WHERE budget_transaction_id=?", (ts, transaction_id)) record_audit_event(conn, source="vue_dashboard", action="reverse", entity_type="budget_transaction", entity_id=transaction_id, new_values={"status": "reversed", "reversal_transaction_id": result["entity_id"]}, user_text_note=payload.get("notes"), created_by="user") conn.commit() return result def create_transfer_preview(conn: Connection, payload: dict) -> dict: from_account_id = str(payload.get("from_account_id") or "").strip() to_account_id = str(payload.get("to_account_id") or "").strip() if from_account_id == to_account_id: raise HTTPException(status_code=422, detail="transfer accounts must differ") for account_id in (from_account_id, to_account_id): if not conn.execute("SELECT 1 FROM budget_accounts WHERE budget_account_id=? AND is_active=1", (account_id,)).fetchone(): raise HTTPException(status_code=422, detail="active transfer accounts required") amount = require_decimal_text(str(payload.get("amount_original") or ""), "amount_original") currency = validate_currency(payload.get("currency_original") or "CHF") fx_status, fx_rate, amount_chf = _fx_fields(conn, currency, amount) transfer = { "transfer_id": payload.get("transfer_id") or new_id("btrf"), "from_account_id": from_account_id, "to_account_id": to_account_id, "amount_original": amount, "currency_original": currency, "fx_rate_to_chf": fx_rate, "amount_chf": amount_chf, "transaction_date": str(payload.get("transaction_date") or now()[:10]), "description": str(payload.get("description") or "Transfer"), "notes": payload.get("notes") or None, "transfer_type": str(payload.get("transfer_type") or "internal_transfer"), "fx_status": fx_status, } return {"preview_id": new_id("preview"), "summary": f"Transfer · {currency} {amount}", "fx_status": fx_status, "warnings": [] if fx_status != "missing" else ["FX fehlt"], "payload": transfer} def confirm_transfer(conn: Connection, payload: dict) -> dict: transfer = create_transfer_preview(conn, payload)["payload"] out_payload = {"account_id": transfer["from_account_id"], "transaction_type": "transfer", "transaction_date": transfer["transaction_date"], "description": transfer["description"], "amount_original": transfer["amount_original"], "currency_original": transfer["currency_original"], "fx_rate_to_chf": transfer.get("fx_rate_to_chf"), "notes": transfer.get("notes")} in_payload = out_payload | {"account_id": transfer["to_account_id"]} out_result = confirm_budget_transaction(conn, out_payload) in_result = confirm_budget_transaction(conn, in_payload) ts = now() conn.execute( "INSERT INTO budget_transfers(transfer_id, from_transaction_id, to_transaction_id, from_account_id, to_account_id, amount_original, currency_original, fx_rate_to_chf, notes, created_at, transfer_type) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (transfer["transfer_id"], out_result["entity_id"], in_result["entity_id"], transfer["from_account_id"], transfer["to_account_id"], transfer["amount_original"], transfer["currency_original"], transfer.get("fx_rate_to_chf"), transfer.get("notes"), ts, transfer.get("transfer_type") or "internal_transfer"), ) audit_id = record_audit_event(conn, source="vue_dashboard", action="confirm", entity_type="budget_transfer", entity_id=transfer["transfer_id"], new_values=transfer, created_by="user") conn.commit() return {"status": "confirmed", "entity_id": transfer["transfer_id"], "audit_id": audit_id, "message": "Transfer gespeichert"} def update_budget_transaction(conn: Connection, transaction_id: str, payload: dict) -> dict: row = conn.execute("SELECT * FROM budget_transactions WHERE budget_transaction_id=?", (transaction_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="transaction not found") forbidden_amount_fields = {"amount_original", "amount_chf", "currency_original", "fx_rate_to_chf"} & set(payload) if forbidden_amount_fields: raise HTTPException(status_code=422, detail="amount corrections require adjustment or reversal") old = row_to_dict(row) category_id = payload.get("category_id", row["category_id"]) if category_id in ("", "uncategorized"): category_id = None 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 not found") description = str(payload.get("description") or row["description"]).strip() payee = payload.get("payee", row["payee"]) merchant_id = payload.get("merchant_id", row["merchant_id"]) notes = payload.get("notes", row["notes"]) ts = now() conn.execute( "UPDATE budget_transactions SET category_id=?, description=?, payee=?, merchant_id=?, notes=?, updated_at=? WHERE budget_transaction_id=?", (category_id, description, payee, merchant_id, notes, ts, transaction_id), ) if "tag_names" in payload: conn.execute("DELETE FROM budget_transaction_tags WHERE budget_transaction_id=?", (transaction_id,)) _insert_tags(conn, transaction_id, payload.get("tag_names") or []) new_values = {"category_id": category_id, "description": description, "payee": payee, "merchant_id": merchant_id, "notes": notes, "tag_names": payload.get("tag_names")} audit_id = record_audit_event(conn, source="vue_dashboard", action="update", entity_type="budget_transaction", entity_id=transaction_id, old_values=old, new_values=new_values, user_text_note=payload.get("change_note"), created_by="user") conn.commit() return {"status": "updated", "entity_id": transaction_id, "audit_id": audit_id, "message": "Buchung aktualisiert"} def list_budget_transactions(conn: Connection, *, month: str | None = None, year: str | None = "2026", account_id: str | None = None, category_id: str | None = None, transaction_type: str | None = None, merchant: str | None = None, tag: str | None = None, status: str | None = None, source_type: str | None = None, search: str | None = None, date_from: str | None = None, date_to: str | None = None) -> list[dict]: clauses = ["1=1"] params: list[str] = [] if date_from: clauses.append("t.transaction_date>=?") params.append(date_from) if date_to: clauses.append("t.transaction_date<=?") params.append(date_to) if month: clauses.append("substr(t.transaction_date,1,7)=?") params.append(month) elif year: clauses.append("substr(t.transaction_date,1,4)=?") params.append(year) if account_id: clauses.append("t.account_id=?") params.append(account_id) if category_id: if category_id in {"__uncategorized", "uncategorized"}: clauses.append("t.category_id IS NULL") else: clauses.append("t.category_id=?") params.append(category_id) if transaction_type: clauses.append("t.transaction_type=?") params.append(transaction_type) if merchant: clauses.append("(lower(COALESCE(t.payee,'')) LIKE lower(?) OR lower(t.description) LIKE lower(?))") params.extend([f"%{merchant}%", f"%{merchant}%"]) if search: clauses.append("(lower(COALESCE(t.payee,'')) LIKE lower(?) OR lower(t.description) LIKE lower(?) OR lower(COALESCE(c.name,'')) LIKE lower(?))") params.extend([f"%{search}%", f"%{search}%", f"%{search}%"]) if source_type: clauses.append("t.source_type=?") params.append(source_type) if status: clauses.append("t.status=?") params.append(status) if tag: clauses.append("EXISTS (SELECT 1 FROM budget_transaction_tags xt JOIN budget_tags xg ON xg.tag_id=xt.tag_id WHERE xt.budget_transaction_id=t.budget_transaction_id AND lower(xg.name)=lower(?))") params.append(tag) rows = conn.execute( f""" SELECT t.*, a.name AS account_name, c.name AS category_name, c.icon AS category_icon, c.color AS category_color, COALESCE(t.payee, t.description, t.source_type, 'Quelle') AS merchant_display_name, substr(t.transaction_date, 1, 10) AS transaction_date_group, COALESCE(li.line_item_count, 0) AS line_item_count, GROUP_CONCAT(DISTINCT tag.name) AS tag_names_csv FROM budget_transactions t JOIN budget_accounts a ON a.budget_account_id=t.account_id LEFT JOIN budget_categories c ON c.category_id=t.category_id LEFT JOIN budget_transaction_tags tt ON tt.budget_transaction_id=t.budget_transaction_id LEFT JOIN budget_tags tag ON tag.tag_id=tt.tag_id AND tag.is_active=1 LEFT JOIN ( SELECT transaction_candidate_id, COUNT(*) AS line_item_count FROM budget_import_line_items GROUP BY transaction_candidate_id ) li ON li.transaction_candidate_id=t.source_candidate_id WHERE {' AND '.join(clauses)} GROUP BY t.budget_transaction_id ORDER BY t.transaction_date DESC, t.created_at DESC LIMIT 200 """, params, ).fetchall() result = [] from jarvis_finance.services.household_review_corrections import transaction_token for r in rows: item = row_to_dict(r) item["transaction_token"] = transaction_token(str(item["budget_transaction_id"])) csv = str(item.pop("tag_names_csv") or "") item["tag_names"] = [x for x in csv.split(",") if x] result.append(item) return result def list_budget_transfers(conn: Connection, *, year: str | None = "2026", account_id: str | None = None, transfer_type: str | None = None) -> list[dict]: clauses = ["1=1"] params: list[str] = [] if year: clauses.append("substr(COALESCE(out_tx.transaction_date, in_tx.transaction_date),1,4)=?") params.append(year) if account_id: clauses.append("(tr.from_account_id=? OR tr.to_account_id=?)") params.extend([account_id, account_id]) if transfer_type: clauses.append("COALESCE(tr.transfer_type, 'internal_transfer')=?") params.append(transfer_type) rows = conn.execute( f""" SELECT tr.*, COALESCE(tr.transfer_type, 'internal_transfer') AS transfer_type, out_tx.transaction_date AS transaction_date, out_tx.description AS description, out_tx.status AS status, from_acc.name AS from_account_name, to_acc.name AS to_account_name FROM budget_transfers tr JOIN budget_transactions out_tx ON out_tx.budget_transaction_id=tr.from_transaction_id JOIN budget_transactions in_tx ON in_tx.budget_transaction_id=tr.to_transaction_id JOIN budget_accounts from_acc ON from_acc.budget_account_id=tr.from_account_id JOIN budget_accounts to_acc ON to_acc.budget_account_id=tr.to_account_id WHERE {' AND '.join(clauses)} ORDER BY out_tx.transaction_date DESC, tr.created_at DESC LIMIT 200 """, params, ).fetchall() return [row_to_dict(r) for r in rows]