Add Gitea health check

services/gitea.py

@@ -0,0 +1,88 @@
+import json
+import ssl
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+
+def _request(url: str, headers: dict[str, str], verify_tls: bool) -> tuple[int, str]:
+    context = None
+    if not verify_tls:
+        context = ssl._create_unverified_context()  # nosec - config-controlled
+
+    req = Request(url, headers=headers)
+    try:
+        with urlopen(req, timeout=10, context=context) as resp:
+            body = resp.read().decode("utf-8")
+            return int(resp.status), body
+    except HTTPError as e:
+        try:
+            body = e.read().decode("utf-8")
+        except Exception:
+            body = ""
+        return int(e.code), body
+    except URLError as e:
+        raise RuntimeError(str(e.reason)) from e
+
+
+def _api_base(cfg: dict[str, Any]) -> str:
+    g_cfg = cfg.get("gitea", {})
+    base = (g_cfg.get("base_url") or "").rstrip("/")
+    return base
+
+
+def get_gitea_health(cfg: dict[str, Any]) -> str:
+    g_cfg = cfg.get("gitea", {})
+    base = _api_base(cfg)
+    verify_tls = g_cfg.get("verify_tls", True)
+    if not base:
+        return "⚠️ Gitea base_url not configured"
+
+    token = (g_cfg.get("token") or "").strip()
+    headers = {"User-Agent": "tg-admin-bot"}
+    if token:
+        headers["Authorization"] = f"token {token}"
+
+    lines = ["🍵 Gitea\n"]
+
+    health_paths = ["/api/healthz", "/api/v1/healthz"]
+    health_status = None
+    health_payload = None
+    for path in health_paths:
+        status, body = _request(f"{base}{path}", headers, verify_tls)
+        if status == 200:
+            health_status = (status, path)
+            try:
+                health_payload = json.loads(body)
+            except json.JSONDecodeError:
+                health_payload = None
+            break
+        if status not in (404, 405):
+            health_status = (status, path)
+            break
+
+    if health_status:
+        status, path = health_status
+        icon = "🟢" if status == 200 else "🔴"
+        if status == 200 and isinstance(health_payload, dict):
+            state = health_payload.get("status") or "ok"
+            checks = health_payload.get("checks") or {}
+            checks_total = len(checks) if isinstance(checks, dict) else 0
+            lines.append(f"{icon} API health: {state} ({checks_total} checks)")
+        else:
+            lines.append(f"{icon} API health: {status} ({path})")
+    else:
+        lines.append("🟡 API health: endpoint not found")
+
+    ver_status, ver_body = _request(f"{base}/api/v1/version", headers, verify_tls)
+    if ver_status == 200:
+        try:
+            payload = json.loads(ver_body)
+        except json.JSONDecodeError:
+            payload = {}
+        version = payload.get("version") or "unknown"
+        lines.append(f"ℹ️ Version: {version}")
+    else:
+        lines.append(f"🟡 Version: HTTP {ver_status}")
+
+    return "\n".join(lines)