feat: PDF-Beleg-Parser blockbasiert (mehrzeilig) + Seiten als Bild

- Neuer Parser erkennt mehrzeilige Auszüge (N26 & Co.): Beschreibung → Value
  Date → Buchungsdatum → Betrag stehen auf eigenen Zeilen; Betrags-Zeile schließt
  eine Buchung ab, Kopf-/Fußzeilen (Name/Adresse/eigene IBAN/Seitenzahl/
  Spaltenköpfe) werden gefiltert, Gegen-IBAN und Zweck werden übernommen
- Alter Ein-Zeilen-Heuristik-Parser bleibt als Fallback
- /documents/preview liefert zusätzlich die Seiten als PNG-Data-URLs
  (PyMuPDF get_pixmap, kein poppler nötig)
- Tests: N26-artige mehrzeilige Fixture + Seiten-Rendering (74 grün);
  am echten 34-Seiten-Auszug 207 Buchungen erkannt

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guido
2026-08-20 17:14:12 +02:00
co-authored by Claude Opus 4.8
parent 49b927e2eb
commit 5f88660bcc
3 changed files with 221 additions and 2 deletions
+2
View File
@@ -26,9 +26,11 @@ async def preview(
raise HTTPException(status_code=400, detail=str(exc))
candidates = ocr.parse_transactions(text)
rows = csv_import.detect_rows(db, user, account, candidates)
pages = ocr.render_pages(data, file.filename or "", file.content_type or "")
return {
"rows": rows,
"text": text[:8000],
"count": len(rows),
"filename": file.filename or "",
"pages": pages,
}
+157 -2
View File
@@ -7,6 +7,7 @@ bewusst Best-Effort – das Ergebnis landet in derselben Prüfansicht wie der
CSV-Import und wird dort korrigiert.
"""
import base64
import io
import re
from datetime import datetime
@@ -64,6 +65,8 @@ def extract_text(data: bytes, filename: str, content_type: str) -> str:
DATE_RE = re.compile(r"\b(\d{4}-\d{2}-\d{2}|\d{1,2}[./]\d{1,2}[./]\d{2,4})\b")
# Betrag: entweder mit Tausender-Gruppen (1'234.56 / 1.234,56) oder schlicht (3000.00).
AMOUNT_RE = re.compile(r"[-+]?(?:\d{1,3}(?:['.,\s]\d{3})+|\d+)[.,]\d{2}(?!\d)")
VALUE_DATE_RE = re.compile(r"value\s*date\s*[:\s]\s*(\d{1,2}[./]\d{1,2}[./]\d{2,4}|\d{4}-\d{2}-\d{2})", re.I)
IBAN_RE = re.compile(r"IBAN[:\s]+([A-Z]{2}[A-Z0-9]{8,32})", re.I)
_DATE_FORMATS = ["%Y-%m-%d", "%d.%m.%Y", "%d.%m.%y", "%d/%m/%Y", "%d/%m/%y"]
@@ -93,8 +96,120 @@ def _parse_amount_auto(s: str) -> int | None:
return -cents if neg else cents
def parse_transactions(text: str) -> list[dict]:
"""Findet Zeilen mit Datum + Betrag und macht daraus Kandidaten-Buchungen."""
def _line_amount(line: str) -> int | None:
"""Betrag, falls die ZEILE im Wesentlichen nur ein Betrag ist (evtl. mit
Währungssymbol/-code), z. B. „-97,00€" oder „1'234.56 CHF". Sonst None."""
s = re.sub(r"\s*(€|EUR|CHF|Fr\.?|\$|USD)\s*$", "", line.strip(), flags=re.I).strip()
if AMOUNT_RE.fullmatch(s):
return _parse_amount_auto(s)
return None
def _line_date(line: str):
"""Datum, falls die ZEILE nur aus einem Datum besteht (Buchungsdatum)."""
s = line.strip()
if DATE_RE.fullmatch(s):
return _norm_date(s)
return None
def _filter_noise(raw_lines: list[str]) -> list[str]:
"""Entfernt wiederkehrende Kopf-/Fußzeilen von Kontoauszügen (Name/Adresse/
eigene IBAN im Seitenkopf, „Issued on" + Ausstellungsdatum, Seitenzahl,
Spaltenüberschriften)."""
lines = [ln for ln in raw_lines if ln]
if not lines:
return []
# Halter-Kopf: die Zeilen bis vor „Issued on" (Name, Adresse, eigene IBAN).
holder: set[str] = set()
for ln in lines[:6]:
if ln.lower().startswith("issued on"):
break
holder.add(ln)
heads = {"description", "booking date", "amount", "account activity"}
out: list[str] = []
skip_date = False
for ln in lines:
low = ln.lower()
if skip_date:
skip_date = False
if _line_date(ln): # das Ausstellungsdatum direkt nach „Issued on"
continue
if ln in holder or low in heads:
continue
if low.startswith("issued on"):
skip_date = True
continue
if re.fullmatch(r"\d+\s*/\s*\d+", ln): # Seitenzahl „1 / 34"
continue
out.append(ln)
return out
def _make_candidate(block: list[str], amount: int) -> dict:
"""Baut aus den angesammelten Beschreibungszeilen + Betrag eine Kandidaten-
Buchung. Buchungsdatum = letzte reine Datumszeile (sonst Value-/Erst-Datum)."""
d = None
used: set[int] = set()
for j in range(len(block) - 1, -1, -1):
dd = _line_date(block[j])
if dd:
d, _ = dd, used.add(j)
break
if d is None:
for j, l in enumerate(block):
vm = VALUE_DATE_RE.search(l)
if vm:
d, _ = _norm_date(vm.group(1)), used.add(j)
break
if d is None:
for j, l in enumerate(block):
gm = DATE_RE.search(l)
if gm:
d, _ = _norm_date(gm.group(1)), used.add(j)
break
iban = None
parts: list[str] = []
for j, l in enumerate(block):
if j in used or VALUE_DATE_RE.search(l):
continue
m = IBAN_RE.search(l)
if m and iban is None:
iban = re.sub(r"[^A-Za-z0-9]", "", m.group(1)).upper()
continue
parts.append(l)
payee = parts[0] if parts else None
purpose = " · ".join(parts[1:]) if len(parts) > 1 else None
return {
"date": d.isoformat() if d else None,
"amount_cents": amount,
"payee": payee[:120] if payee else None,
"purpose": purpose[:200] if purpose else None,
"counterparty_iban": iban,
}
def _parse_blocks(text: str) -> list[dict]:
"""Blockweise: jede Zeile, die im Wesentlichen NUR ein Betrag ist, schließt
eine Buchung ab; die Zeilen davor sind Beschreibung/Datum/IBAN. Passt zu
mehrzeiligen Auszügen (N26 & Co.), bei denen Datum und Betrag eigene Zeilen sind."""
lines = _filter_noise([ln.strip() for ln in text.splitlines()])
rows: list[dict] = []
block: list[str] = []
for ln in lines:
amt = _line_amount(ln)
if amt is None or amt == 0:
block.append(ln)
continue
rows.append(_make_candidate(block, amt))
block = []
return rows
def _parse_lines(text: str) -> list[dict]:
"""Fallback: Datum UND Betrag in derselben Zeile (einfache Tabellen)."""
rows: list[dict] = []
for line in text.splitlines():
line = line.strip()
@@ -125,3 +240,43 @@ def parse_transactions(text: str) -> list[dict]:
"counterparty_iban": None,
})
return rows
def parse_transactions(text: str) -> list[dict]:
"""Kandidaten-Buchungen aus dem Beleg-Text. Erst blockbasiert (mehrzeilig),
sonst der einfache Ein-Zeilen-Fallback."""
rows = _parse_blocks(text)
valid = [r for r in rows if r.get("date") and r.get("amount_cents")]
if valid:
return rows
return _parse_lines(text)
# --- Seiten als Bild (zur Kontrolle in der Prüfansicht) --------------------
def render_pages(data: bytes, filename: str, content_type: str,
max_pages: int = 6, dpi: int = 110) -> list[str]:
"""Rendert die Beleg-Seiten als PNG-Data-URLs (PDF via PyMuPDF, ohne poppler).
Bilder werden unverändert als Data-URL zurückgegeben. Best-Effort."""
name = (filename or "").lower()
out: list[str] = []
is_pdf = name.endswith(".pdf") or content_type == "application/pdf"
try:
if is_pdf:
if fitz is None:
return []
with fitz.open(stream=data, filetype="pdf") as doc:
for k, page in enumerate(doc):
if k >= max_pages:
break
pix = page.get_pixmap(dpi=dpi)
b64 = base64.b64encode(pix.tobytes("png")).decode("ascii")
out.append(f"data:image/png;base64,{b64}")
else:
mime = content_type or "image/png"
b64 = base64.b64encode(data).decode("ascii")
out.append(f"data:{mime};base64,{b64}")
except Exception: # noqa: BLE001 – Vorschau ist optional
return out
return out
+62
View File
@@ -38,6 +38,68 @@ def test_parse_transactions_heuristic():
assert rows[1]["amount_cents"] == 300000 # Schweizer Tausender-Apostroph
N26_TEXT = "\n".join([
"GUIDO OLIVIER SCHMIT",
"Riedstrasse 18, 8908 Hedingen",
"IBAN: DE54100110012138281295 • BIC: NTSBDEB1XXX",
"Issued on",
"20.08.2026",
"1 / 2",
"Description",
"Booking Date",
"Amount",
"La Flamme",
"Mastercard • Bars & Restaurants",
"Value Date 31.01.2026",
"01.02.2026",
"-97,00€",
"Guido Schmit",
"Income",
"IBAN: LT293250031382885526 • BIC: REVOLT21XXX",
"Gesendet von Revolut",
"Value Date 01.02.2026",
"01.02.2026",
"+43,00€",
"Noemi Konig",
"Outgoing Transfers",
"IBAN: DE65120300001207628981 • BIC: BYLADEM1001",
"WG Miete Guido Schmit",
"Value Date 02.02.2026",
"02.02.2026",
"-288,00€",
])
def test_parse_transactions_n26_blocks():
from app.services import ocr
rows = ocr.parse_transactions(N26_TEXT)
assert len(rows) == 3
assert rows[0]["date"] == "2026-02-01"
assert rows[0]["amount_cents"] == -9700
assert rows[0]["payee"] == "La Flamme"
assert rows[1]["amount_cents"] == 4300 # Eingang (+)
assert rows[1]["payee"] == "Guido Schmit"
assert rows[1]["counterparty_iban"] == "LT293250031382885526"
assert rows[2]["amount_cents"] == -28800
assert rows[2]["payee"] == "Noemi Konig"
# Gegen-IBAN der Miete – NICHT die eigene Kopf-IBAN (DE54…).
assert rows[2]["counterparty_iban"] == "DE65120300001207628981"
def test_render_pages_pdf():
from app.services import ocr
doc = fitz.open()
doc.new_page()
pages = ocr.render_pages(doc.tobytes(), "beleg.pdf", "application/pdf")
assert len(pages) == 1
assert pages[0].startswith("data:image/png;base64,")
def test_pdf_document_preview(admin):
acc = _account(admin)
pdf = _make_pdf()