"""HTTP-Client für die VLN-Manager-API. Basis-URL: https://api.flurneuordnung-sachsen.de/v2 POST /person/login Body: {"mail": "...", "password": "..."} Antwort: {"data": {"userauth": "", "id": ...}} Das userauth-Token ist der API-Key (= PERSON.url_token). GET /tgen/auswahl -> {"verfahren": [{vkz, name, zugriff}], "aemter": [{vkz, name}]} Eigene Verfahren ("voll") und — als oFB — die Verfahren der eigenen Landkreise ("lesen"), dazu die Landkreise selbst (Manager seit 06.10.2026; Rückfall: GET /tgen). GET /ogc/collections?vkz=… -> Sammlungen (s. ogc_layers.py), ``vln:schreibbar`` je Sammlung GET /linkliste?vkz=… -> Geodienste eines Verfahrens oder Landkreises (s. linkliste.py) Auth über den Header "X-API-Key". Fehlerformat: RFC 7807 (application/problem+json). 401 = nicht (mehr) angemeldet (AuthError), 403 = angemeldet, aber nicht berechtigt (KeinZugriff) — nur 401 führt zum Abmelden. Es wird QgsBlockingNetworkRequest verwendet, damit Proxy- und Zertifikatseinstellungen aus QGIS automatisch greifen. """ import json from qgis.PyQt.QtCore import QUrl from qgis.PyQt.QtNetwork import QNetworkRequest from qgis.core import QgsBlockingNetworkRequest # Fest verdrahtete Produktiv-API des VLN Managers. DEFAULT_BASE_URL = "https://api.flurneuordnung-sachsen.de/v2" class ApiError(Exception): """Fehler bei der Kommunikation mit der Karten-API. :ivar status: HTTP-Status, wenn der Server geantwortet hat (sonst ``None``). """ def __init__(self, nachricht="", status=None): super().__init__(nachricht) self.status = status class AuthError(ApiError): """Nicht (mehr) angemeldet: API-Key fehlt oder ist abgelaufen (401) — dann neu anmelden.""" class KeinZugriff(ApiError): """Angemeldet, aber für diese Aktion nicht berechtigt (403). Kein Grund abzumelden: eine oFB liest z. B. die Verfahren ihres Landkreises, darf dort aber nicht schreiben. Bis 0.6.0 meldete jedes 403 ab und verwarf den gespeicherten API-Key. """ def fehler_fuer_status(status, text): """Die passende Ausnahme für einen HTTP-Fehlerstatus. :returns: ``AuthError`` (401), ``KeinZugriff`` (403), sonst ``ApiError`` — mit ``status``. """ if status == 401: return AuthError(text, status) if status == 403: return KeinZugriff(text, status) return ApiError(text, status) class KartenApiClient: def __init__(self, base_url=DEFAULT_BASE_URL, api_key=None, mail=None): self.base_url = base_url.rstrip("/") self.api_key = api_key self.mail = mail @property def is_authenticated(self): return bool(self.api_key) def login(self, mail, password): """Meldet den Benutzer an. Die API liefert das userauth-Token (= API-Key), das für alle weiteren Requests verwendet wird.""" data = self._request( "POST", "/person/login", payload={"mail": mail, "password": password}, with_auth=False, ) payload = data.get("data") if isinstance(data, dict) else None if not isinstance(payload, dict): payload = data if isinstance(data, dict) else {} api_key = payload.get("userauth") if not api_key: raise ApiError( "Login-Antwort enthielt kein userauth-Token. " "Erhaltene Antwort: %s" % json.dumps(data)[:300] ) self.api_key = api_key self.mail = mail return api_key def logout(self): self.api_key = None self.mail = None # ------------------------------------------------------------------ # Verfahren (Teilnehmergemeinschaften) # ------------------------------------------------------------------ def get_auswahl(self): """Verfahren und Landkreise zur Auswahl — ``GET /tgen/auswahl``. :returns: ``{"verfahren": [{vkz, name, zugriff}], "aemter": [{vkz, name}]}`` mit ``vkz`` als Text und ``zugriff`` ``"voll"`` oder ``"lesen"`` (eine oFB liest die Verfahren ihres Landkreises). Ein älterer Server kennt die Route nicht (400/404) — dann Rückfall auf :meth:`get_verfahren`, alles ``"voll"``, keine Landkreise. """ try: data = self._request("GET", "/tgen/auswahl") except ApiError as exc: if getattr(exc, "status", None) in (400, 404): return {"verfahren": [dict(v, zugriff="voll") for v in self.get_verfahren()], "aemter": []} raise if not isinstance(data, dict) or not isinstance(data.get("verfahren"), list): raise ApiError("Antwort von /tgen/auswahl hat ein unerwartetes Format: %s" % json.dumps(data)[:200]) verfahren = [{"vkz": str(v.get("vkz")), "name": str(v.get("name") or ""), "zugriff": "lesen" if v.get("zugriff") == "lesen" else "voll"} for v in data["verfahren"] if isinstance(v, dict) and v.get("vkz")] aemter = [{"vkz": str(a.get("vkz")), "name": str(a.get("name") or "")} for a in (data.get("aemter") or []) if isinstance(a, dict) and a.get("vkz")] return {"verfahren": verfahren, "aemter": aemter} def get_verfahren(self): """Liste der Verfahren (TGs) als [{"vkz": ..., "name": ...}, ...], sortiert nach VKZ. Quelle: GET /tgen (Rückfall für ältere Server).""" data = self._request("GET", "/tgen") rows = None if isinstance(data, list): rows = data elif isinstance(data, dict): for key in ("data", "rows"): value = data.get(key) if isinstance(value, list): rows = value break if isinstance(value, dict) and isinstance(value.get("rows"), list): rows = value["rows"] break if rows is None: raise ApiError( "Antwort von /tgen hat ein unerwartetes Format: %s" % json.dumps(data)[:200] ) verfahren = [] for row in rows: if not isinstance(row, dict): continue vkz = row.get("vkz") or row.get("VKZ") name = ( row.get("kurzname") or row.get("Kurzname") or row.get("name") or row.get("Name") or "" ) if vkz: verfahren.append({"vkz": str(vkz), "name": str(name)}) verfahren.sort(key=lambda v: v["vkz"]) return verfahren # ------------------------------------------------------------------ # Karten-Layer (die konkreten Pfade stehen in plugin.DATASETS) # ------------------------------------------------------------------ def load_feature_collection(self, path): """Lädt einen Layer als GeoJSON FeatureCollection (dict).""" data = self._request("GET", path) # Die Karten-Endpunkte liefern die FeatureCollection direkt, # andere Routen verpacken sie in {"data": ...}. if isinstance(data, dict): if data.get("type") == "FeatureCollection": return data inner = data.get("data") if isinstance(inner, dict) and inner.get("type") == "FeatureCollection": return inner raise ApiError( "Antwort von %s ist keine GeoJSON FeatureCollection." % path ) def roh(self, path, with_auth=False): """Eine Datei roh holen (Bytes) — fuer die Schildgrafiken und den Katalog. Anders als :meth:`_request` wird hier nichts als JSON gelesen und keine Anmeldung verlangt: ``/v2/zeichen/...`` ist oeffentlich, damit der Stil auch ohne Schluessel zeichnet. Fehler geben ``b""`` zurueck — ein fehlendes Schild darf nichts abbrechen. """ from qgis.PyQt.QtCore import QUrl from qgis.PyQt.QtNetwork import QNetworkRequest request = QNetworkRequest(QUrl(self.base_url + path)) if with_auth and self.api_key: request.setRawHeader(b"X-API-Key", self.api_key.encode("utf-8")) blocking = QgsBlockingNetworkRequest() if blocking.get(request) != QgsBlockingNetworkRequest.NoError: return b"" antwort = blocking.reply() status = antwort.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute) if status is not None and int(status) >= 400: return b"" return bytes(antwort.content()) def _request(self, method, path, payload=None, with_auth=True, frisch=False): """Ein Aufruf gegen die v2-API. :param frisch: den Zwischenspeicher übergehen. Die Linkliste antwortet mit ``Cache-Control: private, max-age=60``, damit ein Ladelauf nicht alles neu holt. Wer aber gerade im Manager einen Dienst eingetragen hat und dann in QGIS auf „Dienste holen“ klickt, bekäme sonst bis zu eine Minute lang die alte Liste (gemessen 11.09.2026). Bei einer ausdrücklichen Nutzeraktion also immer frisch. """ if with_auth and not self.is_authenticated: raise AuthError("Nicht angemeldet — bitte zuerst einloggen.") request = QNetworkRequest(QUrl(self.base_url + path)) request.setHeader( QNetworkRequest.KnownHeaders.ContentTypeHeader, "application/json" ) request.setRawHeader(b"Accept", b"application/json") if with_auth: request.setRawHeader(b"X-API-Key", self.api_key.encode("utf-8")) body = b"" if payload is not None: body = json.dumps(payload).encode("utf-8") # Der Zwischenspeicher wird über forceRefresh gesteuert, NICHT über # CacheLoadControlAttribute am Request: QgsBlockingNetworkRequest setzt das Attribut # selbst und überschreibt dabei, was man vorher gesetzt hat (gemessen 11.09.2026 — # ein AlwaysNetwork am Request blieb wirkungslos). blocking = QgsBlockingNetworkRequest() if method == "GET": error = blocking.get(request, frisch) elif method == "POST": error = blocking.post(request, body, frisch) elif method == "PUT": error = blocking.put(request, body) else: raise ApiError("Nicht unterstützte HTTP-Methode: %s" % method) reply = blocking.reply() content = bytes(reply.content()) status = reply.attribute( QNetworkRequest.Attribute.HttpStatusCodeAttribute ) status = int(status) if status is not None else None if status in (401, 403): raise fehler_fuer_status( status, "Keine Berechtigung für %s %s (HTTP %s)%s" % (method, path, status, self._error_detail(content)), ) if status is not None and status >= 400: raise ApiError( "%s %s lieferte HTTP %s%s" % (method, path, status, self._error_detail(content)), status, ) if error != QgsBlockingNetworkRequest.NoError: raise ApiError( "%s %s fehlgeschlagen: %s%s" % (method, path, blocking.errorMessage(), self._error_detail(content)), status, ) if not content: return None try: return json.loads(content.decode("utf-8")) except ValueError: raise ApiError( "Antwort von %s ist kein gültiges JSON." % path ) @staticmethod def _error_detail(content): """Liest title/detail aus einer RFC-7807-Fehlerantwort.""" if not content: return "" try: problem = json.loads(content.decode("utf-8")) parts = [ str(problem[k]) for k in ("title", "detail") if problem.get(k) ] if parts: return " — " + ": ".join(parts) except (ValueError, AttributeError): pass return " — " + content[:300].decode("utf-8", "replace")