#!/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 ` (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 Obj.-Nr.: Tel. ... Beleg-Nr. * EUR #A* *Zp l EUR/l # * [ - EUR #A -- optional, nur mit SmartDeal EUR/Liter -- optional] Gesamtbetrag EUR Typ Netto Mwst Brutto A:% ... Mobile Payment -- Zahlungszeitstempel ... (Seite 2) POS:...-...-... TSE-Beleg-Nr: Start/Ende: / ... """ 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.,]+" 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.+?)\s*\n\s*(?P.+?)\s*\n\s*(?P.+?)\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[\d/]+)\s+(?P\d{2})\.(?P\d{2})\.(?P\d{2})\s+(?P\d{2}):(?P\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\d+)\s+(?P.+?)\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" + _ZAHL + r")\s*l\s+(?P" + _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" + _ZAHL + r")\s+(?P" + _ZAHL + r")\s+(?P" + _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\d{2})\.(?P\d{2})\.(?P\d{4})\s+(?P\d{2}):(?P\d{2}):(?P\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\d{2})\.(?P\d{2})\.(?P\d{4})\s+(?P\d{2}):(?P\d{2}):(?P\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, "station_name": 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) # ---- Adresse: Name-/Straßen-/PLZ-Ort-Zeile hintereinander - wie bei # Shell (_parsen() oben), aber ohne Ankerzeile "Obj.-Nr." danach, also # direkt an der PLZ-Ort-Form (5 Ziffern + Text) erkannt. Nicht Pflicht. adresse = suchen(r"([^\n]{2,60})\n([^\n]{2,60})\n(\d{5}\s+[^\n]{2,40})") station_name = adresse.group(1).strip() if adresse else None station_address = ( f"{adresse.group(2).strip()}, {adresse.group(3).strip()}" if adresse else None ) # ---- Gesamtbetrag ------------------------------------------------------ gesamt = suchen(r"(?:Gesamt\w*|Absolut\w*|Endbetrag|Summe|Total)\D{0,20}?(" + _ZAHL + r")\s*(?:€|EUR)?") if not gesamt: raise ValueError("Kein Gesamtbetrag gefunden (weder 'Gesamt' noch 'Absolut')") fuel_total_eur = _de_zahl(gesamt.group(1)) # ---- 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 ------- aufgedruckt = suchen( r"(?:Preis\s*/?\s*(?:je\s*)?Liter|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 main(): if len(sys.argv) != 2: print("Aufruf: python3 shell_beleg_parser.py ", file=sys.stderr) sys.exit(1) pdf_pfad = sys.argv[1] try: text = _text_aus_pdf(pdf_pfad) except Exception as exc: print(f"PDF konnte nicht gelesen werden ({type(exc).__name__}): {exc}", file=sys.stderr) sys.exit(1) try: ergebnis = _parsen(text) except Exception: # Kein Shell-Beleg im bekannten Format - Versuch mit dem # stationsunabhängigen Fallback (_parsen_generisch() oben), bevor # endgültig aufgegeben wird. try: ergebnis = _parsen_generisch(text) except Exception as exc: print(f"Beleg konnte nicht geparst werden ({type(exc).__name__}): {exc}", file=sys.stderr) sys.exit(1) print(json.dumps(ergebnis, ensure_ascii=False)) if __name__ == "__main__": main()