feat: Phase 5 – PDF/Bild-Beleg-Import (lokal, Tesseract-OCR)

Backend:
- services/ocr: PDF-Text via PyMuPDF, OCR (Tesseract) für Scans/Bilder,
  heuristischer Parser (Datum + Betrag je Zeile, CH/DE/US-Beträge).
- routers/documents: /documents/preview (Upload -> Text -> Kandidaten ->
  gemeinsame Prüf-/Buchungslogik des CSV-Imports). csv_import refaktoriert:
  detect_rows() von CSV und Beleg gemeinsam genutzt.
- requirements: PyMuPDF, pytesseract, pillow; install.sh installiert tesseract-ocr
  (+ deu). pytest: Heuristik-Parser und PDF-Vorschau.

Frontend:
- Import-Seite: CSV/PDF-Umschalter; PDF-Modus mit Beleg-Auslesen, extrahiertem
  Text zur Kontrolle und Best-Effort-Hinweis; gleiche Prüfansicht + Buchen.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guido
2026-08-19 20:25:32 +02:00
co-authored by Claude Opus 4.8
parent ddf4f1f5ed
commit 7461fddb7a
13 changed files with 353 additions and 41 deletions
+48 -29
View File
@@ -153,16 +153,45 @@ def _apply_rules(rules: list[CategorizationRule], payee: str, purpose: str) -> i
return None
def build_preview(db: Session, user: User, account: Account, template: ImportTemplate, data: bytes) -> list[dict]:
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
rows = read_rows(data, template)
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 ""
# Identifikatoren aller Konten des Nutzers -> Konto-ID (für Transfer-Erkennung).
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)
@@ -177,39 +206,25 @@ def build_preview(db: Session, user: User, account: Account, template: ImportTem
.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_in_file: set[str] = set()
seen: set[str] = set()
out: list[dict] = []
for i, r in enumerate(rows):
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)
payee = cell("payee")
purpose = cell("purpose")
iban = normalize_iban(cell("counterparty_iban"))
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 = "transfer" if False else ("expense" if (amount or 0) < 0 else "income")
ttype = "expense" if (amount or 0) < 0 else "income"
category_id: int | None = None
to_account_id: int | None = None
@@ -218,11 +233,11 @@ def build_preview(db: Session, user: User, account: Account, template: ImportTem
ttype = "transfer"
to_account_id = other
else:
category_id = _apply_rules(rules, payee, purpose)
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_in_file
seen_in_file.add(h)
is_dup = h in existing or h in seen
seen.add(h)
out.append({
"index": i,
@@ -242,6 +257,10 @@ def build_preview(db: Session, user: User, account: Account, template: ImportTem
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 -------------------------------------------------------------