Einstellungen: Gitea-Actions-Variablen per API übernehmen (Token nur für den Abruf)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

app/services/gitea_import.py

@@ -0,0 +1,96 @@
+"""Einstellungen aus den Gitea-Actions-Variablen übernehmen (statt abzuschreiben).
+
+Liest über die Gitea-API die Repo- (und Organisations-)Variablen – Klartext, nie Secrets – und ordnet sie den Einstellungen der Admin-Plattform zu.
+Der Token wird nur für diesen Abruf genutzt: nicht gespeichert, nicht protokolliert, nicht im Audit-Log.
+"""
+from __future__ import annotations
+
+import httpx
+
+from app.models import Setting
+from app.services import contract
+
+# Gitea-Variable → Einstellung der Admin-Plattform (Rest ergibt sich aus dem Env-Schema des Cores)
+SPECIAL = {"DOMAIN": "base_domain", "PROJECT_ID": "mittwald_project_id", "STACK_ID": "mittwald_stack_id", "REGISTRY": "registry"}
+IGNORED = {"REGISTRY_HOST": "nur für den Docker-Login der CI", "ADMIN_URL": "gehört zur Core-CI (Release-Meldung), nicht in die Einstellungen"}
+SECRETISH = ("KEY", "SECRET", "TOKEN", "PASSWORD")
+
+
+class GiteaError(RuntimeError):
+    pass
+
+
+def _real_get(url: str, token: str):
+    r = httpx.get(url, headers={"Authorization": f"token {token}"}, timeout=20, follow_redirects=True)
+    try:
+        data = r.json()
+    except ValueError:
+        data = None
+    return r.status_code, data
+
+
+def _fetch(base: str, path: str, token: str, get) -> list[dict] | None:
+    out, page = [], 1
+    while True:
+        status, data = get(f"{base}/api/v1{path}?page={page}&limit=50", token)
+        if status in (404, 405):
+            return None                                   # Endpunkt/Scope nicht vorhanden
+        if status in (401, 403):
+            raise GiteaError(f"Gitea verweigert den Zugriff (HTTP {status}) – Token braucht Lesezugriff auf das Repository/die Organisation.")
+        if status != 200 or not isinstance(data, list):
+            raise GiteaError(f"Unerwartete Antwort von Gitea (HTTP {status}).")
+        out += data
+        if len(data) < 50:
+            return out
+        page += 1
+
+
+def fetch_variables(base_url: str, repo: str, token: str, get=_real_get) -> dict[str, str]:
+    """{NAME: Wert}. Repo-Variablen überschreiben Organisations-Variablen."""
+    base = base_url.rstrip("/")
+    if "/" not in repo.strip("/"):
+        raise GiteaError("Repo als „Besitzer/Name“ angeben (z. B. Onyva/transkriptipt-platform).")
+    owner = repo.strip("/").split("/")[0]
+    merged: dict[str, str] = {}
+    found_any = False
+    for path in (f"/orgs/{owner}/actions/variables", f"/repos/{repo.strip('/')}/actions/variables"):
+        items = _fetch(base, path, token, get)
+        if items is None:
+            continue
+        found_any = True
+        for it in items:
+            merged[str(it.get("name") or it.get("Name"))] = str(it.get("data", it.get("Data", "")))
+    if not found_any:
+        raise GiteaError("Die Gitea-Variablen-API ist nicht verfügbar (ältere Gitea-Version oder fehlender Scope) – bitte von Hand eintragen.")
+    return merged
+
+
+def setting_for(name: str) -> str | None:
+    if name in SPECIAL:
+        return SPECIAL[name]
+    for e in contract.env_spec():
+        if e["name"] == name and e["kind"] == "setting" and e["scope"] == "shared":
+            return e.get("setting") or name.lower()
+    return None
+
+
+def preview(db, variables: dict[str, str]) -> list[dict]:
+    """Zeilen: {name, setting, value, current, status: neu|geändert|gleich|ignoriert|unbekannt, note}."""
+    rows = []
+    for name, value in sorted(variables.items()):
+        masked = any(s in name for s in SECRETISH)
+        if name in IGNORED:
+            rows.append({"name": name, "setting": None, "value": "", "current": "", "status": "ignoriert", "note": IGNORED[name]})
+            continue
+        key = setting_for(name)
+        if key is None:
+            rows.append({"name": name, "setting": None, "value": "", "current": "", "status": "unbekannt",
+                         "note": "wird vom aktuellen Core nicht gelesen (Altlast) – nicht übernommen"})
+            continue
+        cur = db.get(Setting, key)
+        cur_v = cur.value if cur else ""
+        value = value.strip()
+        status = "gleich" if cur_v == value else ("neu" if not cur_v else "geändert")
+        rows.append({"name": name, "setting": key, "value": "••••" if masked else value, "raw": value if not masked else None,
+                     "current": "••••" if (masked and cur_v) else cur_v, "status": status, "note": ""})
+    return rows