tobias
d8b12da36d
Das Backend liegt jetzt als custom_components/audi_dashboard/ vor - eine normale Home-Assistant-Integration mit Config-Flow, einer sensor-Plattform und 18 Diensten. Damit ist die App über HACS installierbar; bis das Repo auf GitHub gespiegelt ist (HACS spricht ausschließlich mit GitHub), installiert homeassistant/installationspaket/install.ps1 denselben Ordner ohne HACS. Fünf Installationsschritte entfallen ersatzlos: der pyscript:-Block, der panel_custom:-Block, das Kopieren der Oberfläche nach www/, das langlebige Zugriffstoken (der Verlauf wird direkt über die recorder-API gelesen) und "pip install pypdf" (steht in manifest.json). Das Fahrzeugprofil legt die Integration beim ersten Start aus ihrer Vorlage an. Drei alte Schwächen sind dabei mit erledigt: - Die Nutzlast landet nicht mehr in der Recorder-Datenbank (_unrecorded_attributes - das kann nur eine echte Entität). - Eine laufende Fahrt überlebt einen Neustart (Store statt Arbeitsspeicher); fiel sie während eines Ausfalls ins Ende, schließt nach_neustart_fortsetzen() sie beim letzten aufgezeichneten Zeitpunkt. - Sensor-Zuordnungen wirken sofort - die Zustandsbeobachter werden neu gebunden, der Neustart-Hinweis und der Neustart-Dienst sind weg. Namensvertrag geändert, beide Oberflächen mitgezogen: pyscript.audi_dashboard_x -> sensor.audi_dashboard_x, pyscript.audi_dashboard_y -> audi_dashboard.y. Eine Companion-App vom alten Stand findet nach dem Umstieg nichts mehr und muss neu gebaut werden; das Panel liegt in der Integration und kann nicht driften. Der selbstgebaute Updater entfällt - HACS ist die Update-Mechanik, die Home Assistant kennt. Die Versionierung schrumpft auf eine Quelle: manifest.json. Geprüft am laufenden Testcontainer (Container byteweise identisch mit dem Repo): alle 18 Dienste, Panel, Config-Entry neu laden, Historienimport, echter Shell-Beleg in-process, Neuinstallation im Wegwerf-Container blank mit automatisch nachinstalliertem pypdf. Companion-App: tsc sauber, 112/112 Tests, beide Rauchtests gegen das laufende Backend grün. Belegparser 8/8. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
467 lines
20 KiB
Python
467 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Parser für Shell-Tankbelege (PDF), aufgerufen von belegverarbeitung.py.
|
|
|
|
Diese Datei fehlte beim Bau im Projektordner (siehe Kopfkommentar in
|
|
pyscript/belegverarbeitung.py und README.md, Abschnitt "Bekannte Lücken").
|
|
Sie ist hier gegen echte Shell-eReceipts neu geschrieben und gegen deren
|
|
tatsächlich von pypdf extrahierten Text geprüft - nicht gegen eine
|
|
angenommene Struktur. Regressionstest mit zehn realen Belegen aus vier
|
|
Stationen liegt unter tests/test_shell_beleg_parser.py (Aufruf: python3
|
|
tests/test_shell_beleg_parser.py aus diesem Ordner). Die Belege
|
|
unterscheiden sich u. a. in der Markenzeile ("SHELL STATION" vs.
|
|
"Shell Station" vs. "Shell-Station") und der Rabattbezeichnung
|
|
("V-Power Smart Deal" vs. "ClubSmartRabatt") - deshalb ist die
|
|
Stations- und Rabatterkennung unten bewusst nicht an deren genauen
|
|
Wortlaut gebunden, sondern an feste Beleg-Struktur-Marker (Obj.-Nr.,
|
|
Betragsmuster vor "EUR #A").
|
|
|
|
Abhängigkeit: pypdf (reine Textextraktion, der Beleg ist ein normales
|
|
Text-PDF, kein Scan/Bild - kein OCR nötig). Läuft als eigener Prozess über
|
|
`python3 shell_beleg_parser.py <pfad>` (system-python3 im Container, nicht
|
|
die pyscript-Sandbox - siehe _parser_aufrufen() in belegverarbeitung.py),
|
|
deshalb reicht `pip install pypdf` einmalig im selben Python, das
|
|
`python3` in configuration.yaml/Docker auch sonst benutzt.
|
|
|
|
Kommandozeilen-Schnittstelle wie von belegverarbeitung.py angenommen:
|
|
Ein Pfad zu einer PDF-Datei als einziges Argument, JSON-Objekt auf stdout
|
|
bei Erfolg, Exit-Code ungleich 0 und Fehlertext auf stderr bei Misserfolg
|
|
(z. B. wenn der Beleg keiner der erkannten Vorlagen entspricht).
|
|
|
|
Aufbau eines Shell-eReceipts (aus dem echten Beispiel, Reihenfolge fix):
|
|
SHELL STATION
|
|
<Betreibername, z.B. "A. Zrenner GmbH">
|
|
<Straße>
|
|
<PLZ Ort>
|
|
Obj.-Nr.: <Stations-ID> Tel. ...
|
|
Beleg-Nr. <receipt_no> <TT.MM.JJ> <HH:MM>
|
|
*<article_no> <Produktname> <Listenbetrag> EUR #A*
|
|
*Zp <Zapfsäule> <Liter> l <Listenpreis/l> EUR/l # *
|
|
[<Rabattname> -<Rabattbetrag> EUR #A -- optional, nur mit SmartDeal
|
|
<Rabatt/l> EUR/Liter -- optional]
|
|
Gesamtbetrag <Endbetrag> EUR
|
|
Typ Netto Mwst Brutto
|
|
A:<Mwst-Satz>% <Netto> <Mwst> <Brutto>
|
|
...
|
|
Mobile Payment
|
|
<TT.MM.JJJJ> <HH:MM:SS> -- Zahlungszeitstempel
|
|
...
|
|
(Seite 2) POS:...-...-... TSE-Beleg-Nr: <tse_beleg_nr>
|
|
Start/Ende: <TT.MM.JJJJ HH:MM:SS> / ...
|
|
"""
|
|
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
|
|
try:
|
|
import pypdf
|
|
except ImportError:
|
|
print(
|
|
"pypdf fehlt. Einmalig installieren mit: pip install pypdf "
|
|
"(im selben Python, das diesen Parser ausführt, siehe Kopfkommentar).",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
def _text_aus_pdf(pdf_pfad):
|
|
leser = pypdf.PdfReader(pdf_pfad)
|
|
return "\n".join(seite.extract_text() or "" for seite in leser.pages)
|
|
|
|
|
|
def _de_zahl(text):
|
|
""""103,06" -> 103.06 / "2,399" -> 2.399. Kein Tausenderpunkt bei
|
|
Tankbeträgen dieser Größenordnung zu erwarten, trotzdem robust: Punkte
|
|
vor dem Umwandeln entfernen, nicht nur das Komma ersetzen."""
|
|
if text is None:
|
|
return None
|
|
return float(text.strip().replace(".", "").replace(",", "."))
|
|
|
|
|
|
def _iso(tag, monat, jahr, stunde, minute, sekunde="0"):
|
|
jahr = int(jahr)
|
|
if jahr < 100:
|
|
jahr += 2000
|
|
return datetime.datetime(
|
|
jahr, int(monat), int(tag), int(stunde), int(minute), int(sekunde)
|
|
).isoformat()
|
|
|
|
|
|
_ZAHL = r"[\d.,]+"
|
|
|
|
# Echte Geldangabe: Nachkommastellen sind Pflicht, damit im stationsunabhängigen
|
|
# Parser (_parsen_generisch) keine Artikel-/Belegnummer als Betrag durchgeht.
|
|
_GELD = r"\d{1,3}(?:[ .]\d{3})*,\d{2}"
|
|
_GESAMT_WOERTER = ("GESAMT", "ABSOLUT", "ENDBETRAG", "ZUZAHLEN", "TOTAL", "SUMME")
|
|
# Steuer- und Nettozeilen tragen dieselben Schlüsselwörter ("MWST GESAMT"),
|
|
# nennen aber nie den bezahlten Rechnungsbetrag.
|
|
_STEUER_WOERTER = ("MWST", "UST", "VAT", "STEUER", "NETTO")
|
|
|
|
# Kraftstoffmarken, wie sie im Belegkopf stehen. Bewusst nur dort gesucht (die
|
|
# ersten Zeilen), nicht im ganzen Text: "Total" wäre sonst z. B. auch als
|
|
# Summenzeile ein Treffer. Längere Namen zuerst, damit "TotalEnergies" nicht
|
|
# als "Total" endet. Die Schreibweise hier ist die, die angezeigt wird - auf
|
|
# Bons steht die Marke meist in Großbuchstaben ("SHELL STATION").
|
|
_MARKEN = (
|
|
"TotalEnergies", "Total", "Aral", "Shell", "Eni", "Agip", "Esso", "OMV",
|
|
"Avia", "Turmöl", "Turmoel", "Orlen", "Tamoil", "Westfalen", "Allguth",
|
|
"Classic", "Sprint", "Raiffeisen", "BayWa", "Elan", "HEM", "Star", "JET",
|
|
"bft", "BP", "Q1",
|
|
)
|
|
_STRASSE_WOERTER = r"(?:stra(?:ss|ß)e|str\.|gasse|weg|platz|allee|ring|damm|chaussee)"
|
|
|
|
|
|
def _marke(zeilen):
|
|
"""Markenname aus dem Belegkopf ("SHELL STATION" -> "Shell"). None, wenn
|
|
keine bekannte Marke draufsteht - dann bleibt es beim Betreibernamen."""
|
|
kopf = "\n".join(zeilen[:8])
|
|
for marke in _MARKEN:
|
|
if re.search(r"\b" + re.escape(marke) + r"\b", kopf, re.IGNORECASE):
|
|
return marke
|
|
return None
|
|
|
|
|
|
def _ist_strasse(zeile):
|
|
"""Straßenzeilen tragen eine Hausnummer oder ein Straßenwort - der
|
|
Betreibername ("AUTO B. FRISCHMANN GMBH") tut beides nicht."""
|
|
if not zeile:
|
|
return False
|
|
return bool(re.search(r"\d", zeile) or re.search(_STRASSE_WOERTER, zeile, re.IGNORECASE))
|
|
|
|
|
|
def _tankstelle(marke, strasse, ort, ersatzname=None):
|
|
"""Anzeigename der Tankstelle im Format "Marke, Straße, Ort" (§ Anzeige im
|
|
Frontend, Feld station_name). Fehlende Teile fallen weg statt als leere
|
|
Kommastelle stehen zu bleiben - nicht jeder Beleg nennt eine Straße. Ohne
|
|
erkennbare Marke tritt der Betreibername an ihre Stelle."""
|
|
teile = [t for t in (marke or ersatzname, strasse, ort) if t]
|
|
return ", ".join(teile) if teile else None
|
|
|
|
|
|
def _parsen(text):
|
|
def suchen(muster, *, pflicht=True, flags=re.MULTILINE):
|
|
treffer = re.search(muster, text, flags)
|
|
if treffer is None and pflicht:
|
|
raise ValueError(f"Muster nicht gefunden: {muster!r}")
|
|
return treffer
|
|
|
|
# ---- Station ---------------------------------------------------------
|
|
# Nicht auf den Wortlaut der Markenzeile selbst verankert ("SHELL
|
|
# STATION" vs. "Shell-STATION" - beide an echten Belegen gesehen,
|
|
# offenbar je nach Pächter/Kassensystem unterschiedlich geschrieben).
|
|
# Stattdessen einfach die drei Zeilen direkt vor "Obj.-Nr." nehmen - die
|
|
# Reihenfolge Betreiber/Straße/Ort direkt davor ist auf beiden echten
|
|
# Belegen gleich.
|
|
station = suchen(
|
|
r"(?P<name>.+?)\s*\n\s*(?P<strasse>.+?)\s*\n\s*(?P<ort>.+?)\s*\n\s*Obj\.-Nr\."
|
|
)
|
|
station_id = suchen(r"Obj\.-Nr\.:\s*(\d+)").group(1)
|
|
|
|
# ---- Beleg-Kopf (Beleg-Nr., Datum/Zeit auf dem Kassenbon) ----------
|
|
kopf = suchen(
|
|
r"Beleg-Nr\.\s*(?P<receipt_no>[\d/]+)\s+(?P<tag>\d{2})\.(?P<monat>\d{2})\.(?P<jahr>\d{2})\s+(?P<stunde>\d{2}):(?P<minute>\d{2})"
|
|
)
|
|
receipt_no = kopf.group("receipt_no")
|
|
ts = _iso(kopf.group("tag"), kopf.group("monat"), kopf.group("jahr"),
|
|
kopf.group("stunde"), kopf.group("minute"))
|
|
|
|
# ---- Kraftstoff-Position --------------------------------------------
|
|
position = suchen(
|
|
r"\*(?P<article_no>\d+)\s+(?P<produkt>.+?)\s+" + _ZAHL + r"\s*EUR\s*#A\*"
|
|
)
|
|
article_no = position.group("article_no")
|
|
product_name = position.group("produkt").strip()
|
|
|
|
zapfsaeule = suchen(
|
|
r"\*Zp\s*\d+\s+(?P<liter>" + _ZAHL + r")\s*l\s+(?P<preis>" + _ZAHL + r")\s*EUR/l"
|
|
)
|
|
liters = _de_zahl(zapfsaeule.group("liter"))
|
|
list_price_per_l = _de_zahl(zapfsaeule.group("preis"))
|
|
|
|
# ---- SmartDeal-Rabatt (optional - nicht jeder Tankvorgang hat einen) --
|
|
rabatt_treffer = suchen(r"-(" + _ZAHL + r")\s*EUR\s*#A\s*$", pflicht=False)
|
|
discount = _de_zahl(rabatt_treffer.group(1)) if rabatt_treffer else None
|
|
rabatt_l_treffer = suchen(r"(" + _ZAHL + r")\s*EUR/Liter", pflicht=False)
|
|
discount_per_l = _de_zahl(rabatt_l_treffer.group(1)) if rabatt_l_treffer else None
|
|
|
|
# ---- Beträge ----------------------------------------------------------
|
|
fuel_total_eur = _de_zahl(suchen(r"Gesamtbetrag\s+(" + _ZAHL + r")\s*EUR").group(1))
|
|
steuerzeile = suchen(
|
|
r"A:" + _ZAHL + r"%\s+(?P<netto>" + _ZAHL + r")\s+(?P<mwst>" + _ZAHL + r")\s+(?P<brutto>" + _ZAHL + r")"
|
|
)
|
|
net_eur = _de_zahl(steuerzeile.group("netto"))
|
|
vat_eur = _de_zahl(steuerzeile.group("mwst"))
|
|
receipt_total_eur = _de_zahl(steuerzeile.group("brutto"))
|
|
|
|
# ---- Zahlungszeitstempel (Mobile Payment) - optional, falls anderes
|
|
# Zahlungsverfahren auf dem Beleg steht ---------------------------------
|
|
zahlung = suchen(
|
|
r"Mobile Payment\s*\n+\s*(?P<tag>\d{2})\.(?P<monat>\d{2})\.(?P<jahr>\d{4})\s+(?P<stunde>\d{2}):(?P<minute>\d{2}):(?P<sekunde>\d{2})",
|
|
pflicht=False,
|
|
)
|
|
ts_payment = (
|
|
_iso(zahlung.group("tag"), zahlung.group("monat"), zahlung.group("jahr"),
|
|
zahlung.group("stunde"), zahlung.group("minute"), zahlung.group("sekunde"))
|
|
if zahlung else None
|
|
)
|
|
|
|
# ---- TSE-Signatur (Seite 2) - optional -------------------------------
|
|
tse_nr = suchen(r"TSE-Beleg-Nr:\s*(\d+)", pflicht=False)
|
|
tse_beleg_nr = tse_nr.group(1) if tse_nr else None
|
|
tse_zeit = suchen(
|
|
r"Start/Ende:\s*(?P<tag>\d{2})\.(?P<monat>\d{2})\.(?P<jahr>\d{4})\s+(?P<stunde>\d{2}):(?P<minute>\d{2}):(?P<sekunde>\d{2})",
|
|
pflicht=False,
|
|
)
|
|
ts_tse = (
|
|
_iso(tse_zeit.group("tag"), tse_zeit.group("monat"), tse_zeit.group("jahr"),
|
|
tse_zeit.group("stunde"), tse_zeit.group("minute"), tse_zeit.group("sekunde"))
|
|
if tse_zeit else None
|
|
)
|
|
|
|
# receipt_key: minutengenauer Deduplizierungs-Schlüssel (§6.4/§7.7 Regel 3)
|
|
receipt_key = f"{station_id}_{receipt_no.replace('/', '-')}_{ts[:16]}"
|
|
|
|
return {
|
|
"receipt_key": receipt_key,
|
|
"receipt_no": receipt_no,
|
|
"tse_beleg_nr": tse_beleg_nr,
|
|
"ts": ts,
|
|
"ts_payment": ts_payment,
|
|
"ts_tse": ts_tse,
|
|
"station_id": station_id,
|
|
# Anzeigename "Marke, Straße, Ort" (siehe _tankstelle). Der Ort steht
|
|
# auf dem Bon als "85057 Ingolstadt" - die PLZ gehört nur in die
|
|
# ausführliche station_address darunter, nicht in den Anzeigenamen.
|
|
"station_name": _tankstelle(
|
|
_marke(text.splitlines()),
|
|
station.group("strasse").strip(),
|
|
re.sub(r"^\d{4,5}\s+", "", station.group("ort").strip()),
|
|
ersatzname=station.group("name").strip(),
|
|
),
|
|
"station_address": f"{station.group('strasse').strip()}, {station.group('ort').strip()}",
|
|
"article_no": article_no,
|
|
"product_name": product_name,
|
|
"fuel_type": product_name,
|
|
"liters": liters,
|
|
"list_price_per_l": list_price_per_l,
|
|
"discount": discount,
|
|
"discount_per_l": discount_per_l,
|
|
"fuel_total_eur": fuel_total_eur,
|
|
"receipt_total_eur": receipt_total_eur,
|
|
"net_eur": net_eur,
|
|
"vat_eur": vat_eur,
|
|
}
|
|
|
|
|
|
def _parsen_generisch(text):
|
|
"""Stationsunabhängiger Fallback für Belege, die keinem der oben fest an
|
|
Shells Beleg-Struktur verankerten Muster entsprechen (_parsen() wirft in
|
|
dem Fall ValueError - siehe main()). Arbeitet nach demselben Verfahren,
|
|
das an einem echten Beleg einer bislang unbekannten Tankstelle von Hand
|
|
nachvollzogen wurde: Adresse suchen, Gesamtbetrag suchen ("Gesamt.../
|
|
Absolut..."), Literangabe suchen ("Menge"/"Amount" oder eine an "l"/
|
|
"Liter" hängende Zahl), daraus den tatsächlich bezahlten Preis/Liter
|
|
errechnen (Gesamtbetrag / Liter). Steht daneben auch ein aufgedruckter
|
|
Preis/Liter auf dem Beleg und weicht er vom errechneten ab, wurde ein
|
|
Rabatt gewährt - auf dem Beleg meist als eigener Betrag mit einem Minus
|
|
davor erkennbar, danach wird zuerst gesucht, sonst aus der Preisdifferenz
|
|
hergeleitet. Ob ein so gefundener Rabatt als "SmartDeal" behandelt wird,
|
|
entscheidet einzig station_name (Shell-Logo/-Label im Frontend, siehe
|
|
frontend_veroeffentlichung.py) - kein Sonderfall hier nötig, die Frontend-
|
|
Anzeige hängt nur an einem gesetzten discount, nicht an einem eigenen
|
|
Kennzeichen.
|
|
|
|
Liefert deutlich weniger Felder als _parsen() (kein Bon-Kleingedrucktes
|
|
wie TSE-Nummer/Zahlungszeitstempel) - unproblematisch, denn
|
|
belegverarbeitung.py liest nur receipt_key als echtes Pflichtfeld,
|
|
alles andere per .get()."""
|
|
|
|
def suchen(muster, flags=re.IGNORECASE | re.MULTILINE):
|
|
return re.search(muster, text, flags)
|
|
|
|
zeilen = [z.rstrip() for z in text.splitlines()]
|
|
|
|
# ---- Adresse -----------------------------------------------------------
|
|
# Die PLZ-Ort-Angabe steht nicht zuverlässig auf einer eigenen Zeile (ein
|
|
# österreichischer Beleg führt sie als "SHELL TANKSTELLE, 6450 SÖLDEN"
|
|
# zusammen mit der Filialbezeichnung) und PLZ sind vier- *oder*
|
|
# fünfstellig (AT/DE). Deshalb nicht mehr drei feste Zeilen hintereinander
|
|
# verlangen, sondern im Belegkopf die erste Zeile mit PLZ+Ort suchen; der
|
|
# Name ist die erste Textzeile darüber.
|
|
plz_treffer = None
|
|
plz_zeile = None
|
|
for i, zeile in enumerate(zeilen[:12]):
|
|
treffer = re.search(r"\b\d{4,5}\s+([A-Za-zÄÖÜäöüß][^\n]*)$", zeile)
|
|
if treffer:
|
|
plz_treffer, plz_zeile = treffer, i
|
|
break
|
|
|
|
ort = None
|
|
strasse = None
|
|
betreiber = None
|
|
station_address = None
|
|
if plz_treffer is not None:
|
|
ort = plz_treffer.group(1).strip(" ,.-")
|
|
station_address = zeilen[plz_zeile].strip()
|
|
# Die Straße steht entweder vor der PLZ auf derselben Zeile
|
|
# ("Musterstr. 5, 80331 München") oder auf der Zeile darüber.
|
|
vor_plz = zeilen[plz_zeile][: plz_treffer.start()].strip(" ,.-")
|
|
if _ist_strasse(vor_plz):
|
|
strasse = vor_plz
|
|
for zeile in reversed(zeilen[:plz_zeile]):
|
|
if not zeile.strip():
|
|
continue
|
|
if strasse is None and _ist_strasse(zeile.strip()):
|
|
strasse = zeile.strip()
|
|
continue
|
|
betreiber = zeile.strip()
|
|
break
|
|
if betreiber is None:
|
|
for zeile in zeilen[:5]:
|
|
if zeile.strip():
|
|
betreiber = zeile.strip()
|
|
break
|
|
if station_address and strasse and strasse not in station_address:
|
|
station_address = f"{strasse}, {station_address}"
|
|
station_name = _tankstelle(_marke(zeilen), strasse, ort, ersatzname=betreiber)
|
|
|
|
# ---- Gesamtbetrag ------------------------------------------------------
|
|
# Zeilenweise statt über den ganzen Text, aus zwei Gründen, die beide an
|
|
# einem echten Beleg aufgefallen sind: Bondrucker sperren Überschriften
|
|
# gern buchstabenweise ("G E S A M T BETRAG EUR: 92,60"), und ein frei
|
|
# laufendes Muster fand vorher die Spaltenüberschrift "SUMME-EUR" und
|
|
# las die Artikelnummer der Folgezeile als Betrag. Deshalb: Schlüsselwort
|
|
# auf der leerzeichenfreien Zeile prüfen, Betrag nur als echte Geldangabe
|
|
# (mit Nachkommastellen) akzeptieren und Steuerzeilen ausschließen.
|
|
kandidaten = []
|
|
for zeile in zeilen:
|
|
kompakt = re.sub(r"\s+", "", zeile).upper()
|
|
if not any(wort in kompakt for wort in _GESAMT_WOERTER):
|
|
continue
|
|
if any(wort in kompakt for wort in _STEUER_WOERTER):
|
|
continue
|
|
betraege = re.findall(_GELD, zeile)
|
|
if betraege:
|
|
kandidaten.append(_de_zahl(betraege[-1]))
|
|
if not kandidaten:
|
|
raise ValueError("Kein Gesamtbetrag gefunden (weder 'Gesamt' noch 'Absolut')")
|
|
# Mehrere Treffer (z. B. Summenzeile und Kartenzahlungszeile): der
|
|
# Rechnungsbetrag ist der größte - Teilbeträge sind nie größer.
|
|
fuel_total_eur = max(kandidaten)
|
|
|
|
# ---- Menge in Litern ----------------------------------------------------
|
|
menge = suchen(r"(?:Menge|Amount)\D{0,15}?(" + _ZAHL + r")\s*(?:l\b|L\b|Liter|Ltr\.?)")
|
|
if not menge:
|
|
menge = suchen(r"(" + _ZAHL + r")\s*(?:Liter|Ltr\.?|l)\b")
|
|
if not menge:
|
|
raise ValueError("Keine Literangabe gefunden (weder 'Menge'/'Amount' noch 'X l')")
|
|
liters = _de_zahl(menge.group(1))
|
|
if not liters:
|
|
raise ValueError("Literangabe ist 0 oder unlesbar")
|
|
|
|
berechneter_preis_l = round(fuel_total_eur / liters, 3)
|
|
|
|
# ---- Aufgedruckter Preis/Liter - Abweichung verrät einen Rabatt -------
|
|
# "Preis/L" (ohne ausgeschriebenes "Liter") kommt auf echten Belegen vor,
|
|
# deshalb ist alles nach dem L optional.
|
|
aufgedruckt = suchen(
|
|
r"(?:Preis\s*/?\s*(?:je\s*)?L(?:iter|tr\.?)?|Price\s*/?\s*L(?:iter)?|€\s*/\s*l|EUR\s*/\s*l)\D{0,10}?(" + _ZAHL + r")"
|
|
)
|
|
list_price_per_l = _de_zahl(aufgedruckt.group(1)) if aufgedruckt else berechneter_preis_l
|
|
|
|
discount = None
|
|
discount_per_l = None
|
|
if aufgedruckt and abs(list_price_per_l - berechneter_preis_l) > 0.005:
|
|
minus = suchen(r"-\s*(" + _ZAHL + r")\s*(?:€|EUR)")
|
|
if minus:
|
|
discount = _de_zahl(minus.group(1))
|
|
discount_per_l = round(discount / liters, 3)
|
|
else:
|
|
discount_per_l = round(list_price_per_l - berechneter_preis_l, 3)
|
|
discount = round(discount_per_l * liters, 2)
|
|
|
|
# ---- Zeitstempel: irgendein Datum+Uhrzeit auf dem Beleg ----------------
|
|
zeit = suchen(r"(\d{2})\.(\d{2})\.(\d{2,4})\D{0,10}?(\d{2}):(\d{2})")
|
|
ts = _iso(zeit.group(1), zeit.group(2), zeit.group(3), zeit.group(4), zeit.group(5)) if zeit else None
|
|
|
|
# ---- Belegnummer, sonst deterministischer Ersatzschlüssel --------------
|
|
# Wichtig: kein eingebautes hash() (pro Prozess zufällig gesalzen, siehe
|
|
# PYTHONHASHSEED) - der Parser läuft für jeden Beleg als eigener Prozess
|
|
# (Kopfkommentar oben), ein instabiler Schlüssel würde den §7.7-Regel-3-
|
|
# Dedup-Vergleich in belegverarbeitung.py unbrauchbar machen.
|
|
beleg_nr = suchen(
|
|
r"(?:Beleg-?Nr\.?|Bon-?Nr\.?|Rechnungs-?Nr\.?|Receipt\s*No\.?|Trans(?:aktions)?-?Nr\.?)\D{0,5}?([\dA-Za-z/\-]+)"
|
|
)
|
|
receipt_no = beleg_nr.group(1) if beleg_nr else None
|
|
schluessel_basis = "|".join(
|
|
str(teil) for teil in (receipt_no, ts, station_address, fuel_total_eur, liters) if teil
|
|
) or text[:80]
|
|
receipt_key = "generisch_" + hashlib.sha1(schluessel_basis.encode("utf-8")).hexdigest()[:16]
|
|
|
|
return {
|
|
"receipt_key": receipt_key,
|
|
"receipt_no": receipt_no,
|
|
"tse_beleg_nr": None,
|
|
"ts": ts,
|
|
"ts_payment": None,
|
|
"ts_tse": None,
|
|
"station_id": None,
|
|
"station_name": station_name,
|
|
"station_address": station_address,
|
|
"article_no": None,
|
|
"product_name": None,
|
|
"fuel_type": None,
|
|
"liters": liters,
|
|
"list_price_per_l": list_price_per_l,
|
|
"discount": discount,
|
|
"discount_per_l": discount_per_l,
|
|
"fuel_total_eur": fuel_total_eur,
|
|
"receipt_total_eur": fuel_total_eur,
|
|
"net_eur": None,
|
|
"vat_eur": None,
|
|
}
|
|
|
|
|
|
def beleg_lesen(pdf_pfad):
|
|
"""Liest einen Tankbeleg und gibt die Felder als dict zurück.
|
|
|
|
Der eine Einstiegspunkt für Aufrufer im selben Prozess. Die Integration
|
|
(belege.py) ruft genau diese Funktion im Executor auf - früher lief der
|
|
Parser als eigener Unterprozess über main() unten, weil pyscript keine
|
|
fremden Pakete laden konnte und pypdf deshalb von Hand ins Container-
|
|
Python installiert werden musste. Als echte Integration steht pypdf über
|
|
manifest.json bereit; der Unterprozess und der Handinstallations-Schritt
|
|
entfallen damit ersatzlos.
|
|
|
|
Wirft eine Exception, wenn weder das Shell-Format noch der
|
|
stationsunabhängige Rückfall greifen - der Aufrufer entscheidet, was er
|
|
dem Nutzer davon zeigt."""
|
|
text = _text_aus_pdf(pdf_pfad)
|
|
try:
|
|
return _parsen(text)
|
|
except Exception:
|
|
# Kein Shell-Beleg im bekannten Format - Versuch mit dem
|
|
# stationsunabhängigen Fallback (_parsen_generisch() oben), bevor
|
|
# endgültig aufgegeben wird.
|
|
return _parsen_generisch(text)
|
|
|
|
|
|
def main():
|
|
"""Kommandozeilen-Aufruf, weiterhin nutzbar zum Nachschauen von Hand:
|
|
`python3 shell_beleg_parser.py <pfad-zur-pdf>`."""
|
|
if len(sys.argv) != 2:
|
|
print("Aufruf: python3 shell_beleg_parser.py <pfad-zur-pdf>", file=sys.stderr)
|
|
sys.exit(1)
|
|
try:
|
|
ergebnis = beleg_lesen(sys.argv[1])
|
|
except Exception as exc:
|
|
print(f"Beleg konnte nicht gelesen werden ({type(exc).__name__}): {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(json.dumps(ergebnis, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|