audio_codebuddy/deploy-package/health-monitor/health-push.py

#!/usr/bin/env python3
"""
health-push.py — 通过 push.rrbrr.com WebSocket 推送服务发推送

Usage:
  PUSH_API_URL=https://push.rrbrr.com PUSH_API_KEY=xxx PUSH_USER_ID=admin \
    python3 health-push.py <title> <<< "content"

Env required:
  PUSH_API_URL     push 服务地址，默认 https://push.rrbrr.com
  PUSH_API_KEY     API key（绑定 channel），默认 monitor-api-key-001
  PUSH_USER_ID     接收推送的 userId，默认 admin
  PUSH_CHANNEL     推送 channel，默认 monitor
"""

import os
import sys
import json
import urllib.request
import urllib.error
import ssl


def main():
    if len(sys.argv) < 2:
        print("Usage: health-push.py <title>", file=sys.stderr)
        sys.exit(2)

    title = sys.argv[1]
    content = sys.stdin.read() if not sys.stdin.isatty() else "(empty)"

    url = os.environ.get('PUSH_API_URL', 'https://push.rrbrr.com').rstrip('/')
    api_key = os.environ.get('PUSH_API_KEY', 'monitor-api-key-001')
    user_id = os.environ.get('PUSH_USER_ID', 'admin')
    channel = os.environ.get('PUSH_CHANNEL', 'monitor')

    body = json.dumps({
        'userId': user_id,
        'channel': channel,
        'title': title,
        'content': content[:500],   # 限制长度，H5 显示友好
        'data': {'ts': __import__('time').time()},
    }).encode('utf-8')

    req = urllib.request.Request(
        f'{url}/api/notify',
        data=body,
        method='POST',
        headers={
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json',
        },
    )

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE  # 自签证书先跳过

    try:
        with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
            resp = json.loads(r.read().decode('utf-8'))
            if resp.get('ok'):
                print(f"OK: sent '{title}' to userId={user_id} channel={channel}, online={resp.get('online', 0)}")
                sys.exit(0)
            else:
                print(f"API NOK: {resp}", file=sys.stderr)
                sys.exit(3)
    except urllib.error.HTTPError as e:
        print(f"HTTP {e.code}: {e.read().decode('utf-8', 'replace')[:200]}", file=sys.stderr)
        sys.exit(4)
    except Exception as e:
        print(f"NET FAIL [{type(e).__name__}]: {e}", file=sys.stderr)
        sys.exit(5)


if __name__ == '__main__':
    main()