Beherbergungssteuer: Steuerjournal aus den Rechnungs-PDFs
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>
This commit is contained in:
TheMockTv
commit
80967de26f
12 changed files with 1908 additions and 0 deletions
194
db.py
Normal file
194
db.py
Normal file
|
|
@ -0,0 +1,194 @@
|
|||
# -*- 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
|
||||
Loading…
Add table
Add a link
Reference in a new issue