feat: CSV-Format automatisch erkennen

Bisher musste man bei jedem Import Trennzeichen, Datumsformat und
Spaltenzuordnung von Hand eintragen – die haeufigste Fehlerquelle ueberhaupt:
Ein falsches Datumsformat liess alle Zeilen durchfallen, ein falscher
Dezimaltrenner verzehnfachte Betraege.

Zuerst wird gegen die Bank-Vorlagen geprueft (Kopfzeilen-Signatur); alle vier
echten Ausaege des Nutzers werden so sicher erkannt. Sonst inhaltsbasiert:
welche Spalte enthaelt Daten, welche Betraege, welche den laengsten Text.

Zwei Fallen, die dabei aufgefallen sind und getestet werden:
- Deutsche Datumsangaben stecken voller Punkte, und viele Auszuege fuehren
  Buchungs- UND Valutadatum. Ohne sie herauszurechnen kaeme "." als
  Dezimalzeichen heraus und aus 1.204,70 wuerden 1.20.
- parse_date ist absichtlich nachsichtig; beim Erkennen darf es das nicht
  sein, sonst gewinnt ein Format mit Uhrzeit gegen das richtige ohne.

Ganzzahlen ohne Vorzeichen zaehlen nur halb, damit Referenznummern und Kurse
nicht als Betragsspalte durchgehen.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Guido
2026-08-22 21:17:37 +02:00
co-authored by Claude Opus 5
parent c47525ff7e
commit e6801b0474
3 changed files with 282 additions and 0 deletions
+11
View File
@@ -90,6 +90,17 @@ def _transient(user_id: int, **fields) -> ImportTemplate:
return ImportTemplate(owner_id=user_id, name="_transient", **fields) return ImportTemplate(owner_id=user_id, name="_transient", **fields)
@router.post("/detect")
async def detect_format(
file: UploadFile = File(...),
encoding: str = Form("utf-8"),
_: User = Depends(get_current_user),
) -> dict:
"""Format der Datei erkennen, damit niemand mehr Spalten von Hand zuordnet."""
daten = await file.read()
return csv_import.erkenne_format(daten, encoding)
@router.post("/raw") @router.post("/raw")
async def raw_preview( async def raw_preview(
file: UploadFile = File(...), file: UploadFile = File(...),
+189
View File
@@ -39,6 +39,8 @@ from . import currency
PRESETS: list[dict] = [ PRESETS: list[dict] = [
{ {
"name": "Raiffeisen (CH)", "name": "Raiffeisen (CH)",
# Kopfzeilen, an denen sich die Bank sicher erkennen laesst.
"signature": ("iban", "booked at", "credit/debit amount", "balance"),
"delimiter": ";", "delimiter": ";",
"decimal_sep": ".", "decimal_sep": ".",
"date_format": "%Y-%m-%d %H:%M:%S.%f", "date_format": "%Y-%m-%d %H:%M:%S.%f",
@@ -52,6 +54,7 @@ PRESETS: list[dict] = [
}, },
{ {
"name": "N26", "name": "N26",
"signature": ("booking date", "partner name", "payment reference"),
"delimiter": ",", "delimiter": ",",
"decimal_sep": ".", "decimal_sep": ".",
"date_format": "%Y-%m-%d", "date_format": "%Y-%m-%d",
@@ -68,6 +71,9 @@ PRESETS: list[dict] = [
}, },
{ {
"name": "Revolut", "name": "Revolut",
"signature": ("produkt", "beschreibung", "betrag", "kontostand"),
# Der englische Export derselben Bank.
"signature_alt": ("product", "description", "amount", "balance"),
"delimiter": ",", "delimiter": ",",
"decimal_sep": ".", "decimal_sep": ".",
"date_format": "%Y-%m-%d %H:%M:%S", "date_format": "%Y-%m-%d %H:%M:%S",
@@ -84,6 +90,189 @@ PRESETS: list[dict] = [
] ]
# --- Format erkennen -------------------------------------------------------
# Reihenfolge zaehlt: Das erste Format, das die Beispielwerte vollstaendig
# liest, gewinnt. Mit Uhrzeit zuerst, weil ein Datumsformat OHNE Uhrzeit auch
# auf Zeilen MIT Uhrzeit passt (parse_date ist nachsichtig) – umgekehrt nicht.
DATUMSFORMATE = (
"%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d",
"%d.%m.%Y %H:%M:%S", "%d.%m.%Y", "%d/%m/%Y", "%m/%d/%Y",
)
TRENNZEICHEN = (";", ",", "\t", "|")
# Kopfzeilen-Wörter, die auf ein Feld hindeuten. Der Inhalt entscheidet, aber
# bei mehreren gleich guten Spalten gibt die Beschriftung den Ausschlag.
KOPF_HINWEISE = {
"date": ("buchung", "booked", "datum", "date", "valuta"),
"amount": ("betrag", "amount", "umsatz", "credit", "debit"),
"balance": ("saldo", "balance", "kontostand"),
"payee": ("empfaenger", "empfänger", "partner", "beschreibung", "description",
"text", "name", "auftraggeber"),
"purpose": ("zweck", "reference", "verwendung", "mitteilung", "referenz"),
"counterparty_iban": ("iban",),
}
def _spaltenwerte(zeilen: list[list[str]], i: int) -> list[str]:
return [r[i].strip() for r in zeilen if i < len(r) and r[i].strip()]
def _datumsanteil(werte: list[str], fmt: str) -> float:
"""Streng prüfen, nicht mit ``parse_date``.
``parse_date`` ist absichtlich nachsichtig: Es liest „01.02.2026“ auch mit
dem Format „%d.%m.%Y %H:%M:%S“. Beim Erkennen ist das schädlich – dann
gewinnt ein Format mit Uhrzeit gegen das richtige ohne, und in den
Einstellungen steht hinterher etwas Falsches.
"""
if not werte:
return 0.0
treffer = 0
for w in werte:
try:
datetime.strptime(w.strip(), fmt)
treffer += 1
except (ValueError, TypeError):
pass
return treffer / len(werte)
def _geldanteil(werte: list[str], sep: str) -> float:
"""Anteil der Werte, die wie ein Geldbetrag aussehen.
Reine Ganzzahlen zaehlen nur halb: Referenznummern und Kurse sehen sonst
aus wie Betraege, und eine Spalte „Exchange Rate" als Betrag zu nehmen
waere ein stiller, teurer Fehler.
"""
if not werte:
return 0.0
punkte = 0.0
for w in werte:
if parse_amount(w, sep) is None:
continue
punkte += 1.0 if (sep in w or "-" in w) else 0.5
return punkte / len(werte)
def _kopf_passt(kopf: str, feld: str) -> bool:
k = normalize_text(kopf)
return any(h in k for h in KOPF_HINWEISE.get(feld, ()))
def erkenne_format(data: bytes, encoding: str = "utf-8") -> dict:
"""Aus einer CSV-Datei ableiten, wie sie zu lesen ist.
Zuerst gegen die Bank-Vorlagen: Stimmt die Kopfzeile, ist alles bekannt.
Sonst inhaltsbasiert – welche Spalte enthaelt Daten, welche Betraege, welche
den laengsten Text. Das erspart das Ausfuellen des Formulars, das bisher bei
jedem Import noetig war.
Liefert die Einstellung, eine Begruendung je Feld und wie sicher der Fund
ist. Unsicheres wird gemeldet, nicht verschwiegen.
"""
text = decode_bytes(data, encoding)
erste = next((z for z in text.splitlines() if z.strip()), "")
# Trennzeichen: das haeufigste in der Kopfzeile, das ueberhaupt trennt.
trenner = max(TRENNZEICHEN, key=lambda t: erste.count(t))
if erste.count(trenner) == 0:
trenner = ";"
rohzeilen = [r for r in csv.reader(io.StringIO(text), delimiter=trenner)
if any((c or "").strip() for c in r)]
if not rohzeilen:
return {"ok": False, "grund": "Die Datei enthält keine lesbaren Zeilen."}
kopf = [c.strip() for c in rohzeilen[0]]
zeilen = rohzeilen[1:41] # eine Handvoll reicht zum Erkennen
if not zeilen:
return {"ok": False, "grund": "Die Datei enthält nur eine Kopfzeile."}
kopf_klein = [normalize_text(c) for c in kopf]
for p in PRESETS:
for schluessel in ("signature", "signature_alt"):
sig = p.get(schluessel)
if sig and all(any(s in k for k in kopf_klein) for s in sig):
return {
"ok": True,
"preset": p["name"],
"sicherheit": "hoch",
"begruendung": f"Kopfzeile passt zur Vorlage „{p['name']}“.",
"config": {k: v for k, v in p.items()
if k not in ("name", "signature", "signature_alt")},
"header": kopf,
}
# Zuerst die Datumsspalte: Deutsche Datumsangaben („15.03.2026") stecken
# voller Punkte und würden sonst die Erkennung des Dezimaltrenners kippen.
datum_i, datum_fmt, bestes = None, "%d.%m.%Y", 0.0
for i in range(len(kopf)):
werte = _spaltenwerte(zeilen, i)
for fmt in DATUMSFORMATE:
anteil = _datumsanteil(werte, fmt)
if anteil > bestes or (anteil == bestes and anteil > 0.8 and _kopf_passt(kopf[i], "date")):
datum_i, datum_fmt, bestes = i, fmt, anteil
if datum_i is None or bestes < 0.6:
return {"ok": False, "grund": "Keine Spalte mit Datumswerten gefunden.",
"header": kopf}
# ALLE Datumsspalten heraushalten, nicht nur die gewählte: Viele Auszüge
# führen Buchungs- UND Valutadatum, und beide zählen sonst als Punkte mit.
datumsspalten = {
i for i in range(len(kopf))
if any(_datumsanteil(_spaltenwerte(zeilen, i), f) > 0.8 for f in DATUMSFORMATE)
}
ohne_datum = " ".join(
" ".join(c for j, c in enumerate(r) if j not in datumsspalten) for r in zeilen
)
dezimal = "," if ohne_datum.count(",") > ohne_datum.count(".") and trenner != "," else "."
geld = sorted(
((i, _geldanteil(_spaltenwerte(zeilen, i), dezimal)) for i in range(len(kopf))
if i != datum_i),
key=lambda x: (-x[1], not _kopf_passt(kopf[x[0]], "amount")),
)
geldspalten = [i for i, anteil in geld if anteil > 0.6]
if not geldspalten:
return {"ok": False, "grund": "Keine Spalte mit Beträgen gefunden.", "header": kopf}
betrag_i = next((i for i in geldspalten if _kopf_passt(kopf[i], "amount")), geldspalten[0])
saldo_i = next((i for i in geldspalten
if i != betrag_i and _kopf_passt(kopf[i], "balance")), None)
text_spalten = sorted(
((i, sum(len(w) for w in _spaltenwerte(zeilen, i)) / max(1, len(zeilen)))
for i in range(len(kopf)) if i not in (datum_i, betrag_i, saldo_i)),
key=lambda x: -x[1],
)
lange = [i for i, laenge in text_spalten if laenge >= 3]
payee_i = next((i for i in lange if _kopf_passt(kopf[i], "payee")),
lange[0] if lange else None)
purpose_i = next((i for i in lange
if i != payee_i and _kopf_passt(kopf[i], "purpose")), None)
iban_i = next((i for i in range(len(kopf))
if i not in (datum_i, betrag_i) and _kopf_passt(kopf[i], "counterparty_iban")
and any(len(w) > 14 for w in _spaltenwerte(zeilen, i))), None)
spalten = {"date": datum_i, "amount": betrag_i, "payee": payee_i,
"purpose": purpose_i, "balance": saldo_i, "counterparty_iban": iban_i}
begruendung = " · ".join(
f"{feld}: Spalte {i} „{kopf[i]}“" for feld, i in spalten.items() if i is not None
)
return {
"ok": True,
"preset": None,
"sicherheit": "hoch" if bestes > 0.9 and payee_i is not None else "mittel",
"begruendung": begruendung,
"config": {
"delimiter": trenner, "decimal_sep": dezimal, "date_format": datum_fmt,
"encoding": encoding, "has_header": True, "skip_rows": 0,
"amount_mode": "signed", "column_map": spalten,
},
"header": kopf,
}
# --- Kleine Parser-Helfer -------------------------------------------------- # --- Kleine Parser-Helfer --------------------------------------------------
+82
View File
@@ -0,0 +1,82 @@
"""Automatische Formaterkennung für CSV-Auszüge.
Bisher musste man bei jedem Import Trennzeichen, Datumsformat und
Spaltenzuordnung von Hand eintragen. Das ist die häufigste Fehlerquelle
gewesen: Ein falsches Datumsformat liess alle Zeilen durchfallen, ein
falscher Dezimaltrenner verzehnfachte Beträge.
"""
from app.services.csv_import import erkenne_format, parse_amount
def _erkenne(text: str) -> dict:
return erkenne_format(text.encode("utf-8"))
DEUTSCH = (
"Buchungstag;Wert;Auftraggeber/Empfänger;Verwendungszweck;Umsatz;Saldo\n"
"15.03.2026;15.03.2026;REWE Markt GmbH;Einkauf Filiale 42;-52,30;1.204,70\n"
"18.03.2026;18.03.2026;Arbeitgeber AG;Gehalt März;3.000,00;4.204,70\n"
"20.03.2026;20.03.2026;Stadtwerke;Abschlag Strom;-89,00;4.115,70\n"
)
def test_unbekannte_bank_wird_inhaltsbasiert_erkannt():
r = _erkenne(DEUTSCH)
assert r["ok"] is True
c = r["config"]
assert c["delimiter"] == ";"
assert c["date_format"] == "%d.%m.%Y"
assert c["column_map"]["date"] == 0
assert c["column_map"]["amount"] == 4
assert c["column_map"]["payee"] == 2
assert c["column_map"]["purpose"] == 3
assert c["column_map"]["balance"] == 5
def test_datumsspalten_verfaelschen_den_dezimaltrenner_nicht():
"""„15.03.2026“ steckt voller Punkte – und der Auszug führt zwei solche
Spalten. Ohne sie herauszurechnen käme „.“ als Dezimalzeichen heraus und
aus 1.204,70 würden 1.20 CHF."""
c = _erkenne(DEUTSCH)["config"]
assert c["decimal_sep"] == ","
assert parse_amount("1.204,70", c["decimal_sep"]) == 120470
assert parse_amount("-52,30", c["decimal_sep"]) == -5230
def test_datumsformat_ohne_uhrzeit_gewinnt_wenn_keine_da_ist():
"""parse_date ist absichtlich nachsichtig; beim Erkennen darf es das nicht
sein, sonst steht hinterher ein Format mit Uhrzeit in den Einstellungen."""
c = _erkenne("Tag;Betrag\n01.02.2026;-10,00\n02.02.2026;-20,00\n")["config"]
assert c["date_format"] == "%d.%m.%Y"
def test_englisch_mit_tabulator():
r = _erkenne("Date\tDescription\tAmount\tBalance\n"
"2026-03-15\tCOFFEE SHOP\t-4.50\t120.10\n"
"2026-03-16\tSALARY\t2500.00\t2620.10\n")
c = r["config"]
assert c["delimiter"] == "\t"
assert c["decimal_sep"] == "."
assert c["date_format"] == "%Y-%m-%d"
assert c["column_map"]["payee"] == 1
def test_kurs_und_referenzspalten_werden_nicht_fuer_betraege_gehalten():
"""Ganzzahlen ohne Vorzeichen sind verdächtig: Eine Referenznummer als
Betrag zu nehmen wäre ein stiller, teurer Fehler."""
r = _erkenne("Datum;Referenz;Empfänger;Betrag\n"
"01.03.2026;998877;Coop;-12,50\n"
"02.03.2026;998878;Migros;-8,20\n")
assert r["config"]["column_map"]["amount"] == 3
def test_datei_ohne_datumsspalte_wird_ehrlich_abgelehnt():
r = _erkenne("Name;Wert\nAlpha;12,00\nBeta;9,00\n")
assert r["ok"] is False
assert "Datum" in r["grund"]
def test_leere_datei():
assert _erkenne("")["ok"] is False
assert _erkenne("Nur;Kopfzeile\n")["ok"] is False