695 lines
35 KiB
Python
695 lines
35 KiB
Python
"""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))
|