Plugin_SN_Verfahrensgebiet/functions/eigentuemer_korrektur.py

"""sn_verfahrensgebiet/functions/eigentuemer_korrektur.py – Korrekturen beim Datenabgleich (#56).

Der Bearbeiter korrigiert im Plugin Name, Anschrift oder E-Mail eines
angezeigten Eigentümers. Die Eigentümerdatei (Serienbriefexport) bleibt
unverändert; jede Änderung wird als eigene Zeile an eine Korrektur-CSV
angehängt, standardmäßig ``<Eigentümerdatei>_korrekturen.csv`` im selben Ordner.

Spalten der Korrektur-CSV (UTF-8 mit BOM, Komma, wie der Export):
``Zeitpunkt, EIG_KONTO, PER_ID, Feld, Alter Wert, Neuer Wert``. ``Feld`` ist
der Spaltenname aus dem Export. Gehören mehrere Konten oder Personen-IDs zu
einem Eintrag, stehen sie durch Leerzeichen getrennt in einer Zelle.

**Datenschutz:** Die Korrektur-CSV enthält Personendaten. Sie liegt neben
dem Export und ist genauso zu schützen wie dieser. Das Modul schreibt
nichts auf die Konsole und zeigt keine Dialoge.
"""

from __future__ import annotations

import csv
import os
import re
from dataclasses import replace
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from urllib.parse import unquote, urlparse

from sn_verfahrensgebiet.functions.eigentuemer_suche import Eigentuemer, SPALTEN

#: Korrigierbare Felder: (Schlüssel, Beschriftung). Der Spaltenname kommt aus SPALTEN.
FELDER: List[Tuple[str, str]] = [
    ("anrede", "Anrede"),
    ("titel", "Titel"),
    ("vorname", "Vorname"),
    ("nachname", "Name"),
    ("strasse", "Straße"),
    ("plz", "PLZ"),
    ("wohnort", "Wohnort"),
    ("ortsteil", "Ortsteil"),
    ("email", "E-Mail"),
]
KOPFZEILE = ["Zeitpunkt", "EIG_KONTO", "PER_ID", "Feld", "Alter Wert", "Neuer Wert"]
ENDUNG_KORREKTUR = "_korrekturen.csv"

Aenderung = Tuple[str, str, str]  # (Schlüssel, alter Wert, neuer Wert)


def feldwerte(eigentuemer: Eigentuemer) -> Dict[str, str]:
    """Die korrigierbaren Werte eines Eintrags; mehrere E-Mails durch ``, `` getrennt."""
    werte = {schluessel: getattr(eigentuemer, schluessel, "") for schluessel, _ in FELDER if schluessel != "email"}
    werte["email"] = ", ".join(eigentuemer.emails)
    return werte


def ermittle_aenderungen(eigentuemer: Eigentuemer, neue_werte: Dict[str, str]) -> List[Aenderung]:
    """Vergleicht die Eingaben mit dem Eintrag; Leerraum am Rand und doppelte Leerzeichen zählen nicht."""
    alt = feldwerte(eigentuemer)
    aenderungen: List[Aenderung] = []
    for schluessel, _ in FELDER:
        if schluessel not in neue_werte:
            continue
        neu = " ".join(str(neue_werte[schluessel] or "").split())
        if neu != alt[schluessel]:
            aenderungen.append((schluessel, alt[schluessel], neu))
    return aenderungen


def wende_an(eigentuemer: Eigentuemer, aenderungen: List[Aenderung]) -> Eigentuemer:
    """Liefert den Eintrag mit den neuen Werten (für die Anzeige)."""
    felder = {}
    for schluessel, _alt, neu in aenderungen:
        if schluessel == "email":
            felder["emails"] = tuple(t.strip() for t in neu.split(",") if t.strip())
        else:
            felder[schluessel] = neu
    return replace(eigentuemer, **felder)


def korrektur_pfad(layer_quelle: str) -> Optional[str]:
    """Vorgeschlagene Korrektur-CSV neben der Eigentümerdatei, oder ``None``.

    Versteht die Quellangaben des Textdatei-Providers
    (``file:///D:/ordner/export.csv?type=csv&…``) und des OGR-Providers
    (``D:/ordner/export.csv|layername=export``).
    """
    quelle = (layer_quelle or "").strip()
    if not quelle:
        return None
    if quelle.lower().startswith("file:"):
        pfad = unquote(urlparse(quelle).path)
        if re.match(r"^/[A-Za-z]:", pfad):
            pfad = pfad[1:]
    else:
        pfad = quelle.split("|", 1)[0]
    stamm, endung = os.path.splitext(pfad)
    if endung.lower() != ".csv":
        return None
    return stamm + ENDUNG_KORREKTUR


def schreibe_korrekturen(
    pfad: str,
    eigentuemer: Eigentuemer,
    aenderungen: List[Aenderung],
    zeitpunkt: Optional[datetime] = None,
) -> int:
    """Hängt die Änderungen an die Korrektur-CSV an und gibt die Zahl der Zeilen zurück.

    Legt die Datei mit Kopfzeile an, wenn sie fehlt oder leer ist.

    Raises
    ------
    OSError
        Die Datei lässt sich nicht schreiben.
    """
    if not aenderungen:
        return 0
    stempel = (zeitpunkt or datetime.now()).strftime("%Y-%m-%d %H:%M:%S")
    konten = " ".join(eigentuemer.konten)
    per_ids = " ".join(eigentuemer.per_ids)
    neu_anlegen = not os.path.exists(pfad) or os.path.getsize(pfad) == 0
    with open(pfad, "a", encoding="utf-8-sig" if neu_anlegen else "utf-8", newline="") as datei:
        schreiber = csv.writer(datei, delimiter=",", lineterminator="\r\n")
        if neu_anlegen:
            schreiber.writerow(KOPFZEILE)
        for schluessel, alt, neu in aenderungen:
            schreiber.writerow([stempel, konten, per_ids, SPALTEN[schluessel], alt, neu])
    return len(aenderungen)