89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
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)
|