TheMockTv/rechnungstool
1
0
Fork 0
Code Issues Pull requests Projects Releases 17 Packages Wiki Activity Actions

Amt-Blatt raus, Bedienung ueber die API, Erststart fragt nach zwei Wegen

Browse source 
Das zweite Blatt "fuer das Amt" trug DIESELBE Rechnungsnummer, aber einen
anderen Gesamtbetrag. Mehrere Blaetter zum selben Umsatz mit verschiedenen
Endbetraegen sehen aus, als werde ueber verschiedene Umsaetze abgerechnet -
bei einer Pruefung ein Problem. Ersatzlos entfallen; was die Stadt braucht,
rechnet das Steuerjournal aus den Kenndaten der PDF. Mit dem Blatt fielen
_AmtMarke, gesamtrechnung_betont und der doppelte Renderlauf weg.

Die API kann das Fenster jetzt BEDIENEN - mit Schluessel wie bei einem SSH-Key:
32 Byte Zufall in api_schluessel.txt im gemeinsamen Ordner, den nur das
Steuerjournal und der Betreiber lesen. Ohne gueltigen Schluessel kein Zutritt
(compare_digest). ping und schau_nach bleiben schluessellos, sonst kann ein
aelteres Journal nicht mehr anklopfen. Was einen BELEG schreibt, geht nur im
Testlauf - eine Fernbedienung, die echte Rechnungen ausstellt, hat in einer
Buchhaltung nichts zu suchen. api.py liegt wie immer in beiden Programmen.

Der Erststart fragte nach "der zuletzt vergebenen Nummer" - das holt niemanden
ab, der bei 0 anfaengt. Jetzt zwei Wege: bei 0 anfangen ODER die eigene
Zaehlung weiterfuehren. Unter das schon Vergebene kommt keiner der beiden.

pruef_nummernsperre bediente den Dialog nur halb: nach einer abgelehnten Zahl
bleibt er offen (so soll er), niemand drueckte "Spaeter" - das Programm wartete
in wait_window() ewig. Das war schon vor diesem Umbau so.

Neu: pruef_api.py. Alle 14 Pruefstaende gruen.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V4t48uxDok5rJXhC9bX1Ax
This commit is contained in:
TheMockTv 2026-09-05 18:24:59 +02:00
parent 142dddc647
commit ca9ac212cf
11 changed files with 835 additions and 212 deletions
Download patch file Download diff file Expand all files Collapse all files

181
api.py
View file

@ -18,11 +18,38 @@ 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:
Zwei Nachrichten reichen dafuer:
{"was": "ping"} -> {"antwort": "hallo, bin noch da"}
{"was": "schau_nach"} -> {"antwort": "ok"} (und das Fenster liest nach)
Dazu kommt eine dritte, die das Fenster BEDIENT - damit ein Pruefstand die
Knoepfe druecken kann, ohne dem Menschen am Rechner Maus und Tastatur
wegzunehmen:
{"was": "aktion", "schluessel": "...", "name": "vorschau"} -> {"antwort": {...}}
DER SCHLUESSEL
--------------
Wer bedient, muss sich ausweisen - wie mit einem SSH-Schluessel. Beim ersten
Start legt das Programm eine Zufallszahl in `api_schluessel.txt` **im
gemeinsamen Ordner** ab (dort steht auch das Nummernbuch). Wer diesen Ordner
lesen darf, darf sprechen: das Steuerjournal und der Betreiber - sonst niemand.
Ohne gueltigen Schluessel wird eine Aktion verworfen, und zwar ohne zu
verraten, ob der Name ueberhaupt existiert.
`ping` und `schau_nach` bleiben schluessellos: sie sagen nur "ich lebe" bzw.
"schau in die SQLite" und veraendern nichts. Sonst koennte ein aelteres
Steuerjournal nicht mehr anklopfen.
WAS EINE AKTION DARF
--------------------
Alles, was einen BELEG SCHREIBT (Rechnung, Angebot, Storno), geht nur im
Testlauf (`--test` / `CAMPINGHOF_TEST=1`). Im normalen Betrieb bleiben nur
Aktionen, die nichts in die Buchhaltung schreiben: nachsehen, Felder fuellen,
Vorschau zeigen. Eine Fernbedienung, die echte Rechnungen ausstellt, hat in
einer Buchhaltung nichts zu suchen.
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
@ -42,6 +69,8 @@ hinueber.
import json
import logging
import os
import queue
import secrets
import socket
import socketserver
import sys
@ -50,6 +79,33 @@ import threading
log = logging.getLogger("api")
ANTWORT_HALLO = "hallo, bin noch da"
SCHLUESSEL_DATEI = "api_schluessel.txt"
def schluessel_holen(ordner):
"""Schluessel aus dem Ordner lesen; ist keiner da, einen anlegen.
Der Schluessel ist eine 32 Byte lange Zufallszahl in Hex. Er liegt als
Datei neben dem Nummernbuch, damit beide Programme denselben lesen - eine
Absprache ueber den gemeinsamen Ordner, ohne Server und ohne Passwort, das
sich jemand merken muesste.
"""
if not ordner:
return ""
pfad = os.path.join(ordner, SCHLUESSEL_DATEI)
try:
if os.path.exists(pfad):
wert = open(pfad, encoding="utf-8").read().strip()
if wert:
return wert
wert = secrets.token_hex(32)
with open(pfad, "w", encoding="utf-8") as f:
f.write(wert)
log.info("neuer API-Schluessel angelegt: %s", pfad)
return wert
except OSError as e:
log.warning("API-Schluessel nicht nutzbar (%s)", e)
return ""
class _Handler(socketserver.StreamRequestHandler):
@ -66,14 +122,17 @@ class _Handler(socketserver.StreamRequestHandler):
if not zeile:
return
try:
was = str(json.loads(zeile).get("was", ""))
nachricht = json.loads(zeile)
was = str(nachricht.get("was", ""))
except (ValueError, AttributeError):
was = ""
nachricht, was = {}, ""
if was == "ping":
antwort = ANTWORT_HALLO
elif was == "schau_nach":
antwort = "ok"
self.server.melden() # Fenster benachrichtigen (threadsicher)
elif was == "aktion":
antwort = self._aktion(nachricht)
else:
antwort = "haeh?"
try:
@ -82,12 +141,35 @@ class _Handler(socketserver.StreamRequestHandler):
pass
def _aktion(self, nachricht):
"""Eine Bedienung ausfuehren - nur mit gueltigem Schluessel."""
erwartet = getattr(self.server, "schluessel", "")
gegeben = str(nachricht.get("schluessel", ""))
# compare_digest statt "==": ein Vergleich, der bei der ersten falschen
# Stelle abbricht, verraet ueber die Zeit, wie weit man richtig lag.
if not erwartet or not secrets.compare_digest(erwartet, gegeben):
log.warning("Aktion ohne gueltigen Schluessel abgewiesen")
return "kein zutritt"
if self.server.aktion is None:
return "keine bedienung moeglich"
name = str(nachricht.get("name", ""))
daten = nachricht.get("daten") or {}
log.info("Aktion ueber die API: %s", name)
try:
return self.server.aktion(name, daten)
except Exception as e: # noqa: BLE001 - der Server darf daran nicht sterben
log.warning("Aktion %r fehlgeschlagen: %s", name, e)
return {"fehler": str(e)}
class _Server(socketserver.ThreadingTCPServer):
allow_reuse_address = True
daemon_threads = True
def __init__(self, adresse, bei_push):
def __init__(self, adresse, bei_push, aktion=None, schluessel=""):
self.melden = bei_push
self.aktion = aktion
self.schluessel = schluessel
super().__init__(adresse, _Handler)
@ -99,19 +181,30 @@ class MiniApi:
und holt sie im Fenster mit after() ab.
"""
def __init__(self, programm, bei_push):
def __init__(self, programm, bei_push, aktion=None):
self.programm = programm
self.bei_push = bei_push
self.aktion = aktion # (name, daten) -> Antwort; None = keine Bedienung
self.server = None
self.faden = None
self.port = 0
self.schluessel = ""
def starten(self, buch):
"""Server aufmachen und im Telefonbuch (SQLite) eintragen."""
def starten(self, buch, ordner=None):
"""Server aufmachen und im Telefonbuch (SQLite) eintragen.
Der Schluessel kommt aus dem Ordner des Nummernbuchs - dort liegt er
fuer beide Programme lesbar. Gibt es kein Buch, gibt es keinen
Schluessel und damit auch keine Bedienung von aussen.
"""
if self.server is not None:
return self.port
if ordner is None and buch is not None:
ordner = os.path.dirname(getattr(buch, "pfad", "") or "")
self.schluessel = schluessel_holen(ordner)
try:
self.server = _Server(("127.0.0.1", 0), self.bei_push)
self.server = _Server(("127.0.0.1", 0), self.bei_push,
aktion=self.aktion, schluessel=self.schluessel)
except OSError as e:
log.warning("API konnte nicht starten: %s", e)
self.server = None
@ -150,12 +243,14 @@ class MiniApi:
self.port = 0
def _reden(port, was, zeitgrenze=1.5):
def _reden(port, was, zeitgrenze=1.5, **rest):
"""Eine Zeile hinschicken und die Antwort lesen. None = niemand da."""
nachricht = {"was": was}
nachricht.update(rest)
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"))
s.sendall((json.dumps(nachricht) + chr(10)).encode("utf-8"))
roh = s.makefile("r", encoding="utf-8").readline()
return json.loads(roh).get("antwort", "")
except (OSError, ValueError):
@ -192,3 +287,69 @@ def anklopfen(buch, programm):
if not eintrag or not eintrag.get("port"):
return False
return _reden(eintrag["port"], "schau_nach") == "ok"
def bedienen(buch, programm, name, daten=None, ordner=None, zeitgrenze=15.0):
"""Einen Knopf im anderen Programm druecken. None = niemand da / kein Zutritt.
Fuer Pruefstaende und fuer den Betreiber: damit laesst sich das Fenster
bedienen, ohne dem Menschen am Rechner Maus und Tastatur wegzunehmen.
Der Schluessel kommt aus demselben Ordner wie das Nummernbuch - wer den
lesen darf, darf bedienen.
"""
if buch is None:
return None
try:
eintrag = buch.wer_laeuft(programm)
except Exception as e: # noqa: BLE001
log.debug("Telefonbuch nicht lesbar: %s", e)
return None
if not eintrag or not eintrag.get("port"):
return None
if ordner is None:
ordner = os.path.dirname(getattr(buch, "pfad", "") or "")
return _reden(eintrag["port"], "aktion", zeitgrenze=zeitgrenze,
schluessel=schluessel_holen(ordner), name=name,
daten=daten or {})
class Warteschlange:
"""Aktionen aus dem Netzfaden in den Fensterfaden bringen - und zurueck.
Tkinter darf nur aus SEINEM Faden angefasst werden. Der API-Server lauscht
aber in einem eigenen. Deshalb legt er die Aktion hier ab und wartet auf das
Ergebnis; das Fenster holt sie in seiner after()-Schleife ab, fuehrt sie aus
und legt die Antwort zurueck.
Kommt das Fenster nicht dazu (es haengt in einem Dialog), laeuft die
Wartezeit ab und der Anrufer bekommt eine ehrliche Absage statt einer
Verbindung, die nie antwortet.
"""
def __init__(self):
self.eingang = queue.Queue()
def annehmen(self, name, daten, zeitgrenze=12.0):
"""Aus dem Netzfaden gerufen: Aktion einreihen und auf das Ergebnis warten."""
fertig = threading.Event()
auftrag = {"name": name, "daten": daten, "fertig": fertig, "antwort": None}
self.eingang.put(auftrag)
if not fertig.wait(zeitgrenze):
return {"fehler": "das Fenster hat nicht geantwortet"}
return auftrag["antwort"]
def abarbeiten(self, ausfuehren):
"""Im Fensterfaden gerufen: alles erledigen, was eingegangen ist."""
while True:
try:
auftrag = self.eingang.get_nowait()
except queue.Empty:
return
try:
auftrag["antwort"] = ausfuehren(auftrag["name"], auftrag["daten"])
except Exception as e: # noqa: BLE001 - eine Aktion darf das Fenster nicht killen
log.warning("Aktion %r fehlgeschlagen: %s", auftrag["name"], e)
auftrag["antwort"] = {"fehler": str(e)}
finally:
auftrag["fertig"].set()