from __future__ import annotations import csv import io import json import re import zipfile from dataclasses import dataclass from html import unescape from typing import Any from xml.etree import ElementTree from pydantic import EmailStr, TypeAdapter, ValidationError from sqlalchemy import and_, func, select from sqlalchemy.orm import Session from app.models.contact import Contact from app.models.customer import Customer, CustomerType from app.models.location import Location from app.models.user import User MAX_IMPORT_BYTES = 5 * 1024 * 1024 SOURCE_SYSTEM = "desk4" EMAIL_ADAPTER = TypeAdapter(EmailStr) TARGET_FIELDS: dict[str, str] = { "ignore": "Nicht importieren", "customer_number": "Kundennummer", "customer_name": "Firmenname", "customer_addition": "Zusatz", "customer_street": "Straße", "customer_house_number": "Hausnummer", "customer_postal_code": "Postleitzahl", "customer_city": "Ort", "customer_country": "Land", "customer_phone": "Telefon", "customer_mobile": "Mobiltelefon", "customer_email": "E-Mail", "customer_website": "Website", "customer_vat_id": "Umsatzsteuer-ID", "customer_notes": "Interne Notiz", "contact_salutation": "Anrede", "contact_first_name": "Vorname", "contact_last_name": "Nachname", "contact_full_name": "Ansprechpartner", "contact_function": "Funktion", "contact_phone": "Telefon Ansprechpartner", "contact_mobile": "Mobil Ansprechpartner", "contact_email": "E-Mail Ansprechpartner", "location_name": "Standortbezeichnung", "location_street": "Straße Standort", "location_house_number": "Hausnummer Standort", "location_postal_code": "PLZ Standort", "location_city": "Ort Standort", "location_country": "Land Standort", } ALIASES: dict[str, list[str]] = { "customer_number": ["kundennr", "kunden nr", "kunden-nr", "kunden nummer", "kundennummer", "debitor", "debitorennummer", "nummer"], "customer_name": ["firma", "firmenname", "name", "kunde", "kundenname", "unternehmen"], "customer_addition": ["zusatz", "name 2", "firmenzusatz"], "customer_street": ["strasse", "straße", "anschrift", "adresse", "kunde strasse"], "customer_house_number": ["hausnummer", "hausnr", "nr"], "customer_postal_code": ["plz", "postleitzahl", "zip"], "customer_city": ["ort", "stadt"], "customer_country": ["land"], "customer_phone": ["telefon", "tel", "festnetz"], "customer_mobile": ["mobiltelefon", "mobil", "handy"], "customer_email": ["e-mail", "email", "mail"], "customer_website": ["web", "website", "homepage"], "customer_vat_id": ["ust id", "ustid", "umsatzsteuer", "vat"], "customer_notes": ["notiz", "bemerkung", "interne notiz"], "contact_salutation": ["anrede"], "contact_first_name": ["vorname"], "contact_last_name": ["nachname", "name ansprechpartner"], "contact_full_name": ["ansprechpartner", "kontakt", "kontaktperson"], "contact_function": ["funktion", "position"], "contact_phone": ["telefon ansprechpartner", "kontakt telefon"], "contact_mobile": ["mobil ansprechpartner", "kontakt mobil"], "contact_email": ["email ansprechpartner", "e-mail ansprechpartner", "kontakt email"], "location_name": ["standort", "standortbezeichnung", "filiale"], "location_street": ["standort strasse", "standort straße", "liefer strasse", "lieferadresse"], "location_house_number": ["standort hausnummer"], "location_postal_code": ["standort plz", "liefer plz"], "location_city": ["standort ort", "liefer ort"], "location_country": ["standort land"], } @dataclass(frozen=True) class ParsedImportFile: headers: list[str] rows: list[dict[str, str]] warnings: list[str] def normalize_header(value: str) -> str: value = value.strip().lower().replace("ß", "ss") value = re.sub(r"[_./:;]+", " ", value) value = re.sub(r"\s+", " ", value) return value def clean(value: Any) -> str: if value is None: return "" return str(value).strip() def join_parts(*values: str) -> str: return " ".join(item for item in (clean(value) for value in values) if item) class CustomerImportService: def __init__(self, session: Session) -> None: self.session = session def parse_file(self, filename: str, content: bytes) -> ParsedImportFile: if not content: raise ValueError("Die Importdatei ist leer.") if len(content) > MAX_IMPORT_BYTES: raise ValueError("Die Importdatei ist zu groß.") suffix = filename.lower().rsplit(".", 1)[-1] if "." in filename else "" if suffix == "csv": return self._parse_csv(content) if suffix == "xlsx": return self._parse_xlsx(content) raise ValueError("Nur CSV- und XLSX-Dateien werden unterstützt.") def suggest_mapping(self, headers: list[str]) -> dict[str, str]: mapping: dict[str, str] = {} used: set[str] = set() for header in headers: normalized = normalize_header(header) target = "ignore" for field, aliases in ALIASES.items(): if field in used: continue if normalized in aliases or any(alias in normalized for alias in aliases): target = field used.add(field) break mapping[header] = target return mapping def preview(self, filename: str, content: bytes, mapping: dict[str, str] | None = None) -> dict: parsed = self.parse_file(filename, content) final_mapping = mapping or self.suggest_mapping(parsed.headers) rows = [self._preview_row(index, row, final_mapping) for index, row in enumerate(parsed.rows, start=2)] summary = self._summary(rows) return { "columns": parsed.headers, "target_fields": TARGET_FIELDS, "mapping": final_mapping, "rows": rows, "summary": summary, "warnings": parsed.warnings, } def confirm( self, filename: str, content: bytes, mapping: dict[str, str], row_actions: dict[str, str], ignore_empty_values: bool, user: User, ) -> dict: preview = self.preview(filename, content, mapping) results = [] counters = { "total": len(preview["rows"]), "created_customers": 0, "updated_customers": 0, "skipped_rows": 0, "error_rows": 0, "created_locations": 0, "created_contacts": 0, } for row in preview["rows"]: action = row_actions.get(str(row["row_number"]), row["action"]) if row["action"] == "ERROR": action = "ERROR" if action not in {"NEU_ANLEGEN", "BESTEHENDEN_AKTUALISIEREN", "UEBERSPRINGEN", "ERROR"}: action = row["action"] try: result = self._apply_row(row, action, ignore_empty_values) for key in counters: counters[key] += result.get(key, 0) results.append( { "row_number": row["row_number"], "customer_name": row["recognized"]["customer"].get("name"), "action": action, "result": result["result"], "message": result["message"], } ) except Exception as exc: self.session.rollback() counters["error_rows"] += 1 results.append( { "row_number": row["row_number"], "customer_name": row["recognized"]["customer"].get("name"), "action": action, "result": "FEHLER", "message": str(exc), } ) self.session.commit() return {"summary": counters, "rows": results, "executed_by_user_id": user.id} def _parse_csv(self, content: bytes) -> ParsedImportFile: text = content.decode("utf-8-sig") sample = text[:4096] delimiter = ";" try: delimiter = csv.Sniffer().sniff(sample, delimiters=";,").delimiter except csv.Error: delimiter = ";" if sample.count(";") >= sample.count(",") else "," reader = csv.DictReader(io.StringIO(text), delimiter=delimiter) headers = [clean(item) for item in (reader.fieldnames or []) if clean(item)] if not headers: raise ValueError("Die Importdatei enthält keine Spaltenüberschriften.") rows = [{header: clean(row.get(header)) for header in headers} for row in reader] rows = [row for row in rows if any(row.values())] if not rows: raise ValueError("Die Importdatei enthält keine Datenzeilen.") return ParsedImportFile(headers=headers, rows=rows, warnings=[]) def _parse_xlsx(self, content: bytes) -> ParsedImportFile: if not content.startswith(b"PK"): raise ValueError("Die XLSX-Datei ist ungültig.") try: archive = zipfile.ZipFile(io.BytesIO(content)) except zipfile.BadZipFile as exc: raise ValueError("Die XLSX-Datei ist ungültig.") from exc names = set(archive.namelist()) if any(name.endswith("vbaProject.bin") for name in names): raise ValueError("Makro-Dateien werden nicht unterstützt.") workbook = ElementTree.fromstring(archive.read("xl/workbook.xml")) ns = {"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main", "r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships"} sheets = workbook.findall(".//m:sheet", ns) warnings = [] if len(sheets) > 1: warnings.append("Die XLSX-Datei enthält mehrere Tabellenblätter. Es wurde das erste Tabellenblatt verwendet.") sheet_path = "xl/worksheets/sheet1.xml" shared = self._xlsx_shared_strings(archive) sheet = ElementTree.fromstring(archive.read(sheet_path)) matrix: list[list[str]] = [] for row in sheet.findall(".//m:sheetData/m:row", ns): values: dict[int, str] = {} for cell in row.findall("m:c", ns): ref = cell.attrib.get("r", "A1") col = self._column_index(ref) value = self._xlsx_cell_value(cell, shared, ns) values[col] = value if values: matrix.append([values.get(index, "") for index in range(max(values) + 1)]) if not matrix: raise ValueError("Die XLSX-Datei enthält keine Daten.") headers = [clean(value) for value in matrix[0]] headers = [value for value in headers if value] if not headers: raise ValueError("Die XLSX-Datei enthält keine Spaltenüberschriften.") rows = [] for raw in matrix[1:]: row = {header: clean(raw[index] if index < len(raw) else "") for index, header in enumerate(headers)} if any(row.values()): rows.append(row) if not rows: raise ValueError("Die XLSX-Datei enthält keine Datenzeilen.") return ParsedImportFile(headers=headers, rows=rows, warnings=warnings) def _xlsx_shared_strings(self, archive: zipfile.ZipFile) -> list[str]: if "xl/sharedStrings.xml" not in archive.namelist(): return [] ns = {"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"} root = ElementTree.fromstring(archive.read("xl/sharedStrings.xml")) values = [] for item in root.findall(".//m:si", ns): values.append(unescape("".join(node.text or "" for node in item.findall(".//m:t", ns)))) return values def _xlsx_cell_value(self, cell: ElementTree.Element, shared: list[str], ns: dict[str, str]) -> str: cell_type = cell.attrib.get("t") value_node = cell.find("m:v", ns) if value_node is None: inline = cell.find(".//m:t", ns) return clean(inline.text if inline is not None else "") value = clean(value_node.text) if cell_type == "s" and value.isdigit() and int(value) < len(shared): return shared[int(value)] return value def _column_index(self, ref: str) -> int: letters = re.sub(r"[^A-Z]", "", ref.upper()) index = 0 for char in letters: index = index * 26 + (ord(char) - 64) return max(index - 1, 0) def _mapped_values(self, row: dict[str, str], mapping: dict[str, str]) -> dict[str, str]: values = {field: "" for field in TARGET_FIELDS if field != "ignore"} for column, target in mapping.items(): if target in values and not values[target]: values[target] = clean(row.get(column)) return values def _preview_row(self, row_number: int, row: dict[str, str], mapping: dict[str, str]) -> dict: values = self._mapped_values(row, mapping) recognized = self._recognized(values) errors = self._validate_values(values) matches = self._find_matches(recognized["customer"]) action = "NEU_ANLEGEN" message = "Neuer Kunde" if errors: action = "ERROR" message = "Pflichtfeld oder Eingabe ungültig" elif matches["external_id"]: action = "BESTEHENDEN_AKTUALISIEREN" message = "Bestehender Kunde über Kundennummer gefunden" elif matches["name_postal_code"] or matches["name_city"] or matches["email"]: action = "UEBERSPRINGEN" message = "Mögliche Dublette gefunden" return { "row_number": row_number, "source": row, "recognized": recognized, "errors": errors, "action": action, "message": message, "matches": {key: self._customer_ref(value) for key, value in matches.items() if value}, "allowed_actions": self._allowed_actions(action, bool(matches["external_id"])), } def _recognized(self, values: dict[str, str]) -> dict: customer_notes = "\n".join( item for item in [ values["customer_notes"], f"Zusatz: {values['customer_addition']}" if values["customer_addition"] else "", f"Land: {values['customer_country']}" if values["customer_country"] else "", f"Mobil: {values['customer_mobile']}" if values["customer_mobile"] else "", f"Website: {values['customer_website']}" if values["customer_website"] else "", f"USt-ID: {values['customer_vat_id']}" if values["customer_vat_id"] else "", ] if item ) first_last = join_parts(values["contact_first_name"], values["contact_last_name"]) full_name = values["contact_full_name"] or first_last return { "customer": { "source_system": SOURCE_SYSTEM if values["customer_number"] else None, "external_id": values["customer_number"] or None, "name": values["customer_name"], "street": join_parts(values["customer_street"], values["customer_house_number"]), "postal_code": values["customer_postal_code"], "city": values["customer_city"], "phone": values["customer_phone"] or values["customer_mobile"], "email": values["customer_email"], "notes": customer_notes, }, "location": { "name": values["location_name"] or "Hauptstandort", "street": join_parts(values["location_street"], values["location_house_number"]) or join_parts(values["customer_street"], values["customer_house_number"]), "postal_code": values["location_postal_code"] or values["customer_postal_code"], "city": values["location_city"] or values["customer_city"], }, "contact": { "full_name": full_name, "function": values["contact_function"], "email": values["contact_email"], "phone": values["contact_phone"] or values["contact_mobile"], "notes": values["contact_salutation"], }, } def _validate_values(self, values: dict[str, str]) -> list[str]: errors = [] if not values["customer_number"] and not values["customer_name"]: errors.append("Kundennummer oder Firmenname fehlt.") for label, value in [("Kunden-E-Mail", values["customer_email"]), ("Ansprechpartner-E-Mail", values["contact_email"])]: if value: try: EMAIL_ADAPTER.validate_python(value) except ValidationError: errors.append(f"{label} ist ungültig.") return errors def _find_matches(self, customer: dict) -> dict[str, Customer | None]: external = None if customer.get("external_id"): external = self.session.scalar( select(Customer).where( Customer.source_system == SOURCE_SYSTEM, Customer.external_id == customer["external_id"], ) ) name = customer.get("name") postal_code = customer.get("postal_code") city = customer.get("city") email = customer.get("email") return { "external_id": external, "name_postal_code": self.session.scalar(select(Customer).where(func.lower(Customer.name) == name.lower(), Customer.postal_code == postal_code)) if name and postal_code else None, "name_city": self.session.scalar(select(Customer).where(func.lower(Customer.name) == name.lower(), func.lower(Customer.city) == city.lower())) if name and city else None, "email": self.session.scalar(select(Customer).where(func.lower(Customer.email) == email.lower())) if email else None, } def _allowed_actions(self, action: str, has_external_match: bool) -> list[str]: if action == "ERROR": return ["ERROR"] if has_external_match: return ["BESTEHENDEN_AKTUALISIEREN", "UEBERSPRINGEN"] return ["NEU_ANLEGEN", "UEBERSPRINGEN"] def _customer_ref(self, customer: Customer) -> dict: return {"id": customer.id, "name": customer.name, "external_id": customer.external_id, "postal_code": customer.postal_code, "city": customer.city} def _apply_row(self, row: dict, action: str, ignore_empty_values: bool) -> dict: if action in {"UEBERSPRINGEN", "ERROR"}: return {"skipped_rows": 1 if action == "UEBERSPRINGEN" else 0, "error_rows": 1 if action == "ERROR" else 0, "result": "ÜBERSPRUNGEN" if action == "UEBERSPRINGEN" else "FEHLER", "message": row["message"]} customer_data = row["recognized"]["customer"] matches = self._find_matches(customer_data) customer = matches["external_id"] created_customer = False if action == "BESTEHENDEN_AKTUALISIEREN": if customer is None: raise ValueError("Aktualisierung ist nur mit eindeutiger Kundennummer möglich.") self._update_customer(customer, customer_data, ignore_empty_values) else: if customer is not None: raise ValueError("Kunde mit dieser Kundennummer existiert bereits.") customer = Customer(customer_type=CustomerType.practice, name=customer_data["name"] or customer_data["external_id"]) self._update_customer(customer, customer_data, False) self.session.add(customer) self.session.flush() created_customer = True created_location = self._ensure_location(customer, row["recognized"]["location"]) created_contact = self._ensure_contact(customer, row["recognized"]["contact"]) self.session.flush() return { "created_customers": 1 if created_customer else 0, "updated_customers": 0 if created_customer else 1, "created_locations": 1 if created_location else 0, "created_contacts": 1 if created_contact else 0, "result": "ERFOLGREICH", "message": "Kunde angelegt" if created_customer else "Kunde aktualisiert", } def _update_customer(self, customer: Customer, data: dict, ignore_empty_values: bool) -> None: for key in ["source_system", "external_id", "name", "street", "postal_code", "city", "phone", "email", "notes"]: value = data.get(key) if ignore_empty_values and value in (None, ""): continue setattr(customer, key, value or None) def _ensure_location(self, customer: Customer, data: dict) -> bool: if not any(data.get(key) for key in ["name", "street", "postal_code", "city"]): return False existing = self.session.scalar( select(Location).where( Location.customer_id == customer.id, func.lower(Location.name) == data["name"].lower(), ) ) if existing: return False self.session.add(Location(customer_id=customer.id, **{key: value or None for key, value in data.items()})) return True def _ensure_contact(self, customer: Customer, data: dict) -> bool: if not data.get("full_name"): return False existing = self.session.scalar( select(Contact).where( Contact.customer_id == customer.id, func.lower(Contact.full_name) == data["full_name"].lower(), ) ) if existing: return False self.session.add(Contact(customer_id=customer.id, **{key: value or None for key, value in data.items()})) return True def _summary(self, rows: list[dict]) -> dict: return { "total": len(rows), "new": sum(1 for row in rows if row["action"] == "NEU_ANLEGEN"), "update": sum(1 for row in rows if row["action"] == "BESTEHENDEN_AKTUALISIEREN"), "duplicates": sum(1 for row in rows if row["action"] == "UEBERSPRINGEN" and row["matches"]), "errors": sum(1 for row in rows if row["action"] == "ERROR"), "skip": sum(1 for row in rows if row["action"] == "UEBERSPRINGEN"), } def parse_json_object(value: str | None, fallback: dict | None = None) -> dict: if not value: return fallback or {} parsed = json.loads(value) if not isinstance(parsed, dict): raise ValueError("JSON-Daten müssen ein Objekt sein.") return parsed