"""Validierung vor dem Export und Kennzahlen für Importprüfung.""" from __future__ import annotations from collections import defaultdict from dataclasses import dataclass, field from app.mapping import FieldKey from app.matching import address_no_key, is_address_complete, plate_key, vin_key from app.models import CustomerRecord from app.normalize import clean_text, is_empty from app.salutation import is_valid_letter_salutation @dataclass class ValidationReport: success: bool message: str stats: dict[str, int] = field(default_factory=dict) issues: list[dict] = field(default_factory=list) summary_rows: list[tuple[str, object]] = field(default_factory=list) detail_rows: list[dict] = field(default_factory=list) def _issue_dict(record: CustomerRecord, reasons: list[str]) -> dict: return { "row": record.row_index, "name": record.display_name(), "address_no": clean_text(record.get(FieldKey.ADDRESS_NO)), "plate": clean_text(record.get(FieldKey.PLATE)), "vin": clean_text(record.get(FieldKey.VIN)), "match_status": record.match_status, "street": clean_text(record.get(FieldKey.STREET)), "zip": clean_text(record.get(FieldKey.ZIP)), "city": clean_text(record.get(FieldKey.CITY)), "reasons": reasons, } def _duplicate_keys(records: list[CustomerRecord], keyfn) -> set[str]: buckets: dict[str, list[CustomerRecord]] = defaultdict(list) for record in records: key = keyfn(record) if key: buckets[key].append(record) return {key for key, group in buckets.items() if len(group) > 1} def validate_import( records: list[CustomerRecord], source_count: int, conflicts: list[CustomerRecord], ) -> ValidationReport: missing_street = 0 missing_zip = 0 missing_city = 0 missing_salutation = 0 missing_plate = 0 missing_vin = 0 addresses_complete = 0 salutations_complete = 0 plates_present = 0 vins_present = 0 issues: list[dict] = [] detail_rows: list[dict] = [] duplicate_plates = _duplicate_keys(records, plate_key) duplicate_vins = _duplicate_keys(records, vin_key) duplicate_address_nos = _duplicate_keys(records, address_no_key) conflict_records = list(conflicts) conflict_ids = {id(r) for r in conflicts} for record in records: reasons: list[str] = list(record.issues) if record.conflict and id(record) not in conflict_ids: conflict_records.append(record) if is_empty(record.get(FieldKey.STREET)): missing_street += 1 reasons.append("Straße fehlt") if is_empty(record.get(FieldKey.ZIP)): missing_zip += 1 reasons.append("PLZ fehlt") if is_empty(record.get(FieldKey.CITY)): missing_city += 1 reasons.append("Ort fehlt") if is_address_complete(record): addresses_complete += 1 if is_valid_letter_salutation(record.get(FieldKey.LETTER_SALUTATION)): salutations_complete += 1 else: missing_salutation += 1 reasons.append("Briefanrede fehlt") if plate_key(record): plates_present += 1 else: missing_plate += 1 reasons.append("Kennzeichen fehlt") if vin_key(record): vins_present += 1 else: missing_vin += 1 reasons.append("Fahrgestellnummer fehlt") plate = plate_key(record) vin = vin_key(record) address_no = address_no_key(record) if plate in duplicate_plates: reasons.append("Doppeltes Kennzeichen") if vin in duplicate_vins: reasons.append("Doppelte Fahrgestellnummer") if address_no and address_no in duplicate_address_nos: reasons.append("Doppelte Adressnummer") detail = _issue_dict(record, reasons) detail_rows.append(detail) if reasons: issues.append(detail) for record in conflicts: if all(d.get("row") == record.row_index and d.get("name") == record.display_name() for d in issues): continue if not any(d.get("row") == record.row_index and d.get("name") == record.display_name() for d in issues): reasons = list(record.issues) or ["Konflikt"] issues.append(_issue_dict(record, reasons)) detail_rows.append(_issue_dict(record, reasons)) error_count = len(issues) conflict_count = len({(c.row_index, c.display_name()) for c in conflict_records}) imported = len(records) missing_addresses = imported - addresses_complete stats = { "customers_source": source_count, "customers_imported": imported, "addresses_complete": addresses_complete, "salutations_complete": salutations_complete, "plates_present": plates_present, "vins_present": vins_present, "missing_street": missing_street, "missing_zip": missing_zip, "missing_city": missing_city, "missing_salutation": missing_salutation, "missing_plate": missing_plate, "missing_vin": missing_vin, "duplicate_plates": len(duplicate_plates), "duplicate_vins": len(duplicate_vins), "duplicate_address_nos": len(duplicate_address_nos), "conflicts": conflict_count, "errors": error_count, } summary_rows = [ ("Kunden Quelle", source_count), ("Kunden importiert", imported), ("Adressen vollständig", addresses_complete), ("Briefanreden vollständig", salutations_complete), ("Kennzeichen vorhanden", plates_present), ("Fahrgestellnummern vorhanden", vins_present), ("Fehlende Straße", missing_street), ("Fehlende PLZ", missing_zip), ("Fehlender Ort", missing_city), ("Doppelte Kennzeichen", len(duplicate_plates)), ("Doppelte Fahrgestellnummern", len(duplicate_vins)), ("Doppelte Adressnummern", len(duplicate_address_nos)), ("Konflikte", conflict_count), ("Fehler", error_count), ] hard_fail = ( missing_addresses > 0 or missing_salutation > 0 or missing_plate > 0 or missing_vin > 0 or conflict_count > 0 or len(duplicate_plates) > 0 or len(duplicate_vins) > 0 or imported == 0 ) if hard_fail: lines = [ "IMPORT FEHLGESCHLAGEN", "", f"{source_count} Kunden erkannt", f"{addresses_complete} Adressen vollständig", ] if missing_addresses: lines.append(f"{missing_addresses} Adresse fehlt" if missing_addresses == 1 else f"{missing_addresses} Adressen fehlen") if missing_plate: lines.append(f"{missing_plate} Kennzeichen fehlen") if missing_vin: lines.append(f"{missing_vin} Fahrgestellnummern fehlen") if missing_salutation: lines.append(f"{missing_salutation} Briefanreden fehlen") if conflict_count: lines.append(f"{conflict_count} Konflikte") if duplicate_plates: lines.append(f"{len(duplicate_plates)} doppelte Kennzeichen") if duplicate_vins: lines.append(f"{len(duplicate_vins)} doppelte Fahrgestellnummern") message = "\n".join(lines) return ValidationReport( success=False, message=message, stats=stats, issues=issues, summary_rows=summary_rows, detail_rows=detail_rows, ) return ValidationReport( success=True, message="Verarbeitung abgeschlossen", stats=stats, issues=[], summary_rows=summary_rows, detail_rows=detail_rows, )