109 lines
4.2 KiB
Python
109 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Add monitors to Uptime Kuma via Socket.IO.
|
|
Runs on the host, connects directly to Kuma container's IP.
|
|
Token read from .env - never exposed to chat."""
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
import subprocess
|
|
|
|
try:
|
|
import socketio
|
|
except ImportError:
|
|
venv = os.path.join(os.path.dirname(__file__), ".kuma_venv")
|
|
if not os.path.exists(venv):
|
|
print("Creating virtual environment...")
|
|
subprocess.check_call([sys.executable, "-m", "venv", venv])
|
|
pip = os.path.join(venv, "bin", "pip")
|
|
print("Installing python-socketio[client]...")
|
|
subprocess.check_call([pip, "install", "-q", "python-socketio[client]"])
|
|
python = os.path.join(venv, "bin", "python")
|
|
os.execv(python, [python] + sys.argv)
|
|
|
|
import socketio
|
|
|
|
# Read token from .env
|
|
token = None
|
|
env_file = os.path.join(os.path.dirname(__file__), ".env")
|
|
with open(env_file) as f:
|
|
for line in f:
|
|
if line.startswith("UPTIME_KUMA_API_KEY="):
|
|
token = line.split("=", 1)[1].strip()
|
|
break
|
|
|
|
if not token:
|
|
print("ERROR: UPTIME_KUMA_API_KEY not found in .env")
|
|
sys.exit(1)
|
|
|
|
# Find Kuma's container IP
|
|
result = subprocess.run(
|
|
["docker", "inspect", "uptime-kuma", "-f", "{{.NetworkSettings.Networks.mydocker_default.IPAddress}}"],
|
|
capture_output=True, text=True
|
|
)
|
|
kuma_ip = result.stdout.strip()
|
|
kuma_url = f"http://{kuma_ip}:3001"
|
|
|
|
MONITORS = [
|
|
{"name": "Traefik", "type": "https", "url": "https://traefik.sebson.space"},
|
|
{"name": "Home Assistant", "type": "https", "url": "https://ha.sebson.space"},
|
|
{"name": "Grafana", "type": "https", "url": "https://grafana.sebson.space"},
|
|
{"name": "Bitwarden", "type": "https", "url": "https://bward.sebson.space"},
|
|
{"name": "Immich", "type": "https", "url": "https://immch.sebson.space"},
|
|
{"name": "Jellyfin", "type": "https", "url": "https://jfin.sebson.space"},
|
|
{"name": "Prometheus", "type": "https", "url": "https://prom.sebson.space"},
|
|
{"name": "HA Dashboard", "type": "https", "url": "https://hadash.sebson.space"},
|
|
{"name": "Brana Frontend", "type": "https", "url": "https://brana.sebson.space"},
|
|
{"name": "MikroTik", "type": "ping", "hostname": "192.168.1.1"},
|
|
{"name": "Google DNS", "type": "ping", "hostname": "8.8.8.8"},
|
|
{"name": "Frigate", "type": "http", "url": "http://acemagic:5000"},
|
|
{"name": "Ollama", "type": "http", "url": "http://acemagic:11434"},
|
|
{"name": "Gotify", "type": "http", "url": "http://acemagic:86"},
|
|
{"name": "Dozzle", "type": "http", "url": "http://acemagic:8080"},
|
|
{"name": "OpenSpeedTest", "type": "http", "url": "http://acemagic:3000"},
|
|
{"name": "Gitea", "type": "http", "url": "http://acemagic:3002"},
|
|
{"name": "PostgreSQL", "type": "tcp", "hostname": "acemagic", "port": 5432},
|
|
{"name": "Mosquitto", "type": "tcp", "hostname": "acemagic", "port": 1883},
|
|
{"name": "Redis", "type": "tcp", "hostname": "acemagic", "port": 6379},
|
|
]
|
|
|
|
sio = socketio.Client(logger=False, engineio_logger=False, reconnection=False)
|
|
print(f"Connecting to Kuma at {kuma_url}...")
|
|
sio.connect(kuma_url, wait_timeout=10)
|
|
print("Connected via Socket.IO")
|
|
|
|
r = sio.call("loginByToken", {"token": token}, timeout=10)
|
|
if r.get("ok"):
|
|
print("API token accepted")
|
|
else:
|
|
print(f"Token rejected: {r}")
|
|
sio.disconnect()
|
|
sys.exit(1)
|
|
|
|
print(f"Adding {len(MONITORS)} monitors...")
|
|
added = 0
|
|
for m in MONITORS:
|
|
p = {"name": m["name"], "type": m["type"], "interval": 30, "maxretries": 3, "retryInterval": 15, "notificationIDList": {}, "upsideDown": False}
|
|
if m["type"] in ("https", "http"):
|
|
p["url"] = m["url"]
|
|
if m["type"] == "https":
|
|
p["expiryNotification"] = True
|
|
elif m["type"] == "ping":
|
|
p["hostname"] = m["hostname"]
|
|
elif m["type"] == "tcp":
|
|
p["hostname"] = m["hostname"]
|
|
p["port"] = m["port"]
|
|
try:
|
|
r = sio.call("add", p, timeout=10)
|
|
if r.get("ok") or "Successfully" in str(r.get("msg", "")):
|
|
print(f" OK {m['name']} (id={r.get('monitorID', '?')})")
|
|
added += 1
|
|
else:
|
|
print(f" ? {m['name']} -> {r}")
|
|
except Exception as e:
|
|
print(f" FAIL {m['name']} -> {e}")
|
|
time.sleep(0.2)
|
|
|
|
sio.disconnect()
|
|
print(f"\nDone. {added}/{len(MONITORS)} added.")
|