diff --git a/scripts/kuma-fleet-sync.py b/scripts/kuma-fleet-sync.py index abea618..58e8a6d 100644 --- a/scripts/kuma-fleet-sync.py +++ b/scripts/kuma-fleet-sync.py @@ -141,7 +141,7 @@ def main(): "retryInterval": 60, "resendInterval": 0, "maxretries": 2, - "notificationIDList": {"1": True, "2": True}, + "notificationIDList": {"2": True, "3": True}, # app webhooks (Pushover retired 2026-09-03) "upsideDown": False, "description": "ICMP up/down (kuma-fleet-sync)", "httpBodyEncoding": "json", diff --git a/scripts/kuma-notifications.py b/scripts/kuma-notifications.py new file mode 100644 index 0000000..e991dcd --- /dev/null +++ b/scripts/kuma-notifications.py @@ -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 + python3 kuma-notifications.py monitor-usage # 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()