mitai-jinkendo/backend/data_layer/tandoor_client.py

"""HTTP client for a user's Tandoor instance. Never logs the token."""
from __future__ import annotations

import json
import ssl
import urllib.error
import urllib.request
from typing import Any
from urllib.parse import urljoin, urlparse


class TandoorError(ValueError):
    pass


def normalize_base_url(raw: str | None) -> str:
    url = (raw or "").strip().rstrip("/")
    if not url:
        raise TandoorError("Tandoor-URL fehlt")
    parsed = urlparse(url)
    if parsed.scheme not in ("https", "http") or not parsed.netloc:
        raise TandoorError("Tandoor-URL muss mit https:// oder http:// beginnen")
    if parsed.scheme == "http" and parsed.hostname not in ("localhost", "127.0.0.1"):
        raise TandoorError("Tandoor-URL muss https verwenden")
    return url


def token_hint(token: str | None) -> str | None:
    t = (token or "").strip()
    if len(t) < 4:
        return "gesetzt" if t else None
    return f"…{t[-4:]}"


def public_settings(row: dict[str, Any] | None) -> dict[str, Any]:
    if not row:
        return {
            "configured": False,
            "base_url": "",
            "token_set": False,
            "token_hint": None,
            "last_ok_at": None,
            "last_error": None,
        }
    return {
        "configured": True,
        "base_url": row.get("base_url") or "",
        "token_set": bool((row.get("api_token") or "").strip()),
        "token_hint": token_hint(row.get("api_token")),
        "last_ok_at": row.get("last_ok_at").isoformat() if row.get("last_ok_at") else None,
        "last_error": row.get("last_error"),
    }


def _request(base_url: str, token: str, path: str, timeout: float = 10.0) -> tuple[int, str, str]:
    url = urljoin(base_url.rstrip("/") + "/", path.lstrip("/"))
    req = urllib.request.Request(
        url,
        headers={
            "Authorization": f"Token {token.strip()}",
            "Accept": "application/json",
            "User-Agent": "Mitai-Jinkendo/tandoor-connector",
        },
        method="GET",
    )
    ctx = ssl.create_default_context()

    class _NoRedirect(urllib.request.HTTPRedirectHandler):
        def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ARG002
            return None

    opener = urllib.request.build_opener(_NoRedirect, urllib.request.HTTPSHandler(context=ctx))
    try:
        with opener.open(req, timeout=timeout) as resp:
            body = resp.read().decode("utf-8", errors="replace")
            ctype = resp.headers.get("Content-Type", "")
            return resp.status, body, ctype
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace") if e.fp else ""
        ctype = e.headers.get("Content-Type", "") if e.headers else ""
        return e.code, body, ctype
    except TimeoutError as e:
        raise TandoorError("Tandoor antwortet nicht (Zeitüberschreitung)") from e
    except urllib.error.URLError as e:
        raise TandoorError(f"Tandoor nicht erreichbar: {e.reason}") from e


def probe_connection(base_url: str, token: str) -> dict[str, Any]:
    url = normalize_base_url(base_url)
    tok = (token or "").strip()
    if not tok:
        raise TandoorError("API-Token fehlt")
    status, body, ctype = _request(url, tok, "api/user/")
    if status in (301, 302, 303, 307, 308) or "text/html" in (ctype or "") and status < 400:
        raise TandoorError("Tandoor hat nicht mit JSON geantwortet — URL oder Token prüfen")
    if status in (401, 403):
        raise TandoorError("Token ungültig oder ohne Rechte")
    if status == 404:
        status, body, ctype = _request(url, tok, "api/recipe/?page=1&page_size=1")
    if status in (401, 403):
        raise TandoorError("Token ungültig oder ohne Rechte")
    if status >= 400:
        raise TandoorError(f"Tandoor antwortete mit HTTP {status}")
    if "json" not in (ctype or "") and not (body or "").lstrip().startswith(("{", "[")):
        raise TandoorError("Tandoor hat nicht mit JSON geantwortet — URL oder Token prüfen")
    try:
        data = json.loads(body) if body else {}
    except json.JSONDecodeError as e:
        raise TandoorError("Tandoor-Antwort war kein JSON") from e
    name = None
    if isinstance(data, dict):
        results = data.get("results")
        if isinstance(results, list) and results and isinstance(results[0], dict):
            name = results[0].get("display_name") or results[0].get("username") or results[0].get("name")
        else:
            name = data.get("display_name") or data.get("username") or data.get("name")
        count = data.get("count")
    else:
        count = None
    return {"ok": True, "user": name, "recipe_count": count}