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