AG_QGIS/Plugin_SN_Verfahrensgebiet
3
1
Fork 1
Code Issues 11 Pull Requests Actions Releases 25 Wiki Activity

Funktion angelegt, Button eingefügt

Browse Source
This commit is contained in:
Daniel
2026-08-26 12:19:46 +02:00
parent 2cf0cda199
commit e6332e42ad
5 changed files with 745 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
functions/beteiligung_pruefung.py
+467
View File
@@ -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 ``<Gemarkung>/<Zähler>/<Nenner>/<Folge>``) wird je Zeile
ein zu ALKIS kompatibles Flurstückskennzeichen ``Flurstueck_csv`` gebildet.
Automatische Kaskade je nach Projektzustand:
- **try1**: Ein Layer ``Verfahrensgebiet`` mit befülltem Attribut
``flurstueckskennzeichen`` (Single-Feature-Modell, siehe
:mod:`verfahrensgebiet_edit_service`) ist vorhanden → schneller
Mengenvergleich ohne WFS-Zugriff.
- **try2**: Ein Verfahrensgebiet-Layer ist vorhanden, aber ohne verwertbare
Kennzeichen → 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
``<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).
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)