100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
"""Arcane und Docker auf der Arcane-VM (Phase 3/4, 24.09.2026).
|
|
|
|
Arcane (getarcaneapp/arcane) verwaltet die Docker-Container der VM (NerdQuiz, Rippy …) und hat eine
|
|
eigene Schnittstelle (OpenAPI unter /api/openapi.json). Öffentlich ist nur die eigene Version
|
|
(/api/app-version mit currentVersion, newestVersion, updateAvailable). Alles über Images braucht einen
|
|
API-Schlüssel (Kopfzeile X-API-Key), den man in Arcane anlegt — MC_ARCANE_KEY in /etc/mc2/homelab.env.
|
|
|
|
Docker-Updates laufen über Arcanes eigenen Updater (POST /environments/{id}/updater/run). User-Entscheid
|
|
24.09.: zuerst nur als Probelauf (dryRun); echte Updates erst mit MC_ARCANE_ECHT=1.
|
|
"""
|
|
|
|
import os
|
|
import threading
|
|
import time
|
|
|
|
import httpx
|
|
|
|
ZEITLIMIT = httpx.Timeout(20.0, connect=5.0)
|
|
CACHE_S = 120
|
|
_cache: dict[str, tuple[float, object]] = {}
|
|
_lock = threading.Lock()
|
|
|
|
|
|
def _gemerkt(schluessel: str, holen):
|
|
"""Arcane merkt sich den Update-Stand selbst; die Übersicht fragt ihn höchstens alle zwei Minuten ab."""
|
|
jetzt = time.time()
|
|
with _lock:
|
|
if (eintrag := _cache.get(schluessel)) and jetzt - eintrag[0] < CACHE_S:
|
|
return eintrag[1]
|
|
wert = holen()
|
|
with _lock:
|
|
_cache[schluessel] = (jetzt, wert)
|
|
return wert
|
|
|
|
|
|
def schluessel() -> str:
|
|
return os.environ.get("MC_ARCANE_KEY", "").strip()
|
|
|
|
|
|
def echt() -> bool:
|
|
return os.environ.get("MC_ARCANE_ECHT", "") == "1"
|
|
|
|
|
|
def app_version(url: str) -> dict | None:
|
|
"""Arcanes eigene Version (öffentlich): {"installiert", "neu", "update"} oder None."""
|
|
return _gemerkt(f"version:{url}", lambda: _app_version(url))
|
|
|
|
|
|
def _app_version(url: str) -> dict | None:
|
|
try:
|
|
d = httpx.get(f"{url}/api/app-version", timeout=ZEITLIMIT).json()
|
|
except (httpx.HTTPError, ValueError):
|
|
return None
|
|
if not isinstance(d, dict) or not d.get("currentVersion"):
|
|
return None
|
|
return {"installiert": str(d["currentVersion"]).lstrip("v"), "neu": str(d.get("newestVersion") or "").lstrip("v") or None,
|
|
"update": bool(d.get("updateAvailable"))}
|
|
|
|
|
|
def _get(url: str, pfad: str) -> object:
|
|
r = httpx.get(f"{url}/api{pfad}", headers={"X-API-Key": schluessel()}, timeout=ZEITLIMIT)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def umgebungen(url: str) -> list[dict]:
|
|
daten = _get(url, "/environments?limit=50")
|
|
return [e for e in (daten or {}).get("data") or [] if isinstance(e, dict) and e.get("id") is not None]
|
|
|
|
|
|
def images_mit_update(url: str) -> list[dict]:
|
|
"""Images mit einem neueren Stand in der Registry, über alle Umgebungen: {"name", "aktuell", "neu", "benutzt"}."""
|
|
return _gemerkt(f"images:{url}", lambda: _images_mit_update(url))
|
|
|
|
|
|
def _images_mit_update(url: str) -> list[dict]:
|
|
ergebnis = []
|
|
for umgebung in umgebungen(url):
|
|
daten = _get(url, f"/environments/{umgebung['id']}/images?limit=500")
|
|
for image in (daten or {}).get("data") or []:
|
|
info = image.get("updateInfo") or {}
|
|
if not info.get("hasUpdate"):
|
|
continue
|
|
name = f"{image.get('repo') or '?'}:{image.get('tag') or 'latest'}"
|
|
ergebnis.append({"name": name, "aktuell": info.get("currentVersion") or None,
|
|
"neu": info.get("latestVersion") or None, "umgebung": umgebung["id"],
|
|
"benutzt": [str(u.get("name") if isinstance(u, dict) else u) for u in image.get("usedBy") or []]})
|
|
return ergebnis
|
|
|
|
|
|
def updater(url: str, probelauf: bool) -> dict:
|
|
"""Arcanes Updater für alle Umgebungen. Rückgabe: je Umgebung das Ergebnis (geprüft, aktualisiert, Fehler)."""
|
|
ergebnisse = {}
|
|
for umgebung in umgebungen(url):
|
|
r = httpx.post(f"{url}/api/environments/{umgebung['id']}/updater/run", headers={"X-API-Key": schluessel()},
|
|
json={"dryRun": probelauf}, timeout=httpx.Timeout(600.0, connect=5.0))
|
|
r.raise_for_status()
|
|
ergebnisse[str(umgebung.get("name") or umgebung["id"])] = (r.json() or {}).get("data") or {}
|
|
return ergebnisse
|