# -*- coding: utf-8 -*- """ SQLite-Ablage des Steuerjournals. Die Wahrheit sind die Rechnungs-PDFs – die Datenbank ist nur der schnelle Zwischenspeicher, damit die Oberfläche nicht bei jedem Start alle PDFs neu lesen muss. Ein Scan liest nur PDFs, die neu sind oder sich geändert haben (Pfad + Änderungszeit + Größe). Altbestand (Zeilen aus der alten Excel-Tabelle) und von Hand erfasste Buchungen stehen mit quelle='xlsx' bzw. 'manuell' daneben und werden vom PDF-Scan nicht angefasst. """ import os import logging import sqlite3 from datetime import date, datetime from modell import Buchung log = logging.getLogger("bst.db") SCHEMA = """ CREATE TABLE IF NOT EXISTS buchungen ( id INTEGER PRIMARY KEY AUTOINCREMENT, jahr INTEGER NOT NULL, rechnungsnummer TEXT NOT NULL, datum TEXT NOT NULL, -- ISO: YYYY-MM-DD nachname TEXT NOT NULL DEFAULT '', naechte INTEGER NOT NULL DEFAULT 0, gezahlt REAL NOT NULL DEFAULT 0, satz REAL NOT NULL DEFAULT 5, quelle TEXT NOT NULL DEFAULT 'pdf', art TEXT NOT NULL DEFAULT 'rechnung', storno_zu TEXT NOT NULL DEFAULT '', storno_datum TEXT NOT NULL DEFAULT '', vorgang TEXT NOT NULL DEFAULT '', folge_nummer TEXT NOT NULL DEFAULT '', folge_art TEXT NOT NULL DEFAULT '', pdf_pfad TEXT NOT NULL DEFAULT '', pdf_mtime REAL NOT NULL DEFAULT 0, pdf_groesse INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_datum ON buchungen (jahr, datum); CREATE INDEX IF NOT EXISTS idx_nummer ON buchungen (jahr, rechnungsnummer); -- Der Schluessel ist die DATEI, nicht die Rechnungsnummer. Vorher stand hier -- UNIQUE (jahr, rechnungsnummer): eine zweite Rechnung mit derselben Nummer -- ueberschrieb die erste still, und im Amtsbericht fehlte die Uebernachtung. -- Buchungen von Hand (quelle != 'pdf') haben keinen Pfad und bleiben frei. CREATE UNIQUE INDEX IF NOT EXISTS idx_buchungen_pdf ON buchungen (pdf_pfad) WHERE pdf_pfad <> ''; CREATE TABLE IF NOT EXISTS einstellungen ( schluessel TEXT PRIMARY KEY, wert TEXT NOT NULL ); """ class Journal: def __init__(self, pfad: str): self.pfad = pfad neu = not os.path.exists(pfad) self.con = sqlite3.connect(pfad) self.con.row_factory = sqlite3.Row self._alte_nummern_sperre_loesen() self.con.executescript(SCHEMA) self._spalten_ergaenzen() self.con.commit() log.info("Journal %s (%s)", pfad, "neu angelegt" if neu else "geöffnet") def _alte_nummern_sperre_loesen(self): """Bestehende Journale einmalig auf den neuen Schluessel umbauen. Bis hierher war (jahr, rechnungsnummer) eindeutig - eine zweite Rechnung mit derselben Nummer hat die erste ueberschrieben und fehlte danach im Amtsbericht. ALLES ODER NICHTS: bricht der Umbau ab, bleibt der alte Stand stehen, sonst waere die Tabelle umbenannt und die neue leer. """ da = self.con.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='buchungen'").fetchone() if not da: return alt_sperre = False for zeile in self.con.execute("PRAGMA index_list('buchungen')"): if zeile["unique"] and zeile["origin"] == "u": spalten = [r["name"] for r in self.con.execute(f"PRAGMA index_info('{zeile['name']}')")] if spalten == ["jahr", "rechnungsnummer"]: alt_sperre = True if not alt_sperre: return log.warning("Journal wird umgebaut: Schlüssel war die Rechnungsnummer, " "jetzt die PDF-Datei") try: self.con.execute("BEGIN IMMEDIATE") self.con.execute("ALTER TABLE buchungen RENAME TO buchungen_alt") for befehl in SCHEMA.split(";"): if befehl.strip(): self.con.execute(befehl) # Denselben Pfad kann es mehrfach geben (alter Schluessel war die # Nummer) - die juengste Zeile gewinnt, sonst scheitert der Index. self.con.execute(""" INSERT INTO buchungen (jahr, rechnungsnummer, datum, nachname, naechte, gezahlt, satz, quelle, pdf_pfad, pdf_mtime, pdf_groesse) SELECT jahr, rechnungsnummer, datum, nachname, naechte, gezahlt, satz, quelle, pdf_pfad, pdf_mtime, pdf_groesse FROM buchungen_alt WHERE pdf_pfad = '' OR id IN ( SELECT MAX(id) FROM buchungen_alt WHERE pdf_pfad <> '' GROUP BY pdf_pfad ) """) genommen = self.con.execute("SELECT COUNT(*) FROM buchungen").fetchone()[0] self.con.execute("DROP TABLE buchungen_alt") self.con.execute("COMMIT") except Exception as e: # noqa: BLE001 - hier gibt es keinen halben Umbau self.con.execute("ROLLBACK") log.error("Umbau des Journals fehlgeschlagen, alter Stand bleibt: %s", e) raise log.info("Umbau fertig (%d Buchungen) - doppelte Rechnungsnummern bleiben " "ab jetzt sichtbar", genommen) def _spalten_ergaenzen(self): """Spalten nachziehen, die es in aelteren Journalen noch nicht gab. CREATE TABLE IF NOT EXISTS aendert eine bestehende Tabelle nicht - ohne das hier faenden die neuen Felder in einer alten Datei kein Zuhause. """ da = {z["name"] for z in self.con.execute("PRAGMA table_info('buchungen')")} for name, vorgabe in (("art", "'rechnung'"), ("storno_zu", "''"), ("storno_datum", "''"), ("vorgang", "''"), ("folge_nummer", "''"), ("folge_art", "''")): if name not in da: self.con.execute( f"ALTER TABLE buchungen ADD COLUMN {name} TEXT NOT NULL DEFAULT {vorgabe}") log.info("Journal um Spalte %s ergänzt", name) self.con.commit() def schliessen(self): self.con.close() # ---- Einstellungen ----------------------------------------------------- def hole(self, schluessel: str, standard: str = "") -> str: r = self.con.execute("SELECT wert FROM einstellungen WHERE schluessel=?", (schluessel,)).fetchone() return r["wert"] if r else standard def setze(self, schluessel: str, wert: str): self.con.execute( "INSERT INTO einstellungen (schluessel, wert) VALUES (?, ?) " "ON CONFLICT(schluessel) DO UPDATE SET wert=excluded.wert", (schluessel, str(wert))) self.con.commit() # ---- Lesen ------------------------------------------------------------- def jahre(self): return [r["jahr"] for r in self.con.execute("SELECT DISTINCT jahr FROM buchungen ORDER BY jahr")] def buchungen(self, jahr: int = None, monat: int = None): sql = "SELECT * FROM buchungen" bed, args = [], [] if jahr: bed.append("jahr = ?") args.append(jahr) if monat: bed.append("CAST(strftime('%m', datum) AS INTEGER) = ?") args.append(monat) if bed: sql += " WHERE " + " AND ".join(bed) sql += " ORDER BY datum, rechnungsnummer" return [self._zu_buchung(r) for r in self.con.execute(sql, args)] @staticmethod def _zu_buchung(r: sqlite3.Row) -> Buchung: return Buchung( id=r["id"], jahr=r["jahr"], rechnungsnummer=r["rechnungsnummer"], datum=datetime.strptime(r["datum"], "%Y-%m-%d").date(), nachname=r["nachname"], naechte=r["naechte"], gezahlt=r["gezahlt"], satz=r["satz"], quelle=r["quelle"], pdf_pfad=r["pdf_pfad"], art=(r["art"] if "art" in r.keys() else "rechnung"), storno_zu=(r["storno_zu"] if "storno_zu" in r.keys() else ""), storno_datum=(r["storno_datum"] if "storno_datum" in r.keys() else ""), vorgang=(r["vorgang"] if "vorgang" in r.keys() else ""), folge_nummer=(r["folge_nummer"] if "folge_nummer" in r.keys() else ""), folge_art=(r["folge_art"] if "folge_art" in r.keys() else ""), ) def _pdf_bekannt(self, pfad: str) -> bool: """True, wenn genau diese Datei unverändert schon eingelesen wurde.""" try: st = os.stat(pfad) except OSError: return False r = self.con.execute( "SELECT 1 FROM buchungen WHERE pdf_pfad=? AND pdf_mtime=? AND pdf_groesse=?", (os.path.abspath(pfad), st.st_mtime, st.st_size)).fetchone() return r is not None # ---- Schreiben --------------------------------------------------------- def speichern(self, b: Buchung) -> int: """Legt die Buchung an oder aktualisiert sie (Schlüssel: Jahr + Rechnungsnummer).""" mtime = groesse = 0 if b.pdf_pfad and os.path.exists(b.pdf_pfad): st = os.stat(b.pdf_pfad) mtime, groesse = st.st_mtime, st.st_size cur = self.con.execute( "INSERT INTO buchungen " " (jahr, rechnungsnummer, datum, nachname, naechte, gezahlt, satz, " " quelle, pdf_pfad, pdf_mtime, pdf_groesse, art, storno_zu, " " storno_datum, vorgang, folge_nummer, folge_art) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) " "ON CONFLICT(pdf_pfad) WHERE pdf_pfad <> '' DO UPDATE SET " " jahr=excluded.jahr, rechnungsnummer=excluded.rechnungsnummer, " " datum=excluded.datum, nachname=excluded.nachname, naechte=excluded.naechte, " " gezahlt=excluded.gezahlt, satz=excluded.satz, quelle=excluded.quelle, " " pdf_pfad=excluded.pdf_pfad, pdf_mtime=excluded.pdf_mtime, " " pdf_groesse=excluded.pdf_groesse, art=excluded.art, " " storno_zu=excluded.storno_zu, storno_datum=excluded.storno_datum, " " vorgang=excluded.vorgang, folge_nummer=excluded.folge_nummer, " " folge_art=excluded.folge_art", (b.jahr, b.rechnungsnummer, b.datum.isoformat(), b.nachname, int(b.naechte or 0), float(b.gezahlt or 0), float(b.satz or 0), b.quelle, b.pdf_pfad, mtime, groesse, b.art or "rechnung", b.storno_zu or "", b.storno_datum or "", b.vorgang or "", b.folge_nummer or "", b.folge_art or "")) self.con.commit() return cur.lastrowid def loeschen(self, buchung_id: int): self.con.execute("DELETE FROM buchungen WHERE id=?", (buchung_id,)) self.con.commit() log.info("Buchung %s gelöscht", buchung_id) def leeren(self, quelle: str = None): if quelle: self.con.execute("DELETE FROM buchungen WHERE quelle=?", (quelle,)) else: self.con.execute("DELETE FROM buchungen") self.con.commit() # ---- PDF-Scan ---------------------------------------------------------- def scanne(self, ordner: str, standard_satz: float = 5.0, voll: bool = False): """ Liest den PDF-Ordner ein und schreibt alles Neue/Geänderte ins Journal. voll=False: unveränderte PDFs werden übersprungen (schnell). voll=True : alles neu einlesen (nach Layout-/Satz-Änderungen). Gibt (neu, aktualisiert, uebersprungen, fehler) zurück. """ from pdf_parser import (parse_pdf, berichtigung_daten, keine_buchung, ParserFehler) # lokal: hält modell/db leichtgewichtig neu = akt = uebersprungen = 0 fehler = [] # Berichtigungsblaetter werden gesammelt und ERST AM ENDE angewandt: das # Blatt kann vor "seiner" Rechnung im Ordner liegen, und dann gaebe es # noch nichts zu berichtigen. berichtigungen = [] if not os.path.isdir(ordner): return 0, 0, 0, [(ordner, "Ordner existiert nicht")] for wurzel, _dirs, dateien in os.walk(ordner): for name in sorted(dateien): if not name.lower().endswith(".pdf"): continue pfad = os.path.join(wurzel, name) if not voll and self._pdf_bekannt(pfad): uebersprungen += 1 continue # Berichtigungsblatt: traegt die Nummer der Rechnung und Nullen. # 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. # 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),)) berichtigt = berichtigung_daten(pfad) if berichtigt is not None: berichtigungen.append(berichtigt) continue try: b = parse_pdf(pfad, standard_satz) except ParserFehler as e: log.info("keine Rechnung: %s (%s)", name, e) fehler.append((pfad, str(e))) continue except Exception as e: # noqa: BLE001 log.warning("Fehler bei %s: %s", name, e) fehler.append((pfad, f"{type(e).__name__}: {e}")) continue vorhanden = self.con.execute( "SELECT id FROM buchungen WHERE jahr=? AND rechnungsnummer=?", (b.jahr, b.rechnungsnummer)).fetchone() self.speichern(b) if vorhanden: akt += 1 else: neu += 1 berichtigt_anzahl = self._berichtigungen_anwenden(berichtigungen) log.info("Scan fertig: %d neu, %d aktualisiert, %d unverändert, %d berichtigt, " "%d Fehler", neu, akt, uebersprungen, berichtigt_anzahl, len(fehler)) return neu, akt, uebersprungen, fehler def _berichtigungen_anwenden(self, berichtigungen): """Berichtigte Empfaengerangaben in die zugehoerige Buchung uebernehmen. Nach dem Datum der Berichtigung sortiert - gibt es zu einer Rechnung mehrere Blaetter, gilt das juengste. """ from modell import zerlege_nummer def wann(k): teile = str(k.get("berichtigt_am") or "").split(".") return tuple(reversed(teile)) if len(teile) == 3 else ("",) geaendert = 0 for kopf in sorted(berichtigungen, key=wann): nachname = str(kopf.get("nachname") or "").strip() bezug = str(kopf.get("berichtigt_zu") or kopf.get("rechnungsnummer") or "") if not nachname or not bezug: continue jahr, nummer = zerlege_nummer(bezug, date.today().year) cur = self.con.execute( "UPDATE buchungen SET nachname=? " "WHERE jahr=? AND rechnungsnummer=? AND nachname<>?", (nachname, jahr, nummer, nachname)) if cur.rowcount: geaendert += cur.rowcount log.info("Berichtigung %s-%s: Name jetzt %r", jahr, nummer, nachname) if geaendert: self.con.commit() return geaendert