Shadowrocket 开发者 API 接口与自动化运维

#!/usr/bin/env python3 # shadowrocket_auto_maintain.py import requests, json, time, smtplib API = "http://127.0.0.1:1989/api" TOKEN = "your_developer_token" def refresh_and_switch(): # 刷新订阅 r = requests.post(f"{API}/v1/subscription/refresh", headers={"Authorization": f"Bearer {TOKEN}"}) if r.json().get("success"): print(f"[{time.strftime('%H:%M:%S')}] 订阅已刷新") # 获取所有节点 nodes = requests.get(f"{API}/v1/nodes", headers={"Authorization": f"Bearer {TOKEN}"}).json() # 按延迟排序，选取最优节点 best = min(nodes.get("nodes", []), key=lambda x: x.get("latency", 9999)) # 切换 requests.post(f"{API}/v1/node/switch", json={"node_id": best["id"]}, headers={"Authorization": f"Bearer {TOKEN}"}) print(f"已切换至: {best['name']} ({best['latency']}ms)") # 每6小时执行一次 while True: refresh_and_switch() time.sleep(6 * 3600)

#!/usr/bin/env python3 # shadowrocket_quota_alert.py import requests, logging from datetime import datetime API = "http://127.0.0.1:1989/api" TOKEN = "your_token" QUOTA_GB = 100 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def check_quota(): status = requests.get(f"{API}/v1/status", headers={"Authorization": f"Bearer {TOKEN}"}).json() used = (status.get("up_bytes", 0) + status.get("down_bytes", 0)) / (1024**3) if used >= QUOTA_GB: logging.warning(f"⚠️ 流量配额告警: 已用 {used:.2f}GB / {QUOTA_GB}GB") # 触发企业微信/钉钉/邮件通知 send_alert(f"Shadowrocket节点 {status.get('node')} 流量超限") else: logging.info(f"✓ 当前使用: {used:.2f}GB / {QUOTA_GB}GB") def send_alert(msg): # 企业微信Webhook示例 webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY" requests.post(webhook_url, json={"msgtype": "text", "text": {"content": msg}}) if __name__ == "__main__": check_quota()