monitor.py

#!/usr/bin/env python3
"""Health, usage and optional webhook notifier for CodingPlan proxy."""

import argparse
import json
import os
import sys
import urllib.request


def fetch_json(url, timeout=15):
    with urllib.request.urlopen(url, timeout=timeout) as resp:
        return json.loads(resp.read())


def post_webhook(url, payload, timeout=15):
    data = json.dumps(payload, ensure_ascii=False).encode()
    req = urllib.request.Request(url, data=data, method="POST")
    req.add_header("Content-Type", "application/json")
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return resp.status


def main():
    parser = argparse.ArgumentParser(description="Check CodingPlan proxy health and usage")
    parser.add_argument("--base-url", default=os.environ.get("CODINGPLAN_BASE_URL", "http://127.0.0.1:18999"))
    parser.add_argument("--webhook", default=os.environ.get("CODINGPLAN_WEBHOOK", ""))
    parser.add_argument("--warn-percent", type=float, default=float(os.environ.get("CODINGPLAN_WARN_PERCENT", "80")))
    args = parser.parse_args()

    base = args.base_url.rstrip("/")
    health = fetch_json(f"{base}/health")
    usage = fetch_json(f"{base}/usage")
    summary = usage.get("summary", {})
    percent = float(summary.get("usage_percent") or 0)
    report = {
        "ok": health.get("status") == "ok" and health.get("authenticated"),
        "health": health,
        "usage": summary,
    }
    print(json.dumps(report, ensure_ascii=False, indent=2))

    if args.webhook and (not report["ok"] or percent >= args.warn_percent or summary.get("quota_exhausted")):
        post_webhook(args.webhook, report)

    return 0 if report["ok"] else 2


if __name__ == "__main__":
    sys.exit(main())