erikandClaude Opus 5
3486d3a74e
Drei Fehler, die erst der erweiterte Headless-Test gegen die echte Schnittstelle sichtbar gemacht hat (11.09.2026, alle drei Anbieter plus Vererbung und Stil): 1. WMS-Layer kamen als "Provider is not valid" zurück. Der Rasterprovider erwartet seine Parameter mit "&" getrennt statt mit Leerzeichen wie die Vektorprovider — und "styles" muss dabei sein, auch leer. Beides durchgemessen am wms_geosn_dop-rgb: ohne "styles" ungültig, mit "&"+"styles=" gültig in 0,2 s. Ein fertiger URI wird durchgereicht. 2. "Dienste holen" lieferte bis zu eine Minute lang die alte Liste. Die Schnittstelle antwortet mit "Cache-Control: private, max-age=60", damit ein Ladelauf nicht alles neu holt — bei einem Klick des Nutzers ist das aber falsch: wer gerade im Manager einen Dienst eingetragen hat, will ihn sehen. dienste_laden und stil_anwenden fragen jetzt frisch. 3. Gesteuert wird das über den forceRefresh-Parameter von QgsBlockingNetworkRequest.get/post, NICHT über CacheLoadControlAttribute am Request: das Attribut wird intern überschrieben und blieb wirkungslos. Der Test in tools/headless_dienste_test.py deckt jetzt auch ab, was das Plugin bisher nicht sah, weil alle 44 Altdaten ArcGIS-REST sind: WMS und WFS bauen, Ausblenden wirkt und lässt sich zurücknehmen, ein Stil bleibt beim Ändern ohne QML stehen und lässt sich entfernen. Angelegte Proben räumt der Lauf selbst wieder ab. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
230 lines
8.5 KiB
Python
230 lines
8.5 KiB
Python
"""HTTP-Client für die VLN-Manager-API (Karten-Plugin map.php + Stammdaten).
|
||
|
||
API-Vertrag laut Doku im Wissensdatenbank-Vault
|
||
("API.md", "API – Karten (map.php).md", "API – Schnittstellen (Referenz).md",
|
||
Stand 2026-06-08):
|
||
|
||
Basis-URL: https://api.flurneuordnung-sachsen.de/v2
|
||
|
||
POST /person/login
|
||
Body: {"mail": "...", "password": "..."}
|
||
Antwort: {"data": {"userauth": "<url_token>", "id": ...}}
|
||
Das userauth-Token ist der API-Key (= PERSON.url_token).
|
||
|
||
GET /tgen -> Teilnehmergemeinschaften (Verfahren)
|
||
|
||
GET /maps/<layer>/{vkz} -> GeoJSON FeatureCollection (EPSG:25833)
|
||
PUT /maps/<layer>/{vkz} <- GeoJSON (ersetzt den Layer dieser VKZ)
|
||
Antwort: {"data": {"vkz", "layer", "art", "speicher_id"}, "status": "ok"}
|
||
|
||
Layer: umringe, p41, st, kas, we
|
||
Auth nach Login über den Header "X-API-Key".
|
||
Fehlerformat: RFC 7807 (application/problem+json).
|
||
|
||
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."""
|
||
|
||
|
||
class AuthError(ApiError):
|
||
"""API-Key fehlt, ist abgelaufen oder hat keine Berechtigung (401/403)."""
|
||
|
||
|
||
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_verfahren(self):
|
||
"""Liste der Verfahren (TGs) als [{"vkz": ..., "name": ...}, ...],
|
||
sortiert nach VKZ. Quelle: GET /tgen."""
|
||
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 _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 AuthError(
|
||
"Keine Berechtigung für %s %s (HTTP %s)%s"
|
||
% (method, path, status, self._error_detail(content))
|
||
)
|
||
if error != QgsBlockingNetworkRequest.NoError:
|
||
raise ApiError(
|
||
"%s %s fehlgeschlagen: %s%s"
|
||
% (method, path, blocking.errorMessage(), 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))
|
||
)
|
||
|
||
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")
|