Provisions-Manager 95: Import, Händler-DB, Auszahlungsregeln, Excel-Export, Win95-Oberfläche, Wine-Build
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
adminandClaude Opus 5.5 committed
commit
da2e3c9ba1
25 files changed
+3366
No files matched your search
@@ -0,0 +1,304 @@
|
||||
"""Einlesen der Provisionsabrechnungen (freenet CSV/Excel, Vodafone/Euronics Excel).
|
||||
|
||||
Jedes Format wird über seine Kopfzeile erkannt – Dateiname und Blattname sind egal.
|
||||
Alle Formate werden in dieselbe Struktur `Position` überführt.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import hashlib
|
||||
import io
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from datetime import date, datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
from provtool.money import parse_cent
|
||||
|
||||
FREENET = "freenet"
|
||||
VODAFONE = "vodafone"
|
||||
ANBIETER = {FREENET: "freenet", VODAFONE: "Vodafone"}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Position:
|
||||
anbieter: str
|
||||
zeitraum: str = ""
|
||||
provisionsart: str = ""
|
||||
kennung: str = "" # Unterhändler-Nr. (freenet) bzw. VO-Nr. (Vodafone)
|
||||
datum: date | None = None # Aktivierungs-/Anschaltdatum
|
||||
rufnummer: str = ""
|
||||
imei: str = ""
|
||||
geraet: str = ""
|
||||
vorname: str = ""
|
||||
name: str = ""
|
||||
tarif: str = ""
|
||||
betrag_cent: int = 0
|
||||
netz: str = ""
|
||||
dienst: str = ""
|
||||
aktionscode: str = ""
|
||||
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:
|
||||
datei: str
|
||||
sha256: str
|
||||
format: str
|
||||
positionen: list[Position]
|
||||
warnungen: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
class ImportFehler(Exception):
|
||||
pass
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- Hilfen
|
||||
|
||||
def _norm(s) -> str:
|
||||
"""Kopfzeilen-Vergleich: klein, ohne Leer-/Sonderzeichen, Umlaute vereinheitlicht."""
|
||||
s = str(s or "").strip().lower()
|
||||
for a, b in (("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("ß", "ss"), ("�", "")):
|
||||
s = s.replace(a, b)
|
||||
return "".join(ch for ch in s if ch.isalnum())
|
||||
|
||||
|
||||
def _text(v) -> str:
|
||||
if v is None:
|
||||
return ""
|
||||
if isinstance(v, float) and v.is_integer():
|
||||
v = int(v)
|
||||
return str(v).strip()
|
||||
|
||||
|
||||
def _datum(v) -> date | None:
|
||||
if v is None or v == "":
|
||||
return None
|
||||
if isinstance(v, datetime):
|
||||
return v.date()
|
||||
if isinstance(v, date):
|
||||
return v
|
||||
if isinstance(v, (int, float)): # Excel-Seriennummer
|
||||
return (datetime(1899, 12, 30) + timedelta(days=float(v))).date()
|
||||
s = str(v).strip()
|
||||
for f in ("%d.%m.%Y", "%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%d.%m.%y", "%Y.%m.%d", "%d/%m/%Y"):
|
||||
try:
|
||||
return datetime.strptime(s, f).date()
|
||||
except ValueError:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _sha256(path: Path) -> str:
|
||||
h = hashlib.sha256()
|
||||
with open(path, "rb") as f:
|
||||
for chunk in iter(lambda: f.read(65536), b""):
|
||||
h.update(chunk)
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- Formate
|
||||
|
||||
class Format:
|
||||
key = ""
|
||||
anbieter = ""
|
||||
pflicht: tuple[str, ...] = ()
|
||||
|
||||
def passt(self, header_norm: list[str]) -> bool:
|
||||
return all(_norm(p) in header_norm for p in self.pflicht)
|
||||
|
||||
def zeile(self, row: dict, quelle: str) -> Position | None:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class FreenetFormat(Format):
|
||||
key = "freenet"
|
||||
anbieter = FREENET
|
||||
pflicht = ("Zeitraum", "Verguetungsart", "Unterhaendler", "Betrag")
|
||||
|
||||
def zeile(self, r: dict, quelle: str) -> Position | None:
|
||||
g = lambda *k: next((_text(r.get(_norm(x))) for x in k if _text(r.get(_norm(x)))), "")
|
||||
if not g("Verguetungsart") and not g("Betrag"):
|
||||
return None
|
||||
return Position(
|
||||
anbieter=FREENET,
|
||||
zeitraum=g("Zeitraum"),
|
||||
provisionsart=g("Verguetungsart"),
|
||||
kennung=g("Unterhaendler") or g("Haendlernummer"),
|
||||
datum=_datum(r.get(_norm("Aktiv"))),
|
||||
rufnummer=g("Rufnummer"),
|
||||
imei=g("IMEI_SN"),
|
||||
geraet=g("Geraetename"),
|
||||
vorname=g("Kundenvorname"),
|
||||
name=g("Kundename"),
|
||||
tarif=g("Tarif"),
|
||||
betrag_cent=parse_cent(r.get(_norm("Betrag"))),
|
||||
netz=g("Netz"),
|
||||
dienst=g("Dienst"),
|
||||
aktionscode=g("Aktionscode", "Couponcode"),
|
||||
quelle=quelle,
|
||||
)
|
||||
|
||||
|
||||
class VodafoneFormat(Format):
|
||||
"""Euronics-Provisionsclearing (Vodafone/otelo/Cable), Blatt 'All' bzw. Abrechnungslauf-Blätter."""
|
||||
key = "vodafone"
|
||||
anbieter = VODAFONE
|
||||
pflicht = ("Anschaltdatum", "Artikelbezeichnung", "Vo-Nr")
|
||||
|
||||
def zeile(self, r: dict, quelle: str) -> Position | None:
|
||||
g = lambda *k: next((_text(r.get(_norm(x))) for x in k if _text(r.get(_norm(x)))), "")
|
||||
praemie = r.get(_norm("Prämie in EUR"), r.get(_norm("Praemie in EUR")))
|
||||
if not g("Artikelbezeichnung") and praemie in (None, ""):
|
||||
return None
|
||||
art = g("Artikelbezeichnung")
|
||||
netz = "otelo" if "otelo" in art.lower() else ("Cable" if "cable" in art.lower() else "VF")
|
||||
return Position(
|
||||
anbieter=VODAFONE,
|
||||
zeitraum=g("Abrechnungslauf"),
|
||||
provisionsart=art,
|
||||
kennung=g("Vo-Nr"),
|
||||
datum=_datum(r.get(_norm("Anschaltdatum"))),
|
||||
rufnummer=g("Rufnummer"),
|
||||
imei=g("IMEI", "Kartennummer"),
|
||||
geraet="",
|
||||
vorname="",
|
||||
name=g("Kunde"),
|
||||
tarif=g("Tarif"),
|
||||
betrag_cent=parse_cent(praemie),
|
||||
netz=netz,
|
||||
dienst=g("TarifKey02", "Tarif Key02"),
|
||||
aktionscode=g("Text"),
|
||||
quelle=quelle,
|
||||
)
|
||||
|
||||
|
||||
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]):
|
||||
header = [_norm(c) for c in row]
|
||||
for fmt in FORMATE:
|
||||
if fmt.passt(header):
|
||||
pos: list[Position] = []
|
||||
leer = 0
|
||||
for data in rows[idx + 1:]:
|
||||
if not any(_text(c) for c in data):
|
||||
leer += 1
|
||||
if leer > 20:
|
||||
break
|
||||
continue
|
||||
leer = 0
|
||||
rec = {}
|
||||
for h, v in zip(header, data):
|
||||
if h and h not in rec:
|
||||
rec[h] = v
|
||||
p = fmt.zeile(rec, quelle)
|
||||
if p:
|
||||
p.roh = {str(k): (v.isoformat() if isinstance(v, (date, datetime)) else v)
|
||||
for k, v in zip(row, data) if k not in (None, "")}
|
||||
if blatt:
|
||||
p.roh["_blatt"] = blatt
|
||||
pos.append(p)
|
||||
return fmt, pos
|
||||
return None
|
||||
|
||||
|
||||
def _decode(raw: bytes) -> str:
|
||||
for enc in ("utf-8-sig", "cp1252", "latin-1"):
|
||||
try:
|
||||
return raw.decode(enc)
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
return raw.decode("latin-1", errors="replace")
|
||||
|
||||
|
||||
def _lies_csv(path: Path) -> list[tuple[str, list[list]]]:
|
||||
text = _decode(path.read_bytes())
|
||||
sample = text[:5000]
|
||||
try:
|
||||
dialect = csv.Sniffer().sniff(sample, delimiters=";,\t|")
|
||||
delim = dialect.delimiter
|
||||
except csv.Error:
|
||||
delim = ";"
|
||||
rows = list(csv.reader(io.StringIO(text), delimiter=delim))
|
||||
return [("", rows)]
|
||||
|
||||
|
||||
def _lies_xlsx(path: Path) -> list[tuple[str, list[list]]]:
|
||||
import openpyxl
|
||||
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]
|
||||
finally:
|
||||
wb.close()
|
||||
|
||||
|
||||
def _lies_xls(path: Path) -> list[tuple[str, list[list]]]:
|
||||
import xlrd
|
||||
book = xlrd.open_workbook(str(path))
|
||||
out = []
|
||||
for sh in book.sheets():
|
||||
rows = []
|
||||
for r in range(sh.nrows):
|
||||
row = []
|
||||
for c in range(sh.ncols):
|
||||
cell = sh.cell(r, c)
|
||||
if cell.ctype == xlrd.XL_CELL_DATE:
|
||||
row.append(xlrd.xldate.xldate_as_datetime(cell.value, book.datemode))
|
||||
else:
|
||||
row.append(cell.value)
|
||||
rows.append(row)
|
||||
out.append((sh.name.strip(), rows))
|
||||
return out
|
||||
|
||||
|
||||
def lies_datei(path: str | Path) -> ImportErgebnis:
|
||||
path = Path(path)
|
||||
ext = path.suffix.lower()
|
||||
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.")
|
||||
|
||||
gefunden: list[tuple[str, Format, list[Position]]] = []
|
||||
for name, rows in blaetter:
|
||||
res = _tabellen_aus_rows(rows, path.name, name)
|
||||
if res and res[1]:
|
||||
gefunden.append((name, res[0], res[1]))
|
||||
|
||||
if not gefunden:
|
||||
raise ImportFehler(
|
||||
f"In „{path.name}“ wurde keine bekannte Provisionsliste gefunden.\n"
|
||||
"Erwartet wird eine freenet-Abrechnung (Spalten Verguetungsart/Unterhaendler/Betrag) "
|
||||
"oder eine Vodafone/Euronics-Abrechnung (Spalten Anschaltdatum/Artikelbezeichnung/Vo-Nr)."
|
||||
)
|
||||
|
||||
warnungen: list[str] = []
|
||||
# Vodafone/Euronics: Das Blatt "All" enthält alle Abrechnungsläufe. Sind zusätzlich
|
||||
# Einzelblätter befüllt, würden Positionen doppelt gezählt – daher nur "All" verwenden.
|
||||
alle = [g for g in gefunden if g[0].lower() == "all"]
|
||||
if alle and len(gefunden) > 1:
|
||||
uebrig = [g[0] for g in gefunden if g is not alle[0]]
|
||||
warnungen.append(f"Nur Blatt „All“ übernommen, Einzelblätter ignoriert: {', '.join(uebrig)}")
|
||||
gefunden = alle
|
||||
|
||||
positionen = [p for _, _, ps in gefunden for p in ps]
|
||||
formate = sorted({f.key for _, f, _ in gefunden})
|
||||
ohne_kennung = sum(1 for p in positionen if not p.kennung)
|
||||
if ohne_kennung:
|
||||
warnungen.append(f"{ohne_kennung} Position(en) ohne Unterhändler-/VO-Nummer.")
|
||||
return ImportErgebnis(path.name, _sha256(path), "+".join(formate), positionen, warnungen)
|
||||
Reference in new issue
Block a user