rechnungstool/pruef_kette.py

# -*- coding: utf-8 -*-
"""Pruefstand: haelt die Kette zusammen, und kann ein anderes Programm sie ordnen?

Der Ordner ist die Buchhaltung. Ein zweites Programm (Steuerrechner,
Steuerjournal) muss aus den PDFs allein erkennen koennen, was zusammengehoert:
welche Rechnung wurde storniert, welche ersetzt sie, welches Blatt berichtigt
nur die Anschrift.

Dafuer traegt JEDES Blatt einer Kette dieselbe Vorgangsnummer - die der ersten
Rechnung. Gruppieren heisst dann: nach "vorgang" sortieren, fertig. Ohne den
Schluessel muesste man sich die Kette ueber storno_zu/ersetzt zusammensuchen.

Durchgespielt wird der schlimmste Fall: Rechnung -> Storno -> berichtigte
Rechnung -> Berichtigung der Anschrift -> noch ein Storno -> zweite berichtigte
Rechnung. Alle sechs muessen unter EINEM Schluessel landen und in der Summe
genau die letzte gueltige Rechnung ergeben.
"""
import json
import os
import shutil
import sys
import tempfile

QUELLE = r"C:\claude\rechnungstool"
sys.path.insert(0, QUELLE)

tmp = tempfile.mkdtemp(prefix="kette_pruef_")
ordner = os.path.join(tmp, "rechnungen")
os.makedirs(ordner)

import config  # noqa: E402
shutil.copy(os.path.join(QUELLE, "config.json"), os.path.join(tmp, "config.json"))
config.CONFIG_PATH = os.path.join(tmp, "config.json")
roh = json.load(open(config.CONFIG_PATH, encoding="utf-8"))
roh["rechnung_jahr"], roh["rechnung_zaehler"] = 2026, 0
# Der Erststart-Zustand gehoert dem Pruefstand, nicht der config des
# Entwicklerrechners: sonst laeuft er in einen anderen Zweig, sobald hier
# jemand die Nummer einmal eingerichtet hat.
roh["nummer_eingerichtet"] = False
roh["output_dir"] = ordner
json.dump(roh, open(config.CONFIG_PATH, "w", encoding="utf-8"), ensure_ascii=False)

# Kein Erststart-Dialog im Pruefstand - er wuerde den Lauf haengen lassen.
os.environ["CAMPINGHOF_TEST"] = "1"
import app as A  # noqa: E402
import bestand as BE  # noqa: E402
from tkinter import messagebox  # noqa: E402


# --- Reissleine ---------------------------------------------------------
# Ein Pruefstand darf NIE haengen. Bleibt er in einem Dialog stehen (etwa weil
# eine Aenderung eine neue Frage eingefuehrt hat), bricht er nach zwei Minuten
# von selbst ab, statt ein Fenster offen zu lassen, auf das niemand klickt.
import threading as _th


def _reissleine():
    import os as _os
    print("ABBRUCH | Pruefstand haengt (Zeitgrenze) - vermutlich ein Dialog, "
          "der nicht beantwortet wird")
    _os._exit(2)


_wecker = _th.Timer(120, _reissleine)
_wecker.daemon = True
_wecker.start()

fehler = []


def pruefe(bedingung, text):
    # Die Windows-Konsole ist cp1252 - ein Sonderzeichen im Text darf den Lauf
    # nicht mit einem UnicodeEncodeError abbrechen.
    zeile = ("  OK   | " if bedingung else "FEHLER | ") + text
    kodierung = sys.stdout.encoding or "utf-8"
    print(zeile.encode(kodierung, "replace").decode(kodierung))
    if not bedingung:
        fehler.append(text)


antworten = {"Fertig": False, "Storno wirklich": True,
             "endgültig erstellen": True,
             "Berichtigte Rechnung erstellen": True, "Hausnummer": True}


def askyesno(titel, text="", **kw):
    for schluessel, wert in antworten.items():
        if schluessel in titel:
            return wert
    raise AssertionError(f"unangemeldeter Dialog: {titel!r}")


# Die Freigabe vor dem Beleg (§ 14 UStG) wird bewusst bejaht - geprueft wird hier
# der Ablauf DANACH. Dass ein "Nein" keinen Beleg erzeugt, prueft pruef_freigabe.py.
antworten["jetzt erstellen?"] = True
# Der Erprobungshinweis wird hier bejaht - geprueft wird er in pruef_hinweis.py.
antworten["Erprobungsfassung"] = True
messagebox.askyesno = askyesno
messagebox.showinfo = messagebox.showwarning = messagebox.showerror = \
    lambda *a, **k: None

f = A.RechnungsApp()
f.update()
f.cfg["output_dir"] = ordner
erste = sorted(f.anzahl_vars)[0]


def kunde(strasse="Musterweg 1", tage="4"):
    f.k_vorname.set("Anna")
    f.k_name.set("Beispiel")
    f.k_strasse.set(strasse)
    f.k_plzort.set("12345 Musterstadt")
    f.anzahl_vars[erste].set(tage)
    f.update()


def rechnung_mit(nummer):
    return next(x for x in BE.rechnungen_im_ordner(ordner)
                if x.get("rechnungsnummer") == nummer and x.get("art") != "berichtigung")


# 1. die urspruengliche Rechnung
kunde(tage="4")
f.pdf_erstellen()
# 2. + 3. Storno und berichtigte Rechnung (falsche Naechtezahl)
f._storno_anlegen(rechnung_mit("2026-001"), ordner)
kunde(tage="3")
f.pdf_erstellen()
# 4. Berichtigung der Anschrift auf der berichtigten Rechnung
f._berichtigung_starten(rechnung_mit("2026-003"), ordner)
f.k_strasse.set("Richtige Straße 7")
f.update()
f.pdf_erstellen()
# 5. + 6. dieselbe Rechnung noch einmal stornieren und neu schreiben
f._storno_anlegen(rechnung_mit("2026-003"), ordner)
kunde(strasse="Richtige Straße 7", tage="2")
f.pdf_erstellen()
f.destroy()

alle = BE.rechnungen_im_ordner(ordner)
print("--- was im Ordner liegt ---")
for x in sorted(alle, key=lambda d: (d.get("vorgang", ""), d.get("rechnungsnummer", ""))):
    print(f"       {x.get('vorgang',''):10s} {x.get('art',''):12s} "
          f"{x.get('rechnungsnummer',''):10s} {float(x.get('gesamt') or 0):9.2f}  "
          f"{os.path.basename(x['_pfad'])}")

pruefe(len(alle) == 6, f"sechs Belege im Ordner ({len(alle)})")
pruefe(all(x.get("vorgang") for x in alle), "jeder Beleg trägt eine Vorgangsnummer")

# Genau das macht das andere Programm: nach "vorgang" gruppieren.
ketten = {}
for x in alle:
    ketten.setdefault(x["vorgang"], []).append(x)
pruefe(len(ketten) == 1, f"alle sechs hängen an EINEM Vorgang ({list(ketten)})")
pruefe(list(ketten) == ["2026-001"],
       f"der Schlüssel ist die erste Rechnung ({list(ketten)})")

kette = ketten.get("2026-001", [])
arten = sorted(x.get("art") for x in kette)
pruefe(arten == ["berichtigung", "korrektur", "korrektur", "rechnung", "storno", "storno"],
       f"Arten der Kette stimmen ({arten})")

# Die Summe ueber die Kette muss die letzte gueltige Rechnung sein.
letzte = max((x for x in kette if x.get("art") in ("rechnung", "korrektur")),
             key=lambda d: d["rechnungsnummer"])
summe = round(sum(float(x.get("gesamt") or 0) for x in kette), 2)
pruefe(abs(summe - float(letzte["gesamt"])) < 0.005,
       f"Summe der Kette = letzte gültige Rechnung ({summe:.2f} vs {letzte['gesamt']:.2f})")

# Jeder Beleg zeigt ausserdem einzeln auf seinen Vorgaenger.
for x in kette:
    if x.get("art") == "storno":
        pruefe(bool(x.get("storno_zu")), f"Storno {x['rechnungsnummer']} nennt seine Rechnung")
    elif x.get("art") == "korrektur":
        pruefe(bool(x.get("ersetzt")) and bool(x.get("storno_nummer")),
               f"Korrektur {x['rechnungsnummer']} nennt Rechnung und Storno")
    elif x.get("art") == "berichtigung":
        pruefe(bool(x.get("berichtigt_zu")),
               f"Berichtigung nennt die Rechnung ({x.get('berichtigt_zu')})")

# Gegenprobe mit dem Steuerrechner: er darf nur die Geldbelege sehen.
sys.path.insert(0, r"C:\claude\steuerrechner")
import einnahmen  # noqa: E402
liste, kaputt = einnahmen.scanne(ordner, {"ust_saetze": [7, 19]})
pruefe(len(liste) == 5 and not kaputt,
       f"Steuerrechner liest 5 Belege, kein Fehler ({len(liste)}, {kaputt})")
jahr = round(sum(e.gesamt for e in liste), 2)
pruefe(abs(jahr - float(letzte["gesamt"])) < 0.005,
       f"Jahressumme = letzte gültige Rechnung ({jahr:.2f} vs {letzte['gesamt']:.2f})")


# Wegwerf-Ordner nur bei Fehlern behalten - sonst laeuft das Temp-Verzeichnis
# voll. Ein Pruefstand, der 400 Ordner hinterlaesst, ist selbst ein Problem.
if not fehler:
    import shutil as _sh
    _sh.rmtree(tmp, ignore_errors=True)
print("---")
print(f"Ordner: {ordner}")
print("alles gruen" if not fehler else f"{len(fehler)} Fehler: {fehler}")
sys.exit(1 if fehler else 0)