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