"""SQLite-Datenbank: Händler, deren Kennungen, Auszahlungsregeln und gespeicherte Abrechnungen.""" from __future__ import annotations import hashlib import json import os import shutil import sqlite3 from dataclasses import dataclass from datetime import datetime from pathlib import Path from provtool import regeln from provtool.importer import ImportErgebnis from provtool.money import apply_percent SCHEMA_VERSION = 4 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, airtime_prozent REAL NOT NULL DEFAULT 0, -- 0 = keine Airtime staffel TEXT -- Jahresbonus-Staffel (JSON), NULL = Standard-Staffel ); -- Fest eingestellter Typ einer Provisionsart (sonst automatisch erkannt, siehe regeln.py). CREATE TABLE IF NOT EXISTS art_typ ( provisionsart TEXT PRIMARY KEY, typ TEXT NOT NULL ); -- 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) ); -- Eigene Kennungen (eigene Filiale): diese Positionen werden ausgeblendet und nicht verprovisioniert. CREATE TABLE IF NOT EXISTS eigen ( id INTEGER PRIMARY KEY, anbieter TEXT NOT NULL, wert TEXT NOT NULL, UNIQUE (anbieter, wert) ); 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); -- Abgerechnete Positionen (Fingerabdruck), über alle Abrechnungen hinweg: taucht eine schon -- ausgezahlte Position in einer späteren Datei/Abrechnung wieder auf, wird sie nicht erneut abgerechnet. CREATE TABLE IF NOT EXISTS abgerechnet_fp ( fp TEXT PRIMARY KEY, wann TEXT NOT NULL, abrechnung TEXT NOT NULL ); """ UNZUGEORDNET = 0 # Pseudo-Händler-ID für Positionen ohne bekannte Kennung EIGEN = -2 # Pseudo-Händler-ID für Positionen eigener Kennungen (ausgeblendet) # 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 = "*" FP_FELDER = ("anbieter", "zeitraum", "provisionsart", "kennung", "datum", "rufnummer", "imei", "vorname", "name", "tarif", "dienst", "aktionscode", "betrag_orig") def fingerabdruecke(zeilen: list[dict]) -> list[str]: """Fingerabdruck je Position einer Datei. Gleiche Zeilen innerhalb einer Datei (z. B. zwei gleiche SIM-Karten-Gutschriften) werden durchgezählt, damit beide als eigene Positionen gelten.""" gesehen: dict[str, int] = {} out = [] for z in zeilen: roh = "\x1f".join(str(z.get(k) if z.get(k) is not None else "").strip() for k in FP_FELDER) h = hashlib.sha1(roh.encode("utf-8")).hexdigest()[:24] gesehen[h] = gesehen.get(h, 0) + 1 out.append(f"{h}#{gesehen[h]}") return out class AbrechnungErledigt(Exception): """Die Abrechnung ist als erledigt gekennzeichnet (oder enthält abgerechnete Positionen) und darf so nicht verändert werden.""" 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 airtime_prozent: float = 0.0 staffel: list | None = None # None = Standard-Staffel 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() 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") h_spalten = {r[1] for r in self.con.execute("PRAGMA table_info(haendler)")} if "airtime_prozent" not in h_spalten: self.con.execute("ALTER TABLE haendler ADD COLUMN airtime_prozent REAL NOT NULL DEFAULT 0") if "staffel" not in h_spalten: self.con.execute("ALTER TABLE haendler ADD COLUMN staffel TEXT") if "abgerechnet" not in spalten: # Zeitpunkt der Abrechnung (NULL = offen), Händler zum Zeitpunkt der Abrechnung (eingefroren), # Fingerabdruck (erkennt dieselbe Position in späteren Dateien) self.con.execute("ALTER TABLE position ADD COLUMN abgerechnet TEXT") self.con.execute("ALTER TABLE position ADD COLUMN abgerechnet_hid INTEGER") self.con.execute("ALTER TABLE position ADD COLUMN fp TEXT") a_spalten = {r[1] for r in self.con.execute("PRAGMA table_info(abrechnung)")} if "erledigt" not in a_spalten: self.con.execute("ALTER TABLE abrechnung ADD COLUMN erledigt TEXT") self._fingerabdruecke_nachtragen() self.con.execute("INSERT INTO meta VALUES ('schema_version', ?) " "ON CONFLICT(schluessel) DO UPDATE SET wert=excluded.wert", (str(SCHEMA_VERSION),)) self.con.commit() def _fingerabdruecke_nachtragen(self): """Für Positionen aus älteren Versionen den Fingerabdruck berechnen.""" dateien = [r[0] for r in self.con.execute("SELECT DISTINCT datei_id FROM position WHERE fp IS NULL")] for did in dateien: zeilen = [dict(r) for r in self.con.execute( "SELECT * FROM position WHERE datei_id IS ? ORDER BY reihenfolge", (did,))] for pid, fp in zip([z["id"] for z in zeilen], fingerabdruecke(zeilen)): self.con.execute("UPDATE position SET fp=? WHERE id=? AND fp IS NULL", (fp, pid)) 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: """Spielt eine Sicherung ein. Geprüft wird an einer Kopie (unversehrt, unsere Tabellen, lässt sich öffnen und migrieren) – erst dann ersetzt sie die Datenbank. Schlägt die Prüfung fehl, bleibt alles.""" ziel = Path(ziel) probe = ziel.with_suffix(".pruefung.db") shutil.copyfile(quelle, probe) try: con = sqlite3.connect(str(probe)) try: if con.execute("PRAGMA integrity_check").fetchone()[0] != "ok": raise ValueError("Die Datei ist beschädigt (integrity_check).") tabellen = {r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")} fehlt = {"haendler", "abrechnung", "position"} - tabellen if fehlt: raise ValueError(f"Keine Sicherung des Provisions-Managers (Tabellen fehlen: {', '.join(sorted(fehlt))}).") finally: con.close() Datenbank(probe).close() # öffnet und migriert – wirft, wenn das nicht geht os.replace(probe, ziel) finally: probe.unlink(missing_ok=True) # ------------------------------------------------------------------ Händler @staticmethod def _haendler_aus(r) -> Haendler: return Haendler(r["id"], r["name"], r["ansprechpartner"], r["email"], r["notiz"], r["standard_prozent"], bool(r["aktiv"]), r["airtime_prozent"] or 0.0, regeln.staffel_laden(r["staffel"])) 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 [self._haendler_aus(r) 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() return self._haendler_aus(r) if r else None def haendler_speichern(self, h: Haendler, kennungen: list[tuple[str, str]]) -> int: """Legt an (h.id == 0) oder aktualisiert. kennungen = [(anbieter, wert), ...]""" staffel = regeln.staffel_text(regeln.staffel_pruefen(h.staffel)) if h.staffel is not None else None werte = (h.name, h.ansprechpartner, h.email, h.notiz, h.standard_prozent, int(h.aktiv), h.airtime_prozent or 0.0, staffel) with self.con: if h.id: self.con.execute( "UPDATE haendler SET name=?, ansprechpartner=?, email=?, notiz=?, standard_prozent=?, aktiv=?, " "airtime_prozent=?, staffel=? WHERE id=?", werte + (h.id,)) hid = h.id else: cur = self.con.execute( "INSERT INTO haendler (name, ansprechpartner, email, notiz, standard_prozent, aktiv, " "airtime_prozent, staffel) VALUES (?,?,?,?,?,?,?,?)", werte) 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)} eigene = set(self.eigene_kennungen()) 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 hid = kmap.get((anbieter, kennung), UNZUGEORDNET) if hid == UNZUGEORDNET and (anbieter, kennung) in eigene: return EIGEN return hid return finde # ------------------------------------------------------------------ eigene Kennungen def eigene_kennungen(self) -> list[tuple[str, str]]: return [(r[0], r[1]) for r in self.con.execute("SELECT anbieter, wert FROM eigen ORDER BY anbieter, wert")] def eigene_setzen(self, anbieter: str, wert: str, eigen: bool = True) -> None: wert = (wert or "").strip() with self.con: if eigen: self.con.execute("INSERT OR IGNORE INTO eigen (anbieter, wert) VALUES (?,?)", (anbieter, wert)) else: self.con.execute("DELETE FROM eigen WHERE anbieter=? AND wert=?", (anbieter, wert)) # ------------------------------------------------------------------ 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)) # ------------------------------------------------------------------ Typen der Provisionsarten def art_typen(self) -> dict[str, str]: """Fest eingestellte Typen je Provisionsart (alles andere wird automatisch erkannt).""" return {r[0]: r[1] for r in self.con.execute("SELECT provisionsart, typ FROM art_typ")} def art_typ_setzen(self, provisionsart: str, typ: str | None) -> None: """typ=None: wieder automatisch erkennen.""" if typ is not None and typ not in regeln.TYPEN: raise ValueError(f"Unbekannter Typ: {typ}") with self.con: if typ is None: self.con.execute("DELETE FROM art_typ WHERE provisionsart=?", (provisionsart,)) else: self.con.execute("INSERT INTO art_typ (provisionsart, typ) VALUES (?,?) " "ON CONFLICT(provisionsart) DO UPDATE SET typ=excluded.typ", (provisionsart, typ)) def provisionsarten(self) -> list[tuple[str, str, int]]: """Alle bekannten Provisionsarten: (Provisionsart, automatisch erkannter Typ, Anzahl Positionen).""" arten: dict[str, list] = {} for r in self.con.execute("SELECT provisionsart, dienst, tarif FROM position"): art = r["provisionsart"] or "" e = arten.setdefault(art, [regeln.NORMAL, 0]) erkannt = regeln.erkennen(art, r["dienst"], r["tarif"]) if erkannt != regeln.NORMAL: e[0] = erkannt e[1] += 1 for art in self.art_typen(): arten.setdefault(art, [regeln.erkennen(art), 0]) return sorted(((a, t, n) for a, (t, n) in arten.items()), key=lambda x: x[0].lower()) # ------------------------------------------------------------------ Jahresbonus-Staffel def standard_staffel(self) -> regeln.Staffel: r = self.con.execute("SELECT wert FROM meta WHERE schluessel='staffel_standard'").fetchone() return (regeln.staffel_laden(r[0]) if r else None) or list(regeln.STANDARD_STAFFEL) def standard_staffel_setzen(self, staffel) -> None: text = regeln.staffel_text(regeln.staffel_pruefen(staffel)) with self.con: self.con.execute("INSERT INTO meta VALUES ('staffel_standard', ?) " "ON CONFLICT(schluessel) DO UPDATE SET wert=excluded.wert", (text,)) def staffel_fuer(self, hid: int) -> regeln.Staffel: h = self.haendler(hid) if hid > 0 else None return h.staffel if h and h.staffel is not None else self.standard_staffel() # ------------------------------------------------------------------ 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 count(*) FROM position p WHERE p.abrechnung_id=a.id AND p.abgerechnet IS NOT NULL) AS abgerechnet, " "(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: self._keine_abgerechneten("abrechnung_id=?", aid) with self.con: self.con.execute("DELETE FROM abrechnung WHERE id=?", (aid,)) def _keine_abgerechneten(self, wo: str, wert: int) -> None: """Abgerechnete Positionen dürfen nicht gelöscht werden: ihr Fingerabdruck bliebe in abgerechnet_fp zurück (die Position gälte nach erneutem Import für immer als „doppelt“) – oder, würde man ihn mitlöschen, könnte sie ein zweites Mal ausgezahlt werden. Zuerst die Kennzeichnung aufheben.""" n = self.con.execute(f"SELECT count(*) FROM position WHERE {wo} AND abgerechnet IS NOT NULL", (wert,)).fetchone()[0] if n: raise AbrechnungErledigt(f"{n} Position(en) sind bereits abgerechnet. Zuerst die Kennzeichnung " "„abgerechnet“ aufheben (Rechtsklick auf die Positionen), dann löschen.") 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: r = self.con.execute("SELECT abrechnung_id FROM abrechnung_datei WHERE id=?", (datei_id,)).fetchone() if not r: return self._nicht_erledigt(r[0]) self._keine_abgerechneten("datei_id=?", datei_id) 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,)) self._regeln_anwenden(r[0], set()) # weniger Karten -> Jahresbonus-Staffel der übrigen neu def importieren(self, aid: int, erg: ImportErgebnis) -> int: self._nicht_erledigt(aid) 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): 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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,NULL,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, p.betrag_cent, json.dumps(p.roh, ensure_ascii=False, default=str))) self.con.execute("UPDATE abrechnung SET geaendert=? WHERE id=?", (_now(), aid)) zeilen = [dict(r) for r in self.con.execute( "SELECT * FROM position WHERE datei_id=? ORDER BY reihenfolge", (did,))] for z, fp in zip(zeilen, fingerabdruecke(zeilen)): self.con.execute("UPDATE position SET fp=? WHERE id=?", (fp, z["id"])) neue = {r[0] for r in self.con.execute("SELECT id FROM position WHERE datei_id=?", (did,))} self._regeln_anwenden(aid, neue, auch_manuelle=True) return len(erg.positionen) def positionen(self, aid: int) -> list[dict]: """Alle Positionen mit aktueller Händler-Zuordnung (über die Kennungen) und Typ (regeln.py).""" finde = self.zuordner() fest = self.art_typen() erledigt = self.erledigt(aid) # Fingerabdrücke dieser Abrechnung, die schon abgerechnet sind (in dieser oder einer anderen) schon = dict(self.con.execute( "SELECT f.fp, f.abrechnung FROM abgerechnet_fp f WHERE f.fp IN (SELECT fp FROM position WHERE abrechnung_id=?)", (aid,)).fetchall()) # Offene Doppelte: dieselbe Position (Fingerabdruck) ein zweites Mal offen – z. B. Wochen- und # Monatsdatei in derselben Abrechnung. Die zuerst importierte zählt, die anderen sind „doppelt“. erste: dict[str, tuple[int, str]] = {} for fp, pid, titel in self.con.execute( "SELECT p.fp, p.id, a.titel FROM position p JOIN abrechnung a ON a.id=p.abrechnung_id " "WHERE p.abgerechnet IS NULL AND p.fp IN (SELECT fp FROM position WHERE abrechnung_id=?) " "ORDER BY p.id", (aid,)): erste.setdefault(fp, (pid, titel)) out = [] for r in self.con.execute("SELECT * FROM position WHERE abrechnung_id=? ORDER BY reihenfolge", (aid,)): d = dict(r) if d["abgerechnet"] and d["abgerechnet_hid"] is not None: d["haendler_id"] = d["abgerechnet_hid"] # abgerechnet: Zuordnung eingefroren else: d["haendler_id"] = finde(d["anbieter"], d["kennung"], d["name"], d["haendler_fest"]) if erledigt and d["haendler_id"] > 0: # Erledigt: was dabei nicht abgerechnet wurde, bekommt nachträglich keinen Händler mehr # (sonst stünde es bei einem später angelegten Händler mit 100 % zur Auszahlung) d["haendler_id"] = UNZUGEORDNET d["typ"] = regeln.typ_von(d, fest) d["doppelt"] = None if d["fp"] and not d["abgerechnet"]: if d["fp"] in schon: d["doppelt"] = schon[d["fp"]] elif erste.get(d["fp"], (d["id"],))[0] != d["id"]: d["doppelt"] = erste[d["fp"]][1] out.append(d) return out # ------------------------------------------------------------------ Berechnung nach Regeln def _rechner(self): """Funktion (position, händler_id, typ, karten) -> (auszahlung_cent, prozent). prozent ist None, wenn der Betrag aus der Jahresbonus-Staffel kommt.""" haendler = {h.id: h for h in self.haendler_liste()} art_regeln: dict[int, dict[str, float]] = {} for r in self.con.execute("SELECT haendler_id, provisionsart, prozent FROM regel"): art_regeln.setdefault(r[0], {})[r[1]] = r[2] std_staffel = self.standard_staffel() def rechne(p: dict, hid: int, typ: str, karten: int) -> tuple[int, float | None]: orig = p["betrag_orig"] h = haendler.get(hid) if hid > 0 else None if h is None: # nicht zugeordnet / eigene Filiale: unverändert return orig, 100.0 regel = art_regeln.get(hid, {}).get(p["provisionsart"]) if regel is not None: # ausdrücklich je Provisionsart festgelegt -> hat Vorrang proz = regel elif typ == regeln.AIRTIME: proz = h.airtime_prozent elif typ == regeln.JAHRESBONUS: return regeln.bonus_betrag(h.staffel if h.staffel is not None else std_staffel, karten, orig), None else: proz = h.standard_prozent return apply_percent(orig, proz), proz return rechne @staticmethod def karten_zaehlen(positionen: list[dict]) -> dict[tuple[int, str], int]: """Anzahl Karten je (Händler, Anbieter) für die Jahresbonus-Staffel (Stornos zählen ab, Doppelte nicht mit).""" karten: dict[tuple[int, str], int] = {} for p in positionen: if p["typ"] == regeln.JAHRESBONUS and p["betrag_orig"] and not p["doppelt"]: k = (p["haendler_id"], p["anbieter"]) karten[k] = karten.get(k, 0) + (1 if p["betrag_orig"] > 0 else -1) return {k: max(0, n) for k, n in karten.items()} def _regeln_anwenden(self, aid: int, pids: set[int] | None = None, auch_manuelle: bool = False) -> int: """Berechnet die Auszahlung nach den Händler-Regeln neu: für `pids` (None = alle) und immer für alle Jahresbonus-Positionen (deren Staffel hängt von der Anzahl aller Karten ab). Von Hand geänderte Beträge bleiben, außer auch_manuelle=True. Gibt die Anzahl Änderungen zurück.""" if self.erledigt(aid): return 0 # erledigte Abrechnungen bleiben, wie sie abgerechnet wurden alle = self.positionen(aid) karten = self.karten_zaehlen(alle) rechne = self._rechner() n = 0 with self.con: for p in alle: if p["abgerechnet"]: continue # schon ausgezahlt: Betrag bleibt if pids is not None and p["id"] not in pids and p["typ"] != regeln.JAHRESBONUS: continue if p["manuell"] and not (auch_manuelle and (pids is None or p["id"] in pids)): continue betrag, proz = rechne(p, p["haendler_id"], p["typ"], karten.get((p["haendler_id"], p["anbieter"]), 0)) if betrag != p["betrag"] or proz != p["prozent"] or p["manuell"]: self.con.execute("UPDATE position SET betrag=?, prozent=?, manuell=0 WHERE id=?", (betrag, proz, p["id"])) n += 1 return n def _abrechnungen_von(self, pids: list[int]) -> list[int]: if not pids: return [] marken = ",".join("?" * len(pids)) return [r[0] for r in self.con.execute( f"SELECT DISTINCT abrechnung_id FROM position WHERE id IN ({marken})", list(pids))] 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.""" pids = self._offene(pids) with self.con: for pid in pids: self.con.execute("UPDATE position SET haendler_fest=? WHERE id=?", (hid, pid)) for aid in self._abrechnungen_von(pids): self._regeln_anwenden(aid, set(pids)) def regeln_wiederherstellen(self, pids: list[int]) -> None: """Ausgewählte Positionen wieder nach den Händler-Regeln berechnen (auch von Hand geänderte).""" for aid in self._abrechnungen_von(pids): self._regeln_anwenden(aid, set(pids), auch_manuelle=True) def betrag_setzen(self, pid: int, cent: int) -> None: if not self._offene([pid]): raise AbrechnungErledigt("Die Position ist bereits abgerechnet.") 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: pids = self._offene(pids) 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.""" return self._regeln_anwenden(aid, None, auch_manuelle=auch_manuelle) # ------------------------------------------------------------------ abgerechnet / erledigt def _offene(self, pids) -> list[int]: """Nur Positionen, die noch geändert werden dürfen (nicht abgerechnet, Abrechnung nicht erledigt).""" pids = list(pids) if not pids: return [] marken = ",".join("?" * len(pids)) return [r[0] for r in self.con.execute( f"SELECT p.id FROM position p JOIN abrechnung a ON a.id=p.abrechnung_id " f"WHERE p.id IN ({marken}) AND p.abgerechnet IS NULL AND a.erledigt IS NULL", pids)] def erledigt(self, aid: int) -> str | None: r = self.con.execute("SELECT erledigt FROM abrechnung WHERE id=?", (aid,)).fetchone() return r[0] if r else None def _nicht_erledigt(self, aid: int) -> None: if self.erledigt(aid): raise AbrechnungErledigt("Diese Abrechnung ist als erledigt gekennzeichnet.") def abrechnen(self, pids: list[int], wann: str | None = None) -> int: """Positionen als abgerechnet kennzeichnen: Betrag und Händler werden eingefroren, der Fingerabdruck gemerkt (dieselbe Position wird später nicht noch einmal abgerechnet). Nur Positionen echter Händler, die noch offen und nicht schon anderswo abgerechnet sind.""" wann = wann or _now() ids = set(pids) n = 0 for aid in self._abrechnungen_von(list(ids)): if self.erledigt(aid): continue # erledigt: nichts wird nachträglich abgerechnet titel = self.abrechnung(aid)["titel"] with self.con: for p in self.positionen(aid): if p["id"] not in ids or p["abgerechnet"] or p["doppelt"] or p["haendler_id"] <= 0: continue # innerhalb der Transaktion noch einmal: derselbe Fingerabdruck darf nur einmal abgerechnet sein if p["fp"] and self.con.execute("SELECT 1 FROM abgerechnet_fp WHERE fp=?", (p["fp"],)).fetchone(): continue self.con.execute("UPDATE position SET abgerechnet=?, abgerechnet_hid=? WHERE id=?", (wann, p["haendler_id"], p["id"])) if p["fp"]: self.con.execute("INSERT INTO abgerechnet_fp (fp, wann, abrechnung) VALUES (?,?,?)", (p["fp"], wann, titel)) n += 1 return n def abrechnung_aufheben(self, pids: list[int]) -> int: """Kennzeichnung „abgerechnet“ zurücknehmen (z. B. versehentlich exportiert).""" n = 0 for aid in self._abrechnungen_von(list(pids)): self._nicht_erledigt(aid) with self.con: for pid in pids: r = self.con.execute("SELECT fp, abgerechnet FROM position WHERE id=?", (pid,)).fetchone() if not r or not r["abgerechnet"]: continue self.con.execute("UPDATE position SET abgerechnet=NULL, abgerechnet_hid=NULL WHERE id=?", (pid,)) if r["fp"]: self.con.execute("DELETE FROM abgerechnet_fp WHERE fp=?", (r["fp"],)) n += 1 return n def erledigt_setzen(self, aid: int, erledigt: bool, positionen_aufheben: bool = False) -> int: """Abrechnung als erledigt kennzeichnen (alle offenen Händler-Positionen werden dabei abgerechnet) bzw. wieder öffnen (auf Wunsch samt Kennzeichnung der Positionen). Gibt die Anzahl der dabei (ab)gerechneten Positionen zurück.""" n = 0 if erledigt: wann = _now() n = self.abrechnen([p["id"] for p in self.positionen(aid)], wann) with self.con: self.con.execute("UPDATE abrechnung SET erledigt=?, geaendert=? WHERE id=?", (wann, wann, aid)) else: with self.con: self.con.execute("UPDATE abrechnung SET erledigt=NULL, geaendert=? WHERE id=?", (_now(), aid)) if positionen_aufheben: n = self.abrechnung_aufheben([p["id"] for p in self.positionen(aid)]) return n def touch(self, aid: int) -> None: with self.con: self.con.execute("UPDATE abrechnung SET geaendert=? WHERE id=?", (_now(), aid))