57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
"""Uptime Kuma monitor status via SSH+sqlite3."""
|
|
|
|
import subprocess
|
|
from fastapi import APIRouter
|
|
|
|
router = APIRouter(tags=["kuma"])
|
|
|
|
KUMA_HOST = "pi-5"
|
|
KUMA_CONTAINER = "uptime-kuma"
|
|
|
|
|
|
def _kuma_query(sql: str) -> str:
|
|
"""Run a sqlite3 query against Uptime Kuma's database via SSH."""
|
|
result = subprocess.run(
|
|
["ssh", "-o", "ConnectTimeout=3", KUMA_HOST,
|
|
f'docker exec {KUMA_CONTAINER} sqlite3 /app/data/kuma.db "{sql}"'],
|
|
capture_output=True, text=True, timeout=15)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(result.stderr.strip())
|
|
return result.stdout.strip()
|
|
|
|
|
|
@router.get("/kuma/monitors")
|
|
def kuma_monitors():
|
|
"""List all Uptime Kuma monitors with status."""
|
|
try:
|
|
rows = _kuma_query(
|
|
"SELECT m.id, m.name, m.type, m.active, m.url, m.hostname, m.parent, "
|
|
"COALESCE((SELECT h.status FROM heartbeat h WHERE h.monitor_id=m.id "
|
|
"ORDER BY h.time DESC LIMIT 1), -1) as last_status "
|
|
"FROM monitor m ORDER BY m.parent, m.name"
|
|
)
|
|
if not rows:
|
|
return {"monitors": [], "total": 0, "up": 0, "down": 0}
|
|
|
|
monitors = []
|
|
for row in rows.splitlines():
|
|
parts = row.split("|")
|
|
if len(parts) < 8:
|
|
continue
|
|
mid, name, mtype, active, url, hostname, parent, status = parts[:8]
|
|
monitors.append({
|
|
"id": int(mid),
|
|
"name": name,
|
|
"type": mtype,
|
|
"active": active == "1",
|
|
"url": url or hostname or "",
|
|
"parent": int(parent) if parent and parent != "" else None,
|
|
"status": int(status), # 1=up, 0=down, -1=unknown
|
|
})
|
|
|
|
up = sum(1 for m in monitors if m["status"] == 1 and m["active"])
|
|
down = sum(1 for m in monitors if m["status"] == 0 and m["active"])
|
|
return {"monitors": monitors, "total": len(monitors), "up": up, "down": down}
|
|
except Exception as e:
|
|
return {"monitors": [], "error": str(e)}
|