mrcharles
d237e4a65d
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.
112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
#!/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()
|