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