65eec9937a
Das Rechnungstool schreibt seit heute auch Storno- und Berichtigungsblaetter in denselben Ordner. Zwei Fehler, die daraus entstanden waeren - beide still: 1. Ein Berichtigungsblatt (§ 31 Abs. 5 UStDV) traegt die Nummer DER RECHNUNG, die es berichtigt, und lauter Nullen. Unter dem alten Schluessel (jahr, rechnungsnummer) hat es die echte Buchung ueberschrieben - im Amtsbericht fehlte die Uebernachtung dann. Wird jetzt uebersprungen (pdf_parser.keine_buchung), in beiden Scannern. 2. Der Schluessel ist jetzt die DATEI. Zwei Rechnungen mit derselben Nummer - der Altbestand, bevor das Rechnungstool eine Sperre bekam - ueberschrieben sich sonst gegenseitig. Bestehende Journale werden beim Oeffnen einmalig umgebaut, in einer Transaktion, mit Rueckfall auf den alten Stand. Buchungen von Hand (kein PDF-Pfad) bleiben davon unberuehrt. Stornos zaehlen weiter mit - sie sind negativ und heben die Uebernachtung im Bericht genau so auf, wie es sein soll. pruef_berichtigung.py (neu): schreibt mit dem echten Renderer des Rechnungstools eine Rechnung, ein Berichtigungsblatt dazu und zwei Rechnungen mit derselben Nummer, liest sie ein und prueft das Ergebnis in der Tabelle. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EKvGkdNW1vKdnMPAM9Bp3X
260 lines
11 KiB
Python
260 lines
11 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
|
||
);
|
||
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.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 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(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",
|
||
(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, keine_buchung, 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
|
||
# Berichtigungsblatt: traegt die Nummer der Rechnung und Nullen.
|
||
# Frueher haette es die echte Buchung ueberschrieben.
|
||
if keine_buchung(pfad):
|
||
self.con.execute("DELETE FROM buchungen WHERE pdf_pfad=?",
|
||
(os.path.abspath(pfad),))
|
||
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
|