API_Karte_QGISDemo/vln_karten/api_client.py

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