b77a8cadb2
Some checks failed
Upstream Sync / Sync latest commits from upstream repo (push) Has been cancelled
142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
微信 Chatbot HTTP API 的命令行客户端（供脚本与 Agent 调用）。
|
||
环境变量：WEB2_API_BASE（默认 http://127.0.0.1:8001）、WEB2_API_TOKEN（若后端启用鉴权）。
|
||
|
||
示例：
|
||
python wechat_chatbot_cli.py config-get --json
|
||
python wechat_chatbot_cli.py proxy-status --json
|
||
python wechat_chatbot_cli.py qr-start --json
|
||
python wechat_chatbot_cli.py login-check --w-id <wId> --json
|
||
python wechat_chatbot_cli.py send --target wxid_xxx --text "hello" --json
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
import httpx
|
||
|
||
|
||
def base_url() -> str:
|
||
return os.environ.get("WEB2_API_BASE", "http://127.0.0.1:8001").rstrip("/")
|
||
|
||
|
||
def headers() -> dict[str, str]:
|
||
h = {"Content-Type": "application/json"}
|
||
tok = os.environ.get("WEB2_API_TOKEN", "").strip()
|
||
if tok:
|
||
h["Authorization"] = f"Bearer {tok}"
|
||
return h
|
||
|
||
|
||
def out(obj: object, as_json: bool) -> None:
|
||
if as_json:
|
||
print(json.dumps(obj, ensure_ascii=False, indent=2))
|
||
else:
|
||
print(obj)
|
||
|
||
|
||
def main() -> None:
|
||
p = argparse.ArgumentParser(description="微信 Chatbot API CLI")
|
||
p.add_argument("--json", action="store_true", help="输出 JSON")
|
||
sub = p.add_subparsers(dest="cmd", required=True)
|
||
|
||
sub.add_parser("config-get", help="GET /chatbot/wechat/config")
|
||
|
||
pc = sub.add_parser("config-set", help="POST /chatbot/wechat/config（从 stdin 读 JSON 或文件）")
|
||
pc.add_argument("--file", "-f", help="JSON 文件路径")
|
||
|
||
sub.add_parser("proxy-status", help="GET /chatbot/wechat/proxy/status")
|
||
|
||
sub.add_parser("qr-start", help="POST /chatbot/wechat/qr/start")
|
||
|
||
pl = sub.add_parser("login-check", help="POST /chatbot/wechat/login/check")
|
||
pl.add_argument("--w-id", required=True, dest="w_id")
|
||
|
||
ps = sub.add_parser("send", help="POST /chatbot/wechat/send")
|
||
ps.add_argument("--target", required=True, help="对方 wxid 或群 id")
|
||
ps.add_argument("--text", required=True)
|
||
|
||
pw = sub.add_parser("webhook-register", help="POST /chatbot/wechat/webhook/register")
|
||
pw.add_argument("--url", required=True, help="公网可访问的 webhook 完整 URL")
|
||
|
||
args = p.parse_args()
|
||
bu = base_url()
|
||
h = headers()
|
||
|
||
try:
|
||
with httpx.Client(timeout=120.0) as client:
|
||
if args.cmd == "config-get":
|
||
r = client.get(f"{bu}/chatbot/wechat/config", headers=h)
|
||
r.raise_for_status()
|
||
out(r.json(), args.json)
|
||
elif args.cmd == "config-set":
|
||
raw = sys.stdin.read()
|
||
if getattr(args, "file", None):
|
||
raw = open(args.file, encoding="utf-8").read()
|
||
body = json.loads(raw)
|
||
r = client.post(f"{bu}/chatbot/wechat/config", headers=h, json=body)
|
||
r.raise_for_status()
|
||
out(r.json(), args.json)
|
||
elif args.cmd == "proxy-status":
|
||
r = client.get(f"{bu}/chatbot/wechat/proxy/status", headers=h)
|
||
try:
|
||
out(r.json(), args.json)
|
||
except Exception:
|
||
print(r.text, file=sys.stderr)
|
||
sys.exit(1)
|
||
elif args.cmd == "qr-start":
|
||
r = client.post(f"{bu}/chatbot/wechat/qr/start", headers=h)
|
||
try:
|
||
out(r.json(), args.json)
|
||
except Exception:
|
||
print(r.text, file=sys.stderr)
|
||
sys.exit(1)
|
||
elif args.cmd == "login-check":
|
||
r = client.post(
|
||
f"{bu}/chatbot/wechat/login/check",
|
||
headers=h,
|
||
json={"wId": args.w_id},
|
||
)
|
||
try:
|
||
out(r.json(), args.json)
|
||
except Exception:
|
||
print(r.text, file=sys.stderr)
|
||
sys.exit(1)
|
||
elif args.cmd == "send":
|
||
r = client.post(
|
||
f"{bu}/chatbot/wechat/send",
|
||
headers=h,
|
||
json={"targetId": args.target, "text": args.text},
|
||
)
|
||
try:
|
||
out(r.json(), args.json)
|
||
except Exception:
|
||
print(r.text, file=sys.stderr)
|
||
sys.exit(1)
|
||
elif args.cmd == "webhook-register":
|
||
r = client.post(
|
||
f"{bu}/chatbot/wechat/webhook/register",
|
||
headers=h,
|
||
json={"webhookUrl": args.url},
|
||
)
|
||
try:
|
||
out(r.json(), args.json)
|
||
except Exception:
|
||
print(r.text, file=sys.stderr)
|
||
sys.exit(1)
|
||
except httpx.HTTPStatusError as e:
|
||
print(str(e), file=sys.stderr)
|
||
try:
|
||
print(e.response.text, file=sys.stderr)
|
||
except Exception:
|
||
pass
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|