diff --git a/README.md b/README.md index 0970c97..2240abc 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,11 @@ gekennzeichnet (grau) und nie wieder exportiert – auch nicht, wenn dieselbe Po anderen Datei oder Abrechnung auftaucht. Versehentlich? Rechtsklick ▸ *Kennzeichnung „abgerechnet“ aufheben*. *Datei ▸ Abrechnung erledigt / wieder öffnen* sperrt eine fertige Abrechnung gegen Änderungen. +**Doppelte Positionen:** Taucht dieselbe Position mehrfach auf (z. B. in Wochen- und Monatsdatei oder in zwei +Abrechnungen), zählt nur die zuerst importierte; die anderen stehen als *doppelt* in der Liste und werden +weder exportiert noch bei der Jahresbonus-Staffel mitgezählt. Abgerechnete Positionen lassen sich erst +löschen, wenn ihre Kennzeichnung „abgerechnet“ aufgehoben ist. + **Design:** Standard ist das Design der Handy-Insel-Webtools (Lila, Schrift IBM Plex – liegt dem Programm bei): Kopfzeile mit Menüs und Suche, links Abrechnungen und Händler, oben Kennzahlen, über der Tabelle Filter (nicht zugeordnet, von Hand geändert, doppelt, bereits abgerechnet). Oben rechts „Dunkel“/„Hell“ schaltet diff --git a/provtool/__init__.py b/provtool/__init__.py index 4197078..bcf3bc6 100644 --- a/provtool/__init__.py +++ b/provtool/__init__.py @@ -1,5 +1,5 @@ """Provisions-Manager 95 – Provisionsabrechnung für Unterhändler der Handy-Insel Miesbach.""" -__version__ = "1.7.0" +__version__ = "1.8.0" APP_NAME = "Provisions-Manager 95" APP_ORG = "Handy-Insel Miesbach" diff --git a/provtool/config.py b/provtool/config.py index b0b0e5a..1ab99ed 100644 --- a/provtool/config.py +++ b/provtool/config.py @@ -8,6 +8,7 @@ from __future__ import annotations import json import os +import shutil from pathlib import Path APP_DIR_NAME = "Provisionsmanager95" @@ -41,18 +42,31 @@ DEFAULTS = { def load() -> dict: + """Einstellungen lesen. Ist die Datei unlesbar (z. B. Absturz beim Schreiben), wird sie als + *.defekt.json aufbewahrt und unter "defekt" gemeldet – statt still mit Standardwerten weiterzumachen + (dabei ginge z. B. der Pfad zur Datenbank auf dem Netzlaufwerk verloren).""" data = dict(DEFAULTS) f = _settings_file() if f.exists(): try: data.update(json.loads(f.read_text(encoding="utf-8"))) except (OSError, ValueError): - pass + sicherung = f.with_suffix(".defekt.json") + try: + shutil.copyfile(f, sicherung) + except OSError: + pass + data["defekt"] = str(sicherung) return data def save(data: dict) -> None: - _settings_file().write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") + """Atomar: erst in eine Hilfsdatei, dann umbenennen – ein Absturz hinterlässt nie eine halbe Datei.""" + data = {k: v for k, v in data.items() if k != "defekt"} + f = _settings_file() + tmp = f.with_suffix(".json.tmp") + tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") + os.replace(tmp, f) def db_path(settings: dict | None = None) -> Path: diff --git a/provtool/db.py b/provtool/db.py index 26a31f7..68fcb8d 100644 --- a/provtool/db.py +++ b/provtool/db.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib import json +import os import shutil import sqlite3 from dataclasses import dataclass @@ -134,7 +135,8 @@ def fingerabdruecke(zeilen: list[dict]) -> list[str]: class AbrechnungErledigt(Exception): - """Die Abrechnung ist als erledigt gekennzeichnet und darf nicht mehr verändert werden.""" + """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: @@ -168,8 +170,6 @@ class Datenbank: 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)")} @@ -191,7 +191,8 @@ class Datenbank: 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.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): @@ -215,12 +216,26 @@ class Datenbank: @staticmethod def restore(quelle: str | Path, ziel: str | Path) -> None: - test = sqlite3.connect(str(quelle)) + """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: - test.execute("SELECT count(*) FROM haendler").fetchone() + 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: - test.close() - shutil.copyfile(quelle, ziel) + probe.unlink(missing_ok=True) # ------------------------------------------------------------------ Händler @staticmethod @@ -368,16 +383,6 @@ class Datenbank: 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: @@ -400,9 +405,20 @@ class Datenbank: 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,))) @@ -413,11 +429,14 @@ class Datenbank: 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]) + 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) @@ -450,7 +469,19 @@ class Datenbank: """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()) + 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) @@ -458,9 +489,17 @@ class Datenbank: 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) - # 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 + 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 @@ -493,10 +532,10 @@ class Datenbank: @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).""" + """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"]: + 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()} @@ -599,15 +638,20 @@ class Datenbank: 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 OR REPLACE INTO abgerechnet_fp (fp, wann, abrechnung) VALUES (?,?,?)", + self.con.execute("INSERT INTO abgerechnet_fp (fp, wann, abrechnung) VALUES (?,?,?)", (p["fp"], wann, titel)) n += 1 return n diff --git a/provtool/export.py b/provtool/export.py index 7d2b424..9211b5c 100644 --- a/provtool/export.py +++ b/provtool/export.py @@ -101,6 +101,8 @@ def schreibe_xlsx(ziel: str | Path, positionen: list[dict], *, kopfzeile: bool = ] for c, v in enumerate(werte, start=1): cell = ws.cell(zeile, c, v) + if isinstance(v, str) and v.startswith("="): + cell.data_type = "s" # Text aus der Abrechnung bleibt Text – nie eine Formel cell.font = bold if p.get("_summe") and c in (2, 11) else font ws.cell(zeile, 4).number_format = "DD.MM.YYYY" if isinstance(werte[0], date): @@ -130,9 +132,11 @@ def schreibe_xlsx(ziel: str | Path, positionen: list[dict], *, kopfzeile: bool = def exportiere_haendler(ordner: str | Path, haendler_name: str, positionen: list[dict], titel: str, *, pro_anbieter: bool = False, kopfzeile: bool = False, summe: bool = True, - null_weglassen: bool = True) -> list[Path]: + null_weglassen: bool = True, vergeben: set[str] | None = None) -> list[Path]: """Schreibt die Abrechnung eines Händlers (bei pro_anbieter je Anbieter eine) und – falls es Airtime gibt – die Airtime-Liste. In der Abrechnung steht Airtime nur als eine Summenzeile. + vergeben: Dateinamen (klein geschrieben), die in diesem Export schon geschrieben wurden – gleich + lautende Namen (z. B. Händler „A/B“ und „A:B“) bekommen „ (2)“ statt die Datei zu überschreiben. Gibt die erzeugten Pfade zurück.""" ordner = Path(ordner) if null_weglassen: @@ -142,9 +146,16 @@ def exportiere_haendler(ordner: str | Path, haendler_name: str, positionen: list gruppen: dict[str, list[dict]] = {} for p in positionen: gruppen.setdefault(p["anbieter"] if pro_anbieter else "", []).append(p) + vergeben = set() if vergeben is None else vergeben def name(*teile) -> str: - return dateiname(" - ".join(t for t in (haendler_name,) + teile + (titel,) if t)) + ".xlsx" + basis = dateiname(" - ".join(t for t in (haendler_name,) + teile + (titel,) if t)) + for n in range(1, 1000): + kandidat = basis + (f" ({n})" if n > 1 else "") + ".xlsx" + if kandidat.lower() not in vergeben: + vergeben.add(kandidat.lower()) + return kandidat + raise ValueError(f"Kein freier Dateiname für {basis}") out = [] for anbieter, pos in sorted(gruppen.items()): diff --git a/provtool/gui/app.py b/provtool/gui/app.py index b185796..12c467f 100644 --- a/provtool/gui/app.py +++ b/provtool/gui/app.py @@ -61,11 +61,14 @@ class App: def __init__(self, root: tk.Tk): self.root = root self.settings = config.load() - self.db = Datenbank(config.db_path(self.settings)) + self.db: Datenbank | None = None # wird nach dem Aufbau des Fensters geöffnet (_db_oeffnen) self.aid: int | None = None self.positionen: list[dict] = [] self.haendler_namen: dict[int, str] = {} self.ansicht: int = ALLE + self._alle_gruppen: dict[str, tuple[str, list[dict]]] = {} + self._gruppen: dict[str, tuple[str, list[dict]]] = {} + self._zeilen_tags: dict[str, tuple[str, ...]] = {} self._edit: tk.Entry | None = None self._schliesst = False # Beenden angefordert (offene Dialoge werden gerade geschlossen) self._beendet = False # Datenbank und Fenster sind zu @@ -101,16 +104,51 @@ class App: if self.settings.get("trenner"): root.after(200, lambda: self.pw.sash_place(0, int(self.settings["trenner"]), 0)) + self.db = self._db_oeffnen() abr = self.db.abrechnungen() if abr: self.oeffnen(abr[0]["id"]) else: self._aktualisieren() - if not self.settings.get("assistent_erledigt") and not self.db.haendler_liste(): + if self.settings.pop("defekt", None): + root.after(300, lambda: w.warnung( + root, "Einstellungen", + "Die Einstellungsdatei war unlesbar und wurde durch Standardwerte ersetzt (Kopie: " + f"{config.app_dir() / 'einstellungen.defekt.json'}).\n\nBitte unter Datei > Einstellungen " + "prüfen, ob der Speicherort der Datenbank noch stimmt.")) + elif not self.settings.get("assistent_erledigt") and not self.db.haendler_liste(): root.after(400, self.assistent) elif self.settings.get("tipps_beim_start", True): root.after(500, self.tipp_des_tages) + def _db_oeffnen(self) -> Datenbank: + """Datenbank laut Einstellungen öffnen. Ist ein eigener Speicherort eingestellt (Netzlaufwerk) und + nicht erreichbar, wird gefragt – statt abzustürzen oder dort still eine leere Datenbank anzulegen.""" + while True: + pfad = config.db_path(self.settings) + try: + if self.settings.get("db_pfad") and not pfad.exists(): + raise FileNotFoundError("Die Datei gibt es nicht (Netzlaufwerk nicht verbunden?)") + return Datenbank(pfad) + except Exception as exc: + wahl = w._meldung(self.root, "Datenbank nicht erreichbar", + f"Die Datenbank konnte nicht geöffnet werden:\n{pfad}\n\n{exc}\n\n" + "Standard: die lokale Datenbank dieses PCs verwenden.", + "fehler", [("Standard", "standard"), ("Andere wählen...", "waehlen"), + ("Beenden", "ende")], default=0) + if wahl == "standard": + self.settings["db_pfad"] = None + config.save(self.settings) + elif wahl == "waehlen": + p = filedialog.askopenfilename(parent=self.root, title="Datenbank öffnen", + filetypes=[("Datenbank", "*.db"), ("Alle Dateien", "*.*")]) + if p: + self.settings["db_pfad"] = p + config.save(self.settings) + else: + self.root.destroy() + raise SystemExit(1) from exc + # ================================================================= Aufbau def _kopf(self): """Kopfzeile wie in den Webtools: Logo, Menüs, rechts Suche (modern auch Hell/Dunkel).""" @@ -378,7 +416,7 @@ class App: self.filter = key for k, pill in self.pills.items(): pill.setzen(k == key) - self._tabelle() + self._filtern() def _tasten(self): r = self.root @@ -405,9 +443,10 @@ class App: def _aktualisieren(self, behalte_auswahl: bool = True): self.haendler_namen = {h.id: h.name for h in self.db.haendler_liste()} - if self.aid and self.db.abrechnung(self.aid): + a = self.db.abrechnung(self.aid) if self.aid else None + if a: self.positionen = self.db.positionen(self.aid) - titel = self.db.abrechnung(self.aid)["titel"] + (" – erledigt" if self.db.erledigt(self.aid) else "") + titel = a["titel"] + (" – erledigt" if a["erledigt"] else "") self.rahmen.set_titel(f"{w.APP_NAME} - [{titel}]") self.status.set("info", f" Abrechnung: {titel}") else: @@ -416,20 +455,19 @@ class App: self.rahmen.set_titel(w.APP_NAME) self.status.set("info", " Keine Abrechnung geöffnet – Datei > Neue Abrechnung oder Import") self._abrechnungsliste() - self._seitenkopf() + self._seitenkopf(a) self._haendlerliste(behalte_auswahl) self._tabelle() - def _seitenkopf(self): - """Titel, Status und Knöpfe über den Kennzahlen.""" - a = self.db.abrechnung(self.aid) if self.aid else None + def _seitenkopf(self, a): + """Titel, Status und Knöpfe über den Kennzahlen (a: Zeile der Abrechnung oder None).""" if not a: self.l_status.configure(text="Keine Abrechnung geöffnet".upper() if w.MODERN else "Keine Abrechnung") self.l_titel.configure(text="Willkommen") self.l_info.configure(text="Links „+ Neu“ anlegen oder gleich Dateien importieren.") self.b_erledigt.configure(text="Als erledigt markieren", state="disabled") return - erledigt = self.db.erledigt(self.aid) + erledigt = a["erledigt"] dateien = self.db.dateien(self.aid) status = "erledigt" if erledigt else "offen" self.l_status.configure(text=f"Abrechnung · {status}".upper() if w.MODERN else f"Abrechnung ({status})") @@ -463,7 +501,7 @@ class App: def _suche_ausfuehren(self): self._suche_geplant = None - self._tabelle() + self._filtern() def _gefiltert(self) -> list[dict]: if self.ansicht == ALLE: @@ -521,6 +559,8 @@ class App: return f"{ANBIETER.get(p['anbieter'], p['anbieter'])} · {p['provisionsart'] or '(ohne Provisionsart)'}" def _tabelle(self): + """Baut die Tabelle für die gewählte Ansicht komplett neu auf (nach Datenänderungen). + Suche und Filter blenden danach nur noch Zeilen aus und ein (_filtern) – kein Neuaufbau.""" if self._edit: self._edit.destroy() self._edit = None @@ -528,49 +568,80 @@ class App: offen = {iid for iid in t.get_children() if t.item(iid, "open")} war_leer = not t.get_children() t.delete(*t.get_children()) - such = self.v_suche.get().strip().lower() - pos = self._gefiltert() - if such: - pos = [p for p in pos if such in " ".join(str(p.get(k) or "") for k in ( - "rufnummer", "vorname", "name", "tarif", "provisionsart", "kennung", "zeitraum", "aktionscode", - "imei")).lower()] - pruef = {"unzugeordnet": lambda p: p["haendler_id"] == UNZUGEORDNET, "manuell": lambda p: p["manuell"], - "doppelt": lambda p: p["doppelt"], "abgerechnet": lambda p: p["abgerechnet"]}.get(self.filter) - if pruef: - pos = [p for p in pos if pruef(p)] gruppen: dict[str, list[dict]] = defaultdict(list) - for p in exp.sortiert(pos): + for p in exp.sortiert(self._gefiltert()): gruppen[self._gruppe_von(p)].append(p) - self._gruppen = {} + self._alle_gruppen = {} # giid -> (Name, alle Positionen der Gruppe) + self._zeilen_tags = {} for gi, (gname, plist) in enumerate(sorted(gruppen.items())): giid = f"g{gi}" - self._gruppen[giid] = (gname, plist) - so = sum(p["betrag_orig"] for p in plist) - sa = sum(p["betrag"] for p in plist) - proz = {p["prozent"] for p in plist} - if all(_ist_staffel(p) for p in plist): - pz = "Staffel" - else: - pz = fmt_percent(next(iter(proz))) if len(proz) == 1 and None not in proz else ("gemischt" if so else "") - t.insert("", "end", iid=giid, text=f"{gname} ({len(plist)})", open=(war_leer or giid in offen or bool(such)), - values=("", "", "", "", "", fmt(so), pz, fmt(sa)), tags=("gruppe",)) - for i, p in enumerate(plist): - tags = ["zebra"] if i % 2 else [] + self._alle_gruppen[giid] = (gname, plist) + t.insert("", "end", iid=giid, text="", open=(war_leer or giid in offen), tags=("gruppe",)) + for p in plist: + tags: tuple[str, ...] = () if p["abgerechnet"] or p["doppelt"]: - tags.append("abgerechnet") + tags = ("abgerechnet",) elif p["manuell"]: - tags.append("manuell") + tags = ("manuell",) elif p["betrag"] != p["betrag_orig"]: - tags.append("abw") + tags = ("abw",) + self._zeilen_tags[f"p{p['id']}"] = tags kunde = " ".join(x for x in (p["vorname"], p["name"]) if x) or "(ohne Name)" t.insert(giid, "end", iid=f"p{p['id']}", text=kunde, values=(_datum_de(p["datum"]), p["rufnummer"], p["tarif"], _zeitraum_kurz(p["zeitraum"]), p["kennung"] + (" (fest)" if p["haendler_fest"] is not None else ""), fmt(p["betrag_orig"]), - "abgerechnet" if p["abgerechnet"] or p["doppelt"] else + "abgerechnet" if p["abgerechnet"] else "doppelt" if p["doppelt"] else "manuell" if p["manuell"] else ("Staffel" if _ist_staffel(p) else fmt_percent(p["prozent"])), fmt(p["betrag"])), - tags=tuple(tags)) + tags=tags) + self._filtern() + + def _filtern(self): + """Suche und Filter anwenden: nicht passende Zeilen werden nur ausgehängt (detach), passende wieder + eingehängt – viel schneller als Löschen und Neuanlegen. Gruppenzeilen und Kennzahlen folgen.""" + t = self.ptree + such = self.v_suche.get().strip().lower() + pruef = {"unzugeordnet": lambda p: p["haendler_id"] == UNZUGEORDNET, "manuell": lambda p: p["manuell"], + "doppelt": lambda p: p["doppelt"], "abgerechnet": lambda p: p["abgerechnet"]}.get(self.filter) + + def passt(p: dict) -> bool: + if pruef and not pruef(p): + return False + return not such or such in " ".join(str(p.get(k) or "") for k in ( + "rufnummer", "vorname", "name", "tarif", "provisionsart", "kennung", "zeitraum", "aktionscode", + "imei")).lower() + + self._gruppen = {} # giid -> (Name, sichtbare Positionen) – für Auswahl über die Gruppenzeile + pos: list[dict] = [] + sichtbar_gi = 0 + for giid, (gname, plist) in self._alle_gruppen.items(): + treffer = [] + for p in plist: + iid = f"p{p['id']}" + if passt(p): + t.move(iid, giid, len(treffer)) + t.item(iid, tags=self._zeilen_tags[iid] + (("zebra",) if len(treffer) % 2 else ())) + treffer.append(p) + else: + t.detach(iid) + if not treffer: + t.detach(giid) + continue + t.move(giid, "", sichtbar_gi) + sichtbar_gi += 1 + self._gruppen[giid] = (gname, treffer) + pos += treffer + so = sum(p["betrag_orig"] for p in treffer) + sa = sum(p["betrag"] for p in treffer) + proz = {p["prozent"] for p in treffer} + if all(_ist_staffel(p) for p in treffer): + pz = "Staffel" + else: + pz = fmt_percent(next(iter(proz))) if len(proz) == 1 and None not in proz else ("gemischt" if so else "") + t.item(giid, text=f"{gname} ({len(treffer)})", values=("", "", "", "", "", fmt(so), pz, fmt(sa))) + if such: + t.item(giid, open=True) name = {ALLE: "Alle Positionen", UNZUGEORDNET: "Nicht zugeordnet", EIGEN: "Eigene Kennungen (ausgeblendet, werden nicht exportiert)"}.get( self.ansicht, self.haendler_namen.get(self.ansicht, "")) @@ -616,13 +687,6 @@ class App: return list(dict.fromkeys(pids)) # ================================================================= Aktionen - def _brauche_abrechnung(self, frage_neu=True) -> bool: - if self.aid: - return True - if frage_neu: - return self.neu() - return False - def _vorschlag_titel(self) -> str: heute = date.today() m = heute.month - 1 or 12 @@ -680,6 +744,7 @@ class App: if self._gesperrt(): return bericht = [] + importiert: list[tuple[int, int, list[str]]] = [] # (Zeile im Bericht, datei_id, Hinweise) self.root.configure(cursor="watch") self.root.update() try: @@ -701,15 +766,20 @@ class App: bericht.append((name, erg.format, 0, 0, "übersprungen (bereits importiert)")) continue n = self.db.importieren(self.aid, erg) - did = self.db.dateien(self.aid)[-1]["id"] - doppelt = sum(1 for p in self.db.positionen(self.aid) if p["datei_id"] == did and p["doppelt"]) - hinweise = list(erg.warnungen) + ([f"{doppelt} Pos. bereits abgerechnet (werden nicht erneut " - "exportiert)"] if doppelt else []) - bericht.append((name, erg.format, n, sum(p.betrag_cent for p in erg.positionen), - "; ".join(hinweise) or "OK")) + importiert.append((len(bericht), self.db.dateien(self.aid)[-1]["id"], list(erg.warnungen))) + bericht.append((name, erg.format, n, sum(p.betrag_cent for p in erg.positionen), "")) finally: self.root.configure(cursor="") self._aktualisieren() + # Doppelte (schon abgerechnet oder schon offen vorhanden) je Datei zählen – einmal für alle Dateien + doppelt: dict[int, int] = defaultdict(int) + for p in self.positionen: + if p["doppelt"]: + doppelt[p["datei_id"]] += 1 + for zeile, did, hinweise in importiert: + if doppelt.get(did): + hinweise.append(f"{doppelt[did]} Pos. doppelt bzw. bereits abgerechnet (werden nicht exportiert)") + bericht[zeile] = bericht[zeile][:4] + ("; ".join(hinweise) or "OK",) unbekannt: dict[tuple[str, str], list[int]] = defaultdict(lambda: [0, 0]) for p in self.positionen: if p["haendler_id"] == UNZUGEORDNET: @@ -1071,6 +1141,8 @@ class App: w.info(self.root, "Export", "Diese Abrechnung enthält noch keine Positionen.\n" "Bitte zuerst Dateien importieren.") return + if self._gesperrt(): # erledigt: nichts mehr exportieren (und damit abrechnen) + return # nur offene Positionen echter Händler (nicht: nicht zugeordnet, eigene, schon abgerechnet) offen = [p for p in self.positionen if p["haendler_id"] > 0 and not p["abgerechnet"] and not p["doppelt"]] stat: dict[int, list[int]] = defaultdict(lambda: [0, 0]) @@ -1095,12 +1167,14 @@ class App: config.save(self.settings) erzeugt, exportiert = [], [] zeit = res["zeitraum"] + vergeben: set[str] = set() # Dateinamen dieses Exports – gleich lautende Händler überschreiben sich nicht try: for hid in res["ids"]: pos = [p for p in offen if p["haendler_id"] == hid and (zeit is None or zeit.enthaelt(p))] erzeugt += exp.exportiere_haendler(res["ordner"], self.haendler_namen[hid], pos, res["titel"], pro_anbieter=res["pro_anbieter"], kopfzeile=res["kopfzeile"], - summe=res["summe"], null_weglassen=res["null_weglassen"]) + summe=res["summe"], null_weglassen=res["null_weglassen"], + vergeben=vergeben) exportiert += pos except PermissionError as exc: w.fehler(self.root, "Export", f"Die Datei konnte nicht geschrieben werden:\n{exc.filename}\n\n" @@ -1346,13 +1420,13 @@ class App: try: self.settings["trenner"] = self.pw.sash_coord(0)[0] self.settings["fenster"] = self.rahmen.normale_geometrie() - self.settings["maximiert"] = self.rahmen.maximiert config.save(self.settings) except (tk.TclError, OSError): pass finally: try: - self.db.close() + if self.db is not None: + self.db.close() finally: try: self.root.destroy() diff --git a/provtool/gui/dialogs.py b/provtool/gui/dialogs.py index 7229c7f..ff2bd2d 100644 --- a/provtool/gui/dialogs.py +++ b/provtool/gui/dialogs.py @@ -14,7 +14,6 @@ from provtool.money import fmt, fmt_percent, parse_cent, parse_percent # Auswahl im Händler-Dialog: Schlüssel -> Anzeige KENNUNG_ARTEN = {"freenet": "freenet (Unterhändler-Nr.)", "vodafone": "Vodafone (VO-Nr.)"} -CAPS_TEXT = "Kunden-Nachname komplett in GROSSBUCHSTABEN" CAPS_HAKEN = "Vodafone-Kunden mit Nachnamen in GROSSBUCHSTABEN gehören zu diesem Händler" @@ -295,6 +294,12 @@ class HaendlerDialog(w.Dialog): return if self.v_kwert.get().strip(): self._kennung_dazu() + if self.aktuell and self.aktuell.aktiv and not self.v_aktiv.get() and not w.frage( + self, "Händler deaktivieren", + f"„{name}“ inaktiv setzen?\n\nSeine noch offenen Positionen gelten dann als „nicht zugeordnet“ " + "und zeigen wieder die volle Provision (100 %); exportiert werden sie nicht. Bereits " + "abgerechnete Positionen bleiben, wie sie sind.", ja="Inaktiv setzen", default_ja=False): + return h = Haendler(self.aktuell.id if self.aktuell else 0, name, self.v_ap.get().strip(), self.v_mail.get().strip(), self.t_notiz.get("1.0", "end").strip(), proz, self.v_aktiv.get(), air, self._eigene_staffel if self.v_eigen_staffel.get() else None) @@ -628,7 +633,11 @@ class AbrechnungenDialog(w.Dialog): titel = self.db.abrechnung(aid)["titel"] if w.frage(self, "Löschen", f"Abrechnung „{titel}“ mit allen Positionen und Änderungen endgültig löschen?", default_ja=False): - self.db.abrechnung_loeschen(aid) + try: + self.db.abrechnung_loeschen(aid) + except AbrechnungErledigt as exc: + w.info(self, "Löschen", str(exc)) + return self._laden() self.result = ("geloescht", aid) diff --git a/provtool/gui/win32.py b/provtool/gui/win32.py index 5f50694..b6acb68 100644 --- a/provtool/gui/win32.py +++ b/provtool/gui/win32.py @@ -191,11 +191,6 @@ def ist_bedienbar(fenster: tk.Misc) -> bool: return bool(user32().IsWindowEnabled(h)) if h else True -def letztes_popup(fenster: tk.Misc) -> int | None: - h = hwnd(fenster) - return user32().GetLastActivePopup(h) if h else None - - class Modal: """Macht `fenster` modal zu `besitzer` (Windows-Art). beenden() muss VOR dem Zerstören des Fensters laufen: erst Besitzer wieder freigeben, dann schließen – sonst aktiviert Windows diff --git a/provtool/gui/win95.py b/provtool/gui/win95.py index 4665ea3..a93c1d4 100644 --- a/provtool/gui/win95.py +++ b/provtool/gui/win95.py @@ -593,34 +593,6 @@ class Sunken(tk.Frame): super().__init__(master, **kw) -class Fenster(tk.Frame): - """Ein „Fenster im Fenster“ mit blauer Titelleiste – reine Optik, gliedert die Oberfläche.""" - - def __init__(self, master, titel: str, icon: tk.PhotoImage | None = None, **kw): - if MODERN: # schlichte Überschrift statt blauer Titelleiste - super().__init__(master, bd=0, bg=FACE, **kw) - self.titel_var = tk.StringVar(value=titel) - self.titelleiste = tk.Label(self, textvariable=self.titel_var, bg=FACE, fg=DARK, font=FONT_BOLD, - anchor="w") - self.titelleiste.pack(fill="x", pady=(0, 4)) - self.inhalt = tk.Frame(self, bg=FACE) - self.inhalt.pack(fill="both", expand=True) - return - super().__init__(master, bd=2, relief="raised", bg=FACE, **kw) - self.titelleiste = tk.Frame(self, bg=NAVY, height=18) - self.titelleiste.pack(fill="x", padx=1, pady=1) - if icon is not None: - tk.Label(self.titelleiste, image=icon, bg=NAVY).pack(side="left", padx=(2, 0)) - self.titel_var = tk.StringVar(value=titel) - tk.Label(self.titelleiste, textvariable=self.titel_var, bg=NAVY, fg=WHITE, font=FONT_BOLD, - anchor="w").pack(side="left", fill="x", padx=3) - self.inhalt = tk.Frame(self, bg=FACE) - self.inhalt.pack(fill="both", expand=True, padx=2, pady=2) - - def set_titel(self, t: str): - self.titel_var.set(t) - - class Statusleiste(tk.Frame): def __init__(self, master, felder: list[tuple[str, int]]): super().__init__(master, bg=FACE) @@ -671,7 +643,7 @@ class Karte(tk.Frame): def titelleiste(master, text: str = "") -> tuple[tk.Frame, tk.Label]: - """Win95: blaue Titelleiste oben in einer Karte (wie bei Fenster); rechts passen kleine Knöpfe hinein.""" + """Win95: blaue Titelleiste oben in einer Karte („Fenster im Fenster“); rechts passen kleine Knöpfe hinein.""" f = tk.Frame(master, bg=NAVY) f.pack(fill="x", padx=1, pady=1) lbl = tk.Label(f, text=text, bg=NAVY, fg=WHITE, font=FONT_BOLD, anchor="w", padx=3) @@ -768,35 +740,6 @@ def logo(master, groesse: int = 34) -> tk.Canvas: return c -class ToolbarButton(tk.Label): - """Flacher Werkzeugleisten-Knopf (Office-95-Stil): hebt sich beim Überfahren, senkt sich beim Klick.""" - - def __init__(self, master, text: str, icon: tk.PhotoImage, command, tooltip: str = ""): - super().__init__(master, text=text, image=icon, compound="top", bd=1, relief="flat", bg=FACE, - padx=6, pady=2, font=FONT) - self.command = command - if MODERN: # Hervorhebung durch Farbe statt 3D-Kante - self.configure(bd=0, padx=8, pady=4) - self.bind("", lambda e: self.configure(bg=LIGHT)) - self.bind("", lambda e: self.configure(bg=FACE)) - self.bind("", lambda e: self.configure(bg=SEL_BG)) - self.bind("", self._release) - if tooltip: - Tooltip(self, tooltip) - return - self.bind("", lambda e: self.configure(relief="raised")) - self.bind("", lambda e: self.configure(relief="flat")) - self.bind("", lambda e: self.configure(relief="sunken")) - self.bind("", self._release) - if tooltip: - Tooltip(self, tooltip) - - def _release(self, e): - self.configure(bg=LIGHT) if MODERN else self.configure(relief="raised") - if 0 <= e.x < self.winfo_width() and 0 <= e.y < self.winfo_height() and self.command: - self.command() - - class Tooltip: def __init__(self, widget, text: str): self.widget, self.text, self.tip, self._after = widget, text, None, None diff --git a/provtool/importer.py b/provtool/importer.py index bbe69bf..2f14b9a 100644 --- a/provtool/importer.py +++ b/provtool/importer.py @@ -8,7 +8,9 @@ from __future__ import annotations import csv import hashlib import io -from dataclasses import asdict, dataclass, field +import itertools +from collections.abc import Iterable, Iterator +from dataclasses import dataclass, field from datetime import date, datetime, timedelta from pathlib import Path @@ -18,6 +20,10 @@ FREENET = "freenet" VODAFONE = "vodafone" ANBIETER = {FREENET: "freenet", VODAFONE: "Vodafone"} +# Grenzen gegen versehentlich falsche oder kaputte Dateien (echte Abrechnungen: wenige MB, einige tausend Zeilen) +MAX_BYTES = 100 * 1024 * 1024 # Dateigröße, bei .xlsx auch entpackt +MAX_ZEILEN = 200_000 # Datenzeilen je Tabelle + @dataclass class Position: @@ -39,11 +45,6 @@ class Position: quelle: str = "" roh: dict = field(default_factory=dict) - def as_dict(self) -> dict: - d = asdict(self) - d["datum"] = self.datum.isoformat() if self.datum else None - return d - @dataclass class ImportErgebnis: @@ -183,15 +184,24 @@ FORMATE: list[Format] = [FreenetFormat(), VodafoneFormat()] # --------------------------------------------------------------------------- Tabellen lesen -def _tabellen_aus_rows(rows: list[list], quelle: str, blatt: str = "") -> tuple[Format, list[Position]] | None: - """Sucht in den ersten 40 Zeilen eine bekannte Kopfzeile und liest alle Datenzeilen darunter.""" - for idx, row in enumerate(rows[:40]): +def _tabellen_aus_rows(rows: Iterable[list], quelle: str, blatt: str = "", + nur_kopf: bool = False) -> tuple[Format, list[Position]] | None: + """Sucht in den ersten 40 Zeilen eine bekannte Kopfzeile und liest alle Datenzeilen darunter. + Ohne Kopfzeile (oder mit nur_kopf) wird der Rest der Tabelle gar nicht gelesen.""" + rows = iter(rows) + anfang = list(itertools.islice(rows, 40)) + for idx, row in enumerate(anfang): header = [_norm(c) for c in row] for fmt in FORMATE: if fmt.passt(header): pos: list[Position] = [] + if nur_kopf: + return fmt, pos leer = 0 - for data in rows[idx + 1:]: + for nr, data in enumerate(itertools.chain(anfang[idx + 1:], rows)): + if nr >= MAX_ZEILEN: + raise ImportFehler(f"„{quelle}“ hat mehr als {MAX_ZEILEN // 1000}.000 Zeilen – das ist " + "keine Provisionsabrechnung.") if not any(_text(c) for c in data): leer += 1 if leer > 20: @@ -222,7 +232,7 @@ def _decode(raw: bytes) -> str: return raw.decode("latin-1", errors="replace") -def _lies_csv(path: Path) -> list[tuple[str, list[list]]]: +def _lies_csv(path: Path) -> Iterator[tuple[str, Iterable[list]]]: text = _decode(path.read_bytes()) sample = text[:5000] try: @@ -230,15 +240,20 @@ def _lies_csv(path: Path) -> list[tuple[str, list[list]]]: delim = dialect.delimiter except csv.Error: delim = ";" - rows = list(csv.reader(io.StringIO(text), delimiter=delim)) - return [("", rows)] + yield "", csv.reader(io.StringIO(text), delimiter=delim) -def _lies_xlsx(path: Path) -> list[tuple[str, list[list]]]: +def _lies_xlsx(path: Path) -> Iterator[tuple[str, Iterable[list]]]: + import zipfile + import openpyxl + with zipfile.ZipFile(path) as z: # entpackte Größe prüfen (eine .xlsx ist ein Zip-Archiv) + if sum(i.file_size for i in z.infolist()) > MAX_BYTES: + raise ImportFehler(f"„{path.name}“ ist entpackt größer als {MAX_BYTES // 1024 // 1024} MB.") wb = openpyxl.load_workbook(path, read_only=True, data_only=True) try: - return [(ws.title.strip(), [list(r) for r in ws.iter_rows(values_only=True)]) for ws in wb.worksheets] + for ws in wb.worksheets: # Zeilen werden erst beim Durchlaufen gelesen (nur Blätter mit Kopfzeile ganz) + yield ws.title.strip(), (list(r) for r in ws.iter_rows(values_only=True)) finally: wb.close() @@ -265,19 +280,33 @@ def _lies_xls(path: Path) -> list[tuple[str, list[list]]]: def lies_datei(path: str | Path) -> ImportErgebnis: path = Path(path) ext = path.suffix.lower() + if ext not in (".csv", ".txt", ".xlsx", ".xlsm", ".xls"): + raise ImportFehler(f"Dateityp {ext or '(ohne Endung)'} wird nicht unterstützt.") + if path.stat().st_size > MAX_BYTES: + raise ImportFehler(f"„{path.name}“ ist größer als {MAX_BYTES // 1024 // 1024} MB – das ist keine " + "Provisionsabrechnung.") if ext in (".csv", ".txt"): blaetter = _lies_csv(path) elif ext in (".xlsx", ".xlsm"): blaetter = _lies_xlsx(path) - elif ext == ".xls": - blaetter = _lies_xls(path) else: - raise ImportFehler(f"Dateityp {ext or '(ohne Endung)'} wird nicht unterstützt.") + blaetter = _lies_xls(path) gefunden: list[tuple[str, Format, list[Position]]] = [] + ignoriert: list[str] = [] + erstes = None for name, rows in blaetter: - res = _tabellen_aus_rows(rows, path.name, name) - if res and res[1]: + if erstes is None: + erstes = name + # Vodafone/Euronics: Maßgeblich ist nur die erste Seite ("All") – weitere Blätter mit Abrechnung + # werden nur erkannt (für den Hinweis), aber nicht mehr gelesen + vodafone_schon = any(f.key == VodafoneFormat.key for _, f, _ in gefunden) + res = _tabellen_aus_rows(rows, path.name, name, nur_kopf=vodafone_schon) + if not res: + continue + if vodafone_schon: + ignoriert.append(name) + elif res[1]: gefunden.append((name, res[0], res[1])) if not gefunden: @@ -288,11 +317,10 @@ def lies_datei(path: str | Path) -> ImportErgebnis: ) warnungen: list[str] = [] - # Vodafone/Euronics: Maßgeblich ist nur die erste Seite ("All") – sie enthält alle - # Abrechnungsläufe. Weitere Blätter würden Positionen doppelt zählen. + # Vodafone/Euronics: die erste Seite ("All") enthält alle Abrechnungsläufe, weitere Blätter würden + # Positionen doppelt zählen. if any(f.key == VodafoneFormat.key for _, f, _ in gefunden): - erstes = blaetter[0][0] - uebrig = [g[0] for g in gefunden if g[0] != erstes] + uebrig = [g[0] for g in gefunden if g[0] != erstes] + ignoriert gefunden = [g for g in gefunden if g[0] == erstes] if uebrig: warnungen.append(f"Nur die erste Seite „{erstes}“ übernommen, ignoriert: {', '.join(uebrig)}") diff --git a/tests/test_gui.py b/tests/test_gui.py index a59604c..d189efa 100644 --- a/tests/test_gui.py +++ b/tests/test_gui.py @@ -1123,6 +1123,10 @@ def test_export_zeitraum_kennzeichnen_und_erledigt(lauf_modern, tmp_path, monkey app.ptree.selection_set(app.ptree.get_children()[0]) app.prozent() assert "erledigt" in meldungen[-1] + # ... auch der Export (würde sonst offene Positionen abrechnen) + n = len(meldungen) + app.exportieren() + assert lauf.dialog(ExportDialog) is None and len(meldungen) == n + 1 and "erledigt" in meldungen[-1] lauf.pruefen() @@ -1205,3 +1209,53 @@ def test_startet_maximiert(home): assert lauf.app.rahmen.maximiert, "Programm startet nicht maximiert" finally: lauf.ende() + + +def test_suche_und_filter_blenden_nur_aus(lauf_modern): + """Suche/Filter bauen die Tabelle nicht neu auf: Zeilen werden aus- und wieder eingehängt.""" + lauf = lauf_modern + app = lauf.app + alle = [i for g in app.ptree.get_children() for i in app.ptree.get_children(g)] + assert len(alle) == 3 + app.v_suche.set("Beispiel") + app._suche_ausfuehren() + sichtbar = [i for g in app.ptree.get_children() for i in app.ptree.get_children(g)] + assert len(sichtbar) == 1 and app.ptree.item(sichtbar[0], "text") == "Beispiel KG" + assert all(app.ptree.exists(i) for i in alle), "ausgeblendete Zeilen bleiben erhalten (nur ausgehängt)" + assert app.kpi["pos"][0].cget("text") == "1" + gruppe = app.ptree.parent(sichtbar[0]) + assert "(1)" in app.ptree.item(gruppe, "text") and app.ptree.item(gruppe, "open") + app.ptree.selection_set(gruppe) + assert app._ausgewaehlte_pids() == [int(sichtbar[0][1:])], "Gruppenauswahl = nur sichtbare Zeilen" + app.v_suche.set("") + app._suche_ausfuehren() + assert [i for g in app.ptree.get_children() for i in app.ptree.get_children(g)] == alle + assert app.kpi["pos"][0].cget("text") == "3" + lauf.pruefen() + + +def test_start_mit_unerreichbarer_datenbank_fragt(home, monkeypatch): + """Eingestellter (Netz-)Pfad fehlt: Rückfrage statt Absturz – und es wird dort keine leere DB angelegt.""" + from provtool import config + from provtool.gui import win95 as w + _profil_mit_daten(home) + weg = home / "netz" / "provisionen.db" + fragen = [] + monkeypatch.setattr(w, "_meldung", lambda m, t, text, icon, knoepfe, default=0: fragen.append(text) or "standard") + lauf = Lauf(home, {"assistent_erledigt": True, "tipps_beim_start": False, "db_pfad": str(weg)}) + try: + assert len(fragen) == 1 and str(weg) in fragen[0] + assert not weg.exists() and lauf.app.db.path == config.app_dir() / "provisionen.db" + assert config.load()["db_pfad"] is None, "Wahl „Standard“ wird gespeichert" + assert len(lauf.app.positionen) == 3 + finally: + lauf.ende() + lauf.pruefen() + # „Beenden“ gewählt: Programm endet sauber + monkeypatch.setattr(w, "_meldung", lambda *a, **k: "ende") + (home / "einstellungen.json").write_text(json.dumps({"db_pfad": str(weg), "vollbild": False}), encoding="utf-8") + from provtool.gui import app as A + root = tk.Tk() + with pytest.raises(SystemExit): + A.App(root) + assert not weg.exists() diff --git a/tests/test_kern.py b/tests/test_kern.py index 52f2523..5301424 100644 --- a/tests/test_kern.py +++ b/tests/test_kern.py @@ -468,12 +468,16 @@ def test_abgerechnet_einfrieren_und_doppelt_erkennen(db, freenet_csv): db.haendler_speichern(db.haendler(hid), [("freenet", "111")]) a2 = db.abrechnung_neu("Nov") db.importieren(a2, lies_datei(freenet_csv)) + # (2 davon schon abgerechnet, die 3 anderen sind in „Okt“ noch offen – auch die zählen nur einmal) doppelt = [p for p in db.positionen(a2) if p["doppelt"]] - assert len(doppelt) == 2 and all(p["doppelt"] == "Okt" for p in doppelt) + assert len(doppelt) == 5 and all(p["doppelt"] == "Okt" for p in doppelt) assert db.abrechnen([p["id"] for p in doppelt]) == 0, "doppelte Positionen dürfen nicht erneut abgerechnet werden" - # Kennzeichnung aufheben -> nicht mehr doppelt + # Kennzeichnung aufheben -> die Positionen in „Okt“ sind wieder offen und bleiben die zuerst importierten; + # „Nov“ bleibt doppelt. Erst wenn „Okt“ weg ist, ist „Nov“ frei. assert db.abrechnung_aufheben([p["id"] for p in erste]) == 2 + assert all(p["doppelt"] == "Okt" for p in db.positionen(a2)) + db.abrechnung_loeschen(a1) assert not any(p["doppelt"] for p in db.positionen(a2)) @@ -515,3 +519,180 @@ def test_migration_v3_fingerabdruck(tmp_path, freenet_csv): assert all(p["fp"] for p in d.positionen(aid)) assert len({p["fp"] for p in d.positionen(aid)}) == 5 d.close() + + +# --------------------------------------------------------------------------- Audit-Befunde (Nachtests) + +def test_offene_doppelte_nur_einmal_exportieren_und_abrechnen(db, freenet_csv, tmp_path): + """Wochen- und Monatsdatei mit denselben Zeilen in EINER Abrechnung: die zweite ist „doppelt“ und + wird weder exportiert noch abgerechnet. Zwei gleiche Zeilen in EINER Datei bleiben zwei Positionen.""" + hid = db.haendler_speichern(Haendler(0, "A"), [("freenet", "111")]) + aid = db.abrechnung_neu("Okt") + db.importieren(aid, lies_datei(freenet_csv)) + monat = tmp_path / "monat.csv" + monat.write_bytes(freenet_csv.read_bytes() + freenet_csv.read_bytes().split(b"\r\n", 1)[1]) # Zeilen 2x + db.importieren(aid, lies_datei(monat)) + pos = db.positionen(aid) + assert len(pos) == 15 + erste_datei = [p for p in pos if p["datei_id"] == db.dateien(aid)[0]["id"]] + assert not any(p["doppelt"] for p in erste_datei), "die zuerst importierte Datei zählt" + zweite = [p for p in pos if p["datei_id"] != db.dateien(aid)[0]["id"]] + assert sum(1 for p in zweite if p["doppelt"] == "Okt") == 5, "je Zeile ist nur die erste Wiederholung doppelt" + assert sum(1 for p in zweite if not p["doppelt"]) == 5, "zweite Vorkommen innerhalb der Datei (#2) sind eigene Positionen" + # abrechnen aller Händler-Positionen: Doppelte bleiben offen + n = db.abrechnen([p["id"] for p in pos]) + offen_haendler = [p for p in db.positionen(aid) if p["haendler_id"] == hid and not p["abgerechnet"]] + assert n == 8 and all(p["doppelt"] for p in offen_haendler) + # und noch einmal: derselbe Fingerabdruck wird in der Transaktion erneut geprüft + assert db.abrechnen([p["id"] for p in db.positionen(aid)]) == 0 + + +def test_staffel_ohne_doppelte_und_datei_entfernen_rechnet_neu(db, bonus_csv, tmp_path): + from provtool import regeln + hid = db.haendler_speichern(Haendler(0, "Sub"), [("freenet", "111")]) + aid = db.abrechnung_neu("Okt") + db.importieren(aid, lies_datei(bonus_csv)) + kopie = tmp_path / "bonus2.csv" + kopie.write_bytes(bonus_csv.read_bytes()) + db.importieren(aid, lies_datei(kopie)) + bonus = [p for p in db.positionen(aid) if p["typ"] == regeln.JAHRESBONUS and p["haendler_id"] == hid] + assert len(bonus) == 26 and sum(1 for p in bonus if p["doppelt"]) == 13 + echte = [p for p in bonus if not p["doppelt"]] + assert sorted(p["betrag"] for p in echte) == [300] + [1000] * 12, "Doppelte zählen nicht als Karten (13, nicht 26)" + # zweite Datei (3 Zeilen) zusätzlich mit anderen Rufnummern -> 16 Karten -> Stufe 15: 15 € je Karte + extra = tmp_path / "extra.csv" + extra.write_bytes(("\r\n".join([FREENET_KOPF] + [_pos("Jahresbonus Neuaktivierung", "40", rufnr=f"17000000{i:02d}") + for i in range(3)]) + "\r\n").encode("cp1252")) + db.importieren(aid, lies_datei(extra)) + echte = [p for p in db.positionen(aid) if p["typ"] == regeln.JAHRESBONUS and not p["doppelt"]] + assert sorted(p["betrag"] for p in echte) == [300] + [1500] * 15 + # Datei entfernen -> Staffel der übrigen wieder 10 € je Karte + db.datei_entfernen(db.dateien(aid)[-1]["id"]) + echte = [p for p in db.positionen(aid) if p["typ"] == regeln.JAHRESBONUS and not p["doppelt"]] + assert sorted(p["betrag"] for p in echte) == [300] + [1000] * 12 + + +def test_abgerechnete_positionen_nicht_loeschbar(db, freenet_csv): + """Datei/Abrechnung mit abgerechneten Positionen: Löschen verweigert (sonst Fingerabdruck-Leichen + oder doppelte Auszahlung). Nach Aufheben der Kennzeichnung geht es – und hinterlässt keine Fingerabdrücke.""" + from provtool.db import AbrechnungErledigt + db.haendler_speichern(Haendler(0, "A"), [("freenet", "111")]) + aid = db.abrechnung_neu("Okt") + db.importieren(aid, lies_datei(freenet_csv)) + pos = db.positionen(aid) + assert db.abrechnen([pos[0]["id"]]) == 1 + with pytest.raises(AbrechnungErledigt): + db.datei_entfernen(db.dateien(aid)[0]["id"]) + with pytest.raises(AbrechnungErledigt): + db.abrechnung_loeschen(aid) + assert len(db.positionen(aid)) == 5 + db.abrechnung_aufheben([pos[0]["id"]]) + db.abrechnung_loeschen(aid) + assert db.con.execute("SELECT count(*) FROM abgerechnet_fp").fetchone()[0] == 0 + a2 = db.abrechnung_neu("Nov") + db.importieren(a2, lies_datei(freenet_csv)) + assert not any(p["doppelt"] for p in db.positionen(a2)) + + +def test_erledigt_offene_positionen_bekommen_keinen_haendler_mehr(db, freenet_csv): + """Erledigt: eine damals nicht zugeordnete Position darf einem später angelegten Händler nicht mit + 100 % zur Auszahlung erscheinen; abrechnen() fasst erledigte Abrechnungen nicht an.""" + db.haendler_speichern(Haendler(0, "A"), [("freenet", "111")]) + aid = db.abrechnung_neu("Okt") + db.importieren(aid, lies_datei(freenet_csv)) + db.erledigt_setzen(aid, True) + hid_b = db.haendler_speichern(Haendler(0, "B", standard_prozent=50), [("freenet", "999")]) + p = next(p for p in db.positionen(aid) if p["kennung"] == "999") + assert p["haendler_id"] == UNZUGEORDNET and not p["abgerechnet"] + assert db.abrechnen([p["id"]]) == 0 + # in einer offenen Abrechnung gehört dieselbe Kennung sehr wohl zu B (mit 50 %) + a2 = db.abrechnung_neu("Nov") + db.importieren(a2, lies_datei(freenet_csv)) + q = next(p for p in db.positionen(a2) if p["kennung"] == "999") + assert q["haendler_id"] == hid_b and q["betrag"] == -400 + + +def test_export_text_mit_gleichheitszeichen_bleibt_text(tmp_path): + pos = [{"anbieter": "freenet", "zeitraum": "2026.10.01-2026.10.07", "provisionsart": "=1+1", "kennung": "111", + "datum": "2026-10-05", "rufnummer": "1601", "imei": "", "geraet": "", "vorname": "", + "name": '=HYPERLINK("http://böse.example","Klick")', "tarif": "T", "betrag": 800, "betrag_orig": 800, + "netz": "", "dienst": "", "aktionscode": "=cmd|' /C calc'!A0", "typ": "normal", "reihenfolge": 1}] + datei = export.exportiere_haendler(tmp_path, "A", pos, "Okt")[0] + ws = openpyxl.load_workbook(datei).active + zeile = [c for c in ws[1]] + assert zeile[1].value == "=1+1" and zeile[1].data_type == "s" + assert zeile[8].value.startswith("=HYPERLINK") and zeile[8].data_type == "s" + assert zeile[13].data_type == "s" + summe = next(c for row in ws.iter_rows() for c in row if isinstance(c.value, str) and c.value.startswith("=SUM")) + assert summe.data_type == "f", "unsere Summenformel bleibt eine Formel" + + +def test_export_gleiche_dateinamen_ueberschreiben_sich_nicht(tmp_path): + p = {"anbieter": "freenet", "zeitraum": "", "provisionsart": "X", "kennung": "1", "datum": None, "rufnummer": "", + "imei": "", "geraet": "", "vorname": "", "name": "K", "tarif": "", "betrag": 100, "betrag_orig": 100, + "netz": "", "dienst": "", "aktionscode": "", "typ": "normal", "reihenfolge": 1} + vergeben: set[str] = set() + a = export.exportiere_haendler(tmp_path, "A/B", [p], "Okt", vergeben=vergeben) + b = export.exportiere_haendler(tmp_path, "A:B", [dict(p, betrag=200)], "Okt", vergeben=vergeben) + assert a[0].name == "A_B - Okt.xlsx" and b[0].name == "A_B - Okt (2).xlsx" + assert openpyxl.load_workbook(a[0]).active["K1"].value == 1.0 + assert openpyxl.load_workbook(b[0]).active["K1"].value == 2.0 + + +def test_restore_prueft_sicherung(db, freenet_csv, tmp_path): + import sqlite3 + aid = db.abrechnung_neu("Okt") + db.importieren(aid, lies_datei(freenet_csv)) + db.close() + ziel = db.path + vorher = ziel.read_bytes() + # keine SQLite-Datei + muell = tmp_path / "muell.db" + muell.write_bytes(b"kein sqlite") + with pytest.raises(Exception): + Datenbank.restore(muell, ziel) + # SQLite, aber nicht unsere Tabellen + fremd = tmp_path / "fremd.db" + con = sqlite3.connect(fremd) + con.execute("CREATE TABLE haendler (id INTEGER)") + con.commit() + con.close() + with pytest.raises(ValueError): + Datenbank.restore(fremd, ziel) + assert ziel.read_bytes() == vorher and not ziel.with_suffix(".pruefung.db").exists() + # echte Sicherung + d = Datenbank(ziel) + sicherung = tmp_path / "sicherung.db" + d.backup(sicherung) + d.abrechnung_loeschen(aid) + d.close() + Datenbank.restore(sicherung, ziel) + d = Datenbank(ziel) + assert len(d.positionen(aid)) == 5 + d.close() + + +def test_import_grenzen(tmp_path, freenet_csv, vodafone_xlsx, monkeypatch): + from provtool import importer + monkeypatch.setattr(importer, "MAX_ZEILEN", 3) + with pytest.raises(ImportFehler, match="Zeilen"): + lies_datei(freenet_csv) + monkeypatch.setattr(importer, "MAX_ZEILEN", 200_000) + monkeypatch.setattr(importer, "MAX_BYTES", 10) + with pytest.raises(ImportFehler, match="MB"): + lies_datei(freenet_csv) + monkeypatch.setattr(importer, "MAX_BYTES", vodafone_xlsx.stat().st_size + 1) # gepackt klein, entpackt zu groß + with pytest.raises(ImportFehler, match="entpackt"): + lies_datei(vodafone_xlsx) + + +def test_einstellungen_atomar_und_defekt(tmp_path, monkeypatch): + from provtool import config + monkeypatch.setenv("PROVTOOL_HOME", str(tmp_path)) + config.save({"db_pfad": "Z:/x.db", "defekt": "weg"}) + assert not (tmp_path / "einstellungen.json.tmp").exists() + assert config.load()["db_pfad"] == "Z:/x.db" and "defekt" not in config.load() + (tmp_path / "einstellungen.json").write_text("{ kaputt", encoding="utf-8") + daten = config.load() + assert daten["db_pfad"] is None and daten["defekt"].endswith("einstellungen.defekt.json") + assert (tmp_path / "einstellungen.defekt.json").read_text(encoding="utf-8") == "{ kaputt"