363 lines
17 KiB
Python
363 lines
17 KiB
Python
"""SQLite-Datenbank: Händler, deren Kennungen, Auszahlungsregeln und gespeicherte Abrechnungen."""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import shutil
|
||
import sqlite3
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from provtool.importer import ImportErgebnis
|
||
from provtool.money import apply_percent
|
||
|
||
SCHEMA_VERSION = 2
|
||
|
||
SCHEMA = """
|
||
CREATE TABLE IF NOT EXISTS meta (schluessel TEXT PRIMARY KEY, wert TEXT);
|
||
|
||
CREATE TABLE IF NOT EXISTS haendler (
|
||
id INTEGER PRIMARY KEY,
|
||
name TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
||
ansprechpartner TEXT DEFAULT '',
|
||
email TEXT DEFAULT '',
|
||
notiz TEXT DEFAULT '',
|
||
standard_prozent REAL NOT NULL DEFAULT 100,
|
||
aktiv INTEGER NOT NULL DEFAULT 1
|
||
);
|
||
|
||
-- Eine Kennung ordnet Positionen eines Anbieters einem Händler zu
|
||
-- (freenet: Unterhändlernummer, Vodafone: VO-Nummer).
|
||
CREATE TABLE IF NOT EXISTS kennung (
|
||
id INTEGER PRIMARY KEY,
|
||
haendler_id INTEGER NOT NULL REFERENCES haendler(id) ON DELETE CASCADE,
|
||
anbieter TEXT NOT NULL,
|
||
wert TEXT NOT NULL,
|
||
UNIQUE (anbieter, wert)
|
||
);
|
||
|
||
-- Abweichender Auszahlungs-Prozentsatz je Händler und Provisionsart.
|
||
CREATE TABLE IF NOT EXISTS regel (
|
||
id INTEGER PRIMARY KEY,
|
||
haendler_id INTEGER NOT NULL REFERENCES haendler(id) ON DELETE CASCADE,
|
||
provisionsart TEXT NOT NULL,
|
||
prozent REAL NOT NULL,
|
||
UNIQUE (haendler_id, provisionsart)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS abrechnung (
|
||
id INTEGER PRIMARY KEY,
|
||
titel TEXT NOT NULL,
|
||
erstellt TEXT NOT NULL,
|
||
geaendert TEXT NOT NULL,
|
||
notiz TEXT DEFAULT ''
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS abrechnung_datei (
|
||
id INTEGER PRIMARY KEY,
|
||
abrechnung_id INTEGER NOT NULL REFERENCES abrechnung(id) ON DELETE CASCADE,
|
||
dateiname TEXT NOT NULL,
|
||
sha256 TEXT NOT NULL,
|
||
format TEXT NOT NULL,
|
||
importiert TEXT NOT NULL,
|
||
anzahl INTEGER NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS position (
|
||
id INTEGER PRIMARY KEY,
|
||
abrechnung_id INTEGER NOT NULL REFERENCES abrechnung(id) ON DELETE CASCADE,
|
||
datei_id INTEGER REFERENCES abrechnung_datei(id) ON DELETE CASCADE,
|
||
reihenfolge INTEGER NOT NULL,
|
||
anbieter TEXT NOT NULL,
|
||
zeitraum TEXT, provisionsart TEXT, kennung TEXT, datum TEXT,
|
||
rufnummer TEXT, imei TEXT, geraet TEXT, vorname TEXT, name TEXT, tarif TEXT,
|
||
netz TEXT, dienst TEXT, aktionscode TEXT, quelle TEXT,
|
||
betrag_orig INTEGER NOT NULL, -- Cent, laut Abrechnung des Anbieters
|
||
betrag INTEGER NOT NULL, -- Cent, Auszahlung an den Händler
|
||
prozent REAL, -- zuletzt angewendeter Prozentsatz (NULL = manuell)
|
||
manuell INTEGER NOT NULL DEFAULT 0,
|
||
roh TEXT
|
||
);
|
||
CREATE INDEX IF NOT EXISTS ix_position_abr ON position(abrechnung_id);
|
||
"""
|
||
|
||
UNZUGEORDNET = 0 # Pseudo-Händler-ID für Positionen ohne bekannte Kennung
|
||
|
||
# Sonder-Kennung: Vodafone-Positionen, deren Kundenname komplett in GROSSBUCHSTABEN steht
|
||
# (so erfasst ein Unterhändler seine Kunden). Hat Vorrang vor der VO-Nr.
|
||
CAPS_ANBIETER = "vodafone_caps"
|
||
CAPS_WERT = "*"
|
||
|
||
|
||
def ist_grossgeschrieben(name: str | None) -> bool:
|
||
buchstaben = [c for c in (name or "") if c.isalpha() and c != "ß"]
|
||
return len(buchstaben) >= 2 and all(c.isupper() for c in buchstaben)
|
||
|
||
|
||
@dataclass
|
||
class Haendler:
|
||
id: int
|
||
name: str
|
||
ansprechpartner: str = ""
|
||
email: str = ""
|
||
notiz: str = ""
|
||
standard_prozent: float = 100.0
|
||
aktiv: bool = True
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
class Datenbank:
|
||
def __init__(self, path: str | Path):
|
||
self.path = Path(path)
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
self.con = sqlite3.connect(str(self.path))
|
||
self.con.row_factory = sqlite3.Row
|
||
self.con.execute("PRAGMA foreign_keys = ON")
|
||
self.con.executescript(SCHEMA)
|
||
self._migrieren()
|
||
self.con.execute("INSERT OR IGNORE INTO meta VALUES ('schema_version', ?)", (str(SCHEMA_VERSION),))
|
||
self.con.commit()
|
||
|
||
def _migrieren(self):
|
||
spalten = {r[1] for r in self.con.execute("PRAGMA table_info(position)")}
|
||
if "haendler_fest" not in spalten:
|
||
# NULL = automatisch über Kennungen, sonst fest zugewiesene Händler-ID (0 = bewusst keiner)
|
||
self.con.execute("ALTER TABLE position ADD COLUMN haendler_fest INTEGER")
|
||
self.con.execute("UPDATE meta SET wert=? WHERE schluessel='schema_version'", (str(SCHEMA_VERSION),))
|
||
self.con.commit()
|
||
|
||
def close(self):
|
||
self.con.close()
|
||
|
||
def backup(self, ziel: str | Path) -> None:
|
||
self.con.commit()
|
||
dst = sqlite3.connect(str(ziel))
|
||
with dst:
|
||
self.con.backup(dst)
|
||
dst.close()
|
||
|
||
@staticmethod
|
||
def restore(quelle: str | Path, ziel: str | Path) -> None:
|
||
test = sqlite3.connect(str(quelle))
|
||
try:
|
||
test.execute("SELECT count(*) FROM haendler").fetchone()
|
||
finally:
|
||
test.close()
|
||
shutil.copyfile(quelle, ziel)
|
||
|
||
# ------------------------------------------------------------------ Händler
|
||
def haendler_liste(self, nur_aktive: bool = False) -> list[Haendler]:
|
||
sql = "SELECT * FROM haendler" + (" WHERE aktiv=1" if nur_aktive else "") + " ORDER BY name"
|
||
return [Haendler(r["id"], r["name"], r["ansprechpartner"], r["email"], r["notiz"],
|
||
r["standard_prozent"], bool(r["aktiv"])) for r in self.con.execute(sql)]
|
||
|
||
def haendler(self, hid: int) -> Haendler | None:
|
||
r = self.con.execute("SELECT * FROM haendler WHERE id=?", (hid,)).fetchone()
|
||
if not r:
|
||
return None
|
||
return Haendler(r["id"], r["name"], r["ansprechpartner"], r["email"], r["notiz"],
|
||
r["standard_prozent"], bool(r["aktiv"]))
|
||
|
||
def haendler_speichern(self, h: Haendler, kennungen: list[tuple[str, str]]) -> int:
|
||
"""Legt an (h.id == 0) oder aktualisiert. kennungen = [(anbieter, wert), ...]"""
|
||
with self.con:
|
||
if h.id:
|
||
self.con.execute(
|
||
"UPDATE haendler SET name=?, ansprechpartner=?, email=?, notiz=?, standard_prozent=?, aktiv=? WHERE id=?",
|
||
(h.name, h.ansprechpartner, h.email, h.notiz, h.standard_prozent, int(h.aktiv), h.id))
|
||
hid = h.id
|
||
else:
|
||
cur = self.con.execute(
|
||
"INSERT INTO haendler (name, ansprechpartner, email, notiz, standard_prozent, aktiv) VALUES (?,?,?,?,?,?)",
|
||
(h.name, h.ansprechpartner, h.email, h.notiz, h.standard_prozent, int(h.aktiv)))
|
||
hid = cur.lastrowid
|
||
self.con.execute("DELETE FROM kennung WHERE haendler_id=?", (hid,))
|
||
for anbieter, wert in kennungen:
|
||
wert = wert.strip()
|
||
if not wert:
|
||
continue
|
||
besitzer = self.con.execute("SELECT h.name FROM kennung k JOIN haendler h ON h.id=k.haendler_id "
|
||
"WHERE k.anbieter=? AND k.wert=?", (anbieter, wert)).fetchone()
|
||
if besitzer:
|
||
if anbieter == CAPS_ANBIETER:
|
||
raise ValueError("Die Regel „Vodafone-Kundenname in GROSSBUCHSTABEN“ ist bereits dem "
|
||
f"Händler „{besitzer[0]}“ zugeordnet – sie kann nur einen Händler haben.")
|
||
raise ValueError(f"Die Kennung {wert} ({anbieter}) ist bereits dem Händler „{besitzer[0]}“ zugeordnet.")
|
||
self.con.execute("INSERT INTO kennung (haendler_id, anbieter, wert) VALUES (?,?,?)", (hid, anbieter, wert))
|
||
return hid
|
||
|
||
def haendler_loeschen(self, hid: int) -> None:
|
||
with self.con:
|
||
self.con.execute("DELETE FROM haendler WHERE id=?", (hid,))
|
||
|
||
def kennungen(self, hid: int) -> list[tuple[str, str]]:
|
||
return [(r[0], r[1]) for r in self.con.execute(
|
||
"SELECT anbieter, wert FROM kennung WHERE haendler_id=? ORDER BY anbieter, wert", (hid,))]
|
||
|
||
def kennung_map(self) -> dict[tuple[str, str], int]:
|
||
return {(r[0], r[1]): r[2] for r in self.con.execute(
|
||
"SELECT k.anbieter, k.wert, k.haendler_id FROM kennung k JOIN haendler h ON h.id=k.haendler_id WHERE h.aktiv=1")}
|
||
|
||
def zuordner(self):
|
||
"""Liefert eine Funktion (anbieter, kennung, name, haendler_fest) -> Händler-ID."""
|
||
kmap = self.kennung_map()
|
||
caps = kmap.get((CAPS_ANBIETER, CAPS_WERT))
|
||
aktive = {h.id for h in self.haendler_liste(nur_aktive=True)}
|
||
|
||
def finde(anbieter: str, kennung: str, name: str | None = None, fest: int | None = None) -> int:
|
||
if fest is not None:
|
||
return fest if (fest == UNZUGEORDNET or fest in aktive) else UNZUGEORDNET
|
||
if caps and anbieter == "vodafone" and ist_grossgeschrieben(name):
|
||
return caps
|
||
return kmap.get((anbieter, kennung), UNZUGEORDNET)
|
||
return finde
|
||
|
||
# ------------------------------------------------------------------ Regeln
|
||
def regeln(self, hid: int) -> dict[str, float]:
|
||
return {r[0]: r[1] for r in self.con.execute(
|
||
"SELECT provisionsart, prozent FROM regel WHERE haendler_id=?", (hid,))}
|
||
|
||
def regel_setzen(self, hid: int, provisionsart: str, prozent: float | None) -> None:
|
||
with self.con:
|
||
if prozent is None:
|
||
self.con.execute("DELETE FROM regel WHERE haendler_id=? AND provisionsart=?", (hid, provisionsart))
|
||
else:
|
||
self.con.execute("INSERT INTO regel (haendler_id, provisionsart, prozent) VALUES (?,?,?) "
|
||
"ON CONFLICT(haendler_id, provisionsart) DO UPDATE SET prozent=excluded.prozent",
|
||
(hid, provisionsart, prozent))
|
||
|
||
def prozent_fuer(self, hid: int, provisionsart: str) -> float:
|
||
if hid == UNZUGEORDNET:
|
||
return 100.0
|
||
r = self.con.execute("SELECT prozent FROM regel WHERE haendler_id=? AND provisionsart=?",
|
||
(hid, provisionsart)).fetchone()
|
||
if r:
|
||
return r[0]
|
||
h = self.haendler(hid)
|
||
return h.standard_prozent if h else 100.0
|
||
|
||
# ------------------------------------------------------------------ Abrechnungen
|
||
def abrechnung_neu(self, titel: str) -> int:
|
||
with self.con:
|
||
cur = self.con.execute("INSERT INTO abrechnung (titel, erstellt, geaendert) VALUES (?,?,?)",
|
||
(titel, _now(), _now()))
|
||
return cur.lastrowid
|
||
|
||
def abrechnungen(self) -> list[sqlite3.Row]:
|
||
return list(self.con.execute(
|
||
"SELECT a.*, (SELECT count(*) FROM position p WHERE p.abrechnung_id=a.id) AS anzahl, "
|
||
"(SELECT group_concat(dateiname, ', ') FROM abrechnung_datei d WHERE d.abrechnung_id=a.id) AS dateien "
|
||
"FROM abrechnung a ORDER BY a.geaendert DESC"))
|
||
|
||
def abrechnung(self, aid: int) -> sqlite3.Row | None:
|
||
return self.con.execute("SELECT * FROM abrechnung WHERE id=?", (aid,)).fetchone()
|
||
|
||
def abrechnung_umbenennen(self, aid: int, titel: str) -> None:
|
||
with self.con:
|
||
self.con.execute("UPDATE abrechnung SET titel=?, geaendert=? WHERE id=?", (titel, _now(), aid))
|
||
|
||
def abrechnung_loeschen(self, aid: int) -> None:
|
||
with self.con:
|
||
self.con.execute("DELETE FROM abrechnung WHERE id=?", (aid,))
|
||
|
||
def dateien(self, aid: int) -> list[sqlite3.Row]:
|
||
return list(self.con.execute("SELECT * FROM abrechnung_datei WHERE abrechnung_id=? ORDER BY id", (aid,)))
|
||
|
||
def datei_schon_importiert(self, sha256: str) -> list[str]:
|
||
return [r[0] for r in self.con.execute(
|
||
"SELECT a.titel FROM abrechnung_datei d JOIN abrechnung a ON a.id=d.abrechnung_id WHERE d.sha256=?",
|
||
(sha256,))]
|
||
|
||
def datei_entfernen(self, datei_id: int) -> None:
|
||
with self.con:
|
||
self.con.execute("DELETE FROM position WHERE datei_id=?", (datei_id,))
|
||
self.con.execute("DELETE FROM abrechnung_datei WHERE id=?", (datei_id,))
|
||
|
||
def importieren(self, aid: int, erg: ImportErgebnis) -> int:
|
||
finde = self.zuordner()
|
||
start = self.con.execute("SELECT coalesce(max(reihenfolge), 0) FROM position WHERE abrechnung_id=?",
|
||
(aid,)).fetchone()[0]
|
||
with self.con:
|
||
cur = self.con.execute(
|
||
"INSERT INTO abrechnung_datei (abrechnung_id, dateiname, sha256, format, importiert, anzahl) VALUES (?,?,?,?,?,?)",
|
||
(aid, erg.datei, erg.sha256, erg.format, _now(), len(erg.positionen)))
|
||
did = cur.lastrowid
|
||
for i, p in enumerate(erg.positionen, start=start + 1):
|
||
hid = finde(p.anbieter, p.kennung, p.name)
|
||
proz = self.prozent_fuer(hid, p.provisionsart)
|
||
self.con.execute(
|
||
"INSERT INTO position (abrechnung_id, datei_id, reihenfolge, anbieter, zeitraum, provisionsart, kennung, "
|
||
"datum, rufnummer, imei, geraet, vorname, name, tarif, netz, dienst, aktionscode, quelle, "
|
||
"betrag_orig, betrag, prozent, manuell, roh) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0,?)",
|
||
(aid, did, i, p.anbieter, p.zeitraum, p.provisionsart, p.kennung,
|
||
p.datum.isoformat() if p.datum else None, p.rufnummer, p.imei, p.geraet, p.vorname, p.name,
|
||
p.tarif, p.netz, p.dienst, p.aktionscode, p.quelle, p.betrag_cent,
|
||
apply_percent(p.betrag_cent, proz), proz, json.dumps(p.roh, ensure_ascii=False, default=str)))
|
||
self.con.execute("UPDATE abrechnung SET geaendert=? WHERE id=?", (_now(), aid))
|
||
return len(erg.positionen)
|
||
|
||
def positionen(self, aid: int) -> list[dict]:
|
||
"""Alle Positionen mit aktueller Händler-Zuordnung (über die Kennungen)."""
|
||
finde = self.zuordner()
|
||
out = []
|
||
for r in self.con.execute("SELECT * FROM position WHERE abrechnung_id=? ORDER BY reihenfolge", (aid,)):
|
||
d = dict(r)
|
||
d["haendler_id"] = finde(d["anbieter"], d["kennung"], d["name"], d["haendler_fest"])
|
||
out.append(d)
|
||
return out
|
||
|
||
def haendler_zuweisen(self, pids: list[int], hid: int | None) -> None:
|
||
"""Weist Positionen fest einem Händler zu (hid=None: wieder automatisch zuordnen).
|
||
Nicht manuell geänderte Beträge werden nach den Regeln des neuen Händlers berechnet."""
|
||
with self.con:
|
||
for pid in pids:
|
||
self.con.execute("UPDATE position SET haendler_fest=? WHERE id=?", (hid, pid))
|
||
finde = self.zuordner()
|
||
with self.con:
|
||
for pid in pids:
|
||
r = self.con.execute("SELECT * FROM position WHERE id=?", (pid,)).fetchone()
|
||
if r is None or r["manuell"]:
|
||
continue
|
||
neu_hid = finde(r["anbieter"], r["kennung"], r["name"], r["haendler_fest"])
|
||
proz = self.prozent_fuer(neu_hid, r["provisionsart"])
|
||
self.con.execute("UPDATE position SET betrag=?, prozent=? WHERE id=?",
|
||
(apply_percent(r["betrag_orig"], proz), proz, pid))
|
||
|
||
def betrag_setzen(self, pid: int, cent: int) -> None:
|
||
with self.con:
|
||
self.con.execute("UPDATE position SET betrag=?, manuell=1, prozent=NULL WHERE id=?", (cent, pid))
|
||
|
||
def prozent_anwenden(self, pids: list[int], prozent: float, auch_manuelle: bool = True) -> int:
|
||
n = 0
|
||
with self.con:
|
||
for pid in pids:
|
||
r = self.con.execute("SELECT betrag_orig, manuell FROM position WHERE id=?", (pid,)).fetchone()
|
||
if not r or (r["manuell"] and not auch_manuelle):
|
||
continue
|
||
self.con.execute("UPDATE position SET betrag=?, prozent=?, manuell=0 WHERE id=?",
|
||
(apply_percent(r["betrag_orig"], prozent), prozent, pid))
|
||
n += 1
|
||
return n
|
||
|
||
def neu_berechnen(self, aid: int, auch_manuelle: bool = False) -> int:
|
||
"""Wendet die aktuell hinterlegten Händler-Regeln auf alle (nicht manuellen) Positionen an.
|
||
Nötig z. B. wenn ein Händler erst nach dem Import angelegt wurde."""
|
||
n = 0
|
||
with self.con:
|
||
for p in self.positionen(aid):
|
||
if p["manuell"] and not auch_manuelle:
|
||
continue
|
||
proz = self.prozent_fuer(p["haendler_id"], p["provisionsart"])
|
||
neu = apply_percent(p["betrag_orig"], proz)
|
||
if neu != p["betrag"] or proz != p["prozent"] or p["manuell"]:
|
||
self.con.execute("UPDATE position SET betrag=?, prozent=?, manuell=0 WHERE id=?", (neu, proz, p["id"]))
|
||
n += 1
|
||
return n
|
||
|
||
def touch(self, aid: int) -> None:
|
||
with self.con:
|
||
self.con.execute("UPDATE abrechnung SET geaendert=? WHERE id=?", (_now(), aid))
|