claudetools/.claude/skills/yealink-ymcs/scripts/ymcs.py

#!/usr/bin/env python3
"""ymcs.py — CLI for the Yealink Management Cloud Service (YMCS) v2 API.

Read-only by default; every write (add/delete/reboot/reset/RPS) is gated behind
--confirm. Pairs with the `packetdial` skill for phone provisioning.

  ymcs.py sites|devices|accounts|sipaccounts|device-groups|device-configs
          firmwares|official-firmwares|models|alarms|oplogs|rps-servers|rps-devices
  ymcs.py devices --site <siteId>          # filter a list by site
  ymcs.py add-devices-by-mac --body '{...}' --confirm
  ymcs.py reboot --body '{"ids":[...]}' --confirm
  ymcs.py rps-add --body '{...}' --confirm
  ymcs.py raw POST /v2/dm/listDevices --body '{"skip":0,"limit":50,"autoCount":true}'
"""
import sys, os, json, argparse
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from ymcs_client import YmcsClient, YmcsError


def _emit(obj):
    print(json.dumps(obj, indent=2, default=str))


def _parse_body(s):
    if not s:
        return {}
    try:
        return json.loads(s)
    except json.JSONDecodeError as e:
        print(f"[ERROR] --body is not valid JSON: {e}", file=sys.stderr); sys.exit(2)


def _require_confirm(args, action, detail=""):
    if not getattr(args, "confirm", False):
        print(f"[DRY RUN] Would {action}" + (f": {detail}" if detail else ""))
        print("Refusing to perform a write without --confirm. Re-run with --confirm.")
        sys.exit(0)


def main(argv=None):
    p = argparse.ArgumentParser(prog="ymcs.py", description="Yealink YMCS v2 API client")
    sub = p.add_subparsers(dest="cmd", required=True)

    # reads
    for name in ["sites", "accounts", "device-groups", "device-configs",
                 "firmwares", "official-firmwares", "models", "alarms", "oplogs",
                 "rps-servers", "rps-devices"]:
        sub.add_parser(name)
    d = sub.add_parser("devices"); d.add_argument("--site")

    # writes (gated)
    for name in ["add-devices-by-mac", "add-devices", "del-devices", "reboot", "reset",
                 "rps-add", "rps-del", "add-sipaccount"]:
        w = sub.add_parser(name); w.add_argument("--body", required=True); w.add_argument("--confirm", action="store_true")

    r = sub.add_parser("raw")
    r.add_argument("method", choices=["GET", "POST", "PUT", "PATCH", "DELETE"])
    r.add_argument("path"); r.add_argument("--body"); r.add_argument("--confirm", action="store_true")

    args = p.parse_args(argv)
    try:
        c = YmcsClient()
        if args.cmd == "sites":                 _emit(c.list_sites())
        elif args.cmd == "devices":             _emit(c.list_devices(**({"siteId": args.site} if args.site else {})))
        elif args.cmd == "accounts":            _emit(c.list_accounts())
        elif args.cmd == "device-groups":       _emit(c.list_device_groups())
        elif args.cmd == "device-configs":      _emit(c.list_device_configs())
        elif args.cmd == "firmwares":           _emit(c.list_firmwares())
        elif args.cmd == "official-firmwares":  _emit(c.list_official_firmwares())
        elif args.cmd == "models":              _emit(c.models())
        elif args.cmd == "alarms":              _emit(c.list_alarms())
        elif args.cmd == "oplogs":              _emit(c.list_oplogs())
        elif args.cmd == "rps-servers":         _emit(c.rps_list_servers())
        elif args.cmd == "rps-devices":         _emit(c.rps_list_devices())

        elif args.cmd == "add-devices-by-mac":
            b = _parse_body(args.body); _require_confirm(args, "ADD devices by MAC", json.dumps(b)[:160]); _emit(c.add_devices_by_mac(b))
        elif args.cmd == "add-devices":
            b = _parse_body(args.body); _require_confirm(args, "ADD devices", json.dumps(b)[:160]); _emit(c.add_devices(b))
        elif args.cmd == "del-devices":
            b = _parse_body(args.body); _require_confirm(args, "DELETE devices", json.dumps(b)[:160]); _emit(c.del_devices(b))
        elif args.cmd == "reboot":
            b = _parse_body(args.body); _require_confirm(args, "REBOOT devices", json.dumps(b)[:160]); _emit(c.device_reboot(b))
        elif args.cmd == "reset":
            b = _parse_body(args.body); _require_confirm(args, "FACTORY-RESET devices", json.dumps(b)[:160]); _emit(c.device_reset(b))
        elif args.cmd == "rps-add":
            b = _parse_body(args.body); _require_confirm(args, "RPS add devices", json.dumps(b)[:160]); _emit(c.rps_add_devices(b))
        elif args.cmd == "rps-del":
            b = _parse_body(args.body); _require_confirm(args, "RPS delete devices", json.dumps(b)[:160]); _emit(c.rps_delete_devices(b))
        elif args.cmd == "add-sipaccount":
            b = _parse_body(args.body); _require_confirm(args, "ADD SIP account to device", json.dumps(b)[:160]); _emit(c.add_sip_account(b))

        elif args.cmd == "raw":
            b = _parse_body(args.body) if args.body else None
            path = args.path
            # Recover from Git-Bash MSYS path-conversion (a leading /v2/... arg gets
            # rewritten to C:/Program Files/Git/v2/...). Cut back to the API path.
            if "/v2/" in path:
                path = path[path.index("/v2/"):]
            if args.method != "GET":
                _require_confirm(args, f"{args.method} {path}", json.dumps(b or {})[:160])
            _emit(c.request(args.method, path, body=b))
        else:
            p.error(f"unknown command {args.cmd}")
    except YmcsError as exc:
        print(f"[ERROR] {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()