From e6332e42ad2be89bfe90a3423d432a79df663ced Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 26 Aug 2026 12:19:46 +0200 Subject: [PATCH 1/5] =?UTF-8?q?Funktion=20angelegt,=20Button=20eingef?= =?UTF-8?q?=C3=BCgt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- functions/beteiligung_pruefung.py | 467 +++++++++++++++++++++++++++++ styles/beteiligung_erfolgreich.qml | 22 ++ styles/beteiligung_vergleich.qml | 51 ++++ tests/test_beteiligung_pruefung.py | 157 ++++++++++ ui/tabs/working_tab.py | 48 +++ 5 files changed, 745 insertions(+) create mode 100644 functions/beteiligung_pruefung.py create mode 100644 styles/beteiligung_erfolgreich.qml create mode 100644 styles/beteiligung_vergleich.qml create mode 100644 tests/test_beteiligung_pruefung.py diff --git a/functions/beteiligung_pruefung.py b/functions/beteiligung_pruefung.py new file mode 100644 index 0000000..f277ce3 --- /dev/null +++ b/functions/beteiligung_pruefung.py @@ -0,0 +1,467 @@ +"""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 (Spalten ``FLURSTUECK``, +``VEF_VKZF``, ``FLA_BETEILIGUNG``) mit dem Verfahrensgebiet-Bestand ab. Aus +``FLURSTUECK`` (Format ``///``) 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 → je CSV-Zeile eine WFS-Einzelabfrage gegen ``AX_Flurstueck``. +- **try3**: Kein Verfahrensgebiet-Layer im Projekt → Verfahrensnummer aus + ``VEF_VKZF`` als Projektvariable setzen, Verfahrensgebiet analog zur + bestehenden ALKIS-Pipeline laden (ohne Detailprüfung), 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 import QMessageBox +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_verfahrensgebiet.functions.verfahrensgebiet_alkis import verfahrensgebiet_alkis +from sn_verfahrensgebiet.functions.flurstueck_alkis import ALKIS_SF_URL, FLST_TYPENAME, 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" +ERGEBNIS_LAYER_ERFOLGREICH = "Beteiligung – gefunden" +ERGEBNIS_LAYER_NICHT_GEFUNDEN = "Beteiligung – nicht gefunden" + +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 +# --------------------------------------------------------------------------- + +def baue_flurstueck_csv(flurstueck_wert: str, laenderkennung: str) -> Optional[str]: + """Bildet aus dem CSV-Attribut ``FLURSTUECK`` ein ALKIS-Flurstückskennzeichen. + + ``FLURSTUECK`` hat das Format + ``///``. 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). + + 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_ohne_null = nenner.lstrip("0") + nenner_kz = "____" if not nenner_ohne_null else nenner.zfill(4) + + return f"{laenderkennung}{gemarkung_kz}{flur_kz}{zaehler_kz}{nenner_kz}{folge}" + + +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 aktuell relevanten Verfahrensgebiet-Layer, falls vorhanden.""" + 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 + + treffer = QgsProject.instance().mapLayersByName(VERF_LAYER_NAME) + return treffer[0] if treffer else 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(verf_layer, flurstueck_csv_liste: List[str]) -> Tuple[object, Dict[str, int]]: + verf_menge = get_kennzeichen_menge(verf_layer) + csv_menge = set(flurstueck_csv_liste) + + beide = sorted(verf_menge & csv_menge) + nur_verfahrensgebiet = sorted(verf_menge - csv_menge) + nur_aglb = sorted(csv_menge - verf_menge) + + 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() + + statistik = { + WERT_BEIDE: len(beide), + WERT_NUR_VERFAHRENSGEBIET: len(nur_verfahrensgebiet), + WERT_NUR_AGLB: len(nur_aglb), + } + return layer, statistik + + +# --------------------------------------------------------------------------- +# try2: WFS-Einzelabgleich +# --------------------------------------------------------------------------- + +def _try2_wfs_einzelabgleich( + flurstueck_csv_liste: List[str], +) -> Tuple[Optional[object], object, Dict[str, int]]: + erfolgreich: List[tuple] = [] + nicht_gefunden: List[str] = [] + schema_fields = None + crs = "EPSG:25833" + dienst_erreichbar = False + + for kz in flurstueck_csv_liste: + uri = ( + f"url='{ALKIS_SF_URL}' typename='{FLST_TYPENAME}' srsname='EPSG:25833' " + f"sql=SELECT * FROM AX_Flurstueck " + f"WHERE {FELD_FLURSTUECKSKENNZEICHEN_WFS} = '{kz}'" + ) + wfs = QgsVectorLayer(uri, f"_TmpBet{kz}", "WFS") + if not wfs.isValid(): + continue + dienst_erreichbar = True + + feats = list(wfs.getFeatures()) + if not feats: + nicht_gefunden.append(kz) + continue + + if schema_fields is None: + schema_fields = wfs.fields() + crs = wfs.crs().authid() + erfolgreich.append((wfs.fields(), feats[0])) + + if not dienst_erreichbar: + raise BeteiligungPruefungFehler( + f"Dienst nicht erreichbar – ALKIS-WFS ({ALKIS_SF_URL}) konnte nicht " + f"abgefragt werden. Bitte Netzwerkverbindung prüfen." + ) + + erfolg_layer = None + if schema_fields is not None: + erfolg_layer = QgsVectorLayer(f"Polygon?crs={crs}", ERGEBNIS_LAYER_ERFOLGREICH, "memory") + pr = erfolg_layer.dataProvider() + pr.addAttributes(list(schema_fields)) + erfolg_layer.updateFields() + mem_fields = erfolg_layer.fields() + + neue_features = [] + for wfs_fields, feat in erfolgreich: + new_f = QgsFeature(mem_fields) + new_f.setGeometry(feat.geometry()) + for i in range(wfs_fields.count()): + fname = wfs_fields.field(i).name() + idx = mem_fields.indexFromName(fname) + if idx >= 0: + try: + new_f.setAttribute(idx, feat[fname]) + except Exception: + pass + neue_features.append(new_f) + pr.addFeatures(neue_features) + erfolg_layer.updateExtents() + + ng_layer = QgsVectorLayer( + f"None?field={FELD_FLURSTUECK_CSV}:string", ERGEBNIS_LAYER_NICHT_GEFUNDEN, "memory" + ) + pr = ng_layer.dataProvider() + ng_fields = ng_layer.fields() + neue_ng = [] + for kz in nicht_gefunden: + f = QgsFeature(ng_fields) + f.setAttribute(0, kz) + neue_ng.append(f) + pr.addFeatures(neue_ng) + + statistik = {"erfolgreich": len(erfolgreich), "nicht_gefunden": len(nicht_gefunden)} + return erfolg_layer, ng_layer, statistik + + +# --------------------------------------------------------------------------- +# try3: Bootstrap (Verfahrensnummer aus CSV, VG laden) + 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") + + # Identische BROR-Ladung wie die bestehende ALKIS-Pipeline, ohne die + # nachgelagerte Detailprüfung/Flächenvergleich aus verfahrensgebiet_alkis_komplett. + neuer_vg_layer, status, _feats = verfahrensgebiet_alkis(tab_widget, skip_existing_check=True) + if neuer_vg_layer is None: + # verfahrensgebiet_alkis() hat bereits eine erklärende Messagebox gezeigt. + raise BeteiligungPruefungFehler( + f"Verfahrensgebiet konnte für Verfahrensnummer '{vef_vkzf}' nicht geladen werden." + ) + + erfolg_layer, ng_layer, statistik = _try2_wfs_einzelabgleich(flurstueck_csv_liste) + + layers: List[Tuple[object, Optional[str]]] = [(neuer_vg_layer, "verfahrensgebiet.qml")] + if erfolg_layer is not None: + layers.append((erfolg_layer, "beteiligung_erfolgreich.qml")) + layers.append((ng_layer, None)) + + text = ( + f"Verfahrensnummer aus CSV: {vef_vkzf}\n" + f"Erfolgreich geladen: {statistik['erfolgreich']}\n" + f"Nicht gefunden: {statistik['nicht_gefunden']}" + ) + return BeteiligungErgebnis("try3", layers, statistik, text) + + +# --------------------------------------------------------------------------- +# Ö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(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: + erfolg_layer, ng_layer, statistik = _try2_wfs_einzelabgleich(flurstueck_csv_liste) + layers: List[Tuple[object, Optional[str]]] = [] + if erfolg_layer is not None: + layers.append((erfolg_layer, "beteiligung_erfolgreich.qml")) + layers.append((ng_layer, None)) + text = ( + f"Erfolgreich geladen: {statistik['erfolgreich']}\n" + f"Nicht gefunden: {statistik['nicht_gefunden']}" + ) + return BeteiligungErgebnis("try2", layers, statistik, text) + + return _try3_bootstrap_und_try2(tab_widget, zeilen, spaltennamen, flurstueck_csv_liste) diff --git a/styles/beteiligung_erfolgreich.qml b/styles/beteiligung_erfolgreich.qml new file mode 100644 index 0000000..3691aa3 --- /dev/null +++ b/styles/beteiligung_erfolgreich.qml @@ -0,0 +1,22 @@ + + + + + + + + + + + + 0 + 0 + 2 + diff --git a/styles/beteiligung_vergleich.qml b/styles/beteiligung_vergleich.qml new file mode 100644 index 0000000..1062501 --- /dev/null +++ b/styles/beteiligung_vergleich.qml @@ -0,0 +1,51 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + 0 + 0 + 2 + diff --git a/tests/test_beteiligung_pruefung.py b/tests/test_beteiligung_pruefung.py new file mode 100644 index 0000000..ca36fb3 --- /dev/null +++ b/tests/test_beteiligung_pruefung.py @@ -0,0 +1,157 @@ +"""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 + +# --------------------------------------------------------------------------- +# 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, + FELD_FLURSTUECK_CSV, + CSV_SPALTE_FLURSTUECK, +) + +_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")) + + +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 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( + ["FLURSTUECK", "VEF_VKZF", "FLA_BETEILIGUNG"], + [["1234/123/45/00", "12345", "1"]], + ) + mock_fd.getOpenFileName.return_value = (pfad, "") + zeilen, spalten = lade_beteiligung_csv(None) + self.assertEqual(len(zeilen), 1) + self.assertIn("FLURSTUECK", spalten) + self.assertEqual(zeilen[0]["FLURSTUECK"], "1234/123/45/00") + + +if __name__ == "__main__": + unittest.main() diff --git a/ui/tabs/working_tab.py b/ui/tabs/working_tab.py index 813a10f..9ba2c65 100644 --- a/ui/tabs/working_tab.py +++ b/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. From 2cb6193b98740fb94741598428fc3ea51ee04808 Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 26 Aug 2026 14:30:07 +0200 Subject: [PATCH 2/5] csv-header korrigiert --- functions/beteiligung_pruefung.py | 16 +++++++++------- tests/test_beteiligung_pruefung.py | 9 +++++---- 2 files changed, 14 insertions(+), 11 deletions(-) diff --git a/functions/beteiligung_pruefung.py b/functions/beteiligung_pruefung.py index f277ce3..fdf916d 100644 --- a/functions/beteiligung_pruefung.py +++ b/functions/beteiligung_pruefung.py @@ -6,10 +6,11 @@ Schema). Die Länderkennung wird nicht hartcodiert, sondern über Schwesterplugins für andere Bundesländer dieselbe Logik mit einer anderen Länderkennung wiederverwenden können. -Gleicht eine CSV-Exportliste des BI-Servers (Spalten ``FLURSTUECK``, -``VEF_VKZF``, ``FLA_BETEILIGUNG``) mit dem Verfahrensgebiet-Bestand ab. Aus -``FLURSTUECK`` (Format ``///``) wird je Zeile -ein zu ALKIS kompatibles Flurstückskennzeichen ``Flurstueck_csv`` gebildet. +Gleicht eine CSV-Exportliste des BI-Servers (Kopfzeile +``FLA_VEF_VKZF,FLA_Beteiligung,FLURSTUECK``) mit dem Verfahrensgebiet-Bestand +ab. Aus ``FLURSTUECK`` (Format ``///``) wird +je Zeile ein zu ALKIS kompatibles Flurstückskennzeichen ``Flurstueck_csv`` +gebildet. Automatische Kaskade je nach Projektzustand: @@ -44,8 +45,8 @@ from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import ( ) CSV_SPALTE_FLURSTUECK = "FLURSTUECK" -CSV_SPALTE_VEF_VKZF = "VEF_VKZF" -CSV_SPALTE_BETEILIGUNG = "FLA_BETEILIGUNG" +CSV_SPALTE_VEF_VKZF = "FLA_VEF_VKZF" +CSV_SPALTE_BETEILIGUNG = "FLA_Beteiligung" FELD_FLURSTUECK_CSV = "Flurstueck_csv" FELD_FLURSTUECKSKENNZEICHEN_WFS = "flurstueckskennzeichen" @@ -129,10 +130,11 @@ def lade_beteiligung_csv(parent_widget) -> Optional[Tuple[List[dict], List[str]] def baue_flurstueck_csv(flurstueck_wert: str, laenderkennung: str) -> Optional[str]: """Bildet aus dem CSV-Attribut ``FLURSTUECK`` ein ALKIS-Flurstückskennzeichen. - ``FLURSTUECK`` hat das Format + Das Attribut hat das Format ``///``. 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 ------- diff --git a/tests/test_beteiligung_pruefung.py b/tests/test_beteiligung_pruefung.py index ca36fb3..e676f59 100644 --- a/tests/test_beteiligung_pruefung.py +++ b/tests/test_beteiligung_pruefung.py @@ -51,6 +51,7 @@ from sn_verfahrensgebiet.functions.beteiligung_pruefung import ( BeteiligungPruefungFehler, FELD_FLURSTUECK_CSV, CSV_SPALTE_FLURSTUECK, + CSV_SPALTE_VEF_VKZF, ) _MOD = "sn_verfahrensgebiet.functions.beteiligung_pruefung" @@ -143,14 +144,14 @@ class TestLadeBeteiligungCsv(unittest.TestCase): @patch(f"{_MOD}.QFileDialog") def test_erfolgsfall_gibt_zeilen_und_spalten_zurueck(self, mock_fd): pfad = self._schreibe_csv( - ["FLURSTUECK", "VEF_VKZF", "FLA_BETEILIGUNG"], - [["1234/123/45/00", "12345", "1"]], + [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("FLURSTUECK", spalten) - self.assertEqual(zeilen[0]["FLURSTUECK"], "1234/123/45/00") + self.assertIn(CSV_SPALTE_FLURSTUECK, spalten) + self.assertEqual(zeilen[0][CSV_SPALTE_FLURSTUECK], "1234/123/45/00") if __name__ == "__main__": From 6bc5256d5fa78fe4a4e1ff09825371e530342450 Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 26 Aug 2026 16:08:01 +0200 Subject: [PATCH 3/5] =?UTF-8?q?try2=20und=20try3=20laden=20das=20Verfahren?= =?UTF-8?q?sgebiet=20mit=20Detailpr=C3=BCfung=20anstelle=20der=20Flurst?= =?UTF-8?q?=C3=BCcke=20vom=20wfs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- functions/beteiligung_pruefung.py | 244 ++++++++++--------- functions/verfahrensgebiet_alkis.py | 14 +- functions/verfahrensgebiet_alkis_komplett.py | 66 +++-- tests/test_beteiligung_pruefung.py | 111 ++++++++- 4 files changed, 287 insertions(+), 148 deletions(-) diff --git a/functions/beteiligung_pruefung.py b/functions/beteiligung_pruefung.py index fdf916d..54152c1 100644 --- a/functions/beteiligung_pruefung.py +++ b/functions/beteiligung_pruefung.py @@ -7,7 +7,7 @@ Schwesterplugins für andere Bundesländer dieselbe Logik mit einer anderen Länderkennung wiederverwenden können. Gleicht eine CSV-Exportliste des BI-Servers (Kopfzeile -``FLA_VEF_VKZF,FLA_Beteiligung,FLURSTUECK``) mit dem Verfahrensgebiet-Bestand +``VEF_VKZF,FLA_Beteiligung,FLURSTUECK``) mit dem Verfahrensgebiet-Bestand ab. Aus ``FLURSTUECK`` (Format ``///``) wird je Zeile ein zu ALKIS kompatibles Flurstückskennzeichen ``Flurstueck_csv`` gebildet. @@ -19,10 +19,12 @@ Automatische Kaskade je nach Projektzustand: :mod:`verfahrensgebiet_edit_service`) ist vorhanden → schneller Mengenvergleich ohne WFS-Zugriff. - **try2**: Ein Verfahrensgebiet-Layer ist vorhanden, aber ohne verwertbare - Kennzeichen → je CSV-Zeile eine WFS-Einzelabfrage gegen ``AX_Flurstueck``. + 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, Verfahrensgebiet analog zur - bestehenden ALKIS-Pipeline laden (ohne Detailprüfung), danach wie try2. + ``VEF_VKZF`` als Projektvariable setzen, danach wie try2. """ from __future__ import annotations @@ -32,28 +34,29 @@ 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 import QMessageBox 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 import verfahrensgebiet_alkis -from sn_verfahrensgebiet.functions.flurstueck_alkis import ALKIS_SF_URL, FLST_TYPENAME, FLST_LAYER_NAME +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 = "FLA_VEF_VKZF" +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" -ERGEBNIS_LAYER_ERFOLGREICH = "Beteiligung – gefunden" -ERGEBNIS_LAYER_NICHT_GEFUNDEN = "Beteiligung – nicht gefunden" WERT_BEIDE = "beide" WERT_NUR_VERFAHRENSGEBIET = "nur_verfahrensgebiet" @@ -127,6 +130,38 @@ def lade_beteiligung_csv(parent_widget) -> Optional[Tuple[List[dict], List[str]] # 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. @@ -154,11 +189,10 @@ def baue_flurstueck_csv(flurstueck_wert: str, laenderkennung: str) -> Optional[s gemarkung_kz = gemarkung.zfill(4) flur_kz = "___" zaehler_kz = zaehler.zfill(5) + nenner_kz = _baue_nenner_kennzeichen(nenner) + folge_kz = folge.zfill(2) - nenner_ohne_null = nenner.lstrip("0") - nenner_kz = "____" if not nenner_ohne_null else nenner.zfill(4) - - return f"{laenderkennung}{gemarkung_kz}{flur_kz}{zaehler_kz}{nenner_kz}{folge}" + 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]: @@ -231,14 +265,25 @@ def _flurstueckskennzeichen_geometrien(flst_layer) -> Dict[str, object]: # try1: schneller Listenabgleich # --------------------------------------------------------------------------- -def _try1_listenabgleich(verf_layer, flurstueck_csv_liste: List[str]) -> Tuple[object, Dict[str, int]]: +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) + 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(): @@ -274,6 +319,9 @@ def _try1_listenabgleich(verf_layer, flurstueck_csv_liste: List[str]) -> Tuple[o pr.addFeatures(neue_features) layer.updateExtents() + progress.set_value(3) + progress.close() + statistik = { WERT_BEIDE: len(beide), WERT_NUR_VERFAHRENSGEBIET: len(nur_verfahrensgebiet), @@ -283,87 +331,68 @@ def _try1_listenabgleich(verf_layer, flurstueck_csv_liste: List[str]) -> Tuple[o # --------------------------------------------------------------------------- -# try2: WFS-Einzelabgleich +# try2/try3: ALKIS-Pipeline (silent) + Listenabgleich wie try1 # --------------------------------------------------------------------------- -def _try2_wfs_einzelabgleich( - flurstueck_csv_liste: List[str], -) -> Tuple[Optional[object], object, Dict[str, int]]: - erfolgreich: List[tuple] = [] - nicht_gefunden: List[str] = [] - schema_fields = None - crs = "EPSG:25833" - dienst_erreichbar = False +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. - for kz in flurstueck_csv_liste: - uri = ( - f"url='{ALKIS_SF_URL}' typename='{FLST_TYPENAME}' srsname='EPSG:25833' " - f"sql=SELECT * FROM AX_Flurstueck " - f"WHERE {FELD_FLURSTUECKSKENNZEICHEN_WFS} = '{kz}'" - ) - wfs = QgsVectorLayer(uri, f"_TmpBet{kz}", "WFS") - if not wfs.isValid(): - continue - dienst_erreichbar = True - - feats = list(wfs.getFeatures()) - if not feats: - nicht_gefunden.append(kz) - continue - - if schema_fields is None: - schema_fields = wfs.fields() - crs = wfs.crs().authid() - erfolgreich.append((wfs.fields(), feats[0])) - - if not dienst_erreichbar: - raise BeteiligungPruefungFehler( - f"Dienst nicht erreichbar – ALKIS-WFS ({ALKIS_SF_URL}) konnte nicht " - f"abgefragt werden. Bitte Netzwerkverbindung prüfen." - ) - - erfolg_layer = None - if schema_fields is not None: - erfolg_layer = QgsVectorLayer(f"Polygon?crs={crs}", ERGEBNIS_LAYER_ERFOLGREICH, "memory") - pr = erfolg_layer.dataProvider() - pr.addAttributes(list(schema_fields)) - erfolg_layer.updateFields() - mem_fields = erfolg_layer.fields() - - neue_features = [] - for wfs_fields, feat in erfolgreich: - new_f = QgsFeature(mem_fields) - new_f.setGeometry(feat.geometry()) - for i in range(wfs_fields.count()): - fname = wfs_fields.field(i).name() - idx = mem_fields.indexFromName(fname) - if idx >= 0: - try: - new_f.setAttribute(idx, feat[fname]) - except Exception: - pass - neue_features.append(new_f) - pr.addFeatures(neue_features) - erfolg_layer.updateExtents() - - ng_layer = QgsVectorLayer( - f"None?field={FELD_FLURSTUECK_CSV}:string", ERGEBNIS_LAYER_NICHT_GEFUNDEN, "memory" + 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 ) - pr = ng_layer.dataProvider() - ng_fields = ng_layer.fields() - neue_ng = [] - for kz in nicht_gefunden: - f = QgsFeature(ng_fields) - f.setAttribute(0, kz) - neue_ng.append(f) - pr.addFeatures(neue_ng) + if enriched_layer is None: + raise BeteiligungPruefungFehler( + "Verfahrensgebiet konnte über die ALKIS-Pipeline nicht geladen werden." + ) - statistik = {"erfolgreich": len(erfolgreich), "nicht_gefunden": len(nicht_gefunden)} - return erfolg_layer, ng_layer, statistik + 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, VG laden) + try2 +# try3: Bootstrap (Verfahrensnummer aus CSV) + wie try2 # --------------------------------------------------------------------------- def _lese_vef_vkzf(zeilen: List[dict], spaltennamen: List[str]) -> str: @@ -386,28 +415,10 @@ def _try3_bootstrap_und_try2( vef_vkzf = _lese_vef_vkzf(zeilen, spaltennamen) set_variable("verfahrensnummer", vef_vkzf, scope="project") - # Identische BROR-Ladung wie die bestehende ALKIS-Pipeline, ohne die - # nachgelagerte Detailprüfung/Flächenvergleich aus verfahrensgebiet_alkis_komplett. - neuer_vg_layer, status, _feats = verfahrensgebiet_alkis(tab_widget, skip_existing_check=True) - if neuer_vg_layer is None: - # verfahrensgebiet_alkis() hat bereits eine erklärende Messagebox gezeigt. - raise BeteiligungPruefungFehler( - f"Verfahrensgebiet konnte für Verfahrensnummer '{vef_vkzf}' nicht geladen werden." - ) - - erfolg_layer, ng_layer, statistik = _try2_wfs_einzelabgleich(flurstueck_csv_liste) - - layers: List[Tuple[object, Optional[str]]] = [(neuer_vg_layer, "verfahrensgebiet.qml")] - if erfolg_layer is not None: - layers.append((erfolg_layer, "beteiligung_erfolgreich.qml")) - layers.append((ng_layer, None)) - - text = ( - f"Verfahrensnummer aus CSV: {vef_vkzf}\n" - f"Erfolgreich geladen: {statistik['erfolgreich']}\n" - f"Nicht gefunden: {statistik['nicht_gefunden']}" - ) - return BeteiligungErgebnis("try3", layers, statistik, text) + 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 # --------------------------------------------------------------------------- @@ -446,7 +457,7 @@ def beteiligung_pruefen(tab_widget) -> Optional[BeteiligungErgebnis]: verf_layer = _hole_verfahrensgebiet_layer(tab_widget) if verf_layer is not None and _hat_verwendbare_kennzeichen(verf_layer): - layer, statistik = _try1_listenabgleich(verf_layer, flurstueck_csv_liste) + 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" @@ -455,15 +466,6 @@ def beteiligung_pruefen(tab_widget) -> Optional[BeteiligungErgebnis]: return BeteiligungErgebnis("try1", [(layer, "beteiligung_vergleich.qml")], statistik, text) if verf_layer is not None: - erfolg_layer, ng_layer, statistik = _try2_wfs_einzelabgleich(flurstueck_csv_liste) - layers: List[Tuple[object, Optional[str]]] = [] - if erfolg_layer is not None: - layers.append((erfolg_layer, "beteiligung_erfolgreich.qml")) - layers.append((ng_layer, None)) - text = ( - f"Erfolgreich geladen: {statistik['erfolgreich']}\n" - f"Nicht gefunden: {statistik['nicht_gefunden']}" - ) - return BeteiligungErgebnis("try2", layers, statistik, text) + return _lade_via_alkis_pipeline_und_vergleiche(tab_widget, flurstueck_csv_liste) return _try3_bootstrap_und_try2(tab_widget, zeilen, spaltennamen, flurstueck_csv_liste) diff --git a/functions/verfahrensgebiet_alkis.py b/functions/verfahrensgebiet_alkis.py index be56b8e..c885eec 100644 --- a/functions/verfahrensgebiet_alkis.py +++ b/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()) diff --git a/functions/verfahrensgebiet_alkis_komplett.py b/functions/verfahrensgebiet_alkis_komplett.py index adca779..e20c1d4 100644 --- a/functions/verfahrensgebiet_alkis_komplett.py +++ b/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 @@ -550,14 +565,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 diff --git a/tests/test_beteiligung_pruefung.py b/tests/test_beteiligung_pruefung.py index e676f59..ff79402 100644 --- a/tests/test_beteiligung_pruefung.py +++ b/tests/test_beteiligung_pruefung.py @@ -27,7 +27,7 @@ import sys import tempfile import types import unittest -from unittest.mock import patch +from unittest.mock import patch, MagicMock # --------------------------------------------------------------------------- # Minimales QGIS-Mock-Setup @@ -49,9 +49,12 @@ from sn_verfahrensgebiet.functions.beteiligung_pruefung import ( 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, ) _MOD = "sn_verfahrensgebiet.functions.beteiligung_pruefung" @@ -83,6 +86,30 @@ class TestBaueFlurstueckCsv(unittest.TestCase): 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): @@ -98,6 +125,88 @@ class TestBaueFlurstueckCsvListe(unittest.TestCase): self.assertEqual(zeilen[2][FELD_FLURSTUECK_CSV], "145678___00001____01") +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 = [] From 4a557e66a70a753a0c2b0f7239644126cb821ade Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 26 Aug 2026 16:13:51 +0200 Subject: [PATCH 4/5] =?UTF-8?q?try1-fix:=20direkte=20Pr=C3=BCfung=20ohne?= =?UTF-8?q?=20ALKIS-Pipeline?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- functions/beteiligung_pruefung.py | 21 +++++++-- functions/verfahrensgebiet_alkis_komplett.py | 4 ++ tests/test_beteiligung_pruefung.py | 45 ++++++++++++++++++++ 3 files changed, 67 insertions(+), 3 deletions(-) diff --git a/functions/beteiligung_pruefung.py b/functions/beteiligung_pruefung.py index 54152c1..9fff270 100644 --- a/functions/beteiligung_pruefung.py +++ b/functions/beteiligung_pruefung.py @@ -221,15 +221,26 @@ def baue_flurstueck_csv_liste(zeilen: List[dict], laenderkennung: str) -> List[s # --------------------------------------------------------------------------- def _hole_verfahrensgebiet_layer(tab_widget) -> Optional[object]: - """Liefert den aktuell relevanten Verfahrensgebiet-Layer, falls vorhanden.""" + """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 - treffer = QgsProject.instance().mapLayersByName(VERF_LAYER_NAME) - return treffer[0] if treffer else None + return None def _hat_verwendbare_kennzeichen(verf_layer) -> bool: @@ -272,6 +283,10 @@ def _try1_listenabgleich( 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) diff --git a/functions/verfahrensgebiet_alkis_komplett.py b/functions/verfahrensgebiet_alkis_komplett.py index e20c1d4..f9ea49d 100644 --- a/functions/verfahrensgebiet_alkis_komplett.py +++ b/functions/verfahrensgebiet_alkis_komplett.py @@ -496,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( diff --git a/tests/test_beteiligung_pruefung.py b/tests/test_beteiligung_pruefung.py index ff79402..36352dd 100644 --- a/tests/test_beteiligung_pruefung.py +++ b/tests/test_beteiligung_pruefung.py @@ -55,6 +55,7 @@ from sn_verfahrensgebiet.functions.beteiligung_pruefung import ( CSV_SPALTE_VEF_VKZF, _lade_via_alkis_pipeline_und_vergleiche, _try3_bootstrap_und_try2, + _hole_verfahrensgebiet_layer, ) _MOD = "sn_verfahrensgebiet.functions.beteiligung_pruefung" @@ -125,6 +126,50 @@ class TestBaueFlurstueckCsvListe(unittest.TestCase): 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") From 6a65c9d7062160229e96430c3af32c34c125922a Mon Sep 17 00:00:00 2001 From: Daniel Helbig Date: Wed, 26 Aug 2026 16:19:14 +0200 Subject: [PATCH 5/5] changelog.txt aktualisiert --- changelog.txt | 3 +++ 1 file changed, 3 insertions(+) diff --git a/changelog.txt b/changelog.txt index d6aebb1..3cb5454 100644 --- a/changelog.txt +++ b/changelog.txt @@ -1,3 +1,6 @@ +Version 26.8.1-unstable: +-Vergleich einer CSV aus dem BI-Server mit den Beteiligten Flurstücken lt. ALKIS +--- Version 26.7.1-unstable: - Verfahrensgebiet als shape laden, inkl. Abgleich mit BROBOR-Objekt ---