b2d933cb7c
Sein Fund: "an der storno verweist du auf die alte rechnung, muss aber
lauten neu zu und neue rechnungs nummer. heisst kette ist so: rechnung
storniert weist auf die storno nummer, storno verweist auf neue rechnung.
wer soll denn sonst wissen wo er suchen muss."
Er hat recht: eine Zeile, die nur zurueckzeigt, laesst den Leser stehen.
Jetzt sagt jede Zeile, wo es WEITERGEHT:
Rechnung -> "storniert mit 2026-901"
Storno -> "neue Rechnung 2026-902" (Fall A)
-> "neu abgerechnet mit 2026-343" (Fall B, Doppelung)
-> "Storno zu 2026-900" (Fall C / alte Belege ohne Angabe)
neue R. -> "Neuausstellung"
Dafuer wandert die Folgenummer aus den PDF-Kenndaten bis ins Journal:
Buchung.folge_nummer + folge_art (neu | vorhanden | ""), gelesen aus
korrektur_nummer/ersatz_vorhanden/nur_storno, zwei neue DB-Spalten, alte
Journale bauen sich wie gehabt selbst um.
An seinen ECHTEN Belegen geprueft: Storno 2026-455 -> folge 2026-369
(vorhanden), 2026-456 -> 2026-343 (vorhanden). Die Angabe steht in den
Kenndaten schon drin, sie wurde bisher nur nicht gelesen.
Fenster und Blatt zeigen dieselben Worte. Erklaerungs-PDF um eine
Kettentabelle ergaenzt. Pruefkette gruen (storno_verrechnet inkl.
September-Blatt, doppelte_leistung, nur_storno).
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017bEsoFUk16DfnNA7MjY36H
361 lines
13 KiB
Python
361 lines
13 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
|
||
|
||
|
||
# Blaetter, die nichts abrechnen und deshalb nicht gebucht werden.
|
||
# "proforma" steht aus der ersten Fassung des Angebots-Blattes noch mit drin.
|
||
KEINE_BUCHUNG = ("berichtigung", "angebot", "proforma")
|
||
|
||
|
||
def berichtigung_daten(pfad: str):
|
||
"""Kenndaten eines Berichtigungsblatts (§ 31 Abs. 5 UStDV), sonst None.
|
||
|
||
Das Blatt bucht nichts - es reicht die Angaben zum Empfaenger nach. Genau
|
||
die will das Journal aber uebernehmen: steht auf der Rechnung noch der
|
||
falsche Name, muss der Amtsbericht den berichtigten zeigen.
|
||
"""
|
||
kopf = _aus_metadaten(pfad)
|
||
if not kopf or kopf.get("art") != "berichtigung":
|
||
return None
|
||
return kopf
|
||
|
||
|
||
def keine_buchung(pfad: str) -> bool:
|
||
"""True fuer Blaetter, die keine Uebernachtung abrechnen.
|
||
|
||
Ein Berichtigungsdokument (§ 31 Abs. 5 UStDV) reicht nur fehlende
|
||
Empfaengerangaben nach. Es traegt in den Kenndaten die Nummer DER RECHNUNG,
|
||
die es berichtigt, und lauter Nullen. Wer es einliest, ueberschreibt damit
|
||
die echte Buchung - im Amtsbericht fehlt die Uebernachtung dann.
|
||
"""
|
||
if berichtigung_daten(pfad) is not None:
|
||
return True
|
||
# Das Angebot ist eine unverbindliche Vorab-Aufstellung, kein Umsatz:
|
||
# gemeldet wird eine Uebernachtung erst, wenn sie abgerechnet ist.
|
||
kopf = _aus_metadaten(pfad)
|
||
if kopf and kopf.get("art") in KEINE_BUCHUNG:
|
||
return True
|
||
# Zweite Sicherung ueber den Dateinamen: die Kenndaten sind die erste
|
||
# Quelle (sie ueberleben auch ein Umbenennen), aber fehlen sie einmal,
|
||
# sagt der Name es auch - das Rechnungstool schreibt "Angebot_..." und
|
||
# legt sie in den Unterordner "Angebote".
|
||
name = os.path.basename(pfad).lower()
|
||
ordner = os.path.basename(os.path.dirname(os.path.abspath(pfad))).lower()
|
||
return name.startswith("angebot_") or ordner == "angebote"
|
||
|
||
|
||
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),
|
||
art=str(meta.get("art") or "rechnung"),
|
||
storno_zu=str(meta.get("storno_zu") or ""),
|
||
storno_datum=str(meta.get("storno_datum") or ""),
|
||
vorgang=str(meta.get("vorgang") or ""),
|
||
folge_nummer=str(meta.get("korrektur_nummer") or ""),
|
||
folge_art=("" if meta.get("nur_storno")
|
||
else "vorhanden" if meta.get("ersatz_vorhanden")
|
||
else "neu" if meta.get("korrektur_nummer") else ""),
|
||
)
|
||
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)
|
||
if keine_buchung(pfad):
|
||
continue
|
||
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
|