GuidoandClaude Opus 4.8
204a24d0df
Regeln: - Schlagwörter mit optionaler Betragsbedingung (z. B. „SV = 5.00 → Restaurant"); Betrags-Regeln gewinnen vor betragslosen - Transfer-Schlagwörter (Text → Transfer) auf der Konten-Seite - Konto-Identifikator-Art „Text": erkennt das Zielkonto eines Transfers aus dem Buchungstext (längster passender Text gewinnt) - Migration 0007, Endpunkte, Tests (Betrag/Transfer/Konflikt), gesamt 30 grün Optik: - App-Icon: Münzhaufen (aus Nutzer-SVG) weiß auf grünem Grund, klein gehalten - Farbschema auf pastelliges Grün (Akzent hell/dunkel + Favicon) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
484 lines
18 KiB
Python
484 lines
18 KiB
Python
"""CSV-Import: Datei parsen, Vorschau mit Erkennung (Duplikate, interne Transfers,
|
||
Auto-Kategorisierung), und die eigentliche Verbuchung inkl. Batch für Undo.
|
||
"""
|
||
|
||
import csv
|
||
import hashlib
|
||
import io
|
||
import json
|
||
from datetime import date, datetime
|
||
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import (
|
||
Account,
|
||
AccountIdentifier,
|
||
AmountMode,
|
||
BudgetTreatment,
|
||
CategorizationRule,
|
||
Category,
|
||
CategoryKind,
|
||
IdentifierKind,
|
||
ImportBatch,
|
||
ImportStatus,
|
||
ImportTemplate,
|
||
MatchField,
|
||
RuleSource,
|
||
SAVINGS_ACCOUNT_TYPES,
|
||
Transaction,
|
||
TxnType,
|
||
User,
|
||
)
|
||
from . import currency
|
||
|
||
|
||
# Vorgefertigte Bank-Vorlagen (aus echten Export-Beispielen). Erscheinen im
|
||
# Import als auswählbare Presets und füllen die Format-/Spalten-Einstellungen.
|
||
PRESETS: list[dict] = [
|
||
{
|
||
"name": "Raiffeisen (CH)",
|
||
"delimiter": ";",
|
||
"decimal_sep": ".",
|
||
"date_format": "%Y-%m-%d %H:%M:%S.%f",
|
||
"encoding": "utf-8",
|
||
"has_header": True,
|
||
"skip_rows": 0,
|
||
"amount_mode": "signed",
|
||
# Spalten: 0=eigene IBAN, 1=Datum, 2=Text, 3=Betrag, 4=Saldo, 5=Valuta.
|
||
"column_map": {"date": 1, "amount": 3, "payee": 2,
|
||
"purpose": None, "counterparty_iban": None},
|
||
},
|
||
{
|
||
"name": "Revolut",
|
||
"delimiter": ",",
|
||
"decimal_sep": ".",
|
||
"date_format": "%Y-%m-%d %H:%M:%S",
|
||
"encoding": "utf-8",
|
||
"has_header": True,
|
||
"skip_rows": 0,
|
||
"amount_mode": "signed",
|
||
# Spalten: 0=Typ, 1=Produkt, 2=Start, 3=Abgeschlossen, 4=Beschreibung,
|
||
# 5=Betrag, 6=Gebühr, 7=Währung, 8=Status, 9=Saldo.
|
||
"column_map": {"date": 3, "amount": 5, "payee": 4,
|
||
"purpose": None, "counterparty_iban": None},
|
||
},
|
||
]
|
||
|
||
|
||
# --- Kleine Parser-Helfer --------------------------------------------------
|
||
|
||
|
||
def normalize_text(s: str | None) -> str:
|
||
return (s or "").strip().lower()
|
||
|
||
|
||
def normalize_iban(s: str | None) -> str:
|
||
return "".join(ch for ch in (s or "").upper() if ch.isalnum())
|
||
|
||
|
||
def parse_amount(raw: str | None, decimal_sep: str) -> int | None:
|
||
"""Betragstext -> Cent (mit Vorzeichen). Erkennt Tausender-Trenner und Klammern."""
|
||
s = str(raw or "").strip()
|
||
if not s:
|
||
return None
|
||
neg = ("-" in s) or (s.startswith("(") and s.endswith(")"))
|
||
thousands = "." if decimal_sep == "," else ","
|
||
s2 = s.replace(thousands, "").replace(decimal_sep, ".")
|
||
digits = "".join(ch for ch in s2 if ch.isdigit() or ch == ".")
|
||
if digits in ("", "."):
|
||
return None
|
||
try:
|
||
val = Decimal(digits)
|
||
except InvalidOperation:
|
||
return None
|
||
cents = int((val * 100).to_integral_value(rounding=ROUND_HALF_UP))
|
||
return -cents if neg else cents
|
||
|
||
|
||
def parse_date(raw: str | None, fmt: str) -> date | None:
|
||
s = str(raw or "").strip()
|
||
if not s:
|
||
return None
|
||
try:
|
||
return datetime.strptime(s, fmt).date()
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def decode_bytes(data: bytes, encoding: str) -> str:
|
||
for enc in (encoding, "utf-8-sig", "utf-8", "latin-1"):
|
||
try:
|
||
return data.decode(enc)
|
||
except (UnicodeDecodeError, LookupError):
|
||
continue
|
||
return data.decode("utf-8", errors="replace")
|
||
|
||
|
||
def read_rows(data: bytes, template: ImportTemplate) -> list[list[str]]:
|
||
text = decode_bytes(data, template.encoding)
|
||
delim = (template.delimiter or ";")[0]
|
||
reader = csv.reader(io.StringIO(text), delimiter=delim)
|
||
rows = [r for r in reader if any((c or "").strip() for c in r)]
|
||
start = template.skip_rows + (1 if template.has_header else 0)
|
||
return rows[start:]
|
||
|
||
|
||
def preview_raw(data: bytes, template: ImportTemplate, limit: int = 8) -> dict:
|
||
"""Rohe erste Zeilen (zum Spalten-Zuordnen im Template-Editor)."""
|
||
text = decode_bytes(data, template.encoding)
|
||
delim = (template.delimiter or ";")[0]
|
||
reader = csv.reader(io.StringIO(text), delimiter=delim)
|
||
rows = [r for r in reader if any((c or "").strip() for c in r)]
|
||
header = None
|
||
body_start = template.skip_rows
|
||
if template.has_header and len(rows) > body_start:
|
||
header = rows[body_start]
|
||
body_start += 1
|
||
return {"header": header, "rows": rows[body_start: body_start + limit]}
|
||
|
||
|
||
def import_hash(account_id: int, d: date | None, amount_cents: int | None, payee: str | None) -> str:
|
||
key = f"{account_id}|{d}|{amount_cents}|{normalize_text(payee)}"
|
||
return hashlib.sha256(key.encode("utf-8")).hexdigest()
|
||
|
||
|
||
# --- Vorschau --------------------------------------------------------------
|
||
|
||
|
||
def _rule_sort_key(rule: CategorizationRule) -> tuple:
|
||
"""Reihenfolge der Regelanwendung: Betrags-Regeln (spezifischer) zuerst, dann
|
||
längstes Schlagwort, dann manuelle vor gelernten, dann Priorität/Treffer."""
|
||
return (
|
||
1 if rule.amount_cents is not None else 0,
|
||
len(rule.pattern or ""),
|
||
1 if rule.source == RuleSource.manual else 0,
|
||
rule.priority,
|
||
rule.hits,
|
||
)
|
||
|
||
|
||
def _pattern_hit(rule: CategorizationRule, p: str, z: str) -> bool:
|
||
if rule.match_field == MatchField.payee:
|
||
return rule.pattern in p
|
||
if rule.match_field == MatchField.purpose:
|
||
return rule.pattern in z
|
||
return rule.pattern in p or rule.pattern in z # any: Empfänger ODER Zweck
|
||
|
||
|
||
def _match_rule(
|
||
rules: list[CategorizationRule],
|
||
payee: str,
|
||
purpose: str,
|
||
amount: int | None,
|
||
allowed_kind: CategoryKind | None,
|
||
cat_kind: dict[int, CategoryKind],
|
||
) -> CategorizationRule | None:
|
||
"""Erste passende Regel (Muster + optionale Betragsbedingung). Kategorie-Regeln
|
||
greifen nur, wenn die Kategorie-Art (Einnahme/Ausgabe) zur Buchung passt;
|
||
Transfer-Regeln kennen keine Art."""
|
||
p = normalize_text(payee)
|
||
z = normalize_text(purpose)
|
||
absamt = abs(amount) if amount is not None else None
|
||
for rule in rules: # bereits sortiert (Betrag/längstes/manuell zuerst)
|
||
if not rule.pattern:
|
||
continue
|
||
if rule.amount_cents is not None and (absamt is None or absamt != rule.amount_cents):
|
||
continue
|
||
if rule.is_transfer:
|
||
if _pattern_hit(rule, p, z):
|
||
return rule
|
||
continue
|
||
if not rule.category_id:
|
||
continue
|
||
if allowed_kind is not None and cat_kind.get(rule.category_id) != allowed_kind:
|
||
continue
|
||
if _pattern_hit(rule, p, z):
|
||
return rule
|
||
return None
|
||
|
||
|
||
def _guess_target_account(text_idents: list[tuple[str, int]], payee: str, purpose: str, source_id: int) -> int | None:
|
||
"""Zielkonto eines Transfers aus dem Buchungstext raten: längster passender
|
||
Erkennungs-Text je Konto gewinnt."""
|
||
hay = f"{normalize_text(payee)} {normalize_text(purpose)}"
|
||
best: tuple[int, int] | None = None # (Länge, Konto-ID)
|
||
for val, acc_id in text_idents:
|
||
if val and acc_id != source_id and val in hay and (best is None or len(val) > best[0]):
|
||
best = (len(val), acc_id)
|
||
return best[1] if best else None
|
||
|
||
|
||
def parse_file_to_candidates(data: bytes, template: ImportTemplate) -> list[dict]:
|
||
"""Zerlegt die Datei anhand des Templates in rohe Kandidaten-Zeilen."""
|
||
cmap = json.loads(template.column_map or "{}")
|
||
|
||
def col(field: str) -> int | None:
|
||
idx = cmap.get(field)
|
||
return idx if isinstance(idx, int) and idx >= 0 else None
|
||
|
||
out: list[dict] = []
|
||
for r in read_rows(data, template):
|
||
def cell(field: str) -> str:
|
||
c = col(field)
|
||
return r[c].strip() if c is not None and c < len(r) else ""
|
||
|
||
d = parse_date(cell("date"), template.date_format)
|
||
if template.amount_mode == AmountMode.debit_credit:
|
||
debit = parse_amount(cell("debit"), template.decimal_sep)
|
||
credit = parse_amount(cell("credit"), template.decimal_sep)
|
||
amount = (credit or 0) - (abs(debit) if debit else 0)
|
||
elif template.amount_mode == AmountMode.abs_expense:
|
||
a = parse_amount(cell("amount"), template.decimal_sep)
|
||
amount = -abs(a) if a is not None else None
|
||
else:
|
||
amount = parse_amount(cell("amount"), template.decimal_sep)
|
||
|
||
out.append({
|
||
"date": d.isoformat() if d else None,
|
||
"amount_cents": amount,
|
||
"payee": cell("payee") or None,
|
||
"purpose": cell("purpose") or None,
|
||
"counterparty_iban": normalize_iban(cell("counterparty_iban")) or None,
|
||
})
|
||
return out
|
||
|
||
|
||
def detect_rows(db: Session, user: User, account: Account, candidates: list[dict]) -> list[dict]:
|
||
"""Reichert Kandidaten an: Typ, Auto-Kategorie, interner Transfer, Duplikat.
|
||
|
||
Wird von CSV- UND Beleg-Import genutzt (gleiche Prüf-/Buchungslogik)."""
|
||
idents: dict[str, int] = {} # IBAN/Nummer -> Konto (für Transfer via IBAN)
|
||
text_idents: list[tuple[str, int]] = [] # freie Erkennungs-Texte -> Konto
|
||
for value_norm, acc_id, kind in (
|
||
db.query(AccountIdentifier.value_norm, AccountIdentifier.account_id, AccountIdentifier.kind)
|
||
.join(Account, Account.id == AccountIdentifier.account_id)
|
||
.filter(Account.owner_id == user.id)
|
||
):
|
||
if kind == IdentifierKind.text:
|
||
text_idents.append((value_norm, acc_id))
|
||
else:
|
||
idents[value_norm] = acc_id
|
||
|
||
rules = db.query(CategorizationRule).filter(CategorizationRule.owner_id == user.id).all()
|
||
rules.sort(key=_rule_sort_key, reverse=True) # längstes/manuell zuerst
|
||
cat_kind: dict[int, CategoryKind] = {
|
||
cid: kind
|
||
for cid, kind in db.query(Category.id, Category.type).filter(Category.owner_id == user.id)
|
||
}
|
||
existing = {
|
||
h[0]
|
||
for h in db.query(Transaction.import_hash).filter(
|
||
Transaction.owner_id == user.id, Transaction.import_hash.isnot(None)
|
||
)
|
||
}
|
||
seen: set[str] = set()
|
||
|
||
out: list[dict] = []
|
||
for i, cand in enumerate(candidates):
|
||
d_iso = cand.get("date")
|
||
amount = cand.get("amount_cents")
|
||
payee = cand.get("payee")
|
||
purpose = cand.get("purpose")
|
||
iban = cand.get("counterparty_iban")
|
||
d = date.fromisoformat(d_iso) if d_iso else None
|
||
|
||
valid = d is not None and amount is not None and amount != 0
|
||
ttype = "expense" if (amount or 0) < 0 else "income"
|
||
category_id: int | None = None
|
||
matched_rule_id: int | None = None
|
||
to_account_id: int | None = None
|
||
|
||
other = idents.get(iban) if iban else None
|
||
if other is not None and other != account.id:
|
||
ttype = "transfer"
|
||
to_account_id = other
|
||
else:
|
||
kind = CategoryKind.expense if ttype == "expense" else CategoryKind.income
|
||
rule = _match_rule(rules, payee or "", purpose or "", amount, kind, cat_kind)
|
||
if rule is not None:
|
||
matched_rule_id = rule.id
|
||
if rule.is_transfer:
|
||
ttype = "transfer"
|
||
to_account_id = _guess_target_account(text_idents, payee or "", purpose or "", account.id)
|
||
else:
|
||
category_id = rule.category_id
|
||
|
||
h = import_hash(account.id, d, amount, payee)
|
||
is_dup = h in existing or h in seen
|
||
seen.add(h)
|
||
|
||
out.append({
|
||
"index": i,
|
||
"date": d.isoformat() if d else None,
|
||
"amount_cents": amount,
|
||
"payee": payee or None,
|
||
"purpose": purpose or None,
|
||
"counterparty_iban": iban or None,
|
||
"valid": valid,
|
||
"type": ttype,
|
||
"category_id": category_id,
|
||
"matched_rule_id": matched_rule_id,
|
||
"to_account_id": to_account_id,
|
||
"import_hash": h,
|
||
"is_duplicate": is_dup,
|
||
"include": valid and not is_dup,
|
||
})
|
||
return out
|
||
|
||
|
||
def build_preview(db: Session, user: User, account: Account, template: ImportTemplate, data: bytes) -> list[dict]:
|
||
return detect_rows(db, user, account, parse_file_to_candidates(data, template))
|
||
|
||
|
||
# --- Verbuchen -------------------------------------------------------------
|
||
|
||
|
||
def _make_transaction(db: Session, user: User, account: Account, accounts: dict[int, Account], row: dict, batch_id: int) -> Transaction | None:
|
||
d = date.fromisoformat(row["date"]) if row.get("date") else None
|
||
amount = row.get("amount_cents")
|
||
if d is None or not amount:
|
||
return None
|
||
ttype = row.get("type") or ("expense" if amount < 0 else "income")
|
||
payee = row.get("payee")
|
||
purpose = row.get("purpose")
|
||
notes = None
|
||
|
||
if ttype == "transfer":
|
||
other_id = row.get("to_account_id")
|
||
other = accounts.get(other_id) if other_id else None
|
||
if other is None:
|
||
ttype = "expense" if amount < 0 else "income" # Fallback ohne Zielkonto
|
||
|
||
if ttype == "transfer":
|
||
other = accounts[row["to_account_id"]]
|
||
mag = abs(amount)
|
||
if amount < 0:
|
||
from_acc, to_acc = account, other
|
||
from_amount = mag # in Kontowährung (bekannt aus dem Auszug)
|
||
else:
|
||
from_acc, to_acc = other, account
|
||
# from-Betrag in der Währung des anderen Kontos schätzen.
|
||
conv = currency.convert(db, mag, account.currency, from_acc.currency, d)
|
||
from_amount = conv if conv is not None else mag
|
||
to_amount = mag if amount < 0 else mag
|
||
# Konsistente to_amount-Berechnung:
|
||
if from_acc.currency == to_acc.currency:
|
||
to_amount = from_amount
|
||
else:
|
||
conv = currency.convert(db, from_amount, from_acc.currency, to_acc.currency, d)
|
||
to_amount = conv if conv is not None else from_amount
|
||
# Sparen/Depot-Ziel budgetieren (überschreibbar über row["budget_treatment"]).
|
||
treatment = row.get("budget_treatment")
|
||
budget_cat = row.get("budget_category_id")
|
||
txn = Transaction(
|
||
owner_id=user.id, type=TxnType.transfer, date=d,
|
||
account_id=from_acc.id, to_account_id=to_acc.id,
|
||
amount_cents=from_amount, to_amount_cents=to_amount,
|
||
amount_eur_cents=currency.compute_amount_eur_cents(db, from_amount, from_acc.currency, d),
|
||
budget_treatment=BudgetTreatment(treatment) if treatment in ("budgeted", "neutral") else None,
|
||
budget_category_id=budget_cat,
|
||
description=purpose, payee=payee, notes=notes,
|
||
import_batch_id=batch_id, import_hash=row.get("import_hash"),
|
||
)
|
||
return txn
|
||
|
||
mag = abs(amount)
|
||
txn = Transaction(
|
||
owner_id=user.id,
|
||
type=TxnType.expense if ttype == "expense" else TxnType.income,
|
||
date=d, account_id=account.id, amount_cents=mag,
|
||
amount_eur_cents=currency.compute_amount_eur_cents(db, mag, account.currency, d),
|
||
category_id=row.get("category_id"),
|
||
payee=payee, description=purpose, notes=notes,
|
||
import_batch_id=batch_id, import_hash=row.get("import_hash"),
|
||
)
|
||
return txn
|
||
|
||
|
||
def _learn_rule(db: Session, user: User, row: dict) -> None:
|
||
"""Merkt sich Empfänger -> Kategorie für künftige Importe."""
|
||
cat = row.get("category_id")
|
||
payee = row.get("payee")
|
||
if not cat or not payee or row.get("type") == "transfer":
|
||
return
|
||
pattern = normalize_text(payee)[:255]
|
||
if not pattern:
|
||
return
|
||
existing = (
|
||
db.query(CategorizationRule)
|
||
.filter(
|
||
CategorizationRule.owner_id == user.id,
|
||
CategorizationRule.match_field == MatchField.payee,
|
||
CategorizationRule.pattern == pattern,
|
||
)
|
||
.first()
|
||
)
|
||
if existing is None:
|
||
db.add(CategorizationRule(
|
||
owner_id=user.id, match_field=MatchField.payee, pattern=pattern,
|
||
category_id=cat, source=RuleSource.learned, priority=1, hits=1,
|
||
))
|
||
# Sichtbar machen (Session ist autoflush=False), damit dasselbe Muster ein
|
||
# zweites Mal im selben Import die vorhandene Regel findet statt zu doppeln.
|
||
db.flush()
|
||
else:
|
||
existing.category_id = cat
|
||
existing.hits += 1
|
||
|
||
|
||
def _track_and_learn(db: Session, user: User, row: dict) -> None:
|
||
"""Zählt für die zutreffende Regel Anwendung/Korrektur (Basis fürs Flaggen von
|
||
Schlagwörtern) und lernt anschließend Empfänger -> Kategorie weiter."""
|
||
rid = row.get("matched_rule_id")
|
||
final_cat = row.get("category_id")
|
||
matched_manual_kept = False
|
||
if rid:
|
||
rule = db.get(CategorizationRule, rid)
|
||
if rule is not None and rule.owner_id == user.id:
|
||
rule.applied_count = (rule.applied_count or 0) + 1
|
||
if final_cat != rule.category_id:
|
||
rule.overridden_count = (rule.overridden_count or 0) + 1
|
||
elif rule.source == RuleSource.manual:
|
||
matched_manual_kept = True # Schlagwort greift schon – nicht doppelt lernen
|
||
if not matched_manual_kept:
|
||
_learn_rule(db, user, row)
|
||
|
||
|
||
def commit_import(db: Session, user: User, account: Account, template_id: int | None, filename: str, rows: list[dict]) -> ImportBatch:
|
||
accounts = {a.id: a for a in db.query(Account).filter(Account.owner_id == user.id)}
|
||
batch = ImportBatch(
|
||
owner_id=user.id, account_id=account.id, template_id=template_id,
|
||
filename=(filename or "")[:255], row_count=len(rows),
|
||
imported_count=0, status=ImportStatus.committed,
|
||
)
|
||
db.add(batch)
|
||
db.flush()
|
||
|
||
imported = 0
|
||
for row in rows:
|
||
if not row.get("include", True):
|
||
continue
|
||
txn = _make_transaction(db, user, account, accounts, row, batch.id)
|
||
if txn is None:
|
||
continue
|
||
db.add(txn)
|
||
_track_and_learn(db, user, row)
|
||
imported += 1
|
||
|
||
batch.imported_count = imported
|
||
db.commit()
|
||
db.refresh(batch)
|
||
return batch
|
||
|
||
|
||
def undo_batch(db: Session, user: User, batch: ImportBatch) -> int:
|
||
n = (
|
||
db.query(Transaction)
|
||
.filter(Transaction.owner_id == user.id, Transaction.import_batch_id == batch.id)
|
||
.delete()
|
||
)
|
||
batch.status = ImportStatus.undone
|
||
db.commit()
|
||
return n
|