事情的开始是先前用的server酱，免费版本每日上限5条，额度太少了，用telegram服务器消息接收到的不及时，正好最近有在搞qq的机器人，就想着能否接入，komari探针官方没有带qq机器人的通知接口，但是可以通过JavaScript 代码进行高度扩展，直接上最终成功图
一、准备工作
1.1 创建QQ机器人
首先，登录 QQ机器人管理后台，创建你的机器人应用。
创建完成后，在 开发设置 页面获取两个关键凭证：
- AppID：机器人的唯一标识
- ClientSecret：机器人的密钥
1.2 选择接入方式
QQ机器人官方支持两种事件接收方式：
WebSocket和Webhook
首次接入选择webhook，由于指定用户通知需要知道用户的OpenID，按照官方的描述是唯一身份机制，不同 bot 在单聊场景，获取到的用户唯一识别 openid 不一样，称为 user_openid，在配置中，我觉得有点恶心了，为了获取这个id折腾了不少时间，选择webhook就是为了获取id先
需要准备
1.一个域名
2.一台有公网的服务器（国内服务器的话域名需要备案）
3.服务器提前安装好python和nginx
二、部署Webhook服务
2.1 服务器端代码
在你的服务器上创建 webhook.py，填入你的 AppID 和 AppSecret：
import json
import time
import hmac
import hashlib
import base64
import requests
from flask import Flask, request, jsonify
APP_ID = "你的AppID"
CLIENT_SECRET = "你的AppSecrett"
app = Flask(__name__)
token_cache = None
token_expire = 0
def get_access_token():
"""获取并缓存 Access Token"""
global token_cache, token_expire
now = time.time()
if token_cache and now < token_expire - 60:
return token_cache
url = "https://api.bot.qq.com/app/getAppAccessToken"
resp = requests.post(url, json={"appId": APP_ID, "clientSecret": CLIENT_SECRET})
data = resp.json()
if data.get("err_code", 0) != 0:
raise Exception(f"Token 错误: {data}")
token_cache = data["access_token"]
token_expire = now + int(data.get("expires_in", 7200))
print(f"[Token] 获取成功，有效期至 {time.ctime(token_expire)}")
return token_cache
def generate_ed25519_signature(plain_token, event_ts):
"""
使用 Ed25519 算法生成 Webhook 验证签名
参考: https://bot.q.qq.com/wiki/develop/api-v2/dev-prepare/interface-framework/sign.html
"""
try:
import nacl.signing
import nacl.encoding
except ImportError:
print("[错误] 请安装 pynacl: pip install pynacl")
raise
# 根据 botSecret 生成 seed（需要32字节）
seed = CLIENT_SECRET
while len(seed) < 32:
seed = seed + seed
seed = seed[:32]
# 使用 seed 生成签名密钥
signing_key = nacl.signing.SigningKey(seed.encode('utf-8'))
# 拼接签名体: event_ts + plain_token
msg = str(event_ts) + plain_token
# 生成签名并转为十六进制
signed = signing_key.sign(msg.encode('utf-8'))
signature_hex = signed.signature.hex()
return signature_hex
def reply_message(target_type, target_openid, msg_id, content, title=""):
"""被动回复消息"""
token = get_access_token()
if target_type == "user":
url = f"https://api.bot.qq.com/v2/users/{target_openid}/messages"
else:
url = f"https://api.bot.qq.com/v2/groups/{target_openid}/messages"
full = f"**{title}**\n\n{content}" if title else content
payload = {"msg_id": msg_id, "msg_type": 2, "markdown": {"content": full}}
headers = {"Authorization": f"QQBot {token}", "Content-Type": "application/json"}
resp = requests.post(url, json=payload, headers=headers)
result = resp.json()
if result.get("err_code", 0) != 0:
print(f"[回复失败] {result}")
else:
print(f"[回复成功] msg_id={result.get('id')}")
return result
@app.route("/webhook", methods=["POST"])
def webhook():
event = request.get_json()
print("[收到事件]", json.dumps(event, ensure_ascii=False, indent=2))
op = event.get("op")
# ========== 处理 Webhook 验证请求（op=13）==========
# 官方文档：https://bot.q.qq.com/wiki/develop/api-v2/dev-prepare/interface-framework/event-emit.html
if op == 13:
d = event.get("d", {})
plain_token = d.get("plain_token")
event_ts = d.get("event_ts")
if not plain_token or not event_ts:
return jsonify({"code": 1, "message": "Missing parameters"}), 400
try:
signature = generate_ed25519_signature(plain_token, event_ts)
resp = {"plain_token": plain_token, "signature": signature}
return jsonify(resp)
except Exception as e:
print(f"[验证] 签名生成失败: {e}")
return jsonify({"code": 1, "message": str(e)}), 500
# ========== 处理普通消息事件 ==========
d = event.get("d", {})
t = event.get("t") or event.get("type")
if t not in ("C2C_MESSAGE_CREATE", "GROUP_AT_MESSAGE_CREATE", "GROUP_MESSAGE_CREATE"):
return jsonify({"code": 0})
if t == "C2C_MESSAGE_CREATE":
target_type = "user"
# OpenID 从消息事件中提取[reference:5]
target_openid = d.get("author", {}).get("id") or d.get("user_openid") or d.get("openid")
else:
target_type = "group"
target_openid = d.get("group_openid")
msg_id = d.get("id")
if not target_openid or not msg_id:
print("[错误] 缺少必要字段")
return jsonify({"code": 0})
# 回复用户的 OpenID
reply_content = f"您的 OpenID 是：\n`{target_openid}`\n\n（请人工记录此 ID）"
reply_message(target_type, target_openid, msg_id, reply_content, "🔑 您的 OpenID")
print(f"\n【人工提取】OpenID: {target_openid}\n")
return jsonify({"code": 0})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5001, debug=False)2.2 安装依赖并运行
# 安装依赖
pip install flask requests pynacl
# 启动服务（建议后台运行）
nohup python webhook.py > webhook.log 2>&1 &注意：QQ平台要求 Webhook 地址必须使用 HTTPS，且必须域名，因此需要通过 Nginx 反向代理转发。上面代码使用的端口为5001，则对应反代本地5001端口就行
2.3 Nginx 反向代理配置
这边做个简单的配置示例代码
server {
listen 443 ssl;
server_name your-domain.com;
ssl_certificate /path/to/cert.pem;
ssl_certificate_key /path/to/key.pem;
location /webhook {
proxy_pass http://127.0.0.1:5001;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}如果有安装宝塔面板的话，可以方便点，就不详细说了
三、配置 Webhook 并获取 OpenID
3.1 配置回调地址
- 登录 QQ机器人管理后台
- 进入你的机器人 → 开发设置 → 事件与回调配置
- 选择 Webhook 方式
- 填入回调地址：
https://你的域名/webhook - 点击保存
3.2 理解 Webhook 验证流程
当你保存 Webhook 地址时，QQ平台会发送一个 op=13 的验证请求
相关文档可看 通用数据结构 | QQ 机器人官方文档 事件订阅与通知 | QQ 机器人官方文档
{
"op": 13,
"d": {
"plain_token": "Arq0D5A61EgUu4OxUvOp",
"event_ts": "1725442341"
}
}你的服务器需要：
- 将
event_ts和plain_token拼接：event_ts + plain_token - 使用
Secret作为种子生成 Ed25519 私钥，对拼接字符串签名 - 返回
{"plain_token": plain_token, "signature": 签名}
3.3 发送测试消息
配置保存成功后，在 QQ 中给机器人发送一条消息（单聊）。
你的服务器日志会打印：
[收到事件] {"op":0,"t":"C2C_MESSAGE_CREATE","d":{...}}
【人工提取】OpenID: 825155B4D4DFF336C7FBDC4AFCF34CE4OpenID 是什么？ OpenID 是用户在某个机器人下的唯一标识，格式为32位十六进制字符串。同一个QQ用户在不同机器人下的OpenID是不同的，不能混用。
当你完成到这边的时候，恭喜你，最麻烦的地方完成了，然后回到机器人管理后台将事件与回调配置接入方式改为WebSocket
四、集成到 Komari 通知面板
4.1 完整 JavaScript 代码
获得 OpenID 后，将以下代码配置区替换正常的值然后粘贴到 Komari 管理后台的 设置 → 通知 → JavaScript 代码 中：
// ============================================================
// QQ机器人通知 - Komari 模板
// ============================================================
const CONFIG = {
appId: '你的AppID',
clientSecret: '你的AppSecret',
defaultTargetType: 'user',
defaultTargetOpenId: '你获取到的OpenID', // 替换为实际值
};
// ---------- Access Token 缓存 ----------
let cachedToken = null;
let tokenExpireTime = 0;
async function getAccessToken() {
const now = Date.now();
if (cachedToken && now < tokenExpireTime - 60000) {
return cachedToken;
}
const response = await fetch('https://api.bot.qq.com/app/getAppAccessToken', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
appId: CONFIG.appId,
clientSecret: CONFIG.clientSecret,
}),
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
}
const data = await response.json();
if (data.err_code && data.err_code !== 0) {
throw new Error(`Token错误: ${data.message || data.err_code}`);
}
cachedToken = data.access_token;
tokenExpireTime = now + (parseInt(data.expires_in) || 7200) * 1000;
return cachedToken;
}
// ---------- 核心发送函数 ----------
/**
* 发送消息（Komari 必须实现）
* @param {string} message - 消息正文（支持 Markdown）
* @param {string} title - 消息标题（会加粗显示）
* @returns {Promise<boolean>} 成功返回 true
*/
async function sendMessage(message, title) {
try {
const token = await getAccessToken();
const targetType = CONFIG.defaultTargetType;
const targetOpenId = CONFIG.defaultTargetOpenId;
if (!targetOpenId) {
console.error('[sendMessage] 未配置目标 OpenID');
return false;
}
const fullContent = `**${title}**\n\n${message}`;
const url = targetType === 'user'
? `https://api.bot.qq.com/v2/users/${targetOpenId}/messages`
: `https://api.bot.qq.com/v2/groups/${targetOpenId}/messages`;
const response = await fetch(url, {
method: 'POST',
headers: {
'Authorization': `QQBot ${token}`,
'Content-Type': 'application/json; charset=utf-8',
},
body: JSON.stringify({
msg_type: 2, // Markdown
markdown: { content: fullContent },
}),
});
const result = await response.json();
if (result.err_code && result.err_code !== 0) {
console.error('[sendMessage] 失败:', result);
if (result.err_code === 100001 || result.err_code === 100007) {
cachedToken = null;
}
return false;
}
return true;
} catch (error) {
console.error('[sendMessage] 异常:', error);
return false;
}
}
/**
* 处理 Komari 事件（可选）
* @param {Object} event - 事件对象
* @param {string} event.event - 事件类型 (offline, online, alert 等)
* @param {Array} event.clients - 客户端列表
* @param {string} event.time - 时间
* @param {string} event.message - 详情
* @param {string} event.emoji - Emoji
* @returns {Promise<boolean>}
*/
async function sendEvent(event) {
try {
const formatTime = (timeStr) => {
if (!timeStr) return '未知时间';
const date = new Date(timeStr);
return date.toLocaleString('zh-CN', {
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit',
second: '2-digit',
timeZone: 'Asia/Shanghai',
});
};
let clientInfo = '';
if (event.clients && event.clients.length > 0) {
const names = event.clients.map(c => c.name || c.client || '未知').join('、');
clientInfo = `\n📡 客户端：${names}`;
}
const emoji = event.emoji || '📢';
const title = `${emoji} ${event.event || '通知'}`;
const message = `事件：${event.event || '未知事件'}${clientInfo}\n时间：${formatTime(event.time)}\n详情：${event.message || '无详细信息'}`;
return await sendMessage(message, title);
} catch (error) {
console.error('[sendEvent] 异常:', error);
return false;
}
}4.2 测试通知
保存后，在 Komari 中触发一个通知测试。你应该会在 QQ 中收到机器人发送的 Markdown 格式消息。
五、总结
如果没有收到测试通知，看下接入方案有没有切回WebSocket，还是有问题从头看下文档有没有漏掉的地方，教程到这也结束了，有其他问题联系邮箱 2177367423@qq.com 看到会回复
