Merge pull request 'Beteiligung pruefen' (#34) from beteiligung_pruefen into unstable

Reviewed-on: #34

functions/beteiligung_pruefung.py

@@ -0,0 +1,486 @@
+"""sn_verfahrensgebiet/functions/beteiligung_pruefung.py – Beteiligung prüfen.
+
+AGLB-länderspezifisch (Flurstückskennzeichen-Bildung nach sächsischem ALKIS-
+Schema). Die Länderkennung wird nicht hartcodiert, sondern über
+:func:`sn_basis.functions.laenderkennung.get_laenderkennung` ermittelt, damit
+Schwesterplugins für andere Bundesländer dieselbe Logik mit einer anderen
+Länderkennung wiederverwenden können.
+
+Gleicht eine CSV-Exportliste des BI-Servers (Kopfzeile
+``VEF_VKZF,FLA_Beteiligung,FLURSTUECK``) mit dem Verfahrensgebiet-Bestand
+ab. Aus ``FLURSTUECK`` (Format ``<Gemarkung>/<Zähler>/<Nenner>/<Folge>``) wird
+je Zeile ein zu ALKIS kompatibles Flurstückskennzeichen ``Flurstueck_csv``
+gebildet.
+
+Automatische Kaskade je nach Projektzustand:
+
+- **try1**: Ein Layer ``Verfahrensgebiet`` mit befülltem Attribut
+  ``flurstueckskennzeichen`` (Single-Feature-Modell, siehe
+  :mod:`verfahrensgebiet_edit_service`) ist vorhanden → schneller
+  Mengenvergleich ohne WFS-Zugriff.
+- **try2**: Ein Verfahrensgebiet-Layer ist vorhanden, aber ohne verwertbare
+  Kennzeichen → die Pipeline "Verfahrensgebiet aus ALKIS laden"
+  (:func:`verfahrensgebiet_alkis_komplett`) wird silent (ohne
+  Nutzerrückfragen, aber inkl. Detailprüfung) durchlaufen, um einen
+  vollständigen, aktuellen Bestand zu erzeugen – danach wie try1.
+- **try3**: Kein Verfahrensgebiet-Layer im Projekt → Verfahrensnummer aus
+  ``VEF_VKZF`` als Projektvariable setzen, danach wie try2.
+"""
+
+from __future__ import annotations
+
+import csv
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Tuple
+
+from sn_basis.functions import QgsVectorLayer, QgsProject, QgsFeature, QgsField
+from sn_basis.functions.qt_wrapper import QFileDialog, QVariant
+from sn_basis.functions.variable_wrapper import set_variable
+from sn_basis.functions.laenderkennung import get_laenderkennung
+from sn_basis.functions.dialog_wrapper import create_progress_dialog
+from sn_basis.functions.ly_style_wrapper import apply_style
+
+from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import (
+    verfahrensgebiet_alkis_komplett,
+)
+from sn_verfahrensgebiet.functions.flurstueck_alkis import FLST_LAYER_NAME
+from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import (
+    VERF_LAYER_NAME,
+    get_kennzeichen_menge,
+)
+
+CSV_SPALTE_FLURSTUECK = "FLURSTUECK"
+CSV_SPALTE_VEF_VKZF = "VEF_VKZF"
+CSV_SPALTE_BETEILIGUNG = "FLA_Beteiligung"
+
+FELD_FLURSTUECK_CSV = "Flurstueck_csv"
+FELD_FLURSTUECKSKENNZEICHEN_WFS = "flurstueckskennzeichen"
+
+ERGEBNIS_LAYER_VERGLEICH = "Beteiligung – Vergleich"
+
+WERT_BEIDE = "beide"
+WERT_NUR_VERFAHRENSGEBIET = "nur_verfahrensgebiet"
+WERT_NUR_AGLB = "nur_aglb"
+
+
+class BeteiligungPruefungFehler(Exception):
+    """Erwarteter, dem Nutzer per Messagebox mitzuteilender Fehlerfall."""
+
+
+@dataclass
+class BeteiligungErgebnis:
+    """Ergebnis eines Prüflaufs, das die aufrufende UI ins Projekt einfügt."""
+
+    modus: str
+    layers: List[Tuple[object, Optional[str]]] = field(default_factory=list)
+    statistik: Dict[str, int] = field(default_factory=dict)
+    statistik_text: str = ""
+
+
+# ---------------------------------------------------------------------------
+# CSV laden
+# ---------------------------------------------------------------------------
+
+def lade_beteiligung_csv(parent_widget) -> Optional[Tuple[List[dict], List[str]]]:
+    """Öffnet Dateidialog (CSV-Filter) und liest die BI-Server-Exportliste.
+
+    Returns
+    -------
+    Optional[Tuple[List[dict], List[str]]]
+        ``(zeilen, spaltennamen)`` oder ``None`` bei Nutzerabbruch.
+
+    Raises
+    ------
+    BeteiligungPruefungFehler
+        Bei ungültigem CSV-Format oder fehlendem Attribut ``FLURSTUECK``.
+    """
+    pfad, _ = QFileDialog.getOpenFileName(
+        parent_widget,
+        "CSV-Export aus BI-Server auswählen",
+        "",
+        "CSV-Dateien (*.csv)",
+    )
+    if not pfad:
+        return None  # Nutzerabbruch
+
+    try:
+        with open(pfad, "r", encoding="utf-8-sig", newline="") as f:
+            reader = csv.DictReader(f, delimiter=",")
+            zeilen = list(reader)
+            spaltennamen = reader.fieldnames or []
+    except (OSError, csv.Error, UnicodeDecodeError) as exc:
+        raise BeteiligungPruefungFehler(
+            f"CSV-Format ungültig – Datei konnte nicht gelesen werden: {exc}"
+        ) from exc
+
+    if not spaltennamen or not zeilen:
+        raise BeteiligungPruefungFehler(
+            "CSV-Format ungültig – die Datei ist leer oder enthält keine Kopfzeile."
+        )
+
+    if CSV_SPALTE_FLURSTUECK not in spaltennamen:
+        raise BeteiligungPruefungFehler(
+            f"Die CSV enthält nicht das benötigte Attribut '{CSV_SPALTE_FLURSTUECK}'."
+        )
+
+    return zeilen, spaltennamen
+
+
+# ---------------------------------------------------------------------------
+# Flurstueck_csv-Bildung
+# ---------------------------------------------------------------------------
+
+#: Alte CSV-Exporte führen Buchstaben-Nenner (ALKIS: a, b, c, ...) teils als
+#: numerischen Platzhalter 9001, 9002, ... statt als Buchstabe. 9001 → "a",
+#: 9002 → "b" usw. (Basiswert, auf den 1..26 für a..z addiert wird).
+_NENNER_BUCHSTABEN_BASIS = 9000
+
+
+def _baue_nenner_kennzeichen(nenner: str) -> str:
+    """Bildet den 4-stelligen Nenner-Teil des Flurstückskennzeichens.
+
+    Normalfall: numerischer Nenner, 4-stellig mit führenden Nullen
+    aufgefüllt; Nenner 0/leer → Platzhalter ``"____"``.
+
+    Format-Inkonsistenz alter CSV-Exporte: ein Buchstaben-Nenner (z.B. "a")
+    wird dort als ``9001`` (a), ``9002`` (b), usw. geführt. Wird ein solcher
+    Platzhalterwert erkannt, wird stattdessen der Buchstabe selbst
+    (4-stellig mit führenden Nullen aufgefüllt, z.B. ``"000a"``) verwendet,
+    damit das Kennzeichen mit dem ALKIS-Bestand übereinstimmt.
+    """
+    nenner = (nenner or "").strip()
+    nenner_ohne_null = nenner.lstrip("0")
+    if not nenner_ohne_null:
+        return "____"
+
+    if nenner_ohne_null.isdigit():
+        zahl = int(nenner_ohne_null)
+        if _NENNER_BUCHSTABEN_BASIS < zahl <= _NENNER_BUCHSTABEN_BASIS + 26:
+            buchstabe = chr(ord("a") + (zahl - _NENNER_BUCHSTABEN_BASIS - 1))
+            return buchstabe.zfill(4)
+
+    return nenner.zfill(4)
+
+
+def baue_flurstueck_csv(flurstueck_wert: str, laenderkennung: str) -> Optional[str]:
+    """Bildet aus dem CSV-Attribut ``FLURSTUECK`` ein ALKIS-Flurstückskennzeichen.
+
+    Das Attribut hat das Format
+    ``<Gemarkungsnummer>/<Zähler>/<Nenner>/<Folge>``. Die Flurnummer ist in der
+    CSV nicht enthalten und wird als fixer Platzhalter ``___`` geführt (so wie
+    ALKIS unbesetzte Kennzeichen-Bestandteile selbst mit Unterstrichen füllt).
+    #Für Länderanpassung dieses Verhalten ändern, so dass die Flurnummer extrahiert und berücksichtigt wird, wenn angegeben
+
+    Returns
+    -------
+    Optional[str]
+        Das zusammengesetzte Kennzeichen, oder ``None`` bei ungültigem Format
+        (nicht genau 4 durch ``/`` getrennte Teile).
+    """
+    if not flurstueck_wert:
+        return None
+
+    teile = [t.strip() for t in flurstueck_wert.strip().split("/")]
+    if len(teile) != 4:
+        return None
+
+    gemarkung, zaehler, nenner, folge = teile
+
+    gemarkung_kz = gemarkung.zfill(4)
+    flur_kz = "___"
+    zaehler_kz = zaehler.zfill(5)
+    nenner_kz = _baue_nenner_kennzeichen(nenner)
+    folge_kz = folge.zfill(2)
+
+    return f"{laenderkennung}{gemarkung_kz}{flur_kz}{zaehler_kz}{nenner_kz}{folge_kz}"
+
+
+def baue_flurstueck_csv_liste(zeilen: List[dict], laenderkennung: str) -> List[str]:
+    """Bildet für alle CSV-Zeilen ``Flurstueck_csv`` und gibt die Liste zurück.
+
+    Schreibt das Ergebnis zusätzlich als Attribut ``Flurstueck_csv`` in jede
+    Zeile zurück und gibt die ersten 20 Einträge als DEBUG-Ausgabe aus.
+    """
+    ergebnisse: List[str] = []
+    for zeile in zeilen:
+        wert = zeile.get(CSV_SPALTE_FLURSTUECK, "")
+        fkz = baue_flurstueck_csv(wert, laenderkennung)
+        zeile[FELD_FLURSTUECK_CSV] = fkz or ""
+        if fkz:
+            ergebnisse.append(fkz)
+
+    print(f">>> Flurstueck_csv – {len(ergebnisse)} gebildet, erste 20:")
+    for wert in ergebnisse[:20]:
+        print(f">>>   {wert}")
+
+    return ergebnisse
+
+
+# ---------------------------------------------------------------------------
+# Hilfsfunktionen: Layer-Lookup
+# ---------------------------------------------------------------------------
+
+def _hole_verfahrensgebiet_layer(tab_widget) -> Optional[object]:
+    """Liefert den Verfahrensgebiet-Layer, falls vorhanden.
+
+    Der per Name (``VERF_LAYER_NAME``) identifizierte Layer hat Vorrang vor
+    der Dropdown-Auswahl im Tab: try1/try2/try3 orientieren sich am
+    tatsächlichen "Verfahrensgebiet"-Bestand im Projekt, nicht an einer
+    ggf. anderweitig belegten Nutzerauswahl (z.B. für Grenzpunkte/
+    Umringstatistik). Die Dropdown-Auswahl dient nur als Fallback, falls
+    kein so benannter Layer existiert.
+    """
+    treffer = QgsProject.instance().mapLayersByName(VERF_LAYER_NAME)
+    if treffer:
+        return treffer[0]
+
+    combo = getattr(tab_widget, "verf_layer_combo", None)
+    if combo is not None:
+        layer = combo.currentLayer()
+        if layer is not None and layer.isValid():
+            return layer
+
+    return None
+
+
+def _hat_verwendbare_kennzeichen(verf_layer) -> bool:
+    try:
+        return bool(get_kennzeichen_menge(verf_layer))
+    except Exception:
+        return False
+
+
+def _hole_flurstuecke_layer() -> Optional[object]:
+    treffer = QgsProject.instance().mapLayersByName(FLST_LAYER_NAME)
+    return treffer[0] if treffer else None
+
+
+def _flurstueckskennzeichen_geometrien(flst_layer) -> Dict[str, object]:
+    """Baut ``flurstueckskennzeichen`` → Geometrie-Map aus dem Flurstücke-Layer."""
+    mapping: Dict[str, object] = {}
+    if flst_layer is None:
+        return mapping
+    if flst_layer.fields().indexFromName(FELD_FLURSTUECKSKENNZEICHEN_WFS) < 0:
+        return mapping
+    for feat in flst_layer.getFeatures():
+        try:
+            kz = str(feat[FELD_FLURSTUECKSKENNZEICHEN_WFS] or "").strip()
+        except Exception:
+            kz = ""
+        if kz:
+            mapping[kz] = feat.geometry()
+    return mapping
+
+
+# ---------------------------------------------------------------------------
+# try1: schneller Listenabgleich
+# ---------------------------------------------------------------------------
+
+def _try1_listenabgleich(
+    tab_widget, verf_layer, flurstueck_csv_liste: List[str]
+) -> Tuple[object, Dict[str, int]]:
+    progress = create_progress_dialog(
+        3, "Beteiligung prüfen", "Verfahrensgebiet-Kennzeichen einlesen …"
+    )
+    progress.set_value(0)
+    # Ohne eine dazwischenliegende Nutzerinteraktion (z.B. nach der silent
+    # durchlaufenen ALKIS-Pipeline in try2/try3) holt sich kein anderer
+    # Dialog den Fokus zurück – Dialog daher explizit in den Vordergrund holen.
+    progress.raise_to_front()
+
+    verf_menge = get_kennzeichen_menge(verf_layer)
+    csv_menge = set(flurstueck_csv_liste)
+    progress.set_value(1)
+
+    progress.set_label("CSV-Liste abgleichen …")
+    beide = sorted(verf_menge & csv_menge)
+    nur_verfahrensgebiet = sorted(verf_menge - csv_menge)
+    nur_aglb = sorted(csv_menge - verf_menge)
+    progress.set_value(2)
+
+    progress.set_label("Ergebnislayer erstellen …")
+    flst_layer = _hole_flurstuecke_layer()
+    geom_map = _flurstueckskennzeichen_geometrien(flst_layer)
+    if flst_layer is not None and flst_layer.isValid():
+        crs = flst_layer.crs().authid()
+    elif verf_layer.isValid():
+        crs = verf_layer.crs().authid()
+    else:
+        crs = "EPSG:25833"
+
+    layer = QgsVectorLayer(f"Polygon?crs={crs}", ERGEBNIS_LAYER_VERGLEICH, "memory")
+    pr = layer.dataProvider()
+    pr.addAttributes([
+        QgsField(FELD_FLURSTUECK_CSV, QVariant.String),
+        QgsField("status", QVariant.String),
+    ])
+    layer.updateFields()
+    fields = layer.fields()
+
+    neue_features = []
+    for status, menge in (
+        (WERT_BEIDE, beide),
+        (WERT_NUR_VERFAHRENSGEBIET, nur_verfahrensgebiet),
+        (WERT_NUR_AGLB, nur_aglb),
+    ):
+        for kz in menge:
+            feat = QgsFeature(fields)
+            geom = geom_map.get(kz)
+            if geom is not None:
+                feat.setGeometry(geom)
+            feat.setAttributes([kz, status])
+            neue_features.append(feat)
+
+    pr.addFeatures(neue_features)
+    layer.updateExtents()
+
+    progress.set_value(3)
+    progress.close()
+
+    statistik = {
+        WERT_BEIDE: len(beide),
+        WERT_NUR_VERFAHRENSGEBIET: len(nur_verfahrensgebiet),
+        WERT_NUR_AGLB: len(nur_aglb),
+    }
+    return layer, statistik
+
+
+# ---------------------------------------------------------------------------
+# try2/try3: ALKIS-Pipeline (silent) + Listenabgleich wie try1
+# ---------------------------------------------------------------------------
+
+def _lade_via_alkis_pipeline_und_vergleiche(
+    tab_widget, flurstueck_csv_liste: List[str]
+) -> BeteiligungErgebnis:
+    """Lädt das Verfahrensgebiet silent per ALKIS-Pipeline (inkl. Detailprüfung)
+    und gleicht den Bestand danach wie in try1 gegen die CSV-Liste ab.
+
+    Raises
+    ------
+    BeteiligungPruefungFehler
+        Wenn die ALKIS-Pipeline kein Verfahrensgebiet liefert (Pipeline hat
+        bereits eine erklärende Messagebox gezeigt).
+    """
+    enriched_layer, flst_layer, bror_layer, status = verfahrensgebiet_alkis_komplett(
+        tab_widget, silent=True
+    )
+    if enriched_layer is None:
+        raise BeteiligungPruefungFehler(
+            "Verfahrensgebiet konnte über die ALKIS-Pipeline nicht geladen werden."
+        )
+
+    project = QgsProject.instance()
+
+    if flst_layer is not None and flst_layer.isValid():
+        if not project.mapLayersByName(flst_layer.name()):
+            project.addMapLayer(flst_layer)
+            apply_style(flst_layer, "GIS_Flst_Beschriftung_ALKIS_NAS.qml")
+
+    if bror_layer is not None and bror_layer.isValid():
+        if not project.mapLayer(bror_layer.id()):
+            project.addMapLayer(bror_layer)
+        apply_style(bror_layer, "BauRaumOderBodenordnungsrecht.qml")
+
+    if not project.mapLayer(enriched_layer.id()):
+        project.addMapLayer(enriched_layer)
+    apply_style(enriched_layer, "verfahrensgebiet.qml")
+
+    vergleich_layer, statistik = _try1_listenabgleich(
+        tab_widget, enriched_layer, flurstueck_csv_liste
+    )
+
+    layers: List[Tuple[object, Optional[str]]] = [
+        (enriched_layer, None),
+        (vergleich_layer, "beteiligung_vergleich.qml"),
+    ]
+    if flst_layer is not None:
+        layers.append((flst_layer, None))
+    if bror_layer is not None:
+        layers.append((bror_layer, None))
+
+    text = (
+        f"Übereinstimmend: {statistik[WERT_BEIDE]}\n"
+        f"Nur im Verfahrensgebiet: {statistik[WERT_NUR_VERFAHRENSGEBIET]}\n"
+        f"Nur in der CSV-Liste (AGLB): {statistik[WERT_NUR_AGLB]}"
+    )
+    return BeteiligungErgebnis("try2", layers, statistik, text)
+
+
+# ---------------------------------------------------------------------------
+# try3: Bootstrap (Verfahrensnummer aus CSV) + wie try2
+# ---------------------------------------------------------------------------
+
+def _lese_vef_vkzf(zeilen: List[dict], spaltennamen: List[str]) -> str:
+    if CSV_SPALTE_VEF_VKZF not in spaltennamen:
+        raise BeteiligungPruefungFehler(
+            f"Die CSV enthält nicht das benötigte Attribut '{CSV_SPALTE_VEF_VKZF}'."
+        )
+    for zeile in zeilen:
+        wert = (zeile.get(CSV_SPALTE_VEF_VKZF) or "").strip()
+        if wert:
+            return wert
+    raise BeteiligungPruefungFehler(
+        f"Die CSV enthält keinen befüllten Wert für '{CSV_SPALTE_VEF_VKZF}'."
+    )
+
+
+def _try3_bootstrap_und_try2(
+    tab_widget, zeilen: List[dict], spaltennamen: List[str], flurstueck_csv_liste: List[str]
+) -> BeteiligungErgebnis:
+    vef_vkzf = _lese_vef_vkzf(zeilen, spaltennamen)
+    set_variable("verfahrensnummer", vef_vkzf, scope="project")
+
+    ergebnis = _lade_via_alkis_pipeline_und_vergleiche(tab_widget, flurstueck_csv_liste)
+    ergebnis.modus = "try3"
+    ergebnis.statistik_text = f"Verfahrensnummer aus CSV: {vef_vkzf}\n" + ergebnis.statistik_text
+    return ergebnis
+
+
+# ---------------------------------------------------------------------------
+# Öffentliche API
+# ---------------------------------------------------------------------------
+
+def beteiligung_pruefen(tab_widget) -> Optional[BeteiligungErgebnis]:
+    """Vollständiger Ablauf: CSV laden → Flurstueck_csv bilden → try1/try2/try3.
+
+    Returns
+    -------
+    Optional[BeteiligungErgebnis]
+        ``None`` bei Nutzerabbruch des Dateidialogs. Sonst das Ergebnis mit
+        den ins Projekt einzufügenden Layern und der Statistik.
+
+    Raises
+    ------
+    BeteiligungPruefungFehler
+        Bei einem der dokumentierten Fehlerfälle (CSV-Format, fehlende
+        Spalten, WFS-Dienst nicht erreichbar). Der Aufrufer zeigt daraufhin
+        eine Messagebox an.
+    """
+    geladen = lade_beteiligung_csv(tab_widget)
+    if geladen is None:
+        return None
+    zeilen, spaltennamen = geladen
+
+    laenderkennung = get_laenderkennung("sn")
+    flurstueck_csv_liste = baue_flurstueck_csv_liste(zeilen, laenderkennung)
+    if not flurstueck_csv_liste:
+        raise BeteiligungPruefungFehler(
+            "Aus der CSV konnten keine gültigen Flurstück-Kennzeichen gebildet werden "
+            "(Attribut 'FLURSTUECK' muss 4 durch '/' getrennte Teile enthalten)."
+        )
+
+    verf_layer = _hole_verfahrensgebiet_layer(tab_widget)
+
+    if verf_layer is not None and _hat_verwendbare_kennzeichen(verf_layer):
+        layer, statistik = _try1_listenabgleich(tab_widget, verf_layer, flurstueck_csv_liste)
+        text = (
+            f"Übereinstimmend: {statistik[WERT_BEIDE]}\n"
+            f"Nur im Verfahrensgebiet: {statistik[WERT_NUR_VERFAHRENSGEBIET]}\n"
+            f"Nur in der CSV-Liste (AGLB): {statistik[WERT_NUR_AGLB]}"
+        )
+        return BeteiligungErgebnis("try1", [(layer, "beteiligung_vergleich.qml")], statistik, text)
+
+    if verf_layer is not None:
+        return _lade_via_alkis_pipeline_und_vergleiche(tab_widget, flurstueck_csv_liste)
+
+    return _try3_bootstrap_und_try2(tab_widget, zeilen, spaltennamen, flurstueck_csv_liste)

functions/verfahrensgebiet_alkis.py

@@ -72,11 +72,17 @@ def _check_existing_layer(tab_widget):
         return existing_layers[0], LoadStatus.KEEP
 
 
-def verfahrensgebiet_alkis(tab_widget, skip_existing_check: bool = False):
+def verfahrensgebiet_alkis(tab_widget, skip_existing_check: bool = False, silent: bool = False):
     """Lädt Verfahrensgebiet aus ALKIS-WFS basierend auf Verfahrensnummer.
 
     Args:
         tab_widget: Parent-Widget für Dialoge und Meldungen
+        skip_existing_check: Überspringt die Nachfrage bei bereits vorhandenem
+            Layer (immer neu laden).
+        silent: Automatisierter Modus ohne Nutzerinteraktion. Impliziert
+            ``skip_existing_check=True`` und wählt bei mehreren
+            BROR-Kombinationen automatisch die erste (sortierte) statt den
+            Nutzer per Dialog entscheiden zu lassen.
 
     Returns:
         Tuple[Optional[QgsVectorLayer], LoadStatus, List[QgsFeature]]:
@@ -92,6 +98,8 @@ def verfahrensgebiet_alkis(tab_widget, skip_existing_check: bool = False):
                              "In den Projekteigenschaften ist noch keine Verfahrensnummer eingetragen!")
         return None, LoadStatus.NONE, []
 
+    skip_existing_check = skip_existing_check or silent
+
     # Vorhandenen Layer prüfen (überspringbar für die Shape-Pipeline)
     status = LoadStatus.FIRST
     if not skip_existing_check:
@@ -136,6 +144,10 @@ def verfahrensgebiet_alkis(tab_widget, skip_existing_check: bool = False):
     if len(kombinationen) == 1:
         selected_key = next(iter(kombinationen))
         selected_feats = kombinationen[selected_key]
+    elif silent:
+        # Automatisierter Modus: erste (sortierte) Kombination ohne Nachfrage
+        selected_key = sorted(kombinationen.keys())[0]
+        selected_feats = kombinationen[selected_key]
     else:
         # Mehrere Kombinationen → Nutzerauswahl
         key_list = sorted(kombinationen.keys())

functions/verfahrensgebiet_alkis_komplett.py

@@ -254,6 +254,7 @@ def _vergleiche_flaechen(
 
 def verfahrensgebiet_alkis_komplett(
     tab_widget,
+    silent: bool = False,
 ) -> Tuple[Optional[object], Optional[object], Optional[object], LoadStatus]:
     """Vollständiger VG-Lade-Workflow: ALKIS-VG → Gemarkungen →
     Flurstücke → Verschneidung → Flächenvergleich → opt. Umringsprüfung.
@@ -276,6 +277,15 @@ def verfahrensgebiet_alkis_komplett(
     ----------
     tab_widget :
         Eltern-Widget für alle Dialoge.
+    silent :
+        Automatisierter Modus ohne Nutzerentscheidungen: bestehender
+        VG-Layer wird ohne Nachfrage neu geladen, mehrdeutige
+        BROR-Kombinationen werden automatisch aufgelöst, bei
+        Flächenabweichung wird immer die Detailprüfung gestartet (nie „als
+        VG laden“, nie abbrechen), und die beiden rein informativen
+        Abschluss-Meldungen (Flächenvergleich-OK, Detailprüfung
+        abgeschlossen) werden unterdrückt. Die Fortschrittsdialoge bleiben
+        sichtbar.
 
     Returns
     -------
@@ -305,7 +315,7 @@ def verfahrensgebiet_alkis_komplett(
         if progress.is_canceled():
             return None, None, None, LoadStatus.NONE
 
-        alkis_vg_layer, vg_status, _ = verfahrensgebiet_alkis(tab_widget)
+        alkis_vg_layer, vg_status, _ = verfahrensgebiet_alkis(tab_widget, silent=silent)
         progress.raise_to_front()
 
         if vg_status == LoadStatus.NONE or not alkis_vg_layer:
@@ -426,29 +436,34 @@ def verfahrensgebiet_alkis_komplett(
         if stimmt_ueberein:
             progress.set_value(7)
             progress.close()
-            show_info_dialog(
-                "Flächenvergleich",
-                "Verfahrensgebiet mit erwarteter Geometrie erzeugt",
-                parent=tab_widget,
-            )
+            if not silent:
+                show_info_dialog(
+                    "Flächenvergleich",
+                    "Verfahrensgebiet mit erwarteter Geometrie erzeugt",
+                    parent=tab_widget,
+                )
             abgeschlossen = True
             return enriched_layer, flst_layer, alkis_vg_layer, vg_status
 
         # ------------------------------------------------------------------
-        # Flächenabweichung → 3-Optionen-Dialog
+        # Flächenabweichung → 3-Optionen-Dialog (im silent-Modus: immer
+        # Detailprüfung, ohne Nachfrage)
         # ------------------------------------------------------------------
         progress.close()
         diff = abs(bror_flaeche - vg_flaeche)
-        entscheidung = ask_detailpruefung_oder_vg_laden(
-            tab_widget,
-            "Abweichung erkannt",
-            f"Abweichung zwischen BauRaumOderBodenordnungsrecht-Objekt und abgeleitetem "
-            f"Verfahrensgebiet erkannt.\n\n"
-            f"As: {bror_flaeche:.2f} m²\n"
-            f"Verfahrensgebiet:     {vg_flaeche:.2f} m²\n"
-            f"Differenz:            {diff:.2f} m²\n\n"
-            f"Wie möchten Sie fortfahren?",
-        )
+        if silent:
+            entscheidung = "detailpruefung"
+        else:
+            entscheidung = ask_detailpruefung_oder_vg_laden(
+                tab_widget,
+                "Abweichung erkannt",
+                f"Abweichung zwischen BauRaumOderBodenordnungsrecht-Objekt und abgeleitetem "
+                f"Verfahrensgebiet erkannt.\n\n"
+                f"As: {bror_flaeche:.2f} m²\n"
+                f"Verfahrensgebiet:     {vg_flaeche:.2f} m²\n"
+                f"Differenz:            {diff:.2f} m²\n\n"
+                f"Wie möchten Sie fortfahren?",
+            )
 
         if entscheidung == "abbrechen":
             abgeschlossen = True
@@ -481,6 +496,10 @@ def verfahrensgebiet_alkis_komplett(
             "Schritt 1: Grenzpunkte laden …",
         )
         detail_progress.set_value(0)
+        # Ohne die (im silent-Modus übersprungene) Nachfrage vor diesem Punkt
+        # holt sich kein anderer Dialog mehr den Fokus zurück – Dialog daher
+        # explizit in den Vordergrund holen, sonst bleibt er unsichtbar.
+        detail_progress.raise_to_front()
 
         # Schritt 1: PunktortTA gemarkungsweise laden (Hits + 5000er-Batches)
         punktort_layer = lade_punktort_ta(
@@ -550,14 +569,15 @@ def verfahrensgebiet_alkis_komplett(
         if ergebnis_layer and ergebnis_layer.isValid():
             project.addMapLayer(ergebnis_layer)
 
-        show_info_dialog(
-            "Detailprüfung abgeschlossen",
-            f"Knickpunkte geprüft: {n_knick}\n"
-            f"  Identisch (grün):          {identisch}\n"
-            f"  Kein PunktortTA (rot):      {kein_pta}\n"
-            f"  Fehlende Knickpunkte (orange): {fehlender_knick}",
-            parent=tab_widget,
-        )
+        if not silent:
+            show_info_dialog(
+                "Detailprüfung abgeschlossen",
+                f"Knickpunkte geprüft: {n_knick}\n"
+                f"  Identisch (grün):          {identisch}\n"
+                f"  Kein PunktortTA (rot):      {kein_pta}\n"
+                f"  Fehlende Knickpunkte (orange): {fehlender_knick}",
+                parent=tab_widget,
+            )
 
         abgeschlossen = True
         return enriched_layer, flst_layer, alkis_vg_layer, vg_status

styles/beteiligung_erfolgreich.qml

@@ -0,0 +1,22 @@
+<!DOCTYPE qgis PUBLIC 'http://mrcc.com/qgis.dtd' 'SYSTEM'>
+<qgis layerType="Vector" styleCategories="Symbology" version="4.0.1-Norrköping">
+  <renderer-v2 enableorderby="0" forceraster="0" referencescale="-1" symbollevels="0" type="singleSymbol">
+    <symbols>
+      <symbol alpha="1" clip_to_extent="1" force_rhr="0" frame_rate="10" is_animated="0" name="0" type="fill">
+        <layer class="SimpleFill" enabled="1" locked="0" pass="0">
+          <Option type="Map">
+            <Option name="color" type="QString" value="0,180,0,150,rgb:0,0.7058824,0,0.5882353"/>
+            <Option name="outline_color" type="QString" value="0,120,0,255,rgb:0,0.4705882,0,1"/>
+            <Option name="outline_style" type="QString" value="solid"/>
+            <Option name="outline_width" type="QString" value="0.4"/>
+            <Option name="outline_width_unit" type="QString" value="MM"/>
+            <Option name="style" type="QString" value="solid"/>
+          </Option>
+        </layer>
+      </symbol>
+    </symbols>
+  </renderer-v2>
+  <blendMode>0</blendMode>
+  <featureBlendMode>0</featureBlendMode>
+  <layerGeometryType>2</layerGeometryType>
+</qgis>

styles/beteiligung_vergleich.qml

@@ -0,0 +1,51 @@
+<!DOCTYPE qgis PUBLIC 'http://mrcc.com/qgis.dtd' 'SYSTEM'>
+<qgis layerType="Vector" styleCategories="Symbology" version="4.0.1-Norrköping">
+  <renderer-v2 attr="status" enableorderby="0" forceraster="0" referencescale="-1" symbollevels="0" type="categorizedSymbol">
+    <categories>
+      <category label="beide (Übereinstimmung)" render="true" symbol="0" value="beide"/>
+      <category label="nur im Verfahrensgebiet" render="true" symbol="1" value="nur_verfahrensgebiet"/>
+      <category label="nur in der CSV-Liste (AGLB)" render="true" symbol="2" value="nur_aglb"/>
+    </categories>
+    <symbols>
+      <symbol alpha="1" clip_to_extent="1" force_rhr="0" frame_rate="10" is_animated="0" name="0" type="fill">
+        <layer class="SimpleFill" enabled="1" locked="0" pass="0">
+          <Option type="Map">
+            <Option name="color" type="QString" value="0,180,0,150,rgb:0,0.7058824,0,0.5882353"/>
+            <Option name="outline_color" type="QString" value="0,120,0,255,rgb:0,0.4705882,0,1"/>
+            <Option name="outline_style" type="QString" value="solid"/>
+            <Option name="outline_width" type="QString" value="0.4"/>
+            <Option name="outline_width_unit" type="QString" value="MM"/>
+            <Option name="style" type="QString" value="solid"/>
+          </Option>
+        </layer>
+      </symbol>
+      <symbol alpha="1" clip_to_extent="1" force_rhr="0" frame_rate="10" is_animated="0" name="1" type="fill">
+        <layer class="SimpleFill" enabled="1" locked="0" pass="0">
+          <Option type="Map">
+            <Option name="color" type="QString" value="220,0,0,150,rgb:0.8627451,0,0,0.5882353"/>
+            <Option name="outline_color" type="QString" value="150,0,0,255,rgb:0.5882353,0,0,1"/>
+            <Option name="outline_style" type="QString" value="solid"/>
+            <Option name="outline_width" type="QString" value="0.4"/>
+            <Option name="outline_width_unit" type="QString" value="MM"/>
+            <Option name="style" type="QString" value="solid"/>
+          </Option>
+        </layer>
+      </symbol>
+      <symbol alpha="1" clip_to_extent="1" force_rhr="0" frame_rate="10" is_animated="0" name="2" type="fill">
+        <layer class="SimpleFill" enabled="1" locked="0" pass="0">
+          <Option type="Map">
+            <Option name="color" type="QString" value="255,128,0,150,rgb:1,0.5019608,0,0.5882353"/>
+            <Option name="outline_color" type="QString" value="190,96,0,255,rgb:0.745098,0.3764706,0,1"/>
+            <Option name="outline_style" type="QString" value="solid"/>
+            <Option name="outline_width" type="QString" value="0.4"/>
+            <Option name="outline_width_unit" type="QString" value="MM"/>
+            <Option name="style" type="QString" value="solid"/>
+          </Option>
+        </layer>
+      </symbol>
+    </symbols>
+  </renderer-v2>
+  <blendMode>0</blendMode>
+  <featureBlendMode>0</featureBlendMode>
+  <layerGeometryType>2</layerGeometryType>
+</qgis>

tests/test_beteiligung_pruefung.py

@@ -0,0 +1,312 @@
+"""sn_verfahrensgebiet/tests/test_beteiligung_pruefung.py
+
+Unit-Tests für :mod:`sn_verfahrensgebiet.functions.beteiligung_pruefung`.
+
+Getestete Szenarien:
+
+baue_flurstueck_csv:
+- Normalfall mit führenden Nullen bei Zähler/Nenner
+- Nenner = 0 → Platzhalter "____"
+- Nenner leer → Platzhalter "____"
+- Ungültiges Format (nicht 4 Teile) → None
+- Leerer Wert → None
+
+baue_flurstueck_csv_liste:
+- Bildet Liste nur aus gültigen Zeilen, schreibt Flurstueck_csv in jede Zeile zurück
+
+lade_beteiligung_csv:
+- Dateidialog abgebrochen → None
+- Spalte FLURSTUECK fehlt → BeteiligungPruefungFehler
+- Leere Datei → BeteiligungPruefungFehler
+- Erfolgsfall → (zeilen, spaltennamen)
+"""
+
+import csv
+import os
+import sys
+import tempfile
+import types
+import unittest
+from unittest.mock import patch, MagicMock
+
+# ---------------------------------------------------------------------------
+# Minimales QGIS-Mock-Setup
+# ---------------------------------------------------------------------------
+if "qgis" not in sys.modules:
+    qgis_mod = types.ModuleType("qgis")
+    qgis_core = types.ModuleType("qgis.core")
+    qgis_utils = types.ModuleType("qgis.utils")
+    qgis_utils.iface = types.SimpleNamespace()
+    qgis_utils.plugins = {}
+    qgis_mod.utils = qgis_utils
+    qgis_mod.core = qgis_core
+    sys.modules["qgis"] = qgis_mod
+    sys.modules["qgis.core"] = qgis_core
+    sys.modules["qgis.utils"] = qgis_utils
+
+from sn_verfahrensgebiet.functions.beteiligung_pruefung import (
+    baue_flurstueck_csv,
+    baue_flurstueck_csv_liste,
+    lade_beteiligung_csv,
+    BeteiligungPruefungFehler,
+    BeteiligungErgebnis,
+    FELD_FLURSTUECK_CSV,
+    CSV_SPALTE_FLURSTUECK,
+    CSV_SPALTE_VEF_VKZF,
+    _lade_via_alkis_pipeline_und_vergleiche,
+    _try3_bootstrap_und_try2,
+    _hole_verfahrensgebiet_layer,
+)
+
+_MOD = "sn_verfahrensgebiet.functions.beteiligung_pruefung"
+
+
+class TestBaueFlurstueckCsv(unittest.TestCase):
+    def test_normalfall_mit_fuehrenden_nullen(self):
+        ergebnis = baue_flurstueck_csv("1234/123/45/00", "14")
+        self.assertEqual(ergebnis, "141234___00123004500")
+
+    def test_nenner_null_wird_zu_unterstrichen(self):
+        ergebnis = baue_flurstueck_csv("1234/123/0/00", "14")
+        self.assertEqual(ergebnis, "141234___00123____00")
+
+    def test_nenner_leer_wird_zu_unterstrichen(self):
+        ergebnis = baue_flurstueck_csv("1234//45/00", "14")
+        # Gemarkung 1234, Zähler leer -> "00000"? -> zfill auf leeren String ergibt "00000"
+        self.assertTrue(ergebnis.startswith("141234___00000"))
+
+    def test_gemarkung_ohne_fuehrende_nullen_wird_aufgefuellt(self):
+        ergebnis = baue_flurstueck_csv("12/1/1/00", "14")
+        self.assertTrue(ergebnis.startswith("140012___"))
+
+    def test_ungueltiges_format_gibt_none(self):
+        self.assertIsNone(baue_flurstueck_csv("1234/123/45", "14"))
+        self.assertIsNone(baue_flurstueck_csv("1234/123/45/00/99", "14"))
+
+    def test_leerer_wert_gibt_none(self):
+        self.assertIsNone(baue_flurstueck_csv("", "14"))
+        self.assertIsNone(baue_flurstueck_csv(None, "14"))
+
+    def test_nenner_9001_wird_zu_buchstabe_a(self):
+        ergebnis = baue_flurstueck_csv("1234/123/9001/00", "14")
+        self.assertEqual(ergebnis, "141234___00123000a00")
+
+    def test_nenner_9002_wird_zu_buchstabe_b(self):
+        ergebnis = baue_flurstueck_csv("1234/123/9002/00", "14")
+        self.assertEqual(ergebnis, "141234___00123000b00")
+
+    def test_nenner_9026_wird_zu_buchstabe_z(self):
+        ergebnis = baue_flurstueck_csv("1234/123/9026/00", "14")
+        self.assertEqual(ergebnis, "141234___00123000z00")
+
+    def test_nenner_ausserhalb_buchstaben_bereich_bleibt_numerisch(self):
+        ergebnis = baue_flurstueck_csv("1234/123/9027/00", "14")
+        self.assertEqual(ergebnis, "141234___00123902700")
+
+    def test_nenner_9001_mit_fuehrenden_nullen_wird_erkannt(self):
+        ergebnis = baue_flurstueck_csv("1234/123/09001/00", "14")
+        self.assertEqual(ergebnis, "141234___00123000a00")
+
+    def test_einstellige_folge_wird_auf_2_stellen_aufgefuellt(self):
+        ergebnis = baue_flurstueck_csv("1234/123/45/5", "14")
+        self.assertEqual(ergebnis, "141234___00123004505")
+
+
+class TestBaueFlurstueckCsvListe(unittest.TestCase):
+    def test_nur_gueltige_zeilen_in_liste(self):
+        zeilen = [
+            {CSV_SPALTE_FLURSTUECK: "1234/123/45/00"},
+            {CSV_SPALTE_FLURSTUECK: "ungueltig"},
+            {CSV_SPALTE_FLURSTUECK: "5678/1/0/01"},
+        ]
+        ergebnis = baue_flurstueck_csv_liste(zeilen, "14")
+        self.assertEqual(len(ergebnis), 2)
+        self.assertEqual(zeilen[0][FELD_FLURSTUECK_CSV], "141234___00123004500")
+        self.assertEqual(zeilen[1][FELD_FLURSTUECK_CSV], "")
+        self.assertEqual(zeilen[2][FELD_FLURSTUECK_CSV], "145678___00001____01")
+
+
+class TestHoleVerfahrensgebietLayer(unittest.TestCase):
+    @patch(f"{_MOD}.QgsProject")
+    def test_layer_per_name_hat_vorrang_vor_combo_auswahl(self, mock_project_cls):
+        namens_layer = MagicMock(name="verfahrensgebiet_layer")
+        mock_project = MagicMock()
+        mock_project.mapLayersByName.return_value = [namens_layer]
+        mock_project_cls.instance.return_value = mock_project
+
+        andere_auswahl = MagicMock(name="andere_layer_in_combo")
+        andere_auswahl.isValid.return_value = True
+        tab_widget = MagicMock()
+        tab_widget.verf_layer_combo.currentLayer.return_value = andere_auswahl
+
+        ergebnis = _hole_verfahrensgebiet_layer(tab_widget)
+
+        self.assertIs(ergebnis, namens_layer)
+
+    @patch(f"{_MOD}.QgsProject")
+    def test_combo_auswahl_als_fallback_ohne_namens_layer(self, mock_project_cls):
+        mock_project = MagicMock()
+        mock_project.mapLayersByName.return_value = []
+        mock_project_cls.instance.return_value = mock_project
+
+        combo_layer = MagicMock(name="combo_layer")
+        combo_layer.isValid.return_value = True
+        tab_widget = MagicMock()
+        tab_widget.verf_layer_combo.currentLayer.return_value = combo_layer
+
+        ergebnis = _hole_verfahrensgebiet_layer(tab_widget)
+
+        self.assertIs(ergebnis, combo_layer)
+
+    @patch(f"{_MOD}.QgsProject")
+    def test_kein_layer_gibt_none(self, mock_project_cls):
+        mock_project = MagicMock()
+        mock_project.mapLayersByName.return_value = []
+        mock_project_cls.instance.return_value = mock_project
+
+        tab_widget = MagicMock()
+        tab_widget.verf_layer_combo.currentLayer.return_value = None
+
+        self.assertIsNone(_hole_verfahrensgebiet_layer(tab_widget))
+
+
+class TestLadeViaAlkisPipelineUndVergleiche(unittest.TestCase):
+    @patch(f"{_MOD}._try1_listenabgleich")
+    @patch(f"{_MOD}.apply_style")
+    @patch(f"{_MOD}.QgsProject")
+    @patch(f"{_MOD}.verfahrensgebiet_alkis_komplett")
+    def test_pipeline_fehler_wirft_beteiligung_pruefung_fehler(
+        self, mock_pipeline, mock_project_cls, mock_apply_style, mock_try1
+    ):
+        mock_pipeline.return_value = (None, None, None, None)
+        with self.assertRaises(BeteiligungPruefungFehler):
+            _lade_via_alkis_pipeline_und_vergleiche(object(), ["145678___00001____01"])
+        mock_try1.assert_not_called()
+
+    @patch(f"{_MOD}._try1_listenabgleich")
+    @patch(f"{_MOD}.apply_style")
+    @patch(f"{_MOD}.QgsProject")
+    @patch(f"{_MOD}.verfahrensgebiet_alkis_komplett")
+    def test_ruft_pipeline_silent_auf_und_delegiert_an_try1(
+        self, mock_pipeline, mock_project_cls, mock_apply_style, mock_try1
+    ):
+        tab_widget = object()
+        enriched = MagicMock(name="enriched")
+        enriched.isValid.return_value = True
+        enriched.id.return_value = "vg1"
+        flst = MagicMock(name="flst")
+        flst.isValid.return_value = True
+        flst.name.return_value = "Flurstücke"
+        bror = MagicMock(name="bror")
+        bror.isValid.return_value = True
+        bror.id.return_value = "bror1"
+        mock_pipeline.return_value = (enriched, flst, bror, "FIRST")
+
+        mock_project = MagicMock()
+        mock_project.mapLayersByName.return_value = []
+        mock_project.mapLayer.return_value = None
+        mock_project_cls.instance.return_value = mock_project
+
+        vergleich_layer = MagicMock(name="vergleich")
+        statistik = {"beide": 1, "nur_verfahrensgebiet": 2, "nur_aglb": 3}
+        mock_try1.return_value = (vergleich_layer, statistik)
+
+        flurstueck_csv_liste = ["145678___00001____01"]
+        ergebnis = _lade_via_alkis_pipeline_und_vergleiche(tab_widget, flurstueck_csv_liste)
+
+        mock_pipeline.assert_called_once_with(tab_widget, silent=True)
+        mock_try1.assert_called_once_with(tab_widget, enriched, flurstueck_csv_liste)
+
+        self.assertIsInstance(ergebnis, BeteiligungErgebnis)
+        self.assertEqual(ergebnis.modus, "try2")
+        self.assertEqual(ergebnis.statistik, statistik)
+        self.assertIn("Übereinstimmend: 1", ergebnis.statistik_text)
+        layers = [lyr for lyr, _ in ergebnis.layers]
+        self.assertIn(enriched, layers)
+        self.assertIn(vergleich_layer, layers)
+        self.assertIn(flst, layers)
+        self.assertIn(bror, layers)
+
+
+class TestTry3BootstrapUndTry2(unittest.TestCase):
+    @patch(f"{_MOD}._lade_via_alkis_pipeline_und_vergleiche")
+    @patch(f"{_MOD}.set_variable")
+    def test_setzt_verfahrensnummer_und_ergaenzt_statistik_text(
+        self, mock_set_variable, mock_lade
+    ):
+        basis_ergebnis = BeteiligungErgebnis(
+            modus="try2",
+            layers=[],
+            statistik={"beide": 0, "nur_verfahrensgebiet": 0, "nur_aglb": 0},
+            statistik_text="Übereinstimmend: 0\n...",
+        )
+        mock_lade.return_value = basis_ergebnis
+
+        zeilen = [{CSV_SPALTE_VEF_VKZF: "12345", CSV_SPALTE_FLURSTUECK: "1234/1/1/00"}]
+        spaltennamen = [CSV_SPALTE_VEF_VKZF, CSV_SPALTE_FLURSTUECK]
+
+        ergebnis = _try3_bootstrap_und_try2(object(), zeilen, spaltennamen, ["kz1"])
+
+        mock_set_variable.assert_called_once_with("verfahrensnummer", "12345", scope="project")
+        self.assertEqual(ergebnis.modus, "try3")
+        self.assertTrue(ergebnis.statistik_text.startswith("Verfahrensnummer aus CSV: 12345"))
+
+
+class TestLadeBeteiligungCsv(unittest.TestCase):
+    def setUp(self):
+        self._tmp_files = []
+
+    def tearDown(self):
+        for pfad in self._tmp_files:
+            try:
+                os.remove(pfad)
+            except OSError:
+                pass
+
+    def _schreibe_csv(self, header, zeilen):
+        fd, pfad = tempfile.mkstemp(suffix=".csv")
+        self._tmp_files.append(pfad)
+        with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
+            writer = csv.writer(f)
+            writer.writerow(header)
+            for zeile in zeilen:
+                writer.writerow(zeile)
+        return pfad
+
+    @patch(f"{_MOD}.QFileDialog")
+    def test_dialog_abgebrochen_gibt_none(self, mock_fd):
+        mock_fd.getOpenFileName.return_value = ("", "")
+        self.assertIsNone(lade_beteiligung_csv(None))
+
+    @patch(f"{_MOD}.QFileDialog")
+    def test_fehlende_flurstueck_spalte_wirft_fehler(self, mock_fd):
+        pfad = self._schreibe_csv(["FOO", "BAR"], [["1", "2"]])
+        mock_fd.getOpenFileName.return_value = (pfad, "")
+        with self.assertRaises(BeteiligungPruefungFehler) as ctx:
+            lade_beteiligung_csv(None)
+        self.assertIn(CSV_SPALTE_FLURSTUECK, str(ctx.exception))
+
+    @patch(f"{_MOD}.QFileDialog")
+    def test_leere_datei_wirft_fehler(self, mock_fd):
+        fd, pfad = tempfile.mkstemp(suffix=".csv")
+        os.close(fd)
+        self._tmp_files.append(pfad)
+        mock_fd.getOpenFileName.return_value = (pfad, "")
+        with self.assertRaises(BeteiligungPruefungFehler):
+            lade_beteiligung_csv(None)
+
+    @patch(f"{_MOD}.QFileDialog")
+    def test_erfolgsfall_gibt_zeilen_und_spalten_zurueck(self, mock_fd):
+        pfad = self._schreibe_csv(
+            [CSV_SPALTE_VEF_VKZF, "FLA_Beteiligung", CSV_SPALTE_FLURSTUECK],
+            [["12345", "1", "1234/123/45/00"]],
+        )
+        mock_fd.getOpenFileName.return_value = (pfad, "")
+        zeilen, spalten = lade_beteiligung_csv(None)
+        self.assertEqual(len(zeilen), 1)
+        self.assertIn(CSV_SPALTE_FLURSTUECK, spalten)
+        self.assertEqual(zeilen[0][CSV_SPALTE_FLURSTUECK], "1234/123/45/00")
+
+
+if __name__ == "__main__":
+    unittest.main()

ui/tabs/working_tab.py

@@ -23,6 +23,10 @@ from sn_verfahrensgebiet.functions.flurstueck_alkis import (
 )
 from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import verfahrensgebiet_alkis_komplett
 from sn_verfahrensgebiet.functions.verfahrensgebiet_shape_komplett import verfahrensgebiet_shape_komplett
+from sn_verfahrensgebiet.functions.beteiligung_pruefung import (
+    beteiligung_pruefen,
+    BeteiligungPruefungFehler,
+)
 from sn_verfahrensgebiet.functions.grenzpunkt_alkis import grenzpunkte_gemarkungsweise, JOINED_LAYER_NAME
 from sn_verfahrensgebiet.functions.gemarkung_alkis import (
     gemarkung_schluessel_und_namen_von_vg_geometrie,
@@ -72,6 +76,15 @@ class WorkingTab(QWidget):
         row_verf_buttons.addWidget(self.btn_verf_shape)
         layout.addLayout(row_verf_buttons)
 
+        row_beteiligung_buttons = QHBoxLayout()
+        self.btn_beteiligung_pruefen = QPushButton("Beteiligung prüfen")
+        self.btn_beteiligung_pruefen.setToolTip(
+            "CSV-Export aus dem BI-Server verwenden. benötigte Spalten: "
+            "FLA_Verfahren-VEF_VKZF, FLA-Flurstuecke-Flurstueck, FLA_Beteiligung"
+        )
+        row_beteiligung_buttons.addWidget(self.btn_beteiligung_pruefen)
+        layout.addLayout(row_beteiligung_buttons)
+
         # Abschnitt 1b: Verfahrensgebiet-Layer auswählen
         row_layer_label = QHBoxLayout()
         lbl_layer = QLabel("Oder: Verfahrensgebiet-Layer auswählen")
@@ -171,6 +184,7 @@ class WorkingTab(QWidget):
         self.btn_verf_aus_liste.clicked.connect(self.handle_verf_aus_liste)
         self.btn_verf_shape.clicked.connect(self.handle_verf_shape)
         # self.btn_flurst_shape.clicked.connect(self.handle_flurst_shape)
+        self.btn_beteiligung_pruefen.clicked.connect(self.handle_beteiligung_pruefen)
 
         # Beim Init: aktuelle Verfahrensgebiet-Layer-ID laden und im Dropdown anzeigen
         self._restore_verf_layer_selection()
@@ -278,6 +292,40 @@ class WorkingTab(QWidget):
         finally:
             self._set_busy(self.btn_verf_shape, False, "Aus Shape laden")
 
+    def handle_beteiligung_pruefen(self) -> None:
+        """Gleicht eine BI-Server-CSV-Beteiligungsliste mit dem Verfahrensgebiet ab.
+
+        Delegiert an :func:`~sn_verfahrensgebiet.functions.beteiligung_pruefung.beteiligung_pruefen`.
+        Je nach Projektzustand (try1/try2/try3) werden ein oder mehrere
+        Ergebnis-Layer ins Projekt eingefügt, gestylt und eine Statistik als
+        Messagebox angezeigt.
+        """
+        self._set_busy(self.btn_beteiligung_pruefen, True, "Beteiligung prüfen")
+        try:
+            try:
+                ergebnis = beteiligung_pruefen(self)
+            except BeteiligungPruefungFehler as exc:
+                QMessageBox.critical(self, "Beteiligung prüfen", str(exc))
+                return
+
+            if ergebnis is None:
+                return  # Nutzerabbruch des Dateidialogs
+
+            project = QgsProject.instance()
+            for layer, style_name in ergebnis.layers:
+                if layer is None or not layer.isValid():
+                    continue
+                if not project.mapLayer(layer.id()):
+                    project.addMapLayer(layer)
+                if style_name:
+                    apply_style(layer, style_name)
+
+            iface.mapCanvas().refresh()
+            QMessageBox.information(self, "Beteiligung prüfen", ergebnis.statistik_text)
+
+        finally:
+            self._set_busy(self.btn_beteiligung_pruefen, False, "Beteiligung prüfen")
+
     def _set_busy(self, button: QPushButton, busy: bool, original_text: str = "Aus ALKIS laden") -> None:
         """Setzt den Busy-State eines Buttons.