KAIandClaude Opus 5
d79dff1093
Der Server liefert seit heute die Sammlung widmung-<vkz>-punkt|-linie mit einem Stil, der
die Schilder selbst malt, dazu blattzeichen-<vkz> mit den neuen Zeichen am Weg. Das Plugin
kannte beides nicht -- die Gruppenzuordnung fiel auf "Weitere", und die Seitenleiste
filterte das heraus.
* GRUPPEN kennt jetzt widmung ("Widmungen & Verkehrszeichen", eigener Haken in der
Seitenleiste) und blattzeichen (laeuft in der Gruppe der Maßnahmenblaetter mit).
* Neues Modul zeichen_katalog.py: holt /v2/zeichen/katalog.json (oeffentlich, wird neben den
Grafiken abgelegt, damit die Palette auch ohne Netz oeffnet), laedt fehlende Schilder nach
<Profil>/svg/vln/<code>.svg, traegt den Ordner in die SVG-Suchpfade ein und schreibt im
Stil das Adressen-Praefix auf "vln/" um. Fehlt eine Datei, bleibt die Serveradresse stehen
-- so zeichnet QGIS sie weiter aus dem Netz, im Druck und offline stimmt es trotzdem.
* Beim Laden eines Zeichen-Layers werden die Grafiken bereitgestellt, BEVOR der Stil greift;
sonst zeigt der Ausdruck auf eine Datei, die es lokal noch nicht gibt.
* api_client.roh() holt eine Datei roh und ohne Anmeldung; Fehler geben b"" zurueck, ein
fehlendes Schild darf nichts abbrechen.
Headless gemessen (QGIS 4.2, offscreen): Katalog 890 Zeichen, drei Grafiken geholt, ein
unbekannter Code sauber als fehlend gemeldet, Suchpfad eingetragen, Stil lokalisiert.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
252 lines
9.5 KiB
Python
252 lines
9.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 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 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")
|