feat(monitoring): kuma notification tooling; app webhooks replace pushover [#435]

kuma-notifications.py (list/delete/usage). Deleted unused Pushover provider 'KNEL Alerts' per Charles — 211/211 monitors already on the ultix-sidecar + Ultix-mini app webhooks. fleet-sync creation payload updated to the live notification IDs.
2026-09-03 11:38:38 -05:00
parent 4c2726b116
commit d237e4a65d
2 changed files with 112 additions and 1 deletions
@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""kuma-notifications.py — Uptime Kuma notification provider management [#435]
+
+List, delete, and inspect notification providers over the socket.io API.
+Read-only by default; --delete mutates (deleting a notification detaches it
+from all monitors server-side).
+
+Env (from ~/.creds/uptime-kuma.env): UPTIME_KUMA_URL, UPTIME_KUMA_TOKEN
+
+Usage:
+  python3 kuma-notifications.py list
+  python3 kuma-notifications.py delete <id>
+  python3 kuma-notifications.py monitor-usage <id>   # monitors still referencing it
+"""
+import argparse
+import json
+import os
+import time
+
+import socketio
+
+
+def kuma_connect():
+    sio = socketio.Client(reconnection=False)
+    state = {}
+
+    @sio.on("notificationList")
+    def _nl(d):
+        state["nl"] = d
+
+    sio.connect(
+        os.environ["UPTIME_KUMA_URL"], transports=["websocket"], wait_timeout=30
+    )
+    r = sio.call("loginByToken", os.environ["UPTIME_KUMA_TOKEN"], timeout=30)
+    if not (isinstance(r, dict) and r.get("ok")):
+        raise SystemExit(f"login failed: {r}")
+    sio.emit("getNotifications")
+    for _ in range(25):
+        if "nl" in state:
+            break
+        time.sleep(1)
+    return sio, state.get("nl", [])
+
+
+def get_monitor_list(sio):
+    state = {}
+
+    @sio.on("monitorList")
+    def _ml(d):
+        state["ml"] = d
+
+    sio.emit("monitorList")
+    for _ in range(25):
+        if "ml" in state:
+            break
+        time.sleep(1)
+    return state.get("ml", {})
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("cmd", choices=["list", "delete", "monitor-usage"])
+    ap.add_argument("id", nargs="?", type=int)
+    ap.add_argument("--yes", action="store_true", help="skip confirmation on delete")
+    args = ap.parse_args()
+
+    sio, notifs = kuma_connect()
+    try:
+        if args.cmd == "list":
+            for n in notifs:
+                print(
+                    f'{n.get("id")}\t{n.get("name")}\ttype={n.get("type")}\t'
+                    f'active={n.get("active")}'
+                )
+            if not notifs:
+                print("(no notification providers configured)")
+            return
+
+        if args.id is None:
+            raise SystemExit("id required for this subcommand")
+
+        if args.cmd == "monitor-usage":
+            mons = get_monitor_list(sio)
+            users = [
+                f'{m.get("id")}:{m.get("name")}'
+                for m in mons.values()
+                if (m.get("notificationIDList") or {}).get(str(args.id))
+            ]
+            print(f"{len(users)} monitors reference notification {args.id}")
+            for u in users:
+                print(" ", u)
+            return
+
+        if args.cmd == "delete":
+            target = next((n for n in notifs if n.get("id") == args.id), None)
+            if not target:
+                raise SystemExit(f"no notification with id {args.id}")
+            if not args.yes:
+                print(f"refusing to delete {target.get('name')} without --yes")
+                return
+            r = sio.call("deleteNotification", args.id, timeout=30)
+            ok = isinstance(r, dict) and r.get("ok")
+            print(f"delete notification {args.id} ({target.get('name')}): "
+                  f"{'OK' if ok else json.dumps(r)}")
+            return
+    finally:
+        sio.disconnect()
+
+
+if __name__ == "__main__":
+    main()