# -*- coding: utf-8 -*- """Das gemeinsame Nummernbuch von Rechnungstool und Steuerjournal. WARUM ES DAS GIBT ----------------- Bis hierher wusste jedes Programm nur, was in SEINEM Ordner lag. Das Rechnungstool sucht die naechste freie Nummer in den PDFs im Rechnungsordner - im Steuerjournal stehen aber auch Buchungen, zu denen es gar keine PDF gibt: der Altbestand aus der Excel-Mappe und die frueher von Hand eingetragenen Zeilen. Deren Nummern kannte das Rechnungstool nicht und hat sie ein zweites Mal vergeben. Jede Rechnungsnummer darf es aber nur einmal geben (Paragraph 14 Abs. 4 Nr. 4 UStG). Deshalb fuehren beide Programme ab jetzt EIN gemeinsames Buch: eine SQLite-Datei in einem Ordner, den beide kennen. Wer eine Nummer vergibt oder eine Buchung anlegt, traegt sie dort ein; wer eine neue Nummer sucht, schaut zuerst hier nach. Die Datei ist ein REGISTER, keine zweite Buchhaltung: die Wahrheit bleiben die Rechnungs-PDFs. Geht das Buch verloren, fehlt keine Rechnung - es muss nur neu gefuellt werden (beide Programme tun das beim naechsten Einlesen von selbst). Diese Datei liegt in BEIDEN Programmen gleich. Wer sie aendert, kopiert sie hinueber - sonst schreiben zwei Fassungen in dieselbe Datei. """ import logging import os import sqlite3 import sys import time log = logging.getLogger("gemeinsam") DATEINAME = "campinghof-nummernbuch.sqlite3" SCHEMA = """ CREATE TABLE IF NOT EXISTS nummern ( nummer TEXT PRIMARY KEY, -- '2026-442', immer die volle Nummer jahr INTEGER NOT NULL DEFAULT 0, datum TEXT NOT NULL DEFAULT '', -- ISO JJJJ-MM-TT nachname TEXT NOT NULL DEFAULT '', vorname TEXT NOT NULL DEFAULT '', naechte INTEGER NOT NULL DEFAULT 0, zwischensumme REAL NOT NULL DEFAULT 0, steuer_satz REAL NOT NULL DEFAULT 0, steuer_betrag REAL NOT NULL DEFAULT 0, gesamt REAL NOT NULL DEFAULT 0, art TEXT NOT NULL DEFAULT 'rechnung', -- rechnung|storno|korrektur|berichtigung vorgang TEXT NOT NULL DEFAULT '', -- Klammer um Rechnung+Storno+Korrektur storno_zu TEXT NOT NULL DEFAULT '', -- welche Rechnung hebt dieser Storno auf ersetzt TEXT NOT NULL DEFAULT '', -- welche Rechnung ersetzt diese hier berichtigt_zu TEXT NOT NULL DEFAULT '', -- zu welcher Rechnung gehoert das Blatt vorlage TEXT NOT NULL DEFAULT '', -- '' oder 'montage' quelle TEXT NOT NULL DEFAULT '', -- rechnungstool|journal herkunft TEXT NOT NULL DEFAULT '', -- pdf|xlsx|manuell pdf_pfad TEXT NOT NULL DEFAULT '', freigabe TEXT NOT NULL DEFAULT '', -- ISO-Zeitstempel der -- ausdruecklichen Freigabe -- durch den Betreiber geaendert TEXT NOT NULL DEFAULT '' -- ISO-Zeitstempel ); CREATE INDEX IF NOT EXISTS idx_nummern_jahr ON nummern (jahr); -- Wer gerade laeuft. Eine Zeile je Programm, mit PID und EXE-Namen: ob es -- wirklich noch laeuft, beantwortet Windows (prozess_laeuft), nicht ein -- Zeitstempel. Ein abgestuerztes Programm hinterlaesst hier nur eine -- Karteileiche, die beim naechsten Nachsehen weggeraeumt wird. 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 '' ); -- Der Briefkasten. Das Rechnungstool legt hier ab, dass es eine Rechnung -- geschrieben hat; das Journal liest die neuen Zeilen und liest die PDFs nach. CREATE TABLE IF NOT EXISTS nachrichten ( id INTEGER PRIMARY KEY AUTOINCREMENT, zeit TEXT NOT NULL DEFAULT '', absender TEXT NOT NULL DEFAULT '', text TEXT NOT NULL DEFAULT '' ); """ FELDER = ("jahr", "datum", "nachname", "vorname", "naechte", "zwischensumme", "steuer_satz", "steuer_betrag", "gesamt", "art", "vorgang", "storno_zu", "ersetzt", "berichtigt_zu", "vorlage", "quelle", "herkunft", "pdf_pfad", "freigabe") def standard_ordner(): """Wo das Nummernbuch liegt, wenn niemand etwas anderes einstellt. WARUM ES DIESEN FESTEN ORT GIBT (07.09.2026) -------------------------------------------- Vorher fragte jedes Programm beim ersten Start nach einem Ordner - und der Nutzer musste in BEIDEN denselben eintragen. Sein Einwand, wortwoertlich: "wie soll das rechnungstool und steuertool wissen wo es ist, das muss im ... ravokk liegen und dann zusammen, sonst kommt das eine nicht an das andere ran" Er hat recht: die beiden sind ueber genau zwei Dinge verbunden - diese SQLite-Datei und die kleine API, die "hier ist etwas Neues" ruft. Wenn schon der Ort geraten werden muss, ist die Verbindung nur so gut wie die Aufmerksamkeit dessen, der zweimal denselben Pfad tippt. Ein falscher Buchstabe faellt erst auf, wenn eine Rechnungsnummer zum zweiten Mal vergeben wurde. Deshalb: ein fester Ort, den beide von selbst kennen. Gefragt wird nur noch, wer ihn ausdruecklich verschieben will (Menue -> Gemeinsamer Ordner) - etwa auf ein Netzlaufwerk, wenn die Programme auf zwei Rechnern laufen. LOCALAPPDATA und nicht "Dokumente": dort liegen die Programmdaten ohnehin, und niemand raeumt den Ordner versehentlich auf. Das Buch ist ein REGISTER, kein Beleg - geht es verloren, fehlt keine Rechnung, es wird beim naechsten Einlesen neu gefuellt. """ heim = os.environ.get("LOCALAPPDATA") or os.path.expanduser("~") return os.path.join(heim, "ravokk") def standard_anlegen(): """Den festen Ordner anlegen und zurueckgeben - oder '' wenn das nicht geht.""" ordner = standard_ordner() try: os.makedirs(ordner, exist_ok=True) return ordner except OSError as e: log.warning("Gemeinsamer Ordner %s liess sich nicht anlegen: %s", ordner, e) return "" def pfad_zur_datei(ordner): """Vollstaendiger Pfad der gemeinsamen Datei in diesem Ordner.""" return os.path.join(ordner or "", DATEINAME) def ordner_taugt(ordner): """Kann in diesem Ordner wirklich gearbeitet werden? Wird ausprobiert, nicht geraten - ein nicht verbundenes Netzlaufwerk sieht sonst gut aus, bis die erste Rechnung geschrieben wird.""" if not ordner or not os.path.isdir(ordner): return False, "Den Ordner gibt es nicht." probe = os.path.join(ordner, ".schreibprobe") try: with open(probe, "w", encoding="ascii") as f: f.write("x") os.remove(probe) except OSError as e: return False, f"In den Ordner kann nicht geschrieben werden ({e.strerror or e})." return True, "" # --------------------------------------------------------------------------- # Briefkasten und Herzschlag: das Rechnungstool sagt dem Steuerjournal Bescheid # # Kein Programm durchsucht dauernd den Rechnungsordner. Stattdessen legt das # LAUFENDE Steuerjournal einen Herzschlag im gemeinsamen Ordner ab, und das # Rechnungstool schaut nach dem Schreiben einer Rechnung nur kurz hinein: laeuft # es, wirft es eine Nachricht in den Briefkasten. Das Journal sieht genau EINE # Datei nach (os.stat) und liest die Rechnung dann von selbst ein. # # Ist das Journal zu, passiert nichts - es liest den Ordner beim naechsten Start # ohnehin komplett. Die Nachricht ist eine Anklopfhilfe, keine Datenquelle. def prozess_laeuft(pid, name=""): """Laeuft unter dieser Prozessnummer wirklich noch etwas? Die Antwort kommt von Windows, nicht aus einer Uhrzeit und nicht aus einer Fenstersuche. PIDs werden wiederverwendet - deshalb wird zusaetzlich der Name der EXE verglichen, wenn einer eingetragen wurde. Sonst haette ein fremder Prozess mit derselben Nummer das andere Programm "am Leben" gehalten. """ if not pid: return False if not sys.platform.startswith("win"): try: os.kill(int(pid), 0) return True except (OSError, ValueError): return False import ctypes from ctypes import wintypes PROCESS_QUERY_LIMITED = 0x1000 k32 = ctypes.WinDLL("kernel32", use_last_error=True) k32.OpenProcess.restype = wintypes.HANDLE griff = k32.OpenProcess(PROCESS_QUERY_LIMITED, False, int(pid)) if not griff: return False try: if not name: return True puffer = ctypes.create_unicode_buffer(4096) laenge = wintypes.DWORD(len(puffer)) if not k32.QueryFullProcessImageNameW(griff, 0, puffer, ctypes.byref(laenge)): return False return os.path.basename(puffer.value).lower() == str(name).lower() finally: k32.CloseHandle(griff) def datei_stand(ordner): """(Aenderungszeit, Groesse) der gemeinsamen Datei - oder None. Damit sieht ein Programm mit EINEM os.stat, ob sich im Buch etwas getan hat, ohne dafuer die Datenbank zu oeffnen. """ try: st = os.stat(pfad_zur_datei(ordner)) except OSError: return None return st.st_mtime, st.st_size def journal_laeuft(ordner): """Ist das Steuerjournal gerade offen?""" if not ordner or not os.path.exists(pfad_zur_datei(ordner)): return False try: return Nummernbuch(ordner).laeuft("journal") except Exception as e: # noqa: BLE001 - eine Auskunft darf nichts umwerfen log.debug("Anmeldung nicht lesbar: %s", e) return False class Nummernbuch: """Das gemeinsame Register. Wird zum Schreiben kurz geoeffnet und wieder geschlossen - zwei Programme haben die Datei sonst dauerhaft in der Hand, und auf einem Netzlaufwerk gibt das gesperrte Dateien.""" def __init__(self, ordner): self.ordner = ordner self.pfad = pfad_zur_datei(ordner) def _verbindung(self): 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") # Ebenso die Freigabe-Spalte: ein Buch aus der Fassung vor dem 06.09.2026 # kennt sie noch nicht. Alte Zeilen bleiben leer - dort wurde nie gefragt, # und ein nachtraeglich erfundener Zeitstempel waere schlimmer als keiner. spalten_n = {z["name"] for z in con.execute("PRAGMA table_info('nummern')")} if spalten_n and "freigabe" not in spalten_n: con.execute("ALTER TABLE nummern ADD COLUMN freigabe TEXT NOT NULL DEFAULT ''") return con # ---- Lesen ------------------------------------------------------------- def nummern(self, jahr=None): """Alle vergebenen Nummern (als Menge), auf Wunsch nur eines Jahres.""" with self._verbindung() as con: if jahr: zeilen = con.execute("SELECT nummer FROM nummern WHERE jahr=?", (int(jahr),)) else: zeilen = con.execute("SELECT nummer FROM nummern") return {z["nummer"] for z in zeilen} def hoechster_zaehler(self, jahr): """Groesster Zaehler dieses Jahres im Buch (0, wenn nichts drinsteht). Aus '2026-442' wird 442. Nummern anderer Reihen (Angebote 2026-A001) zaehlen NICHT mit - die haben mit der Rechnungsnummer nichts zu tun. """ hoch = 0 for nummer in self.nummern(jahr): teil = str(nummer).split("-")[-1] if teil.isdigit(): hoch = max(hoch, int(teil)) return hoch def eintrag(self, nummer): with self._verbindung() as con: z = con.execute("SELECT * FROM nummern WHERE nummer=?", (str(nummer),)).fetchone() return dict(z) if z else None def alle(self): with self._verbindung() as con: return [dict(z) for z in con.execute("SELECT * FROM nummern ORDER BY nummer")] def belegt(self, nummer): return self.eintrag(nummer) is not None # ---- Schreiben --------------------------------------------------------- def eintragen(self, nummer, **werte): """Nummer eintragen oder ihre Angaben auffrischen. Eine schon eingetragene Nummer wird NICHT als Fehler behandelt: beide Programme sehen dieselbe Rechnung (das Rechnungstool beim Schreiben, das Journal beim Einlesen der PDF) und sollen sich gegenseitig ergaenzen duerfen. """ import datetime nummer = str(nummer or "").strip() if not nummer: return False daten = {f: werte.get(f) for f in FELDER if werte.get(f) is not None} daten["geaendert"] = datetime.datetime.now().isoformat(timespec="seconds") spalten = ["nummer"] + list(daten) platzhalter = ",".join("?" * len(spalten)) setzen = ",".join(f"{k}=excluded.{k}" for k in daten) with self._verbindung() as con: con.execute( f"INSERT INTO nummern ({','.join(spalten)}) VALUES ({platzhalter}) " f"ON CONFLICT(nummer) DO UPDATE SET {setzen}", [nummer] + list(daten.values())) log.info("Nummernbuch: %s eingetragen (%s)", nummer, daten.get("quelle", "")) return True def storniert(self): """Nummern, die durch einen Storno aufgehoben sind.""" with self._verbindung() as con: return {z["storno_zu"] for z in con.execute("SELECT storno_zu FROM nummern WHERE storno_zu <> ''")} # ---- Wer laeuft, und der Briefkasten ----------------------------------- 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, port, seit) VALUES (?,?,?,?,?) " "ON CONFLICT(programm) DO UPDATE SET pid=excluded.pid, exe=excluded.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, 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: con.execute("DELETE FROM laeuft WHERE programm=?", (str(programm),)) def laeuft(self, programm): """Laeuft dieses Programm noch? Karteileichen werden weggeraeumt.""" return self.wer_laeuft(programm) is not None def melden(self, absender, text): """Eine Zeile in den Briefkasten legen.""" import datetime with self._verbindung() as con: con.execute("INSERT INTO nachrichten (zeit, absender, text) VALUES (?,?,?)", (datetime.datetime.now().isoformat(timespec="seconds"), str(absender), str(text))) log.info("Nachricht von %s: %s", absender, text) return True def letzte_nachricht(self): """Hoechste Nachrichten-Nummer (0, wenn der Kasten leer ist).""" with self._verbindung() as con: z = con.execute("SELECT MAX(id) AS m FROM nachrichten").fetchone() return int(z["m"] or 0) def nachrichten_ab(self, seit_id): """Alle Nachrichten NACH dieser Nummer: (Liste von Texten, neue Nummer).""" with self._verbindung() as con: zeilen = list(con.execute( "SELECT id, absender, text FROM nachrichten WHERE id > ? ORDER BY id", (int(seit_id or 0),))) if not zeilen: return [], int(seit_id or 0) return [z["text"] for z in zeilen], int(zeilen[-1]["id"]) def aufraeumen(self, behalten=200): """Alte Nachrichten wegwerfen - der Briefkasten ist kein Archiv.""" with self._verbindung() as con: con.execute("DELETE FROM nachrichten WHERE id <= " "(SELECT MAX(id) FROM nachrichten) - ?", (int(behalten),)) def entfernen(self, nummer): """Nur fuer den Rueckweg: ein verworfener Storno gibt seine Nummer zurueck.""" with self._verbindung() as con: con.execute("DELETE FROM nummern WHERE nummer=?", (str(nummer),)) log.info("Nummernbuch: %s entfernt", nummer)