erik
091ef281b0
Lädt Verfahrensgebiet, Plan 41, Karte alter Stand (KAS) und Wertermittlung (WE) je VKZ vollständig aus KARTE_OBJEKT (Listen-Endpunkt mit Paging) und schreibt sie per PUT zurück. Einmaliger Login (mail/password -> userauth), API-Key persistiert in QSettings; Verfahrens-Auswahl in der Toolbar wird je QGIS-Projekt gemerkt. Gemischte Geometrietypen werden beim Laden in Punkte-/Linien-/Flächen-Layer gesplittet und beim Hochladen wieder vereint. Qt5/Qt6-kompatibel (QGIS 3.22+ und QGIS 4). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
256 lines
9.2 KiB
Python
256 lines
9.2 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_layer_complete(self, layer_key, vkz, page_size=2000):
|
||
"""Lädt ALLE Objekte eines Layers für eine VKZ aus KARTE_OBJEKT.
|
||
|
||
Nutzt wie das Web-GIS (0_map, db_p41) den Listen-Endpunkt
|
||
/maps/<layer>?vkz=…, aber mit limit/offset-Paging, bis keine
|
||
weitere Seite mehr kommt — so ist der Abruf auch bei mehr als
|
||
2000 Objekten garantiert vollständig."""
|
||
from urllib.parse import quote
|
||
|
||
features = []
|
||
offset = 0
|
||
while True:
|
||
data = self._request(
|
||
"GET",
|
||
"/maps/%s?vkz=%s&limit=%d&offset=%d"
|
||
% (quote(str(layer_key)), quote(str(vkz)), page_size, offset),
|
||
)
|
||
if not isinstance(data, dict) or data.get("type") != "FeatureCollection":
|
||
raise ApiError(
|
||
"Antwort von /maps/%s ist keine GeoJSON FeatureCollection."
|
||
% layer_key
|
||
)
|
||
page = data.get("features", [])
|
||
features.extend(page)
|
||
if len(page) < page_size:
|
||
break
|
||
offset += page_size
|
||
return {"type": "FeatureCollection", "features": features}
|
||
|
||
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 save_feature_collection(self, path, feature_collection):
|
||
"""Schreibt eine GeoJSON FeatureCollection zurück an die API
|
||
(PUT ersetzt den kompletten Layer der jeweiligen VKZ)."""
|
||
return self._request("PUT", path, payload=feature_collection)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Intern
|
||
# ------------------------------------------------------------------
|
||
|
||
def _request(self, method, path, payload=None, with_auth=True):
|
||
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")
|
||
|
||
blocking = QgsBlockingNetworkRequest()
|
||
if method == "GET":
|
||
error = blocking.get(request)
|
||
elif method == "POST":
|
||
error = blocking.post(request, body)
|
||
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")
|