9fd5dbb0df
All checks were successful
Validation Dashboard / Tests backend et build frontend (push) Successful in 58s
117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
||
import json
|
||
import os
|
||
import secrets
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
|
||
GITEA_URL = os.environ.get("GITEA_URL", "https://git.recette.santinova-soft.org").rstrip("/")
|
||
GITEA_USER = os.environ.get("GITEA_USERNAME", "manus-admin")
|
||
DASHBOARD_URL = os.environ.get("DASHBOARD_URL", "https://dashboard-santinova.recette.santinova-soft.org").rstrip("/")
|
||
DASHBOARD_APP_DIR = Path(os.environ.get("DASHBOARD_APP_DIR", "/opt/manus-deploy/apps/manus-dashboard"))
|
||
ENV_FILE = DASHBOARD_APP_DIR / ".env"
|
||
WEBHOOK_ENDPOINT = f"{DASHBOARD_URL}/api/webhook/gitea"
|
||
|
||
|
||
def read_env() -> dict[str, str]:
|
||
values: dict[str, str] = {}
|
||
if ENV_FILE.exists():
|
||
for line in ENV_FILE.read_text(encoding="utf-8").splitlines():
|
||
if line and not line.lstrip().startswith("#") and "=" in line:
|
||
key, value = line.split("=", 1)
|
||
values[key.strip()] = value.strip()
|
||
return values
|
||
|
||
|
||
def update_env(key: str, value: str) -> None:
|
||
lines = ENV_FILE.read_text(encoding="utf-8").splitlines() if ENV_FILE.exists() else []
|
||
updated: list[str] = []
|
||
found = False
|
||
for line in lines:
|
||
if line.startswith(f"{key}="):
|
||
updated.append(f"{key}={value}")
|
||
found = True
|
||
else:
|
||
updated.append(line)
|
||
if not found:
|
||
updated.append(f"{key}={value}")
|
||
temporary = ENV_FILE.with_suffix(".env.next")
|
||
temporary.write_text("\n".join(updated).rstrip() + "\n", encoding="utf-8")
|
||
temporary.chmod(0o600)
|
||
temporary.replace(ENV_FILE)
|
||
|
||
|
||
def api(token: str, method: str, path: str, payload: dict | None = None):
|
||
data = json.dumps(payload).encode() if payload is not None else None
|
||
request = urllib.request.Request(
|
||
f"{GITEA_URL}/api/v1{path}",
|
||
method=method,
|
||
data=data,
|
||
headers={"Authorization": f"token {token}", "Content-Type": "application/json"},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=20) as response:
|
||
body = response.read()
|
||
return json.loads(body) if body else None
|
||
except urllib.error.HTTPError as error:
|
||
detail = error.read().decode(errors="replace")
|
||
raise RuntimeError(f"Gitea {method} {path} : HTTP {error.code} {detail[:300]}") from error
|
||
|
||
|
||
def ensure_dashboard_secret(values: dict[str, str]) -> tuple[dict[str, str], bool]:
|
||
if values.get("WEBHOOK_SECRET"):
|
||
return values, False
|
||
update_env("WEBHOOK_SECRET", secrets.token_hex(32))
|
||
return read_env(), True
|
||
|
||
|
||
def main() -> None:
|
||
values, generated = ensure_dashboard_secret(read_env())
|
||
token = values.get("GITEA_TOKEN", "")
|
||
secret = values.get("WEBHOOK_SECRET", "")
|
||
if not token or not secret:
|
||
raise RuntimeError("GITEA_TOKEN et WEBHOOK_SECRET sont requis dans le .env serveur.")
|
||
if generated:
|
||
subprocess.run(["docker", "compose", "up", "-d", "--force-recreate", "dashboard"], cwd=DASHBOARD_APP_DIR, check=True)
|
||
time.sleep(8)
|
||
identity = api(token, "GET", "/user")
|
||
if identity.get("login") != GITEA_USER:
|
||
raise RuntimeError("Le jeton Gitea n’appartient pas au compte attendu.")
|
||
repositories = api(token, "GET", "/user/repos?limit=100&page=1") or []
|
||
configured = 0
|
||
for repository in repositories:
|
||
owner = repository.get("owner", {}).get("login")
|
||
name = repository.get("name")
|
||
if owner != GITEA_USER or not name:
|
||
continue
|
||
hooks = api(token, "GET", f"/repos/{owner}/{name}/hooks") or []
|
||
existing = next((hook for hook in hooks if hook.get("config", {}).get("url") == WEBHOOK_ENDPOINT), None)
|
||
payload = {
|
||
"active": True,
|
||
"branch_filter": "main",
|
||
"config": {"url": WEBHOOK_ENDPOINT, "content_type": "json", "secret": secret},
|
||
"events": ["push", "workflow_run"],
|
||
}
|
||
if existing:
|
||
api(token, "PATCH", f"/repos/{owner}/{name}/hooks/{existing['id']}", payload)
|
||
else:
|
||
api(token, "POST", f"/repos/{owner}/{name}/hooks", {"type": "gitea", **payload})
|
||
configured += 1
|
||
print(f"WEBHOOK_OK {owner}/{name}")
|
||
if configured == 0:
|
||
raise RuntimeError("Aucun dépôt Gitea n’a été configuré.")
|
||
print(f"GITEA_WEBHOOKS_CONFIGURED {configured}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except Exception as exc:
|
||
print(f"GITEA_WEBHOOK_SETUP_FAILED: {exc}", file=sys.stderr)
|
||
raise SystemExit(1)
|