changelog.txt aktualisiert

Reviewed-on: #54

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
 ---

functions/__init__.py

@@ -21,6 +21,7 @@ from .ly_metadata_wrapper import (
 )
 from .ly_style_wrapper import apply_style
 from .dialog_wrapper import ask_yes_no, ask_overwrite_append_cancel_custom, ask_ergaenzen_entfernen_ersetzen_abbrechen, ask_detailpruefung_oder_vg_laden
+from .laenderkennung import get_laenderkennung, LAENDERKENNUNGEN
 
 from .message_wrapper import (
     _get_message_bar,

functions/laenderkennung.py

@@ -0,0 +1,51 @@
+"""
+sn_basis/functions/laenderkennung.py – ALKIS-Länderkennung je Plugin-Präfix.
+
+AGLB-länderspezifische Funktionen (z.B. Flurstückskennzeichen-Bildung) sollen
+nicht hart auf Sachsen ("14") verdrahtet sein, sondern die Länderkennung über
+dieses generische Mapping ermitteln. Für neue Bundesländer genügt ein weiterer
+Eintrag in ``LAENDERKENNUNGEN`` – die aufrufende Logik bleibt unverändert.
+"""
+
+#: Mapping von sn_*-Plugin-Präfix (ohne Unterstrich) auf die zweistellige
+#: ALKIS-Länderkennung. Aktuell nur Sachsen befüllt; weitere Bundesländer
+#: können hier ergänzt werden, sobald Schwesterplugins dafür existieren.
+LAENDERKENNUNGEN: dict[str, str] = {
+    "sn": "14",  # Sachsen
+}
+
+
+def get_laenderkennung(plugin_praefix: str | None = None) -> str:
+    """Ermittelt die ALKIS-Länderkennung für einen Plugin-Präfix.
+
+    Parameters
+    ----------
+    plugin_praefix : str, optional
+        Plugin-Präfix ohne Unterstrich, z.B. ``"sn"``. Wird kein Präfix
+        übergeben, wird er aus dem Package-Namen des aufrufenden Moduls
+        abgeleitet (``__name__`` des Aufrufers, erstes ``_``-Segment).
+
+    Returns
+    -------
+    str
+        Zweistellige Länderkennung, z.B. ``"14"``.
+
+    Raises
+    ------
+    KeyError
+        Wenn für den ermittelten Präfix keine Länderkennung hinterlegt ist.
+    """
+    if not plugin_praefix:
+        import inspect
+
+        frame = inspect.stack()[1]
+        caller_module = frame.frame.f_globals.get("__name__", "")
+        plugin_praefix = caller_module.split(".")[0].split("_")[0]
+
+    plugin_praefix = plugin_praefix.lower()
+    if plugin_praefix not in LAENDERKENNUNGEN:
+        raise KeyError(
+            f"Keine Länderkennung für Plugin-Präfix '{plugin_praefix}' hinterlegt. "
+            f"Bekannte Präfixe: {sorted(LAENDERKENNUNGEN)}"
+        )
+    return LAENDERKENNUNGEN[plugin_praefix]

modules/Datenabruf.py

@@ -19,7 +19,7 @@ Designprinzipien
 
 from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple
 
-from urllib.parse import urlparse, parse_qsl, urlencode, urlunparse
+from urllib.parse import urlparse, parse_qsl, urlencode, urlunparse, quote
 import json
 import time
 
@@ -132,6 +132,7 @@ class Datenabruf:
             # außer bei WMS (WMS bleibt unverändert).
             use_bbox = (raumfilter != "ohne") and (str(provider).upper() != "WMS")
             url = self._build_provider_url(link=link, provider=str(provider), bbox_geom=bbox_geom if use_bbox else None)
+            print(f"[Datenabruf] URL für {thema!r} ({provider}): {url}")
 
             # 2b) Fachdaten abrufen
             features, error_msg = self._fetch_features(
@@ -309,8 +310,11 @@ class Datenabruf:
                 if callable(extent_to_bbox):
                     bbox_str = extent_to_bbox(extent_obj)
                 else:
-                    # Fallback: einfache xmin/ymin/xmax/ymax-Extraktion (duck-typing)
-                    if hasattr(extent_obj, "xmin") and callable(getattr(extent_obj, "xmin")):
+                    # Fallback: xmin/ymin/xmax/ymax-Extraktion (duck-typing)
+                    # QgsRectangle verwendet xMinimum()/yMinimum()/xMaximum()/yMaximum()
+                    if hasattr(extent_obj, "xMinimum") and callable(getattr(extent_obj, "xMinimum")):
+                        bbox_str = f"{extent_obj.xMinimum()},{extent_obj.yMinimum()},{extent_obj.xMaximum()},{extent_obj.yMaximum()}"
+                    elif hasattr(extent_obj, "xmin") and callable(getattr(extent_obj, "xmin")):
                         bbox_str = f"{extent_obj.xmin()},{extent_obj.ymin()},{extent_obj.xmax()},{extent_obj.ymax()}"
                     elif isinstance(extent_obj, (tuple, list)) and len(extent_obj) == 4:
                         bbox_str = f"{extent_obj[0]},{extent_obj[1]},{extent_obj[2]},{extent_obj[3]}"
@@ -325,12 +329,26 @@ class Datenabruf:
         if provider_norm == "WFS":
             query_params.setdefault("service", "WFS")
             query_params.setdefault("request", "GetFeature")
-            query_params.setdefault("outputFormat", "application/json")
-            if bbox_str:
-                query_params.setdefault("BBOX", bbox_str)
+            query_params.setdefault("VERSION", "2.0.0")
+            # WFS 2.0: 'typename' → 'TYPENAMES'
+            if "typename" in query_params and "TYPENAMES" not in query_params:
+                query_params["TYPENAMES"] = query_params.pop("typename")
+            # resultType=hits: nur Anzahl vom Server abfragen, kein GML-Download
+            query_params.setdefault("resultType", "hits")
+            # CRS in OGC-URN-Format (wie QGIS Datenquellenverwaltung)
+            srsname_urn: Optional[str] = None
             if crs_authid:
-                query_params.setdefault("SRSNAME", crs_authid)
-            new_query = urlencode(query_params, doseq=True)
+                if crs_authid.upper().startswith("EPSG:"):
+                    srsname_urn = f"urn:ogc:def:crs:EPSG::{crs_authid.split(':', 1)[1]}"
+                else:
+                    srsname_urn = crs_authid
+            if bbox_str:
+                # 5. Komponente: CRS-URN anhängen (Standard für WFS 2.0)
+                bbox_with_crs = f"{bbox_str},{srsname_urn}" if srsname_urn else bbox_str
+                query_params.setdefault("BBOX", bbox_with_crs)
+            if srsname_urn:
+                query_params.setdefault("SRSNAME", srsname_urn)
+            new_query = urlencode(query_params, doseq=True, quote_via=quote, safe=':/')
             rebuilt = parsed._replace(query=new_query)
             return urlunparse(rebuilt)
 
@@ -389,14 +407,14 @@ class Datenabruf:
                     query_params.setdefault("inSR", srid)
                     query_params.setdefault("outSR", srid)
 
-            new_query = urlencode(query_params, doseq=True)
+            new_query = urlencode(query_params, doseq=True, quote_via=quote, safe=':/')
             rebuilt = parsed_rest._replace(query=new_query)
             return urlunparse(rebuilt)
 
         # Default: generischer bbox-Parameter (nur wenn vorhanden)
         if bbox_str:
             query_params.setdefault("bbox", bbox_str)
-        new_query = urlencode(query_params, doseq=True)
+        new_query = urlencode(query_params, doseq=True, quote_via=quote, safe=':/')
         rebuilt = parsed._replace(query=new_query)
         return urlunparse(rebuilt)
 
@@ -573,7 +591,33 @@ class Datenabruf:
             # Sonst: gib das gesamte JSON als einzelnes Objekt zurück
             return [parsed], None
         except json.JSONDecodeError:
-            # Nicht-JSON-Antwort (z. B. GML). Wenn QGIS verfügbar, versuche GML via temporärer Datei + OGR
+            # Nicht-JSON-Antwort (GML/XML). Für WFS resultType=hits: Anzahl aus XML lesen.
+            if prov == "WFS":
+                try:
+                    import xml.etree.ElementTree as ET
+                    root = ET.fromstring(response_text)
+                    # WFS 2.0: numberOfMatched; WFS 1.x: numberOfFeatures
+                    n_str = (
+                        root.get("numberOfMatched")
+                        or root.get("numberMatched")
+                        or root.get("numberOfFeatures")
+                        or root.get("numberOfReturned")
+                    )
+                    if n_str is not None:
+                        try:
+                            count = int(n_str)
+                            return [{}] * count, None
+                        except (ValueError, TypeError):
+                            pass
+                    # Fallback: wfs:member-Elemente zählen
+                    members = [
+                        c for c in root
+                        if c.tag.lower().endswith(("member", "featuremember"))
+                    ]
+                    return [{}] * len(members), None
+                except Exception as exc:
+                    return [], f"WFS-XML-Antwort konnte nicht geparst werden: {exc}"
+            # Nicht-WFS, nicht-JSON: GML via temporärer Datei + OGR
             if getattr(qgiscore, "QGIS_AVAILABLE", False):
                 try:
                     import tempfile
@@ -588,5 +632,4 @@ class Datenabruf:
                     return [], "GML-Antwort konnte nicht als Layer geladen werden"
                 except Exception as exc:
                     return [], f"Fehler beim Parsen von GML: {exc}"
-            # Wenn alles fehlschlägt:
             return [], "Antwort konnte nicht als JSON oder GML geparst werden"

modules/LayerLoader.py

@@ -122,7 +122,33 @@ class LayerLoader:
 
         try:
             if provider_lower == "wfs":
-                uri = link if link.strip().lower().startswith("url=") else f"url={link}"
+                raw_link = link.strip()
+                if raw_link.lower().startswith("url="):
+                    raw_link = raw_link[4:]
+                # QGIS WFS-Provider erwartet die Basis-URL und 'typename' als
+                # separaten URI-Parameter. GetFeature-spezifische Parameter
+                # (request, outputFormat, resultType, BBOX, …) werden entfernt.
+                try:
+                    from urllib.parse import urlparse as _up, parse_qsl as _pqsl, urlunparse as _uu, urlencode as _ue
+                    _parsed = _up(raw_link)
+                    _params = dict(_pqsl(_parsed.query, keep_blank_values=True))
+                    _typename = (
+                        _params.pop("typename", None)
+                        or _params.pop("TYPENAME", None)
+                        or _params.pop("TYPENAMES", None)
+                    )
+                    for _k in ("request", "REQUEST", "outputFormat", "OUTPUTFORMAT",
+                               "resultType", "RESULTTYPE", "BBOX", "bbox",
+                               "STARTINDEX", "COUNT"):
+                        _params.pop(_k, None)
+                    _params.setdefault("service", "WFS")
+                    _params.setdefault("version", "2.0.0")
+                    _base_url = _uu(_parsed._replace(query=_ue(_params)))
+                    uri = f"url={_base_url}"
+                    if _typename:
+                        uri += f" typename={_typename}"
+                except Exception:
+                    uri = link if link.strip().lower().startswith("url=") else f"url={link}"
                 layer = QgsVectorLayer(uri, thema, "WFS")
             elif provider_lower == "wms":
                 uri = link if link.strip().lower().startswith("url=") else f"url={link}"

tests/test_laenderkennung.py

@@ -0,0 +1,31 @@
+"""sn_basis/tests/test_laenderkennung.py
+
+Unit-Tests für :mod:`sn_basis.functions.laenderkennung`.
+"""
+
+import unittest
+
+from sn_basis.functions.laenderkennung import get_laenderkennung, LAENDERKENNUNGEN
+
+
+class TestGetLaenderkennung(unittest.TestCase):
+    def test_bekannter_praefix_gibt_kennung_zurueck(self):
+        self.assertEqual(get_laenderkennung("sn"), "14")
+
+    def test_praefix_gross_klein_unabhaengig(self):
+        self.assertEqual(get_laenderkennung("SN"), "14")
+
+    def test_unbekannter_praefix_wirft_keyerror(self):
+        with self.assertRaises(KeyError):
+            get_laenderkennung("xx")
+
+    def test_ohne_argument_wird_aus_aufrufendem_modul_abgeleitet(self):
+        # __name__ dieses Testmoduls beginnt mit "sn_basis" -> Präfix "sn"
+        self.assertEqual(get_laenderkennung(), "14")
+
+    def test_mapping_enthaelt_sachsen(self):
+        self.assertEqual(LAENDERKENNUNGEN.get("sn"), "14")
+
+
+if __name__ == "__main__":
+    unittest.main()