Steuerrechnungstool/gemeinsam.py

# -*- 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 '',
    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")


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")
        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 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)