Beteiligungsprüfung portiert (ohne automatisches Hintergrund-Laden)

Portiert aus origin/unstable (Feature "Beteiligung prüfen": Abgleich einer
BI-Server-CSV-Exportliste mit dem ALKIS-Flurstücksbestand des
Verfahrensgebiets), bewusst OHNE den dortigen silent-Modus für
verfahrensgebiet_alkis()/verfahrensgebiet_alkis_komplett():

- try1 (Verfahrensgebiet-Layer mit Kennzeichen vorhanden) bleibt unverändert
  – schneller Mengenvergleich ohne WFS-Zugriff.
- try2 (Layer vorhanden, aber ohne verwertbare Kennzeichen) und try3 (kein
  Layer vorhanden) laden nicht mehr automatisch/silent über die komplette
  ALKIS-Pipeline, sondern fordern den Nutzer per Meldung auf, das
  Verfahrensgebiet zunächst regulär über "Aus ALKIS laden" zu laden und die
  Beteiligungsprüfung danach erneut zu starten (landet dann automatisch bei
  try1). Grund: ein automatischer Durchlauf ohne Nutzerinteraktion
  (mehrdeutige BROR-Treffer, Flächenabweichungen) hat sich in der Praxis
  nicht als zuverlässig genug erwiesen.
- verfahrensgebiet_alkis.py/verfahrensgebiet_alkis_komplett.py bleiben dadurch
  unangetastet – kein silent-Parameter, kein Konflikt mit dem heutigen
  Zwei-Dialog-Split und dem Landkreis-Fallback.
- Neuer Button "Beteiligung prüfen" in working_tab.py, delegiert an
  beteiligung_pruefen().

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

changelog.txt

@@ -8,6 +8,9 @@
 - Bugfix: Erfolgsmeldung "Layer wurde ... geladen (inkl. Flurstücksverschneidung)" erscheint nicht mehr fälschlich bei "Nur Objekte laden" (da dort keine Verschneidung stattfindet)
 - Verfahrensgebiet-spezifische ALKIS-Dialoge aus sn_basis nach sn_verfahrensgebiet verschoben (functions/verfahrensgebiet_dialoge.py)
 ---
+Version 26.8.1-unstable:
+- Neu: "Beteiligung prüfen" – Abgleich einer BI-Server-CSV-Exportliste mit dem ALKIS-Flurstücksbestand des Verfahrensgebiets (try1: schneller Listenabgleich bei vorhandenem Kennzeichen-Bestand; try2/try3: Aufforderung, das Verfahrensgebiet zunächst über "Aus ALKIS laden" zu laden, statt automatischem Hintergrund-Laden)
+---
 Version 26.7.1-unstable:
 - Verfahrensgebiet als shape laden, inkl. Abgleich mit BROBOR-Objekt
 ---

functions/beteiligung_pruefung.py

@@ -0,0 +1,452 @@
+"""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.
+
+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 → statt einer automatischen (nutzerlosen) ALKIS-Ladepipeline
+  wird der Nutzer per Meldung aufgefordert, das Verfahrensgebiet zunächst
+  regulär über "Aus ALKIS laden" neu zu laden und die Beteiligungsprüfung
+  danach erneut zu starten (landet dann automatisch bei try1). Ein
+  automatischer Durchlauf hat sich in der Praxis nicht als zuverlässig genug
+  erwiesen (mehrdeutige BROR-Treffer, Flächenabweichungen etc. lassen sich
+  nur im interaktiven Ablauf sinnvoll auflösen).
+- **try3**: Kein Verfahrensgebiet-Layer im Projekt → Verfahrensnummer aus
+  ``VEF_VKZF`` wird als Projektvariable übernommen, danach wie try2 (Meldung
+  statt automatischem Laden).
+"""
+
+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_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)
+    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: Aufforderung statt automatischer ALKIS-Pipeline
+# ---------------------------------------------------------------------------
+#
+# Ein automatischer (nutzerloser) Durchlauf der kompletten Ladepipeline
+# (mehrdeutige BROR-Treffer, Flächenabweichungen, Detailprüfung) hat sich in
+# der Praxis nicht als zuverlässig genug erwiesen – deshalb hier bewusst
+# KEIN silent-Aufruf von verfahrensgebiet_alkis_komplett(), sondern eine
+# Aufforderung an den Nutzer, den regulären interaktiven Ablauf
+# ("Aus ALKIS laden") selbst zu starten. Ein erneuter Klick auf
+# "Beteiligung prüfen" landet danach automatisch bei try1, sobald der Layer
+# verwertbare Kennzeichen hat.
+
+def _fordere_manuelles_laden_an(vef_vkzf: Optional[str] = None) -> None:
+    """Bricht try2/try3 mit einer Aufforderung an den Nutzer ab (statt eines
+    automatischen Ladevorgangs)."""
+    hinweis = (
+        "Bitte laden Sie das Verfahrensgebiet jetzt über 'Aus ALKIS laden' "
+        "und starten Sie die Beteiligungsprüfung danach erneut."
+    )
+    if vef_vkzf:
+        raise BeteiligungPruefungFehler(
+            f"Es ist noch kein Verfahrensgebiet geladen. Die Verfahrensnummer "
+            f"{vef_vkzf} wurde aus der CSV übernommen. {hinweis}"
+        )
+    raise BeteiligungPruefungFehler(
+        f"Der vorhandene Verfahrensgebiet-Layer enthält keine verwertbaren "
+        f"Flurstückskennzeichen. {hinweis}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# try3: Bootstrap (Verfahrensnummer aus CSV) + Aufforderung 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(tab_widget, zeilen: List[dict], spaltennamen: List[str]) -> None:
+    vef_vkzf = _lese_vef_vkzf(zeilen, spaltennamen)
+    set_variable("verfahrensnummer", vef_vkzf, scope="project")
+    _fordere_manuelles_laden_an(vef_vkzf)
+
+
+# ---------------------------------------------------------------------------
+# Ö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) oder wenn kein Verfahrensgebiet mit verwertbaren
+        Kennzeichen vorliegt (try2/try3 – der Nutzer wird dann aufgefordert,
+        das Verfahrensgebiet zunächst über "Aus ALKIS laden" zu laden und
+        die Beteiligungsprüfung danach erneut zu starten). 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:
+        _fordere_manuelles_laden_an()
+        return None  # unreachable – _fordere_manuelles_laden_an() wirft immer
+
+    _try3_bootstrap(tab_widget, zeilen, spaltennamen)
+    return None  # unreachable – _try3_bootstrap() wirft immer

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,304 @@
+"""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)
+
+try2/try3 (kein automatisches Laden mehr, siehe Modul-Docstring):
+- Verfahrensgebiet-Layer ohne Kennzeichen vorhanden → Aufforderung statt Auto-Laden
+- Kein Verfahrensgebiet-Layer vorhanden → Verfahrensnummer aus CSV wird gesetzt,
+  dann Aufforderung statt Auto-Laden
+"""
+
+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,
+    beteiligung_pruefen,
+    BeteiligungPruefungFehler,
+    FELD_FLURSTUECK_CSV,
+    CSV_SPALTE_FLURSTUECK,
+    CSV_SPALTE_VEF_VKZF,
+    _fordere_manuelles_laden_an,
+    _try3_bootstrap,
+    _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 TestFordereManuellesLadenAn(unittest.TestCase):
+    """try2/try3 laden nicht mehr automatisch, sondern fordern den Nutzer auf,
+    den regulären interaktiven Ablauf ('Aus ALKIS laden') selbst zu starten."""
+
+    def test_ohne_vef_vkzf_nennt_fehlende_kennzeichen(self):
+        with self.assertRaises(BeteiligungPruefungFehler) as ctx:
+            _fordere_manuelles_laden_an()
+        self.assertIn("keine verwertbaren", str(ctx.exception))
+        self.assertIn("Aus ALKIS laden", str(ctx.exception))
+
+    def test_mit_vef_vkzf_nennt_verfahrensnummer(self):
+        with self.assertRaises(BeteiligungPruefungFehler) as ctx:
+            _fordere_manuelles_laden_an("12345")
+        self.assertIn("12345", str(ctx.exception))
+        self.assertIn("Aus ALKIS laden", str(ctx.exception))
+
+
+class TestTry3Bootstrap(unittest.TestCase):
+    @patch(f"{_MOD}.set_variable")
+    def test_setzt_verfahrensnummer_und_fordert_manuelles_laden_an(self, mock_set_variable):
+        zeilen = [{CSV_SPALTE_VEF_VKZF: "12345", CSV_SPALTE_FLURSTUECK: "1234/1/1/00"}]
+        spaltennamen = [CSV_SPALTE_VEF_VKZF, CSV_SPALTE_FLURSTUECK]
+
+        with self.assertRaises(BeteiligungPruefungFehler) as ctx:
+            _try3_bootstrap(object(), zeilen, spaltennamen)
+
+        mock_set_variable.assert_called_once_with("verfahrensnummer", "12345", scope="project")
+        self.assertIn("12345", str(ctx.exception))
+
+
+class TestBeteiligungPruefenDispatch(unittest.TestCase):
+    """Dispatch-Logik in beteiligung_pruefen(): try1 unverändert, try2/try3
+    fordern jetzt auf statt automatisch zu laden."""
+
+    @patch(f"{_MOD}._hat_verwendbare_kennzeichen", return_value=False)
+    @patch(f"{_MOD}._hole_verfahrensgebiet_layer")
+    @patch(f"{_MOD}.baue_flurstueck_csv_liste")
+    @patch(f"{_MOD}.get_laenderkennung", return_value="14")
+    @patch(f"{_MOD}.lade_beteiligung_csv")
+    def test_layer_ohne_kennzeichen_fordert_manuelles_laden_an(
+        self, mock_lade_csv, mock_laenderkennung, mock_liste, mock_hole_layer, mock_hat_kz
+    ):
+        mock_lade_csv.return_value = ([{}], ["FLURSTUECK"])
+        mock_liste.return_value = ["145678___00001____01"]
+        mock_hole_layer.return_value = MagicMock(name="verf_layer_ohne_kz")
+
+        with self.assertRaises(BeteiligungPruefungFehler) as ctx:
+            beteiligung_pruefen(object())
+        self.assertIn("keine verwertbaren", str(ctx.exception))
+
+    @patch(f"{_MOD}.set_variable")
+    @patch(f"{_MOD}._hole_verfahrensgebiet_layer", return_value=None)
+    @patch(f"{_MOD}.baue_flurstueck_csv_liste")
+    @patch(f"{_MOD}.get_laenderkennung", return_value="14")
+    @patch(f"{_MOD}.lade_beteiligung_csv")
+    def test_kein_layer_setzt_verfahrensnummer_und_fordert_laden_an(
+        self, mock_lade_csv, mock_laenderkennung, mock_liste, mock_hole_layer, mock_set_variable
+    ):
+        zeilen = [{CSV_SPALTE_VEF_VKZF: "12345", CSV_SPALTE_FLURSTUECK: "1234/1/1/00"}]
+        mock_lade_csv.return_value = (zeilen, [CSV_SPALTE_VEF_VKZF, CSV_SPALTE_FLURSTUECK])
+        mock_liste.return_value = ["145678___00001____01"]
+
+        with self.assertRaises(BeteiligungPruefungFehler) as ctx:
+            beteiligung_pruefen(object())
+
+        mock_set_variable.assert_called_once_with("verfahrensnummer", "12345", scope="project")
+        self.assertIn("12345", str(ctx.exception))
+
+
+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

@@ -26,6 +26,10 @@ from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import (
     BROR_LAYER_NAME,
 )
 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,
@@ -75,6 +79,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")
@@ -174,6 +187,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()
@@ -285,6 +299,42 @@ 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`.
+        Ist noch kein (verwertbares) Verfahrensgebiet geladen, fordert die
+        Funktion den Nutzer per Meldung auf, es zunächst über "Aus ALKIS
+        laden" zu laden – es gibt kein automatisches Laden im Hintergrund.
+        Sonst 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.