PFVCluster/scripts/kuma-maintenance.py

#!/usr/bin/env python3
"""kuma-maintenance.py — Uptime Kuma maintenance windows from the CLI [#769]

Companion to kuma-add-ping.py (same connect pattern). Creates "single"
strategy maintenance windows so change work has a Kuma record (founder
golden rule: any VM reboot ships behind a maintenance window).

Env (from ~/.creds/uptime-kuma.env):
  UPTIME_KUMA_URL, UPTIME_KUMA_TOKEN

Usage:
  python3 kuma-maintenance.py open --title "cm rollback 5109" --minutes 30 \
      [--desc "ticket #769"] [--monitor "monitor name" ...]
  python3 kuma-maintenance.py list
  python3 kuma-maintenance.py delete <maintenance-id>

Prints the maintenance id on open (last line: "id=<n>") for scripting.
"""
import argparse
import datetime
import os
import sys

import socketio


def kuma_connect():
    sio = socketio.Client(reconnection=False)
    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}")
    return sio


def fmt(dt):
    return dt.strftime("%Y-%m-%d %H:%M")


def open_window(args):
    sio = kuma_connect()
    start = datetime.datetime.now() - datetime.timedelta(minutes=1)
    end = start + datetime.timedelta(minutes=args.minutes + 1)
    obj = {
        "title": args.title,
        "description": args.desc or "",
        "strategy": "single",
        "active": True,
        "intervalDay": 1,
        "dateRange": [fmt(start), fmt(end)],
        "timeRange": [],
        "weekdays": [],
        "daysOfMonth": [],
        "timezone": "LOCAL",
    }
    attempts = [
        ("obj", lambda: sio.call("addMaintenance", obj, timeout=30)),
        ("obj+monitors", lambda: sio.call(
            "addMaintenance", obj, args.monitor or [], timeout=30)),
    ]
    result = None
    for label, fn in attempts:
        try:
            result = fn()
        except Exception as exc:  # noqa: BLE001 - probe different API shapes
            result = (False, str(exc))
        if isinstance(result, dict) and result.get("ok"):
            break
        if isinstance(result, (list, tuple)) and result and result[0] is True:
            break
    else:
        print(f"FAIL addMaintenance ({label}): {result}", file=sys.stderr)
        sys.exit(1)
    mid = None
    if isinstance(result, dict):
        mid = result.get("maintenanceID") or result.get("id") or (
            result.get("maintenance") or {}).get("id")
    elif isinstance(result, (list, tuple)):
        body = result[1] if len(result) > 1 else None
        if isinstance(body, dict):
            mid = body.get("maintenanceID") or body.get("id") or (
                body.get("maintenance") or {}).get("id")
    if mid is None and args.monitor:
        # best-effort attach for API shapes that create-then-attach
        try:
            sio.call("addMonitorMaintenance", mid, args.monitor, timeout=30)
        except Exception:  # noqa: BLE001 - attach is best-effort
            pass
    print(f"opened '{args.title}' {fmt(start)} -> {fmt(end)}")
    print(f"id={mid}")
    sio.disconnect()


def list_windows(_args):
    sio = kuma_connect()
    state = {}

    @sio.on("maintenanceList")
    def _ml(d):
        state["ml"] = d

    sio.emit("getMaintenanceList")
    for _ in range(20):
        if "ml" in state:
            break
        sio.sleep(0.5)
    items = state.get("ml") or []
    if isinstance(items, dict):
        items = list(items.values())
    for m in items:
        print(f"id={m.get('id')} active={m.get('active')} "
              f"strategy={m.get('strategy')} title={m.get('title')!r}")
    sio.disconnect()


def delete_window(args):
    sio = kuma_connect()
    attempts = [
        lambda: sio.call("deleteMaintenance", int(args.mid), timeout=30),
        lambda: sio.call("deleteMaintenance", {"id": int(args.mid)}, timeout=30),
    ]
    for fn in attempts:
        try:
            r = fn()
        except Exception as exc:  # noqa: BLE001 - probe different API shapes
            r = (False, str(exc))
        if (isinstance(r, dict) and r.get("ok")) or (
            isinstance(r, (list, tuple)) and r and r[0] is True
        ):
            print(f"deleted {args.mid}")
            sio.disconnect()
            return
    print(f"FAIL delete: {r}", file=sys.stderr)
    sys.exit(1)


def main():
    p = argparse.ArgumentParser(description=__doc__)
    sub = p.add_subparsers(dest="cmd", required=True)

    po = sub.add_parser("open", help="open a single-strategy window")
    po.add_argument("--title", required=True)
    po.add_argument("--minutes", type=int, default=30)
    po.add_argument("--desc", default="")
    po.add_argument("--monitor", action="append", default=[])
    po.set_defaults(fn=open_window)

    pl = sub.add_parser("list", help="list maintenance windows")
    pl.set_defaults(fn=list_windows)

    pd = sub.add_parser("delete", help="delete a window by id")
    pd.add_argument("mid")
    pd.set_defaults(fn=delete_window)

    args = p.parse_args()
    args.fn(args)


if __name__ == "__main__":
    main()