Plugin_SN_Verfahrensgebiet/functions/eigentuemer_suche.py

"""sn_verfahrensgebiet/functions/eigentuemer_suche.py – Eigentümersuche in einer CSV (#56).

Die Eigentümerdaten kommen als CSV-Export (Serienbriefexport, Kopfzeile siehe
:data:`SPALTEN`), den der Anwender von Hand als Layer ins Projekt lädt. Das
Plugin speichert davon nichts, es liest nur den Layer.

Gesucht wird über Nachname und Vorname, ohne Rücksicht auf Groß- und
Kleinschreibung und Umlautschreibweise (``Müller`` findet ``Mueller``). Ein
leeres Feld schränkt nicht ein. Zeilen mit gleichem Namen und gleicher Anschrift
werden zu einem :class:`Eigentuemer` zusammengefasst; dieselbe Person mit einer
anderen Anschrift bleibt ein eigener Eintrag.

**Datenschutz:** Das Modul schreibt keine Personendaten in Ausgaben oder
Meldungen. Es zeigt keine Dialoge; Fehler kommen als
:class:`EigentuemerSucheFehler` mit einer Meldung ohne Personendaten zurück.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple

from sn_basis.functions import QgsProject

#: Spalten des Serienbriefexports, die das Plugin auswertet.
SPALTEN = {
    "konto": "EIG_KONTO",
    "per_id": "PER_ID",
    "verstorben": "PER_VERSTORBEN",
    "anrede": "Anrede",
    "titel": "Titel",
    "vorname": "Vorname",
    "nachname": "Nachname",
    "ortsteil": "OT",
    "strasse": "STRASSE",
    "plz": "PLZ",
    "wohnort": "WOHNORT",
    "email": "Email",
    "rechtsperson": "PER_RECHTSPERSON",
}
PFLICHTSPALTEN = ("Nachname", "Vorname")

_NEIN_WERTE = {"", "0", "n", "nein", "f", "false", "falsch", "null"}
_UMLAUTE = str.maketrans({"ä": "ae", "ö": "oe", "ü": "ue"})


class EigentuemerSucheFehler(Exception):
    """Eingabe- oder Layerfehler; der Text enthält keine Personendaten."""


@dataclass(frozen=True)
class Eigentuemer:
    """Ein Treffer der Eigentümersuche, zusammengefasst über gleiche Zeilen."""

    anrede: str = ""
    titel: str = ""
    vorname: str = ""
    nachname: str = ""
    strasse: str = ""
    plz: str = ""
    wohnort: str = ""
    ortsteil: str = ""
    emails: Tuple[str, ...] = ()
    verstorben: bool = False
    rechtsperson: Tuple[str, ...] = ()
    per_ids: Tuple[str, ...] = ()
    konten: Tuple[str, ...] = ()
    feature_ids: Tuple[int, ...] = ()

    @property
    def name(self) -> str:
        """``Nachname, Titel Vorname`` für Listen."""
        vorne = " ".join(t for t in (self.titel, self.vorname) if t)
        return ", ".join(t for t in (self.nachname, vorne) if t)

    @property
    def anschrift(self) -> str:
        """Einzeilige Anschrift, z. B. ``Hauptstraße 1, 01234 Musterstadt, OT Beispiel``."""
        ort = " ".join(t for t in (self.plz, self.wohnort) if t)
        ortsteil = self.ortsteil
        if ortsteil and not ortsteil.upper().startswith("OT"):
            ortsteil = f"OT {ortsteil}"
        return ", ".join(t for t in (self.strasse, ort, ortsteil) if t)

    @property
    def hinweise(self) -> str:
        """Kennzeichen für die Trefferliste (nicht für das Anzeigefenster)."""
        return "verstorben" if self.verstorben else ""

    @property
    def rechtsperson_text(self) -> str:
        """Rohwert(e) von ``PER_RECHTSPERSON`` (bekannt: 0, 1, 5, 9), noch nicht entschlüsselt."""
        return ", ".join(self.rechtsperson)


def _text(wert) -> str:
    """Attributwert als bereinigter Text; QGIS-NULL und ``None`` werden leer."""
    if wert is None:
        return ""
    ist_null = getattr(wert, "isNull", None)
    if callable(ist_null):
        try:
            if ist_null():
                return ""
        except Exception:
            pass
    text = " ".join(str(wert).split())
    return "" if text == "NULL" else text


def _plz(text: str) -> str:
    """Stellt führende Nullen wieder her.

    QGIS erkennt beim Laden einer CSV Feldtypen und liest ``01067`` dann als
    Ganzzahl ``1067`` – in Sachsen beginnt fast jede PLZ mit 0.
    """
    if text.endswith(".0") and text[:-2].isdigit():
        text = text[:-2]
    return text.zfill(5) if text.isdigit() and len(text) < 5 else text


def _ist_gesetzt(wert) -> bool:
    return _text(wert).casefold() not in _NEIN_WERTE


def _vergleichstext(text: str) -> str:
    """Für die Suche: Groß-/Kleinschreibung, ß und Umlautschreibweise angleichen."""
    return " ".join(text.split()).casefold().translate(_UMLAUTE)


def _spalten_des_layers(layer) -> Dict[str, str]:
    """Ordnet die Spaltennamen des Layers den Schlüsseln aus :data:`SPALTEN` zu.

    Verglichen wird ohne Groß-/Kleinschreibung und ohne Byte-Order-Mark, das
    beim Laden einer UTF-8-CSV an der ersten Spalte hängen bleiben kann.
    """
    vorhanden = {}
    for name in layer.fields().names():
        vorhanden[str(name).replace("", "").strip().casefold()] = name
    return {
        schluessel: vorhanden[spalte.casefold()]
        for schluessel, spalte in SPALTEN.items()
        if spalte.casefold() in vorhanden
    }


def ist_eigentuemerlayer(layer) -> bool:
    """``True``, wenn der Layer die Spalten ``Nachname`` und ``Vorname`` hat."""
    try:
        spalten = _spalten_des_layers(layer)
    except Exception:
        return False
    return "nachname" in spalten and "vorname" in spalten


def eigentuemerlayer_im_projekt() -> List[object]:
    """Alle Layer des Projekts, die als Eigentümerdatei taugen, nach Namen sortiert."""
    layer = [lyr for lyr in QgsProject.instance().mapLayers().values() if ist_eigentuemerlayer(lyr)]
    return sorted(layer, key=lambda lyr: lyr.name().casefold())


def _gruppenschluessel(werte: Dict[str, str]) -> Tuple[str, ...]:
    return tuple(werte[k].casefold() for k in ("nachname", "vorname", "strasse", "plz", "wohnort", "ortsteil"))


def suche_eigentuemer(layer, nachname: str, vorname: str) -> List[Eigentuemer]:
    """Sucht Eigentümer, deren Nach- und Vorname die Eingaben enthalten.

    Raises
    ------
    EigentuemerSucheFehler
        Beide Felder sind leer oder dem Layer fehlen die Pflichtspalten.
    """
    such_nachname = _vergleichstext(nachname or "")
    such_vorname = _vergleichstext(vorname or "")
    if not such_nachname and not such_vorname:
        raise EigentuemerSucheFehler("Bitte einen Namen oder Vornamen eingeben.")

    spalten = _spalten_des_layers(layer)
    if "nachname" not in spalten or "vorname" not in spalten:
        raise EigentuemerSucheFehler(
            f"Der Layer „{layer.name()}“ hat nicht die Spalten "
            f"{' und '.join(PFLICHTSPALTEN)}. Ist es der Serienbriefexport?"
        )

    gruppen: Dict[Tuple[str, ...], dict] = {}
    for feature in layer.getFeatures():
        werte = {k: _text(feature[spalte]) for k, spalte in spalten.items()}
        for k in SPALTEN:
            werte.setdefault(k, "")
        werte["plz"] = _plz(werte["plz"])
        if such_nachname and such_nachname not in _vergleichstext(werte["nachname"]):
            continue
        if such_vorname and such_vorname not in _vergleichstext(werte["vorname"]):
            continue

        gruppe = gruppen.setdefault(_gruppenschluessel(werte), {
            "werte": werte, "emails": [], "per_ids": [], "konten": [], "feature_ids": [],
            "verstorben": False, "rechtsperson": [],
        })
        for k in ("anrede", "titel"):
            if not gruppe["werte"][k] and werte[k]:
                gruppe["werte"][k] = werte[k]
        for liste, wert in (("emails", werte["email"]), ("per_ids", werte["per_id"]),
                            ("konten", werte["konto"]), ("rechtsperson", werte["rechtsperson"])):
            if wert and wert not in gruppe[liste]:
                gruppe[liste].append(wert)
        gruppe["feature_ids"].append(feature.id())
        gruppe["verstorben"] = gruppe["verstorben"] or _ist_gesetzt(werte["verstorben"])

    treffer = []
    for gruppe in gruppen.values():
        w = gruppe["werte"]
        treffer.append(Eigentuemer(
            anrede=w["anrede"], titel=w["titel"], vorname=w["vorname"], nachname=w["nachname"],
            strasse=w["strasse"], plz=w["plz"], wohnort=w["wohnort"], ortsteil=w["ortsteil"],
            emails=tuple(gruppe["emails"]),
            verstorben=gruppe["verstorben"], rechtsperson=tuple(gruppe["rechtsperson"]),
            per_ids=tuple(gruppe["per_ids"]), konten=tuple(gruppe["konten"]),
            feature_ids=tuple(gruppe["feature_ids"]),
        ))
    return sorted(treffer, key=lambda e: (e.nachname.casefold(), e.vorname.casefold(), e.wohnort.casefold()))


def hole_layer(layer_id: Optional[str]):
    """Layer zur ID aus dem Projekt oder ``None``."""
    if not layer_id:
        return None
    return QgsProject.instance().mapLayer(layer_id)