470 lines
17 KiB
Python
470 lines
17 KiB
Python
"""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
|
||
``FLA_VEF_VKZF,FLA_Beteiligung,FLURSTUECK``) mit dem Verfahrensgebiet-Bestand
|
||
ab. Aus ``FLURSTUECK`` (Format ``<Gemarkung>/<Zähler>/<Nenner>/<Folge>``) wird
|
||
je Zeile ein zu ALKIS kompatibles Flurstückskennzeichen ``Flurstueck_csv``
|
||
gebildet.
|
||
|
||
Automatische Kaskade je nach Projektzustand:
|
||
|
||
- **try1**: Ein Layer ``Verfahrensgebiet`` mit befülltem Attribut
|
||
``flurstueckskennzeichen`` (Single-Feature-Modell, siehe
|
||
:mod:`verfahrensgebiet_edit_service`) ist vorhanden → schneller
|
||
Mengenvergleich ohne WFS-Zugriff.
|
||
- **try2**: Ein Verfahrensgebiet-Layer ist vorhanden, aber ohne verwertbare
|
||
Kennzeichen → 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 = "FLA_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.
|
||
|
||
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_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)
|