# -*- 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', pdf_pfad TEXT NOT NULL DEFAULT '', pdf_mtime REAL NOT NULL DEFAULT 0, pdf_groesse INTEGER NOT NULL DEFAULT 0, UNIQUE (jahr, rechnungsnummer) ); CREATE INDEX IF NOT EXISTS idx_datum ON buchungen (jahr, datum); 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.con.executescript(SCHEMA) self.con.commit() log.info("Journal %s (%s)", pfad, "neu angelegt" if neu else "geöffnet") 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"], ) 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) " "VALUES (?,?,?,?,?,?,?,?,?,?,?) " "ON CONFLICT(jahr, rechnungsnummer) DO UPDATE SET " " 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", (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)) 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, ParserFehler # lokal: hält modell/db leichtgewichtig neu = akt = uebersprungen = 0 fehler = [] 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 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 log.info("Scan fertig: %d neu, %d aktualisiert, %d unverändert, %d Fehler", neu, akt, uebersprungen, len(fehler)) return neu, akt, uebersprungen, fehler