GuidoandClaude Opus 4.8
f375fc5eaa
Backup & Export: - routers/backup: /backup/db (Admin, konsistentes SQLite-Online-Backup), /export/transactions.csv|json und /export/all.json (owner-gescoped). - Einstellungen: Bereich „Backup & Export" mit Download-Buttons. docs/BACKUP.md (Wiederherstellung + LXC-Sicherung). pytest für Export + Admin-Schutz. Import-Verbesserungen (Nutzer-Feedback zum Ablauf): - Klarer Hinweis, dass die Zuordnungstabelle nur Beispielzeilen zeigt und „Vorschau erstellen" die vollständige Prüfliste + Buchen-Button erzeugt. - Zusammenfassung (erkannt/gebucht) und Warnung, wenn keine gültige Zeile erkannt wurde (Hinweis auf Format/Bank-Vorlage). - Fix: Raiffeisen-Preset hat has_header=true (der Export hat eine Kopfzeile). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
393 lines
14 KiB
Python
393 lines
14 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,
|
|
ImportBatch,
|
|
ImportStatus,
|
|
ImportTemplate,
|
|
MatchField,
|
|
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 _apply_rules(rules: list[CategorizationRule], payee: str, purpose: str) -> int | None:
|
|
p = normalize_text(payee)
|
|
z = normalize_text(purpose)
|
|
for rule in rules: # bereits nach Priorität sortiert
|
|
hay = p if rule.match_field == MatchField.payee else z
|
|
if rule.pattern and rule.pattern in hay and rule.category_id:
|
|
return rule.category_id
|
|
return 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] = {}
|
|
for value_norm, acc_id in (
|
|
db.query(AccountIdentifier.value_norm, AccountIdentifier.account_id)
|
|
.join(Account, Account.id == AccountIdentifier.account_id)
|
|
.filter(Account.owner_id == user.id)
|
|
):
|
|
idents[value_norm] = acc_id
|
|
|
|
rules = (
|
|
db.query(CategorizationRule)
|
|
.filter(CategorizationRule.owner_id == user.id)
|
|
.order_by(CategorizationRule.priority.desc(), CategorizationRule.hits.desc())
|
|
.all()
|
|
)
|
|
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
|
|
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:
|
|
category_id = _apply_rules(rules, payee or "", purpose or "")
|
|
|
|
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,
|
|
"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, priority=1, hits=1,
|
|
))
|
|
else:
|
|
existing.category_id = cat
|
|
existing.hits += 1
|
|
|
|
|
|
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)
|
|
_learn_rule(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
|