TheMockTv/Steuerrechnungstool
1
0
Fork 0
Code Issues Pull requests Projects Releases 21 Packages Wiki Activity Actions

Mini-API, Push vom Rechnungstool, Proforma wird nicht gebucht

Browse source 
Das Journal macht beim Start einen winzigen Server auf 127.0.0.1 auf und traegt
Port und PID in die gemeinsame SQLite ein (Tabelle "laeuft"). Das Rechnungstool
sieht daran, dass jemand zuhoert, und klopft nach jeder Rechnung an: "schau in
die sqlite". Dann wird sofort nachgelesen, statt bis zum naechsten Durchgang zu
warten. Faellt die API aus, findet die Wache die Meldung trotzdem - ein os.stat
auf die eine Datei, alle zwei Sekunden und sofort beim Zurueckklicken.

Die Proforma-Rechnung wird NICHT gebucht: keine_buchung() kennt jetzt
Berichtigung und Proforma. Der Scan fragt wieder keine_buchung() - seit dem
Berichtigungs-Umbau tat er das nicht mehr, und die Proforma waere als Buchung
im Journal gelandet (der Pruefstand hat es gefunden).

Pruefstand pruef_wache.py: 31 Pruefungen ueber BEIDE Programme - Anmeldung mit
PID, ping ("hallo, bin noch da"), genau eine Datei im gemeinsamen Ordner, Push
nach Rechnung und nach Storno, Einlesen samt Anzeige, Rueckfallebene ohne API,
tote PID wird weggeraeumt, kein Melden an ein geschlossenes Journal.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GYeAfLtccFrbU3MTj1MMTx
This commit is contained in:
TheMockTv 2026-09-03 21:24:12 +02:00
parent 31bb0c600b
commit dbe95e1ec6
6 changed files with 319 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

194
api.py Normal file
View file

@ -0,0 +1,194 @@
# -*- coding: utf-8 -*-
"""Mini-API zwischen Rechnungstool und Steuerjournal.
WARUM
-----
Beide Programme stehen nebeneinander offen. Schreibt das Rechnungstool eine
Rechnung, soll sie im Journal SOFORT auftauchen - nicht erst beim naechsten
Einlesen und nicht nach einer Wartezeit.
Seine Ansage dazu: *"machst ein kleinen py server der hin und her pingt mit
einem push vom rechnungstool: eh schau mal in die sqlite"* und *"die muessen ja
sehen das die beiden auf sind"*.
WIE
---
Jedes Programm macht beim Start einen winzigen Server auf **127.0.0.1** mit
einem Port, den das Betriebssystem aussucht (Port 0). Wo es lauscht, traegt es
in die gemeinsame SQLite ein - die ist das Telefonbuch. Wer jemanden erreichen
will, schlaegt dort Port und PID nach, prueft die PID und ruft an.
Zwei Nachrichten reichen:
{"was": "ping"} -> {"antwort": "hallo, bin noch da"}
{"was": "schau_nach"} -> {"antwort": "ok"} (und das Fenster liest nach)
Der Push traegt keine Daten. Er sagt nur "schau in die SQLite" - dort steht die
Meldung ohnehin. Damit gibt es genau EINE Wahrheit, und wenn der Anruf nicht
durchgeht (Programm zu, Port belegt), findet das andere die Meldung beim
naechsten Nachsehen trotzdem.
SICHERHEIT
----------
Gebunden wird ausschliesslich auf 127.0.0.1 - von aussen ist da nichts zu
erreichen, und Windows fragt bei Loopback nicht nach der Firewall. Angenommen
werden nur die zwei Woerter oben; alles andere wird verworfen. Die API fuehrt
keine Befehle aus und uebertraegt keine Daten.
Diese Datei liegt in BEIDEN Programmen gleich. Wer sie aendert, kopiert sie
hinueber.
"""
import json
import logging
import os
import socket
import socketserver
import sys
import threading
log = logging.getLogger("api")
ANTWORT_HALLO = "hallo, bin noch da"
class _Handler(socketserver.StreamRequestHandler):
timeout = 3
def handle(self):
# Nur der eigene Rechner darf reden - gebunden ist ohnehin nur Loopback.
if self.client_address[0] not in ("127.0.0.1", "::1"):
return
try:
zeile = self.rfile.readline(4096).decode("utf-8", "replace").strip()
except OSError:
return
if not zeile:
return
try:
was = str(json.loads(zeile).get("was", ""))
except (ValueError, AttributeError):
was = ""
if was == "ping":
antwort = ANTWORT_HALLO
elif was == "schau_nach":
antwort = "ok"
self.server.melden() # Fenster benachrichtigen (threadsicher)
else:
antwort = "haeh?"
try:
self.wfile.write((json.dumps({"antwort": antwort}) + "\n").encode("utf-8"))
except OSError:
pass
class _Server(socketserver.ThreadingTCPServer):
allow_reuse_address = True
daemon_threads = True
def __init__(self, adresse, bei_push):
self.melden = bei_push
super().__init__(adresse, _Handler)
class MiniApi:
"""Der kleine Server eines Programms.
`bei_push` wird aus einem FREMDEN Faden gerufen - Tkinter darf man von dort
nicht anfassen. Der Aufrufer legt die Meldung deshalb in eine Warteschlange
und holt sie im Fenster mit after() ab.
"""
def __init__(self, programm, bei_push):
self.programm = programm
self.bei_push = bei_push
self.server = None
self.faden = None
self.port = 0
def starten(self, buch):
"""Server aufmachen und im Telefonbuch (SQLite) eintragen."""
if self.server is not None:
return self.port
try:
self.server = _Server(("127.0.0.1", 0), self.bei_push)
except OSError as e:
log.warning("API konnte nicht starten: %s", e)
self.server = None
return 0
self.port = self.server.server_address[1]
self.faden = threading.Thread(target=self.server.serve_forever,
name=f"api-{self.programm}", daemon=True)
self.faden.start()
log.info("API von %s laeuft auf 127.0.0.1:%s", self.programm, self.port)
self.eintragen(buch)
return self.port
def eintragen(self, buch):
"""Port und PID ins Telefonbuch. Ohne Buch gibt es keine API-Auskunft."""
if buch is None or not self.port:
return
try:
buch.anmelden(self.programm, os.getpid(),
os.path.basename(sys.executable), self.port)
except Exception as e: # noqa: BLE001
log.warning("Eintrag ins Telefonbuch fehlgeschlagen: %s", e)
def stoppen(self, buch=None):
if buch is not None:
try:
buch.abmelden(self.programm)
except Exception: # noqa: BLE001 - beim Schliessen nichts erzwingen
pass
if self.server is not None:
try:
self.server.shutdown()
self.server.server_close()
except Exception: # noqa: BLE001
pass
self.server = None
self.port = 0
def _reden(port, was, zeitgrenze=1.5):
"""Eine Zeile hinschicken und die Antwort lesen. None = niemand da."""
try:
with socket.create_connection(("127.0.0.1", int(port)), timeout=zeitgrenze) as s:
s.settimeout(zeitgrenze)
s.sendall((json.dumps({"was": was}) + "\n").encode("utf-8"))
roh = s.makefile("r", encoding="utf-8").readline()
return json.loads(roh).get("antwort", "")
except (OSError, ValueError):
return None
def erreichbar(buch, programm):
"""Meldet sich das andere Programm? ("hallo, bin noch da")
Zuerst die PID - die beantwortet Windows und kostet nichts. Erst wenn die
stimmt, wird ueberhaupt angerufen.
"""
if buch is None:
return False
try:
eintrag = buch.wer_laeuft(programm)
except Exception as e: # noqa: BLE001
log.debug("Telefonbuch nicht lesbar: %s", e)
return False
if not eintrag or not eintrag.get("port"):
return False
return _reden(eintrag["port"], "ping") == ANTWORT_HALLO
def anklopfen(buch, programm):
"""Push: "schau mal in die sqlite". True, wenn es angekommen ist."""
if buch is None:
return False
try:
eintrag = buch.wer_laeuft(programm)
except Exception as e: # noqa: BLE001
log.debug("Telefonbuch nicht lesbar: %s", e)
return False
if not eintrag or not eintrag.get("port"):
return False
return _reden(eintrag["port"], "schau_nach") == "ok"

50
app.py
View file

@ -10,6 +10,7 @@ Start: Doppelklick auf "Beherbergungssteuer starten.bat" oder `python app.py`
"""
import os
import queue
import sys
import logging
from datetime import date
@ -42,6 +43,7 @@ if __name__ == "__main__":
_wurzel.destroy()
sys.exit(0)
import api
import db
import gemeinsam
import ordnerwahl
@ -82,6 +84,9 @@ class App(tk.Tk):
self.journal = db.Journal(JOURNAL)
self.buch = None # gemeinsames Nummernbuch, siehe gemeinsam.py
# Der eigene kleine Server. Klopft das Rechnungstool an, wird sofort
# nachgesehen - ohne auf den naechsten Durchgang der Wache zu warten.
self.api = api.MiniApi("journal", self._push_empfangen)
self.satz = float(self.journal.hole("satz", str(STANDARD_SATZ)))
self.pdf_ordner = self.journal.hole("pdf_ordner", "")
self.jahr = tk.IntVar(value=int(self.journal.hole("jahr", str(date.today().year))))
@ -107,6 +112,8 @@ class App(tk.Tk):
# Beide Programme laufen nebeneinander: schreibt das Rechnungstool eine
# Rechnung, meldet es das hier herein - siehe _wache_jetzt().
self._briefkasten_stand = "?" # "?" = noch nicht nachgesehen
self._push_kasten = queue.Queue() # Zettel aus dem API-Faden
self.after(250, self._push_abholen)
self._nachricht_stand = 0 # bis hierher sind Meldungen gelesen
self.after(600, self._wache)
# Beim Zurueckklicken sofort nachsehen statt zwei Sekunden zu warten.
@ -338,18 +345,43 @@ class App(tk.Tk):
self._buch_oeffnen()
if self.buch is None:
return
try:
self.buch.anmelden("journal", os.getpid(), os.path.basename(sys.executable))
except Exception as e: # noqa: BLE001
log.warning("Anmeldung fehlgeschlagen: %s", e)
# Die Mini-API aufmachen und mit Port und PID ins Telefonbuch: daran
# sieht das Rechnungstool, dass jemand zuhoert, und kann anklopfen.
self.api.starten(self.buch)
if not self.api.port:
try:
self.buch.anmelden("journal", os.getpid(),
os.path.basename(sys.executable))
except Exception as e: # noqa: BLE001
log.warning("Anmeldung fehlgeschlagen: %s", e)
def _abmelden(self):
if self.buch is None:
return
self.api.stoppen(self.buch)
def _push_empfangen(self):
"""Ruft die API aus IHREM Faden - Tkinter darf man von dort nicht anfassen.
Also nur einen Zettel in die Warteschlange legen; das Fenster holt ihn
gleich ab (_push_abholen).
"""
self._push_kasten.put(1)
def _push_abholen(self):
"""Im Fenster-Faden: liegt ein Zettel da, sofort in der SQLite nachsehen."""
try:
self.buch.abmelden("journal")
except Exception as e: # noqa: BLE001 - beim Schliessen nichts erzwingen
log.debug("Abmelden fehlgeschlagen: %s", e)
geklopft = False
while True:
try:
self._push_kasten.get_nowait()
geklopft = True
except queue.Empty:
break
if geklopft:
self._wache_jetzt()
except Exception as e: # noqa: BLE001 - darf das Fenster nie umwerfen
log.warning("Push: %s", e)
finally:
self.after(250, self._push_abholen)
def _buch_oeffnen(self):
ordner = self.journal.hole("gemeinsam_ordner", "")

13
db.py
View file

@ -242,7 +242,7 @@ class Journal:
Gibt (neu, aktualisiert, uebersprungen, fehler) zurück.
"""
from pdf_parser import (parse_pdf, berichtigung_daten,
from pdf_parser import (parse_pdf, berichtigung_daten, keine_buchung,
ParserFehler) # lokal: hält modell/db leichtgewichtig
neu = akt = uebersprungen = 0
@ -266,11 +266,16 @@ class Journal:
# Frueher haette es die echte Buchung ueberschrieben. Es bucht
# nichts - aber die berichtigten Angaben zum Gast gehoeren
# uebernommen, sonst steht im Amtsbericht weiter der falsche Name.
berichtigt = berichtigung_daten(pfad)
if berichtigt is not None:
# Blaetter, die nichts abrechnen: das Berichtigungsblatt und die
# Proforma. keine_buchung() kennt beide - hier NICHT direkt
# berichtigung_daten() fragen, sonst rutscht die Proforma durch
# und steht als Buchung im Journal (gefunden am 03.09.2026).
if keine_buchung(pfad):
self.con.execute("DELETE FROM buchungen WHERE pdf_pfad=?",
(os.path.abspath(pfad),))
berichtigungen.append(berichtigt)
berichtigt = berichtigung_daten(pfad)
if berichtigt is not None:
berichtigungen.append(berichtigt)
continue
try:
b = parse_pdf(pfad, standard_satz)

48
gemeinsam.py
View file

@ -67,6 +67,7 @@ CREATE TABLE IF NOT EXISTS laeuft (
programm TEXT PRIMARY KEY, -- 'journal' | 'rechnungstool'
pid INTEGER NOT NULL DEFAULT 0,
exe TEXT NOT NULL DEFAULT '',
port INTEGER NOT NULL DEFAULT 0, -- wo seine Mini-API lauscht (127.0.0.1)
seit TEXT NOT NULL DEFAULT ''
);
@ -191,6 +192,11 @@ class Nummernbuch:
con = sqlite3.connect(self.pfad, timeout=10)
con.row_factory = sqlite3.Row
con.executescript(SCHEMA)
# Spalte nachziehen: CREATE TABLE IF NOT EXISTS aendert eine bestehende
# Tabelle nicht, und ein Buch aus der Fassung ohne API hat sie noch nicht.
spalten = {z["name"] for z in con.execute("PRAGMA table_info('laeuft')")}
if spalten and "port" not in spalten:
con.execute("ALTER TABLE laeuft ADD COLUMN port INTEGER NOT NULL DEFAULT 0")
return con
# ---- Lesen -------------------------------------------------------------
@ -248,17 +254,37 @@ class Nummernbuch:
con.execute("SELECT storno_zu FROM nummern WHERE storno_zu <> ''")}
# ---- Wer laeuft, und der Briefkasten -----------------------------------
def anmelden(self, programm, pid, exe):
"""Eintragen: dieses Programm ist offen."""
def anmelden(self, programm, pid, exe, port=0):
"""Eintragen: dieses Programm ist offen - und wo seine API lauscht.
Die Tabelle ist das Telefonbuch der beiden Programme: PID (laeuft es
ueberhaupt?) und Port (wo erreiche ich es?).
"""
import datetime
with self._verbindung() as con:
con.execute(
"INSERT INTO laeuft (programm, pid, exe, seit) VALUES (?,?,?,?) "
"INSERT INTO laeuft (programm, pid, exe, port, seit) VALUES (?,?,?,?,?) "
"ON CONFLICT(programm) DO UPDATE SET pid=excluded.pid, exe=excluded.exe, "
"seit=excluded.seit",
(str(programm), int(pid), str(exe),
"port=excluded.port, seit=excluded.seit",
(str(programm), int(pid), str(exe), int(port or 0),
datetime.datetime.now().isoformat(timespec="seconds")))
log.info("angemeldet: %s (PID %s)", programm, pid)
log.info("angemeldet: %s (PID %s, Port %s)", programm, pid, port)
def wer_laeuft(self, programm):
"""Telefonbuch-Eintrag eines laufenden Programms - oder None.
Geprueft wird die PID (Windows antwortet), nicht ein Zeitstempel. Eine
Karteileiche wird gleich weggeraeumt.
"""
with self._verbindung() as con:
z = con.execute("SELECT programm, pid, exe, port FROM laeuft WHERE programm=?",
(str(programm),)).fetchone()
if not z:
return None
if not prozess_laeuft(z["pid"], z["exe"]):
self.abmelden(programm)
return None
return dict(z)
def abmelden(self, programm):
with self._verbindung() as con:
@ -266,15 +292,7 @@ class Nummernbuch:
def laeuft(self, programm):
"""Laeuft dieses Programm noch? Karteileichen werden weggeraeumt."""
with self._verbindung() as con:
z = con.execute("SELECT pid, exe FROM laeuft WHERE programm=?",
(str(programm),)).fetchone()
if not z:
return False
if prozess_laeuft(z["pid"], z["exe"]):
return True
self.abmelden(programm)
return False
return self.wer_laeuft(programm) is not None
def melden(self, absender, text):
"""Eine Zeile in den Briefkasten legen."""

7
pdf_parser.py
View file

@ -198,7 +198,12 @@ def keine_buchung(pfad: str) -> bool:
die es berichtigt, und lauter Nullen. Wer es einliest, ueberschreibt damit
die echte Buchung - im Amtsbericht fehlt die Uebernachtung dann.
"""
return berichtigung_daten(pfad) is not None
if berichtigung_daten(pfad) is not None:
return True
# Die Proforma ist eine unverbindliche Vorab-Aufstellung, kein Umsatz:
# gemeldet wird eine Uebernachtung erst, wenn sie abgerechnet ist.
kopf = _aus_metadaten(pfad)
return bool(kopf and kopf.get("art") == "proforma")
def parse_pdf(pfad: str, standard_satz: float = STANDARD_SATZ) -> Buchung:

38
pruef_wache.py
View file

@ -87,6 +87,16 @@ pruefe(gemeinsam.journal_laeuft(gemeinsam_ordner),
"nach der Anmeldung sieht man an der PID, dass es laeuft")
pruefe(os.listdir(gemeinsam_ordner) == [gemeinsam.DATEINAME],
f"im gemeinsamen Ordner liegt genau EINE Datei ({os.listdir(gemeinsam_ordner)})")
# ---- 1b. Die Mini-API: Port im Telefonbuch, und sie meldet sich -----------
import api # noqa: E402
eintrag = gemeinsam.Nummernbuch(gemeinsam_ordner).wer_laeuft("journal")
pruefe(bool(eintrag and eintrag.get("port")),
f"das Journal hat Port und PID im Telefonbuch ({eintrag})")
pruefe(journal.api.port > 0, f"sein kleiner Server laeuft ({journal.api.port})")
pruefe(api.erreichbar(gemeinsam.Nummernbuch(gemeinsam_ordner), "journal"),
"auf ein ping antwortet es: hallo, bin noch da")
pruefe(gemeinsam.prozess_laeuft(os.getpid(), os.path.basename(sys.executable)),
"die PID-Pruefung erkennt den eigenen Prozess")
pruefe(not gemeinsam.prozess_laeuft(999999, ""),
@ -146,8 +156,15 @@ texte, _ = gemeinsam.Nummernbuch(gemeinsam_ordner).nachrichten_ab(0)
pruefe(any("Rechnung 2026-001 erstellt" in t for t in texte),
f"die Meldung steht in der SQLite ({texte})")
# ---- 4. Das Journal liest von selbst nach --------------------------------
pruefe(journal._wache_jetzt() is True, "die Wache sieht die Meldung")
# ---- 4. Der PUSH kommt an: das Journal liest ohne Wache nach -------------
# Nicht _wache_jetzt() aufrufen - genau das soll der Push von selbst ausloesen.
for _ in range(40):
if not journal._push_kasten.empty():
break
time.sleep(0.05)
pruefe(not journal._push_kasten.empty(),
"der Push des Rechnungstools ist im Fenster angekommen")
journal._push_abholen()
journal.update()
buchungen = journal.journal.buchungen(jahr=2026)
pruefe(len(buchungen) == 1 and buchungen[0].nachname == "Meldung",
@ -163,6 +180,13 @@ pruefe(len(zeilen) == 1, f"das Monatsblatt zeigt sie ({len(zeilen)} im Monat {mo
# ---- 5. Ohne neue Meldung passiert nichts --------------------------------
pruefe(journal._wache_jetzt() is False, "ohne neue Meldung wird nicht eingelesen")
# ---- 5b. Gegenseitig: auch das Rechnungstool ist erreichbar --------------
pruefe(tool.api.port > 0, f"das Rechnungstool hat auch einen kleinen Server ({tool.api.port})")
pruefe(api.erreichbar(gemeinsam.Nummernbuch(gemeinsam_ordner), "rechnungstool"),
"und meldet sich ebenfalls mit: hallo, bin noch da")
pruefe(api.anklopfen(gemeinsam.Nummernbuch(gemeinsam_ordner), "rechnungstool"),
"beim Rechnungstool laesst sich anklopfen (die beiden sehen sich gegenseitig)")
# ---- 6. Auch der Storno meldet sich --------------------------------------
import bestand as BE # noqa: E402
@ -176,6 +200,16 @@ storno = [b for b in journal.journal.buchungen(jahr=2026) if b.storno_zu]
pruefe(len(storno) == 1 and storno[0].storno_zu == "2026-001",
f"der Storno steht mit seinem Bezug im Journal ({[(b.voll_nummer, b.storno_zu) for b in storno]})")
# ---- 6b. Ist die API weg, findet die Meldung trotzdem ihren Weg ---------
journal.api.stoppen() # nur den Server zumachen, Eintrag bleibt stehen
pruefe(not api.erreichbar(gemeinsam.Nummernbuch(gemeinsam_ordner), "journal"),
"ohne Server antwortet niemand mehr auf das ping")
vorher_id = gemeinsam.Nummernbuch(gemeinsam_ordner).letzte_nachricht()
tool._journal_benachrichtigen("Probe ohne API")
pruefe(gemeinsam.Nummernbuch(gemeinsam_ordner).letzte_nachricht() > vorher_id,
"die Meldung steht trotzdem in der SQLite (Rueckfallebene)")
pruefe(journal._wache_jetzt() is True, "und die Wache findet sie beim naechsten Nachsehen")
# ---- 7. Ist das Journal zu, wird nicht gemeldet --------------------------
journal._abmelden()
pruefe(not gemeinsam.journal_laeuft(gemeinsam_ordner),