93 lines
4 KiB
Python
93 lines
4 KiB
Python
"""Lieferweg von Hugin für Integrationen (/api/v1/delivery, Integrations-Key hi_…).
|
|
|
|
Bifröst nutzt die Endpunkte für Plattformen: resolve (auflösen ohne Übersetzung), source (Version im
|
|
Quellformat) und feedback; dazu packages (Katalog) und usage (Nutzung melden).
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
# Austauschbar für Tests (z. B. httpx.MockTransport oder der TestClient von Hugin)
|
|
transport_factory: Any = None
|
|
|
|
|
|
class HuginError(Exception):
|
|
"""Für Menschen lesbar; geht in Meldungen der Oberfläche oder als Tool-Fehler an die KI."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Source:
|
|
archive: bytes
|
|
checksum: str # aus der Kopfzeile X-Hugin-Checksum
|
|
|
|
|
|
def ready(url: str, timeout: float) -> str:
|
|
"""Erreichbarkeit der Instanz ohne Key (/readyz); Fehlertext oder leer."""
|
|
transport = transport_factory() if transport_factory else None
|
|
try:
|
|
with httpx.Client(timeout=timeout, transport=transport, follow_redirects=False) as http:
|
|
response = http.get(url.rstrip("/") + "/readyz")
|
|
except httpx.HTTPError as e:
|
|
return f"Hugin ist nicht erreichbar ({type(e).__name__})."
|
|
if response.status_code != 200:
|
|
try:
|
|
detail = response.json().get("status", "")
|
|
except ValueError:
|
|
detail = ""
|
|
return f"Hugin meldet {response.status_code}{': ' + str(detail)[:200] if detail else ''}."
|
|
return ""
|
|
|
|
|
|
class Client:
|
|
def __init__(self, url: str, key: str, timeout: float):
|
|
transport = transport_factory() if transport_factory else None
|
|
self.http = httpx.Client(base_url=url.rstrip("/") + "/api/v1/delivery", timeout=timeout,
|
|
headers={"Authorization": f"Bearer {key}"}, transport=transport,
|
|
follow_redirects=False)
|
|
|
|
def __enter__(self) -> "Client":
|
|
return self
|
|
|
|
def __exit__(self, *exc) -> None:
|
|
self.http.close()
|
|
|
|
def request(self, method: str, path: str, **kwargs) -> httpx.Response:
|
|
try:
|
|
response = self.http.request(method, path, **kwargs)
|
|
except httpx.HTTPError as e:
|
|
raise HuginError(f"Hugin ist nicht erreichbar ({type(e).__name__}).") from e
|
|
if response.status_code in (401, 403):
|
|
raise HuginError("Hugin lehnt den Key ab; stimmt der Integrations-Key (hi_…) des Teams?")
|
|
if response.status_code >= 400:
|
|
try:
|
|
detail = response.json().get("detail", "")
|
|
except ValueError:
|
|
detail = ""
|
|
raise HuginError(f"Hugin meldet {response.status_code}: {str(detail)[:300] or 'Fehler'}")
|
|
return response
|
|
|
|
def status(self) -> dict[str, Any]:
|
|
return self.request("GET", "/status").json()
|
|
|
|
def packages(self) -> list[dict[str, Any]]:
|
|
return self.request("GET", "/packages").json()
|
|
|
|
def resolve(self, packages: list[str], installed: dict[str, str]) -> dict[str, Any]:
|
|
return self.request("POST", "/resolve", json={"packages": packages, "installed": installed,
|
|
"mandatory": True}).json()
|
|
|
|
def source(self, package_id: str, version: str, max_bytes: int) -> Source:
|
|
response = self.request("GET", f"/packages/{package_id}/versions/{version}/source")
|
|
if len(response.content) > max_bytes:
|
|
raise HuginError(f"Das Paket {package_id} {version} ist größer als {max_bytes // 1024} KB "
|
|
"(AIW_HUGIN_MAX_PACKAGE_KB).")
|
|
return Source(response.content, response.headers.get("x-hugin-checksum", ""))
|
|
|
|
def feedback(self, package_id: str, version: str, kind: str, text: str) -> dict[str, Any]:
|
|
return self.request("POST", "/feedback", json={"package": package_id, "version": version, "kind": kind,
|
|
"text": text, "client": "ai-workspace"}).json()
|
|
|
|
def usage(self, client_name: str, installed: dict[str, str]) -> None:
|
|
self.request("POST", "/usage", json={"repo": client_name, "installed": installed})
|