80967de26f
Ersetzt die Excel-Mappe des Campinghofs. Die Rechnungs-PDFs des Rechnungstools sind die Quelle; SQLite ist nur Cache, Excel nur noch Import des Altbestands. Bericht fuers Amt als PDF im Briefkopf der Rechnungen (Firmendaten aus der config.json des Rechnungstools). - pdf_parser: liest Datum, Nummer, Name, Naechte, Zwischensumme, Satz - db: Journal + Scan mit Cache (Pfad/mtime/Groesse) - xlsx_io: Import der alten Tabelle, vereinheitlicht Rechnungsnummern - bericht_pdf: Monats- und Jahresbericht (reportlab) - GUI: Monatsreiter, Suche, Anhaken + Papierkorb, Kennzahlen-Leiste Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
194 lines
7.7 KiB
Python
194 lines
7.7 KiB
Python
# -*- 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
|