"""Belegverarbeitung für Tankbelege (SPECIFICATION.md §7.7). Beleg-Upload direkt in der App statt Postfachanbindung - kein IMAP-Zugriff. Das eigentliche Lesen der PDF macht shell_beleg_parser.py daneben; hier steht, was mit dem Ergebnis passiert. ZWEI REGELN, DIE DEN AUFBAU ERKLÄREN ------------------------------------ 1. Liter, Kosten und Ersparnis kommen vom Beleg, Kilometerstand und gefahrene Distanz kommen immer vom Fahrzeug (§7.7 Regel 2). Ein Beleg-Nachtrag rührt odometer_km/distance_km deshalb nicht an. 2. price_per_l wird grundsätzlich aus fuel_total_eur ÷ liters berechnet, nie vom Beleg übernommen. fuel_total_eur ist dabei immer der reale, bereits rabattierte Betrag; die SmartDeal-Ersparnis steht separat in discount. Mit tank_id trägt der Upload die Belegdaten in einen bestehenden Tankvorgang nach - egal ob der ursprünglich manuell oder automatisch angelegt wurde. Passt das Belegdatum nicht zum erfassten Tankvorgang (anderer Kalendertag), wird nichts übernommen, sondern ein Fehler veröffentlicht: sonst landet in der Oberfläche versehentlich ein falscher Beleg an einem bestehenden Eintrag. Ohne tank_id legt der Upload KEINEN eigenständigen Tankvorgang an, sondern parst nur und veröffentlicht das Ergebnis - die Oberfläche übernimmt die Werte in den gerade offenen Entwurf und speichert erst auf einen expliziten Klick. """ from __future__ import annotations import base64 import datetime import logging import os from typing import TYPE_CHECKING from . import shell_beleg_parser from . import tankerkennung from .tankerkennung import leerer_tankvorgang from .veroeffentlichung import zustand_oder_none from .verlauf import als_kilometerstand, zeit_normalisiert if TYPE_CHECKING: from .koordinator import Koordinator _LOGGER = logging.getLogger(__name__) def preis_je_liter(fuel_total_eur: float | None, liters: float | None) -> float | None: if not liters or fuel_total_eur is None: return None return round(fuel_total_eur / liters, 3) def _speichern_und_lesen(ordner: str, dateiname: str, rohdaten: bytes) -> tuple[str, dict]: """Läuft komplett im Executor: Datei schreiben und PDF lesen sind beides blockierende Arbeit.""" os.makedirs(ordner, exist_ok=True) pfad = os.path.join(ordner, dateiname) with open(pfad, "wb") as datei: datei.write(rohdaten) return pfad, shell_beleg_parser.beleg_lesen(pfad) async def hochladen(k: Koordinator, daten: dict) -> None: pdf_base64 = daten.get("pdf_base64") tank_id = daten.get("tank_id") if not pdf_base64: _LOGGER.warning("Beleg-Upload ohne pdf_base64 aufgerufen") return dateiname = daten.get("dateiname") or ( f"beleg_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf" ) # Alles bis zum geparsten Ergebnis in einem try: eine unbehandelte # Ausnahme hier (kaputtes Base64, volle Festplatte, unlesbares PDF) wäre # für die Oberfläche unsichtbar - ein abgelehnter Dienstaufruf landet dort # nur in der Browser-Konsole, angezeigt wird nichts. try: rohdaten = base64.b64decode(pdf_base64) pdf_pfad, geparst = await k.hass.async_add_executor_job( _speichern_und_lesen, k.ablage.belege_ordner, dateiname, rohdaten ) except Exception as fehler: # noqa: BLE001 - jede Ursache endet hier gleich _LOGGER.error("Beleg konnte nicht verarbeitet werden: %s", fehler) await k.beleg_ergebnis_veroeffentlichen( {"fehler": "Beleg konnte nicht gelesen werden", "tank_id": tank_id} ) return receipt_key = geparst.get("receipt_key") if not receipt_key: _LOGGER.error("Parser lieferte keinen receipt_key") await k.beleg_ergebnis_veroeffentlichen( {"fehler": "Beleg konnte nicht gelesen werden", "tank_id": tank_id} ) return # §7.7 Regel 3: derselbe Beleg (receipt_key, minutengenau) erzeugt keinen # zweiten Datensatz. vorhandener = await k.ablage.tankvorgang_nach_receipt_key(receipt_key) if vorhandener is not None and vorhandener.get("tank_id") != tank_id: _LOGGER.info("Beleg %s bereits vorhanden, übersprungen", receipt_key) # vorhandener_tank_id zeigt auf den Tankvorgang, der den Beleg schon # trägt. Die Oberflächen springen damit dorthin, statt nur "bereits # erfasst" zu melden - beim Teilen aus Mail ist genau das die Frage, # die der Nutzer hat: welcher Vorgang ist das denn? # # Bewusst ein EIGENES Feld statt tank_id umzudeuten: tank_id ist die # Eingabe des Aufrufs (leer beim Neuanlegen, gesetzt beim Nachtragen), # und das Panel entscheidet daran, wo es den Fehler anzeigt. await k.beleg_ergebnis_veroeffentlichen( { "fehler": "Beleg bereits erfasst", "tank_id": tank_id, "vorhandener_tank_id": vorhandener.get("tank_id"), } ) return beleg_felder = _beleg_felder(geparst, pdf_pfad) if not tank_id: await k.beleg_ergebnis_veroeffentlichen({**beleg_felder, "tank_id": None}) _LOGGER.info("Beleg %s geparst, wartet auf Speichern in der Oberfläche", receipt_key) return await _nachtragen(k, tank_id, geparst, beleg_felder, receipt_key) def _beleg_felder(geparst: dict, pdf_pfad: str) -> dict: liters = geparst.get("liters") fuel_total_eur = geparst.get("fuel_total_eur") # real, bereits rabattiert return { "receipt_key": geparst.get("receipt_key"), "receipt_no": geparst.get("receipt_no"), "tse_beleg_nr": geparst.get("tse_beleg_nr"), "ts": geparst.get("ts"), "ts_payment": geparst.get("ts_payment"), "ts_tse": geparst.get("ts_tse"), "station_id": geparst.get("station_id"), "station_name": geparst.get("station_name"), "station_address": geparst.get("station_address"), "article_no": geparst.get("article_no"), "product_name": geparst.get("product_name"), "fuel_type": geparst.get("fuel_type"), "liters": liters, "fuel_total_eur": fuel_total_eur, "price_per_l": preis_je_liter(fuel_total_eur, liters), "discount": geparst.get("discount"), # SmartDeal-Ersparnis "discount_per_l": geparst.get("discount_per_l"), "list_price_per_l": geparst.get("list_price_per_l"), "receipt_total_eur": geparst.get("receipt_total_eur"), "net_eur": geparst.get("net_eur"), "vat_eur": geparst.get("vat_eur"), "receipt_file": pdf_pfad, } async def _nachtragen( k: Koordinator, tank_id: str, geparst: dict, beleg_felder: dict, receipt_key: str ) -> None: vorhanden = await k.ablage.tankvorgang_nach_id(tank_id) if not vorhanden: _LOGGER.warning("Tankvorgang %s für Beleg-Nachtrag nicht gefunden", tank_id) await k.beleg_ergebnis_veroeffentlichen( {"fehler": "Tankvorgang nicht gefunden", "tank_id": tank_id} ) return vorhandenes_datum = (vorhanden.get("ts") or "")[:10] belegdatum = (geparst.get("ts") or "")[:10] if vorhandenes_datum and belegdatum and vorhandenes_datum != belegdatum: _LOGGER.warning( "Belegdatum %s passt nicht zu Tankvorgang %s (Datum %s)", belegdatum, tank_id, vorhandenes_datum, ) await k.beleg_ergebnis_veroeffentlichen( {"fehler": "Falsches Belegdatum", "tank_id": tank_id} ) return aenderungen = dict(beleg_felder) aenderungen["status"] = "vollständig" if await k.ablage.tankvorgang_aktualisieren(tank_id, aenderungen): await k.tankvorgang_nachbereiten() await k.beleg_ergebnis_veroeffentlichen({"erfolg": True, "tank_id": tank_id}) _LOGGER.info("Beleg %s in Tankvorgang %s nachgetragen", receipt_key, tank_id) else: await k.beleg_ergebnis_veroeffentlichen( {"fehler": "Tankvorgang nicht gefunden", "tank_id": tank_id} ) # ------------------------------------------------ Tankvorgänge aus dem Formular async def tankvorgang_manuell(k: Koordinator, daten: dict) -> None: """Erfassung aus dem Formular "Neuer Tankvorgang". receipt_key/receipt_file sind optional: kommen sie mit (weil zuvor ein Beleg ohne tank_id in den Entwurf übernommen wurde), zählt der Tankvorgang als per Beleg erfasst, sonst als manuell.""" liter = daten.get("liter") kosten = daten.get("kosten") if liter is None or kosten is None: _LOGGER.warning("liter und kosten sind Pflichtfelder") return ts_roh = daten.get("ts") ts = zeit_normalisiert(ts_roh) if ts_roh else datetime.datetime.now(datetime.UTC) # Eine 0 oder ein geleertes Feld heissen "kenne ich nicht" - siehe # als_kilometerstand(). odometer_km = als_kilometerstand(daten.get("km")) if odometer_km is None: odometer_km = _km_aus_sensor_wenn_jetzt(k, ts) distanz = daten.get("distanz") if distanz is None: # Vorschlag aus TANK_DISTANZ_SENSOR oder, ohne ihn, der Differenz zum # letzten Tankvorgang - aber wie odometer_km selbst frei # überschreibbar, ein übergebener Wert hat deshalb immer Vorrang. distanz = await tankerkennung.distanz_seit_tankung( k, odometer_km, jetzt=_ist_jetzt(ts) ) receipt_key = daten.get("receipt_key") tankvorgang = leerer_tankvorgang( ts.isoformat(), "beleg" if receipt_key else "manual", ) tankvorgang.update({ "receipt_key": receipt_key, "liters": liter, "fuel_total_eur": kosten, "price_per_l": preis_je_liter(kosten, liter), "discount": daten.get("ersparnis"), "station_name": daten.get("station"), "odometer_km": odometer_km, "distance_km": distanz, "fuel_type": daten.get("kraftstoff"), "status": "vollständig", "receipt_file": daten.get("receipt_file"), }) await k.ablage.tankvorgang_anhaengen(tankvorgang) await k.tankvorgang_nachbereiten() _LOGGER.info("Tankvorgang %s angelegt", tankvorgang["tank_id"]) # Wie nah der eingetragene Zeitpunkt an "jetzt" liegen muss, damit der # aktuelle Tachostand als Ersatz taugt. SENSOR_ERSATZ_FENSTER = datetime.timedelta(hours=2) def _km_aus_sensor(k: Koordinator) -> float | None: wert = zustand_oder_none(k.hass, k.zuordnung.werte.KM_SENSOR) try: return float(wert) # type: ignore[arg-type] except (TypeError, ValueError): return None def _ist_jetzt(ts: datetime.datetime | None) -> bool: """Liegt der eingetragene Zeitpunkt nah genug an jetzt, dass die Momentanwerte der Sensoren zu ihm passen?""" if ts is None: return False jetzt = datetime.datetime.now(datetime.UTC) return abs((jetzt - ts).total_seconds()) <= SENSOR_ERSATZ_FENSTER.total_seconds() def _km_aus_sensor_wenn_jetzt( k: Koordinator, ts: datetime.datetime | None ) -> float | None: """Der aktuelle Tachostand - aber nur fuer einen Tankvorgang von jetzt. Der Sensor kennt genau einen Wert: den von diesem Augenblick. Fuer eine Tankung, die gerade passiert ist, ist das der richtige Ersatz und der Normalfall - der Nutzer muss nichts eintippen. Fuer eine nachgetragene Tankung von vorgestern ist es der falsche Wert, und er richtet zweifachen Schaden an: die gefahrene Distanz wird gegen einen zu hohen Stand gerechnet, und die Datumspruefung der Verbrauchskorrektur verwirft das Intervall, weil der gespeicherte Stand nicht zum Tacho des eingetragenen Zeitpunkts passt. Dann lieber gar kein Kilometerstand - genau das, was eine 0 im Formular ausdruecken soll.""" if not _ist_jetzt(ts): _LOGGER.info( "Tankvorgang auf %s datiert - der aktuelle Tachostand passt dazu " "nicht und wird nicht eingesetzt", ts.isoformat(), ) return None return _km_aus_sensor(k) async def tankvorgang_aktualisieren(k: Koordinator, daten: dict) -> None: """Bearbeitet einen bestehenden Tankvorgang (Einzelbeleg-Seite, egal ob ursprünglich per Beleg oder manuell angelegt). Nur tatsächlich in `daten` enthaltene Schlüssel werden geändert - fehlt ein Feld ganz (statt nur leer/None zu sein), bleibt der bisherige Wert unangetastet. Das bestehende Bearbeiten-Formular (Einzelbeleg-Seite) schickt ohnehin immer alle Felder mit, verhält sich also unverändert; ein schlankerer Aufrufer wie der CSV-Import (csv_import.py) kann so gezielt nur einzelne Spalten aktualisieren, ohne den Rest des Datensatzes (odometer_km, discount, fuel_type, receipt_key, ...) stillschweigend auf None zu setzen.""" tank_id = daten.get("tank_id") if not tank_id: _LOGGER.warning("tankvorgang_aktualisieren ohne tank_id aufgerufen") return # edited_fields fortschreiben statt ersetzen: was frueher von Hand gesetzt # wurde, bleibt geschuetzt. vorhanden = await k.ablage.tankvorgaenge_lesen() alt = next((t for t in vorhanden if t.get("tank_id") == tank_id), None) hand = set((alt or {}).get("edited_fields") or []) aenderungen: dict = {} if "ts" in daten: ts_normalisiert = zeit_normalisiert(daten.get("ts")) aenderungen["ts"] = ts_normalisiert.isoformat() if ts_normalisiert else daten.get("ts") if "liter" in daten or "kosten" in daten: liter = daten.get("liter") kosten = daten.get("kosten") aenderungen["liters"] = liter aenderungen["fuel_total_eur"] = kosten aenderungen["price_per_l"] = preis_je_liter(kosten, liter) if liter is not None: # Von Hand eingetippte Liter sind der Beleg, nur ohne PDF - und # damit dieselbe Wahrheit wie ein geparster Beleg. Ohne diesen # Vermerk haelt verbrauchskorrektur.beleg_liter() sie fuer den # Anstieg des Gebers und verwirft sie als Zirkelschluss; der # Verbrauchsfaktor entstuende nie. # # Nebenwirkung, bewusst in Kauf genommen: edited_fields schuetzt # das Feld vor der Automatik. Wird spaeter doch noch ein PDF zu # demselben Tankvorgang hochgeladen, bleibt die getippte Zahl # stehen - der Mensch hatte den Beleg in der Hand. hand.add("liters") if "km" in daten: aenderungen["odometer_km"] = als_kilometerstand(daten.get("km")) if "ersparnis" in daten: aenderungen["discount"] = daten.get("ersparnis") if "station" in daten: aenderungen["station_name"] = daten.get("station") if "distanz" in daten: aenderungen["distance_km"] = daten.get("distanz") if "kraftstoff" in daten: aenderungen["fuel_type"] = daten.get("kraftstoff") if daten.get("receipt_key"): aenderungen["receipt_key"] = daten["receipt_key"] aenderungen["receipt_file"] = daten.get("receipt_file") aenderungen["source"] = "beleg" if hand: aenderungen["edited_fields"] = sorted(hand) if await k.ablage.tankvorgang_aktualisieren(tank_id, aenderungen): await k.tankvorgang_nachbereiten() _LOGGER.info("Tankvorgang %s aktualisiert", tank_id) else: _LOGGER.warning("Tankvorgang %s nicht gefunden", tank_id) async def tankvorgang_loeschen(k: Koordinator, tank_id: str | None) -> None: if not tank_id: _LOGGER.warning("tankvorgang_loeschen ohne tank_id aufgerufen") return if await k.ablage.tankvorgang_loeschen(tank_id): await k.tankvorgang_nachbereiten() _LOGGER.info("Tankvorgang %s gelöscht", tank_id) else: _LOGGER.warning("Tankvorgang %s nicht gefunden", tank_id)