3 Commits
Author SHA1 Message Date
Kerim AtilganandClaude Opus 5.5 2275b92188 Resize-Zeiger blieb im Inhalt hängen (vererbt von der Trennlinie): Bereiche mit eigenem Zeiger; Version 1.8.1
Build / linux (push) Successful in 1m55s
Build / windows-exe (push) Successful in 8m7s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 16:33:09 +01:00
Kerim AtilganandClaude Opus 5.5 b85a75857e Updater: Prüfung beim Start und unter Hilfe, Setup herunterladen, DB sichern, still installieren, Neustart
Build / linux (push) Successful in 1m56s
Build / windows-exe (push) Successful in 8m6s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 15:18:22 +01:00
Kerim AtilganandClaude Opus 5.5 968c68ac1a Fixes aus den Audits (Fable/Codex): Doppelte, Staffel, Export-Sperre, Netz-DB, atomare Einstellungen, Formeln, Wiederherstellung, Importgrenzen, schnellere Suche; toter Code entfernt; Version 1.8.0
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 14:58:51 +01:00
14 changed files with 808 additions and 179 deletions

No files matched your search

+10
View File
@@ -86,11 +86,21 @@ 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
das dunkle Design um. Das Windows-95-Design gibt es weiter unter *Datei ▸ Einstellungen ▸ Darstellung*.
**Updates:** Das Programm schaut beim Start auf Gitea nach einer neueren Version (abschaltbar unter
*Datei ▸ Einstellungen*, von Hand unter *Hilfe ▸ Nach Updates suchen*). Die per Setup installierte
Windows-Version lädt das neue Setup, sichert die Datenbank (`vor-update-<alte Version>.db` im Einstellungsordner),
installiert still und startet neu. Die einzelne .exe und Linux öffnen stattdessen die Download-Seite.
**Linux:** `Provisionsmanager95-linux-x86_64-<Version>.tar.gz` entpacken und `Provisionsmanager95` starten
(ab Debian 12 / Ubuntu 24.04). Daten liegen unter `~/.config/Provisionsmanager95`.
+8
View File
@@ -43,3 +43,11 @@ Name: "{autodesktop}\Provisions-Manager"; Filename: "{app}\Provisionsmanager.exe
[Run]
Filename: "{app}\Provisionsmanager.exe"; Description: "{cm:LaunchProgram,Provisions-Manager}"; Flags: nowait postinstall skipifsilent
; Update aus dem Programm (/VERYSILENT /NEUSTART=1): danach wieder starten
Filename: "{app}\Provisionsmanager.exe"; Flags: nowait; Check: NachUpdateStarten
[Code]
function NachUpdateStarten: Boolean;
begin
Result := ExpandConstant('{param:NEUSTART|0}') = '1';
end;
+1 -1
View File
@@ -1,5 +1,5 @@
"""Provisions-Manager 95 – Provisionsabrechnung für Unterhändler der Handy-Insel Miesbach."""
__version__ = "1.7.0"
__version__ = "1.8.1"
APP_NAME = "Provisions-Manager 95"
APP_ORG = "Handy-Insel Miesbach"
+15 -1
View File
@@ -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):
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:
+69 -25
View File
@@ -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:
test.close()
shutil.copyfile(quelle, ziel)
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
@@ -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:
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
+13 -2
View File
@@ -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()):
+233 -59
View File
@@ -5,14 +5,17 @@ import os
import re
import subprocess
import sys
import tempfile
import threading
import tkinter as tk
import traceback
import webbrowser
from collections import defaultdict
from datetime import date
from pathlib import Path
from tkinter import filedialog
from provtool import config
from provtool import __version__, config, updater
from provtool import export as exp
from provtool import regeln
from provtool.db import EIGEN, UNZUGEORDNET, AbrechnungErledigt, Datenbank
@@ -61,11 +64,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 +107,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)."""
@@ -197,6 +238,7 @@ class App:
hi.add_command(label="Kurzanleitung", accelerator="F1", command=self.hilfe)
hi.add_command(label="Rundgang durch das Programm", command=self.rundgang)
hi.add_command(label="Tipp des Tages...", command=self.tipp_des_tages)
hi.add_command(label="Nach Updates suchen...", command=self.updates_pruefen)
hi.add_separator()
hi.add_command(label="Einrichtungs-Assistent...", command=self.assistent)
hi.add_separator()
@@ -218,7 +260,9 @@ class App:
# ---------------- Seitenleiste: Abrechnungen und Händler
# Win95: zwei eigene Flächen mit blauer Titelleiste auf dem Desktop
links = tk.Frame(pw, bg=w.FACE if m else w.BG)
# Eigener Pfeil-Zeiger für beide Bereiche: Sonst erben sie den Zeiger der Trennlinie, und der
# bleibt unter Windows hängen (Resize-Zeiger mitten im Inhalt, v. a. maximiert).
links = tk.Frame(pw, bg=w.FACE if m else w.BG, cursor="arrow")
innen = tk.Frame(links, bg=w.FACE if m else w.BG)
innen.pack(fill="both", expand=True, padx=12 if m else 0, pady=(16, 10) if m else 0)
seite = dict(style="Seite.Treeview") if m else {}
@@ -260,7 +304,8 @@ class App:
pw.add(links, minsize=240, width=320)
# ---------------- Inhalt: Kopf der Abrechnung, Kennzahlen, Positionen
rechts = tk.Frame(pw, bg=w.BG)
rechts = tk.Frame(pw, bg=w.BG, cursor="arrow")
self._bereiche = (links, rechts)
innen = tk.Frame(rechts, bg=w.BG)
innen.pack(fill="both", expand=True, padx=(16, 20) if m else 0, pady=(16, 0) if m else 0)
@@ -378,7 +423,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 +450,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 +462,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 +508,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 +566,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 +575,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 +694,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,7 +751,9 @@ class App:
if self._gesperrt():
return
bericht = []
self.root.configure(cursor="watch")
importiert: list[tuple[int, int, list[str]]] = [] # (Zeile im Bericht, datei_id, Hinweise)
for f in (self.root, *self._bereiche):
f.configure(cursor="watch")
self.root.update()
try:
for f in dateien:
@@ -701,15 +774,22 @@ 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="")
for f in self._bereiche:
f.configure(cursor="arrow")
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 +1151,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 +1177,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"
@@ -1173,9 +1257,95 @@ class App:
self.db_wechseln(res["pfad"])
except Exception as exc:
w.fehler(self.root, "Einstellungen", f"Datenbank konnte nicht geöffnet werden:\n{exc}")
self.settings["update_pruefen"] = res["update"]
config.save(self.settings)
if res["design"] != w.DESIGN:
self.design_wechseln(res["design"])
# ================================================================= Updates
@staticmethod
def _installiert() -> bool:
"""Läuft die per Setup installierte Windows-Version? Nur die kann sich selbst aktualisieren."""
return (bool(getattr(sys, "frozen", False)) and sys.platform.startswith("win")
and Path(sys.executable).name.lower() == "provisionsmanager.exe")
def _im_hintergrund(self, arbeit, fertig):
"""arbeit() im Thread (nur Netzwerk/Dateien, kein Tk), danach fertig(ergebnis) im Tk-Thread."""
erg: dict = {}
def lauf():
try:
erg["wert"] = arbeit()
except Exception as exc:
erg["fehler"] = exc
threading.Thread(target=lauf, daemon=True).start()
def warten():
if self._beendet or self._schliesst:
return
if not erg:
self.root.after(200, warten)
else:
fertig(erg)
self.root.after(200, warten)
def updates_pruefen(self, still: bool = False):
"""Auf Gitea nach einer neueren Version suchen. still = automatisch beim Start: nur melden, wenn es
eine gibt (keine Meldung bei Fehler oder wenn aktuell)."""
if getattr(self, "_update_laeuft", False):
return
self._update_laeuft = True
def fertig(erg):
self._update_laeuft = False
self._update_melden(erg, still)
self._im_hintergrund(updater.neueste, fertig)
def _update_melden(self, erg: dict, still: bool):
if "fehler" in erg:
if not still:
w.warnung(self.root, "Updates", f"Konnte nicht nach Updates suchen:\n{erg['fehler']}")
return
info = erg["wert"]
if not updater.ist_neuer(info["version"], __version__):
if not still:
w.info(self.root, "Updates", f"Sie haben die neueste Version ({__version__}).")
return
if still and modal.offen(): # z. B. Tipp des Tages offen – später noch einmal
self.root.after(3000, lambda: self._update_melden(erg, still))
return
text = f"Version {info['version']} ist verfügbar (installiert: {__version__})."
if self._installiert() and info["setup_url"]:
if w.frage(self.root, "Update verfügbar",
text + "\n\nJetzt herunterladen und installieren? Die Datenbank wird vorher gesichert, das "
"Programm beendet sich kurz und startet danach von selbst wieder.",
ja="Jetzt installieren", nein="Später"):
self._update_installieren(info)
elif w.frage(self.root, "Update verfügbar", text + "\n\nDownload-Seite öffnen?",
ja="Seite öffnen", nein="Später"):
webbrowser.open(info["seite"])
def _update_installieren(self, info: dict):
self.status.set("info", f" Lade Version {info['version']} herunter ...")
ordner = Path(tempfile.gettempdir()) / "Provisionsmanager-Update"
def fertig(erg):
if "fehler" in erg:
self.status.set("info", "")
w.fehler(self.root, "Update", f"Download fehlgeschlagen:\n{erg['fehler']}")
return
try:
self.db.backup(config.app_dir() / f"vor-update-{__version__}.db")
except Exception as exc:
if not w.frage(self.root, "Update", f"Die Datenbank konnte nicht gesichert werden:\n{exc}\n\n"
"Trotzdem installieren?", ja="Installieren", nein="Abbrechen", default_ja=False):
return
# NEUSTART=1: das Setup startet das Programm danach wieder (installer.iss)
subprocess.Popen([str(erg["wert"]), "/VERYSILENT", "/SUPPRESSMSGBOXES", "/NORESTART",
"/CLOSEAPPLICATIONS", "/NEUSTART=1"], close_fds=True)
self.beenden()
self._im_hintergrund(lambda: updater.herunterladen(info, ordner), fertig)
def design_umschalten(self):
self.design_wechseln("modern" if w.DUNKEL else "dunkel")
@@ -1346,12 +1516,12 @@ 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:
if self.db is not None:
self.db.close()
finally:
try:
@@ -1376,12 +1546,16 @@ def main():
except Exception:
pass
dateien = [a for a in sys.argv[1:] if Path(a).exists()]
erster = True
while True:
root = tk.Tk(className="Provisionsmanager95")
app = App(root)
if dateien:
root.after(200, lambda: app.importieren(dateien))
dateien = []
if erster and app.settings.get("update_pruefen", True): # nur im echten Start, nicht in Tests
root.after(3000, lambda: app.updates_pruefen(still=True))
erster = False
root.mainloop()
if not app.neustart: # sonst: Design wurde umgeschaltet – Fenster neu aufbauen
break
+15 -2
View File
@@ -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):
try:
self.db.abrechnung_loeschen(aid)
except AbrechnungErledigt as exc:
w.info(self, "Löschen", str(exc))
return
self._laden()
self.result = ("geloescht", aid)
@@ -1026,6 +1035,10 @@ class EinstellungenDialog(w.Dialog):
tk.Radiobutton(g2, text="Klassisch (Windows 95, Legacy)", variable=self.v_design, value="win95").pack(
anchor="w")
tk.Label(g2, text="Das Fenster wird nach OK mit dem neuen Design neu aufgebaut.", fg=w.SHADOW).pack(anchor="w")
g3 = w.gruppe(self, "Updates")
g3.pack(fill="x", padx=10, pady=(0, 10))
self.v_update = tk.BooleanVar(value=settings.get("update_pruefen", True))
tk.Checkbutton(g3, text="Beim Start nach neuen Versionen suchen", variable=self.v_update).pack(anchor="w")
self.knopfleiste([("OK", self._ok), ("Abbrechen", self.abbrechen)]).pack(pady=(0, 10))
def _waehlen(self):
@@ -1036,7 +1049,7 @@ class EinstellungenDialog(w.Dialog):
self.v.set(p)
def _ok(self):
self.result = dict(pfad=self.v.get().strip(), design=self.v_design.get())
self.result = dict(pfad=self.v.get().strip(), design=self.v_design.get(), update=self.v_update.get())
self.destroy()
-5
View File
@@ -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
+1 -58
View File
@@ -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("<Enter>", lambda e: self.configure(bg=LIGHT))
self.bind("<Leave>", lambda e: self.configure(bg=FACE))
self.bind("<ButtonPress-1>", lambda e: self.configure(bg=SEL_BG))
self.bind("<ButtonRelease-1>", self._release)
if tooltip:
Tooltip(self, tooltip)
return
self.bind("<Enter>", lambda e: self.configure(relief="raised"))
self.bind("<Leave>", lambda e: self.configure(relief="flat"))
self.bind("<ButtonPress-1>", lambda e: self.configure(relief="sunken"))
self.bind("<ButtonRelease-1>", 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
+52 -24
View File
@@ -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)}")
+82
View File
@@ -0,0 +1,82 @@
"""Updates: neuestes Release auf Gitea finden und das Windows-Setup herunterladen.
Nur Netzwerk und Dateien – kein Tk (läuft im Hintergrund-Thread). Das Installieren startet die App (gui/app.py).
"""
from __future__ import annotations
import json
import re
import urllib.request
from pathlib import Path
from urllib.parse import urlparse
HOST = "git.handy-insel.de"
API = f"https://{HOST}/api/v1/repos/kerimatilgan/provisionsmanager95/releases/latest"
SETUP = re.compile(r"Provisionsmanager-Setup-[\d.]+\.exe")
MAX_BYTES = 200 * 1024 * 1024
class UpdateFehler(Exception):
pass
def version_tupel(v: str) -> tuple[int, ...]:
"""'v1.10.0' -> (1, 10, 0); Unlesbares -> ()"""
m = re.fullmatch(r"v?(\d+(?:\.\d+)*)", (v or "").strip())
return tuple(int(x) for x in m[1].split(".")) if m else ()
def ist_neuer(neu: str, alt: str) -> bool:
return version_tupel(neu) > version_tupel(alt) and bool(version_tupel(alt))
def _eigene_url(url: str) -> bool:
"""Nur über HTTPS vom eigenen Gitea laden (die Antwort des Servers ist hier die Vertrauensgrenze)."""
u = urlparse(url)
return u.scheme == "https" and u.hostname == HOST
def neueste(timeout: float = 8) -> dict:
"""Neuestes Release: {"version", "seite", "setup_url", "groesse"} (setup_url None, wenn kein Setup dabei)."""
with urllib.request.urlopen(API, timeout=timeout) as r:
daten = json.loads(r.read(1024 * 1024).decode("utf-8"))
setup = next((a for a in daten.get("assets", []) if SETUP.fullmatch(a.get("name", ""))), None)
url = setup.get("browser_download_url") if setup else None
if url and not _eigene_url(url):
url = None
seite = daten.get("html_url") or ""
return {"version": str(daten.get("tag_name", "")).lstrip("v"),
"seite": seite if _eigene_url(seite) else f"https://{HOST}/kerimatilgan/provisionsmanager95/releases",
"setup_url": url, "groesse": int(setup.get("size") or 0) if setup else 0}
def herunterladen(info: dict, ordner: Path, timeout: float = 60) -> Path:
"""Setup nach `ordner` laden und prüfen (Größe wie angekündigt, Windows-Programm). Gibt den Pfad zurück."""
url = info.get("setup_url")
if not url or not _eigene_url(url):
raise UpdateFehler("Für diese Version gibt es kein Setup zum Herunterladen.")
ordner.mkdir(parents=True, exist_ok=True)
ziel = ordner / Path(urlparse(url).path).name
if not SETUP.fullmatch(ziel.name):
raise UpdateFehler(f"Unerwarteter Dateiname: {ziel.name}")
teil = ziel.with_suffix(".teil")
n = 0
try:
with urllib.request.urlopen(url, timeout=timeout) as r, teil.open("wb") as f:
if not _eigene_url(r.geturl()): # Weiterleitung auf einen fremden Server
raise UpdateFehler("Download wurde auf einen fremden Server umgeleitet.")
while block := r.read(1 << 16):
n += len(block)
if n > MAX_BYTES:
raise UpdateFehler("Download ist unerwartet groß.")
f.write(block)
except BaseException:
teil.unlink(missing_ok=True)
raise
with teil.open("rb") as f:
kopf = f.read(2)
if kopf != b"MZ" or (info.get("groesse") and n != info["groesse"]):
teil.unlink(missing_ok=True)
raise UpdateFehler("Download ist unvollständig oder beschädigt – bitte später erneut versuchen.")
teil.replace(ziel)
return ziel
+61
View File
@@ -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,60 @@ 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()
def test_inhalt_erbt_nicht_den_zeiger_der_trennlinie(lauf_modern):
"""Der Resize-Zeiger der Trennlinie blieb unter Windows im ganzen Inhalt hängen (Kinder erben den Zeiger
der Trennlinie). Beide Bereiche haben daher einen eigenen Zeiger."""
app = lauf_modern.app
assert [str(app.root.nametowidget(str(p)).cget("cursor")) for p in app.pw.panes()] == ["arrow", "arrow"]
+248 -2
View File
@@ -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,245 @@ 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"
# --------------------------------------------------------------------------- Updates
class _Antwort:
"""Ersatz für urlopen(): liefert Bytes, kennt seine (End-)URL."""
def __init__(self, daten: bytes, url: str):
import io
self._f, self._url = io.BytesIO(daten), url
def read(self, n=-1):
return self._f.read(n)
def geturl(self):
return self._url
def __enter__(self):
return self
def __exit__(self, *a):
return False
def test_update_version_vergleich():
from provtool import updater
assert updater.ist_neuer("1.10.0", "1.9.2") and updater.ist_neuer("v1.8.0", "1.7.0")
assert not updater.ist_neuer("1.7.0", "1.7.0") and not updater.ist_neuer("1.6.9", "1.7.0")
assert not updater.ist_neuer("kaputt", "1.7.0") and not updater.ist_neuer("1.8.0", "")
def test_update_neueste_nimmt_nur_eigenes_setup(monkeypatch):
import json
from provtool import updater
basis = "https://git.handy-insel.de/kerimatilgan/provisionsmanager95/releases/download/v1.8.0/"
release = {"tag_name": "v1.8.0", "html_url": "https://git.handy-insel.de/x/releases/tag/v1.8.0", "assets": [
{"name": "Provisionsmanager95.exe", "browser_download_url": basis + "Provisionsmanager95.exe", "size": 5},
{"name": "Provisionsmanager-Setup-1.8.0.exe", "browser_download_url": basis + "Provisionsmanager-Setup-1.8.0.exe",
"size": 7}]}
monkeypatch.setattr(updater.urllib.request, "urlopen",
lambda url, timeout=0: _Antwort(json.dumps(release).encode(), url))
info = updater.neueste()
assert info == {"version": "1.8.0", "seite": release["html_url"],
"setup_url": basis + "Provisionsmanager-Setup-1.8.0.exe", "groesse": 7}
release["assets"][1]["browser_download_url"] = "https://boese.example/Provisionsmanager-Setup-1.8.0.exe"
assert updater.neueste()["setup_url"] is None, "Setup von fremdem Server darf nicht geladen werden"
def test_update_download_wird_geprueft(monkeypatch, tmp_path):
from provtool import updater
url = "https://git.handy-insel.de/r/releases/download/v1.8.0/Provisionsmanager-Setup-1.8.0.exe"
inhalt = {"daten": b"MZ" + b"x" * 98, "url": url}
monkeypatch.setattr(updater.urllib.request, "urlopen", lambda u, timeout=0: _Antwort(inhalt["daten"], inhalt["url"]))
setup = updater.herunterladen({"setup_url": url, "groesse": 100}, tmp_path)
assert setup.name == "Provisionsmanager-Setup-1.8.0.exe" and setup.read_bytes().startswith(b"MZ")
for daten, ende, groesse in ((b"MZ" + b"x" * 10, url, 100), # unvollständig
(b"PK" + b"x" * 98, url, 100), # kein Windows-Programm
(b"MZ" + b"x" * 98, "https://boese.example/a.exe", 100)): # umgeleitet
inhalt.update(daten=daten, url=ende)
setup.unlink(missing_ok=True)
with pytest.raises(updater.UpdateFehler):
updater.herunterladen({"setup_url": url, "groesse": groesse}, tmp_path)
assert not list(tmp_path.iterdir()), "halber Download bleibt liegen"
with pytest.raises(updater.UpdateFehler):
updater.herunterladen({"setup_url": "http://git.handy-insel.de/a/Provisionsmanager-Setup-1.8.0.exe"}, tmp_path)