"""SQLite-Datenbank: Händler, deren Kennungen, Auszahlungsregeln und gespeicherte Abrechnungen.""" from __future__ import annotations import hashlib import json 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 und darf nicht mehr 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() 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") 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("UPDATE meta SET wert=? WHERE schluessel='schema_version'", (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: test = sqlite3.connect(str(quelle)) try: test.execute("SELECT count(*) FROM haendler").fetchone() finally: test.close() shutil.copyfile(quelle, ziel) # ------------------------------------------------------------------ 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() def prozent_fuer(self, hid: int, provisionsart: str) -> float: if hid <= 0: # nicht zugeordnet / eigene Filiale 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 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: 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: r = self.con.execute("SELECT abrechnung_id FROM abrechnung_datei WHERE id=?", (datei_id,)).fetchone() if r: self._nicht_erledigt(r[0]) 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: 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() schon = dict(self.con.execute("SELECT fp, abrechnung FROM abgerechnet_fp").fetchall()) 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"]) d["typ"] = regeln.typ_von(d, fest) # schon anderswo abgerechnet (dieselbe Position in einer früheren Datei/Abrechnung)? d["doppelt"] = schon.get(d["fp"]) if d["fp"] and not d["abgerechnet"] else None 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).""" karten: dict[tuple[int, str], int] = {} for p in positionen: if p["typ"] == regeln.JAHRESBONUS and p["betrag_orig"]: 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)): 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 self.con.execute("UPDATE position SET abgerechnet=?, abgerechnet_hid=? WHERE id=?", (wann, p["haendler_id"], p["id"])) if p["fp"]: self.con.execute("INSERT OR REPLACE 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))