Der Parser nahm die erste Zeile unter "Rechnung an:" als Namen. Steht im Rechnungstool eine Anrede, war das "Herr" statt "Mustermann". - Neue Rechnungen: Kenndaten kommen aus den PDF-Metadaten (JSON), inkl. getrenntem Nachnamen. Nichts wird mehr geraten. - Alte Rechnungen ohne Metadaten: der Adressblock wird gesammelt, reine Anrede-Zeilen (Herr/Frau/Familie/Firma) werden uebersprungen, fuehrende Anreden abgeschnitten. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
309 lines
11 KiB
Python
309 lines
11 KiB
Python
# -*- 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<name>.+?)\s*\(\s*(?P<satz>[\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
|