308 lines
10 KiB
Python
308 lines
10 KiB
Python
"""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: Maßgeblich ist nur die erste Seite ("All") – sie 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]
|
||
gefunden = [g for g in gefunden if g[0] == erstes]
|
||
if uebrig:
|
||
warnungen.append(f"Nur die erste Seite „{erstes}“ übernommen, ignoriert: {', '.join(uebrig)}")
|
||
if not gefunden:
|
||
raise ImportFehler(f"Auf der ersten Seite von „{path.name}“ wurde keine Vodafone-Abrechnung gefunden.")
|
||
|
||
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)
|