# -*- coding: utf-8 -*- """ Liest eine Rechnungs-PDF des Rechnungstools (Campinghof Bartl) und macht daraus eine Buchung fürs Steuerjournal. Der PDF-Text ist zeilenweise aufgebaut: erst das Label, dann in der nächsten Zeile der Wert. Rechnung an: Familie Peter Meier Musterweg 12 Rechnungsnummer: 2026-238 Rechnungsdatum: 12.07.2026 Zeitraum: 3 Nächte ... Zwischensumme: 87,00 € zzgl. Beherbergungssteuer (5 %): 4,35 € Gesamtbetrag: 91,35 € Nur Seite 1 wird gelesen – Seite 2 ist die 1:1-Kopie fürs Amt. """ import os import re import json import logging from datetime import datetime, date from pypdf import PdfReader from modell import Buchung, STANDARD_SATZ, zahl_aus_text, runde, zerlege_nummer log = logging.getLogger("bst.pdf") # Labels, hinter denen der Wert in der nächsten Zeile steht RE_NUMMER = re.compile(r"^\s*Rechnungsnummer:\s*$", re.I) RE_DATUM = re.compile(r"^\s*Rechnungsdatum:\s*$", re.I) RE_ANREISE = re.compile(r"^\s*Anreise:\s*$", re.I) RE_ABREISE = re.compile(r"^\s*Abreise:\s*$", re.I) RE_ZEITRAUM = re.compile(r"^\s*Zeitraum:\s*$", re.I) RE_KUNDE = re.compile(r"^\s*Rechnung an:\s*$", re.I) RE_ZWISCHEN = re.compile(r"^\s*Zwischensumme:\s*$", re.I) RE_GESAMT = re.compile(r"^\s*Gesamtbetrag:\s*$", re.I) # "zzgl. Beherbergungssteuer (5 %):" – Satz steht in der Klammer RE_AUFSCHLAG = re.compile(r"^\s*zzgl\.\s*(?P.+?)\s*\(\s*(?P[\d.,]+)\s*%\s*\):\s*$", re.I) RE_NAECHTE = re.compile(r"(\d+)") class ParserFehler(Exception): """Die Datei ist keine (lesbare) Rechnung dieses Tools.""" def _text_seite1(pfad: str) -> str: reader = PdfReader(pfad) if not reader.pages: raise ParserFehler("PDF hat keine Seiten") return reader.pages[0].extract_text() or "" def _naechster_wert(zeilen, i: int) -> str: """Erste nicht-leere Zeile nach Index i.""" for z in zeilen[i + 1:]: if z.strip(): return z.strip() return "" def _block_nach(zeilen, i: int, hoechstens: int = 5): """ Die Zeilen des Adressblocks nach 'Rechnung an:' – bis zum nächsten Label (z.B. 'Rechnungsnummer:'). Anrede, Name, Zusatz, Straße, PLZ/Ort. """ block = [] for z in zeilen[i + 1:]: if not z.strip(): continue if z.strip().endswith(":"): # nächstes Label -> Adressblock zu Ende break block.append(z.strip()) if len(block) >= hoechstens: break return block def _datum(text: str) -> date: for fmt in ("%d.%m.%Y", "%d.%m.%y", "%Y-%m-%d"): try: return datetime.strptime(text.strip(), fmt).date() except ValueError: continue raise ParserFehler(f"Datum nicht lesbar: {text!r}") RECHTSFORM = {"gmbh", "ag", "kg", "ug", "gbr", "ohg", "mbh", "e.k.", "ek", "eg", "se", "gmbh & co. kg", "verein", "e.v."} NAMENSZUSATZ = {"van", "von", "de", "der", "den", "di", "du", "le", "la", "el", "al", "ten", "ter", "zu", "vom", "zum", "zur", "mc", "mac", "o'"} # Das Rechnungstool hat ein eigenes Feld "Anrede (optional)" über "Name / Firma". # Steht dort etwas, ist es die erste Zeile unter "Rechnung an:" – und darf NICHT # als Name durchgehen. ANREDE = {"herr", "herrn", "frau", "familie", "fam", "eheleute", "firma", "hr", "fr", "dr", "prof", "mr", "mrs", "ms", "und", "&", "an", "die", "z.h.", "zhd"} def _nur_anrede(zeile: str) -> bool: """True, wenn die Zeile ausschließlich aus Anrede-Wörtern besteht ('Herr', 'Familie').""" teile = str(zeile or "").split() return bool(teile) and all(t.lower().strip(".,") in ANREDE for t in teile) def _ohne_anrede(zeile: str) -> str: """Führende Anrede-Wörter abschneiden: 'Firma Muster GmbH' -> 'Muster GmbH'.""" teile = str(zeile or "").split() while teile and teile[0].lower().strip(".,") in ANREDE: teile.pop(0) return " ".join(teile) def nachname_aus_zeile(zeile: str) -> str: """ 'Familie Peter Meier' -> 'Meier' 'Anna Schmidt-Lorenz' -> 'Schmidt-Lorenz' 'Meier, Peter' -> 'Meier' 'Familie van Dijk' -> 'van Dijk' (Namenszusatz bleibt dran) 'Hentschke Bau GmbH' -> 'Hentschke Bau GmbH' (Firma komplett) 'Firma Muster GmbH' -> 'Muster GmbH' (Anrede vorn fällt weg) """ z = " ".join(_ohne_anrede(zeile).split()) if not z: return "" if "," in z: # "Nachname, Vorname" return z.split(",", 1)[0].strip() teile = z.split() if teile[-1].lower().strip(".,") in RECHTSFORM: return z # Firma – der ganze Name gehört dazu if len(teile) >= 2 and teile[-2].lower().strip(".") in NAMENSZUSATZ: return " ".join(teile[-2:]) # "van Dijk" return teile[-1] def name_aus_block(zeilen) -> str: """ Aus dem Adressblock unter 'Rechnung an:' die Zeile mit Name/Firma holen. Reine Anrede-Zeilen ('Herr', 'Familie') werden übersprungen. """ for z in zeilen: if z.strip() and not _nur_anrede(z): return nachname_aus_zeile(z) return "" def _aus_metadaten(pfad: str): """ Rechnungen des Rechnungstools tragen ihre Kenndaten als JSON in den PDF-Metadaten (Feld "Subject"). Dann muss nichts geraten werden. Gibt None zurück, wenn die PDF das nicht hat (ältere Rechnungen). """ try: info = PdfReader(pfad).metadata or {} roh = info.get("/Subject") or "" if not roh.strip().startswith("{"): return None daten = json.loads(roh) except Exception as e: # noqa: BLE001 - fremde/kaputte Metadaten log.debug("%s: keine lesbaren Metadaten (%s)", os.path.basename(pfad), e) return None if daten.get("quelle") != "rechnungstool" or not daten.get("rechnungsnummer"): return None return daten def parse_pdf(pfad: str, standard_satz: float = STANDARD_SATZ) -> Buchung: """Rechnungs-PDF -> Buchung. Wirft ParserFehler, wenn Pflichtfelder fehlen.""" meta = _aus_metadaten(pfad) if meta: datum = _datum(str(meta.get("datum", ""))) jahr, nummer = zerlege_nummer(meta.get("rechnungsnummer"), datum.year) b = Buchung( datum=datum, rechnungsnummer=nummer, jahr=jahr, nachname=str(meta.get("nachname") or "").strip(), naechte=int(zahl_aus_text(meta.get("naechte")) or 0), gezahlt=runde(meta.get("zwischensumme") or 0), satz=float(meta.get("steuer_satz") or 0) or standard_satz, quelle="pdf", pdf_pfad=os.path.abspath(pfad), ) log.debug("aus Metadaten: %s | %s | %s | %s Nächte | %.2f €", os.path.basename(pfad), b.voll_nummer, b.nachname, b.naechte, b.gezahlt) return b text = _text_seite1(pfad) zeilen = text.splitlines() voll_nr = datum_txt = "" anreise = abreise = "" naechte = 0 kunde = "" zwischensumme = None gesamtbetrag = None steuer_betrag_pdf = None satz = None for i, z in enumerate(zeilen): if RE_NUMMER.match(z): voll_nr = _naechster_wert(zeilen, i) elif RE_DATUM.match(z): datum_txt = _naechster_wert(zeilen, i) elif RE_ANREISE.match(z): anreise = _naechster_wert(zeilen, i) elif RE_ABREISE.match(z): abreise = _naechster_wert(zeilen, i) elif RE_ZEITRAUM.match(z): m = RE_NAECHTE.search(_naechster_wert(zeilen, i)) if m: naechte = int(m.group(1)) elif RE_KUNDE.match(z): kunde = name_aus_block(_block_nach(zeilen, i)) elif RE_ZWISCHEN.match(z) and zwischensumme is None: zwischensumme = zahl_aus_text(_naechster_wert(zeilen, i)) elif RE_GESAMT.match(z) and gesamtbetrag is None: gesamtbetrag = zahl_aus_text(_naechster_wert(zeilen, i)) else: m = RE_AUFSCHLAG.match(z) if m and steuer_betrag_pdf is None: satz = zahl_aus_text(m.group("satz")) steuer_betrag_pdf = zahl_aus_text(_naechster_wert(zeilen, i)) if not voll_nr or not datum_txt: raise ParserFehler("keine Rechnungsnummer/kein Rechnungsdatum gefunden") datum = _datum(datum_txt) jahr, nummer = zerlege_nummer(voll_nr, datum.year) if not nummer: raise ParserFehler(f"Rechnungsnummer nicht lesbar: {voll_nr!r}") # Basis der Steuer = Zwischensumme. Fehlt sie (Rechnung ohne Beherbergungs- # steuer), ist der Gesamtbetrag die Zwischensumme. gezahlt = zwischensumme if zwischensumme is not None else gesamtbetrag if gezahlt is None: raise ParserFehler("weder Zwischensumme noch Gesamtbetrag gefunden") if satz is None: satz = 0.0 if steuer_betrag_pdf is None else standard_satz # Nächte notfalls aus An-/Abreise ableiten if not naechte and anreise and abreise: try: naechte = (_datum(abreise) - _datum(anreise)).days except ParserFehler: naechte = 0 b = Buchung( datum=datum, rechnungsnummer=nummer, jahr=jahr, nachname=kunde, # kommt schon fertig aus name_aus_block() naechte=max(0, naechte), gezahlt=runde(gezahlt), satz=satz, quelle="pdf", pdf_pfad=os.path.abspath(pfad), ) # Gegenprobe: unsere Rechnung gegen den in der PDF ausgewiesenen Betrag if steuer_betrag_pdf is not None and abs(b.steuer - steuer_betrag_pdf) > 0.02: log.warning("%s: Steuer laut PDF %.2f, berechnet %.2f (Basis %.2f, Satz %s%%)", os.path.basename(pfad), steuer_betrag_pdf, b.steuer, b.gezahlt, satz) log.debug("gelesen: %s | %s | %s | %s Nächte | %.2f € | Steuer %.2f €", os.path.basename(pfad), b.voll_nummer, b.nachname, b.naechte, b.gezahlt, b.steuer) return b def scanne_ordner(ordner: str, standard_satz: float = STANDARD_SATZ, rekursiv: bool = True): """ Liest alle PDFs unter `ordner`. Gibt (buchungen, fehler) zurück – fehler = Liste von (pfad, grund). """ buchungen, fehler = [], [] if not os.path.isdir(ordner): return buchungen, [(ordner, "Ordner existiert nicht")] for wurzel, _dirs, dateien in os.walk(ordner): for name in sorted(dateien): if not name.lower().endswith(".pdf"): continue pfad = os.path.join(wurzel, name) try: buchungen.append(parse_pdf(pfad, standard_satz)) except ParserFehler as e: log.info("übersprungen: %s (%s)", name, e) fehler.append((pfad, str(e))) except Exception as e: # noqa: BLE001 - kaputte/verschlüsselte PDF log.warning("Fehler bei %s: %s", name, e) fehler.append((pfad, f"{type(e).__name__}: {e}")) if not rekursiv: break log.info("Scan %s: %d Rechnungen, %d übersprungen", ordner, len(buchungen), len(fehler)) return buchungen, fehler