audi-app/homeassistant/data/shell_beleg_parser.py

#!/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 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<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,
        "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 main():
    if len(sys.argv) != 2:
        print("Aufruf: python3 shell_beleg_parser.py <pfad-zur-pdf>", file=sys.stderr)
        sys.exit(1)
    pdf_pfad = sys.argv[1]
    try:
        text = _text_aus_pdf(pdf_pfad)
        ergebnis = _parsen(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()