9dff73e7bc
🔴 致命（5个）: - Bug1: onebot11.json 格式错误（旧 {http:{},ws:{}} → network.httpServers/httppClients） - Bug2: 端口映射错位（WebUI 6099, OneBot API 6097 独立端口） - Bug3: 变量名含空格 NAP CAT_HOST → NAPCAT_HOST（14脚本+2SKILL+README） - Bug4: heredoc 单引号阻止变量展开 - Bug5: process.py 缺失 _get_high_risk_words() 方法 🟡 重要（3个）: - Bug6: onebot11.json 默认不存在（cat > 而非 edit） - Bug7: API 须登录后才响应（新增扫码引导步骤） - Bug8: qqrebot systemd 依赖 napcat 服务 🟢 中等（3个）: - Bug9: Skills 占位符缺 YOUR_NAPCAT_HOST sed 覆盖 - Bug10: venv 不可用需降级方案 - Bug11: 端口 25580 冲突处理 已验证全链路：NapCat + qqrebot + 插件 + OpenClaw Gateway + AI 回复
112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
QQ 主动发信脚本 - 通过 go-cqhttp HTTP API 发送消息到 QQ
|
|
用于 qq-agent 主动推送消息（非回复场景）
|
|
|
|
用法:
|
|
# 发送私聊消息
|
|
python3 qq_send_msg.py --private YOUR_ADMIN_QQ --message "服务器已重启完成"
|
|
|
|
# 发送群聊消息
|
|
python3 qq_send_msg.py --group YOUR_BOT_QQ --message "系统维护通知: ..."
|
|
|
|
# 从文件读取消息内容
|
|
python3 qq_send_msg.py --private YOUR_ADMIN_QQ --file /tmp/report.txt
|
|
|
|
# 快速发送（私聊+短时间内多条消息带上合并开关）
|
|
python3 qq_send_msg.py --private YOUR_ADMIN_QQ --message "你好" --auto_escape
|
|
|
|
注意:
|
|
- 管理员 QQ 号: YOUR_ADMIN_QQ
|
|
- 默认发送到管理员私聊
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import argparse
|
|
import requests
|
|
import os
|
|
|
|
# go-cqhttp HTTP API 地址（与 qqrebot 配置文件一致）
|
|
CQHTTP_URL = "http://YOUR_NAPCAT_HOST:25570"
|
|
|
|
# 默认接收用户（管理员）
|
|
DEFAULT_USER = "YOUR_ADMIN_QQ"
|
|
|
|
|
|
def send_private_msg(user_id, message, auto_escape=False):
|
|
"""发送私聊消息"""
|
|
url = f"{CQHTTP_URL}/send_private_msg"
|
|
payload = {
|
|
"user_id": int(user_id),
|
|
"message": message,
|
|
"auto_escape": auto_escape,
|
|
}
|
|
resp = requests.post(url, json=payload, timeout=10)
|
|
data = resp.json()
|
|
if data.get("status") == "ok":
|
|
return True, data
|
|
return False, data
|
|
|
|
|
|
def send_group_msg(group_id, message, auto_escape=False):
|
|
"""发送群聊消息"""
|
|
url = f"{CQHTTP_URL}/send_group_msg"
|
|
payload = {
|
|
"group_id": int(group_id),
|
|
"message": message,
|
|
"auto_escape": auto_escape,
|
|
}
|
|
resp = requests.post(url, json=payload, timeout=10)
|
|
data = resp.json()
|
|
if data.get("status") == "ok":
|
|
return True, data
|
|
return False, data
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="发送 QQ 消息")
|
|
target = parser.add_mutually_exclusive_group(required=False)
|
|
target.add_argument("--private", type=str, default=DEFAULT_USER, nargs="?",
|
|
const=DEFAULT_USER, help="接收用户 QQ 号 (默认: 管理员)")
|
|
target.add_argument("--group", type=str, help="目标群号")
|
|
|
|
content = parser.add_mutually_exclusive_group(required=True)
|
|
content.add_argument("--message", help="消息内容")
|
|
content.add_argument("--file", help="从文件读取消息内容")
|
|
|
|
parser.add_argument("--auto_escape", action="store_true",
|
|
help="是否转义 CQ 码 (默认不转义)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.file:
|
|
with open(args.file, "r") as f:
|
|
message = f.read()
|
|
else:
|
|
message = args.message
|
|
|
|
try:
|
|
if args.group:
|
|
ok, result = send_group_msg(args.group, message, args.auto_escape)
|
|
target_desc = f"群 {args.group}"
|
|
else:
|
|
ok, result = send_private_msg(args.private, message, args.auto_escape)
|
|
target_desc = f"用户 {args.private}"
|
|
|
|
if ok:
|
|
print(f"✅ 消息已发送到 {target_desc}")
|
|
else:
|
|
print(f"❌ 发送失败 ({target_desc}): {json.dumps(result, ensure_ascii=False)}")
|
|
sys.exit(1)
|
|
except requests.exceptions.ConnectionError:
|
|
print(f"❌ 连接失败: 无法连接到 go-cqhttp ({CQHTTP_URL})")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"❌ 发送异常: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|