rechnungstool/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

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

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


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

    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)