Compare commits
4
Commits
03b48f063c
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2275b92188 | ||
|
|
b85a75857e | ||
|
|
968c68ac1a | ||
|
|
72677ebed7 |
No files matched your search
@@ -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`.
|
||||
|
||||
|
||||
@@ -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,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"
|
||||
+16
-2
@@ -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:
|
||||
|
||||
+70
-26
@@ -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
|
||||
|
||||
+13
-2
@@ -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()):
|
||||
|
||||
+234
-60
@@ -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,13 +1516,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()
|
||||
@@ -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
|
||||
|
||||
+16
-3
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
|
||||
@@ -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
@@ -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
@@ -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)}")
|
||||
|
||||
@@ -122,8 +122,8 @@ def ausfuehren(ergebnisdatei: str) -> int:
|
||||
# 2) Assistent komplett, Import, Export, Rundgang, Beenden
|
||||
csv = basis / "freenet.csv"
|
||||
_beispiel_csv(csv)
|
||||
# Win95-Design: prüft auch den nativen Fensterrahmen (Szenario 1 läuft im modernen Standard)
|
||||
s = _Szenario("Assistent, Import, Export (Win95)", basis / "b", '{"design": "win95"}')
|
||||
# Win95-Design im normalen Fenster: prüft den nativen Rahmen (Szenario 1: modern, startet maximiert)
|
||||
s = _Szenario("Assistent, Import, Export (Win95)", basis / "b", '{"design": "win95", "vollbild": false}')
|
||||
wiz = lambda: s.dialog(Einrichtungsassistent) # noqa: E731
|
||||
seite = lambda n: (lambda: wiz() is not None and wiz().nr == n) # noqa: E731
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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
@@ -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)
|
||||
Reference in new issue
Block a user