"""Parser für DKB (Deutsche Kreditbank)-Kontoauszüge. Realer Fixture-Abgleich (Task 7, Schritt 4): Der eigentliche Auszugszeitraum steht nicht als "vom X bis Y"-Satz im Text (dieser Wortlaut kommt nur in einem separaten quartalsweisen Zinsabschluss-Absatz mit abweichendem Zeitraum vor); er wird daher aus den Datumsangaben der beiden Kontostand-Zeilen abgeleitet. Buchungszeilen haben nur eine Datumsspalte, die ohne Leerzeichen direkt am Buchungstext klebt, und der Betrag trägt ein optionales führendes Minus statt eines nachgestellten S/H-Kürzels. Der Anfangssaldo trägt den Zusatz ", Auszug Nr. N", der Endsaldo den Zusatz "um HH:MM Uhr" – beides unterscheidet die echten Saldo-Zeilen von einer abweichend formulierten Zwischensumme in einem quartalsweisen Rechnungsabschluss-Anhang. """ import re from datetime import date from pathlib import Path import pdfplumber from app.parsers.base import (ParsedStatement, ParsedTransaction, ParserError, parse_german_amount, parse_german_date) BANK = "dkb" RE_IBAN = re.compile(r"\b([A-Z]{2}\d{2}(?:\s?\d{4}){4,5}(?:\s?\d{1,2})?)\b") # "Kontostand am 01.06.2020, Auszug Nr. 1 1.234,56" RE_OPENING = re.compile(r"Kontostand\s+am\s+(\d{2}\.\d{2}\.\d{4}),\s*Auszug\s+Nr\.\s*\d+\s+(-?[\d.,]+)", re.I) # "Kontostand am 30.06.2020 um 12:00 Uhr -2.345,67" RE_CLOSING = re.compile(r"Kontostand\s+am\s+(\d{2}\.\d{2}\.\d{4})\s+um\s+\d{2}:\d{2}\s*Uhr\s+(-?[\d.,]+)", re.I) # Buchungszeile: "01.06.2020Überweisung -123,45" (Datum direkt am Text, kein Leerzeichen) RE_TX_HEAD = re.compile(r"^(\d{2}\.\d{2}\.\d{4})(\S.*?)\s+(-?[\d.,]+)$") # Wiederkehrender Seitenfuß (Bank-Impressum) und Folgeseiten-Kopf; rein # strukturelle Muster (keine Kontodaten), damit dieser Block nicht als # Verwendungszweck der jeweils letzten Buchung einer Seite eingesammelt wird. RE_NOISE = re.compile( r"(?:Kreditbank AG" # Impressum-Zeile 1 r"|Taubenstraße" # Impressum-Zeile 2 (Anschrift) r"|^\d{5}\s+Berlin\b" # Impressum-Zeile 3 (PLZ Ort) r"|Handelsregister" # Impressum-Zeile 4 r"|^Ein Unternehmen" # Impressum-Zeile 5 r"|Landesbank" # Impressum-Zeile 6 r"|^Kontoauszug\s+\d+/\d+\s+Seite" # Folgeseiten-Kopf r"|^Girokonto\s+\d" # Folgeseiten-Kopf (Kontozeile) r"|^Datum\s+Erläuterung" # Spaltenüberschrift r")" ) def _is_noise(line: str) -> bool: return bool(RE_NOISE.search(line)) # --- Empfänger/Zweck-Split-Heuristik (D-T5-1/D-T5-2, siehe # .superpowers/sdd/parser-vs-csv-audit.md) -------------------------------- # extras[0] enthält "" ohne # Trenner. Die folgenden Regeln bestimmen, wo der Empfänger endet. _LEGAL_FORM_TOKENS = { "ag", "gmbh", "ev", "eg", "kg", "bv", "sca", "sarl", "se", "ltd", "inc", "ek", "ohg", "gbr", "stiftung", "bank", } _CONNECTOR_TOKENS = {"et", "cie"} _REF_KEYWORDS = {"rechnung", "vertrag", "kundennummer", "kassenzeichen", "rg", "nr"} _REF_KEYWORD_PREFIXES = ("kd.-nr", "kdnr", "beitragsnr") _RE_TOKEN_DIGIT = re.compile(r"\d") _MAX_LEGAL_FORM_SCAN = 7 def _normalize_token(tok: str) -> str: return tok.strip(",;").lower().replace(".", "") def _is_legal_form(tok: str) -> bool: return _normalize_token(tok) in _LEGAL_FORM_TOKENS def _is_connector(tok: str) -> bool: return _normalize_token(tok) in _CONNECTOR_TOKENS def _is_reference_keyword(tok: str) -> bool: norm = _normalize_token(tok) return norm in _REF_KEYWORDS or norm.startswith(_REF_KEYWORD_PREFIXES) def _split_counterparty(line: str) -> tuple[str, str]: """Trennt Empfänger von Zweck-Beginn in der ersten Fortsetzungszeile (extras[0]) einer DKB-Buchung. Gibt (counterparty, purpose_prefix) zurück; purpose_prefix ist der abgeschnittene Rest (kann leer sein). Reihenfolge (D-T5-1): 1. Rechtsform-Grenze (erste ~7 Tokens; "et Cie"-Ketten bis zum letzten Rechtsform-Token in Folge). 2. Referenz-Grenze (erstes Token mit Ziffer oder Referenz-Schlüsselwort, frühestens ab Token 1 -> counterparty nie leer). 3. Kein Schnitt (reine Wort-Folge, semantisch nicht trennbar). D-T5-2: "siehe Anlage"-Zeilen -> counterparty leer, ganze Zeile in den Zweck. """ line = line.strip() if not line: return "", "" if line.lower().startswith("siehe anlage"): return "", line tokens = line.split() n = len(tokens) for i in range(min(n, _MAX_LEGAL_FORM_SCAN)): if not _is_legal_form(tokens[i]): continue last = i j = i + 1 while j < n and (_is_connector(tokens[j]) or _is_legal_form(tokens[j])): if _is_legal_form(tokens[j]): last = j j += 1 cut = last + 1 return " ".join(tokens[:cut]), " ".join(tokens[cut:]) for idx in range(1, n): if _RE_TOKEN_DIGIT.search(tokens[idx]) or _is_reference_keyword(tokens[idx]): return " ".join(tokens[:idx]), " ".join(tokens[idx:]) return line, "" def parse(path: Path) -> ParsedStatement: with pdfplumber.open(path) as pdf: text = "\n".join((page.extract_text() or "") for page in pdf.pages) opening = RE_OPENING.search(text) closing = RE_CLOSING.search(text) if not opening or not closing: raise ParserError("DKB: Anfangs-/Endsaldo nicht gefunden") period_start = parse_german_date(opening.group(1)) period_end = parse_german_date(closing.group(1)) iban_m = RE_IBAN.search(text) txs: list[ParsedTransaction] = [] pending: dict | None = None for line in text.splitlines(): line = line.strip() if not line: continue # Schlusssaldo erreicht: laufende Buchung abschließen und Sammlung beenden # (danach folgen nur noch Summen, Rechnungsabschluss und Hinweise). if RE_CLOSING.search(line): break m = RE_TX_HEAD.match(line) if m: if pending: txs.append(_finish(pending)) booking = parse_german_date(m.group(1)) pending = { "booking": booking, "value": booking, "head": m.group(2).strip(), "amount": parse_german_amount(m.group(3)), "extra": [], } continue if pending and not RE_OPENING.search(line) and not _is_noise(line): pending["extra"].append(line) if pending: txs.append(_finish(pending)) return ParsedStatement(bank=BANK, iban=iban_m.group(1).replace(" ", "") if iban_m else None, period_start=period_start, period_end=period_end, opening_balance=parse_german_amount(opening.group(2)), closing_balance=parse_german_amount(closing.group(2)), transactions=txs) def _finish(p: dict) -> ParsedTransaction: if p["extra"]: counterparty, remainder = _split_counterparty(p["extra"][0]) purpose_parts = [p["head"]] if remainder: purpose_parts.append(remainder) purpose_parts.extend(p["extra"][1:]) else: counterparty = "" purpose_parts = [p["head"]] purpose = " ".join(purpose_parts).strip() return ParsedTransaction(booking_date=p["booking"], value_date=p["value"], amount=p["amount"], purpose=purpose, counterparty=counterparty)