97 lines
4.3 KiB
Python
97 lines
4.3 KiB
Python
"""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
|