码桶
发现社区成员的开源项目
worker.js14.8 KB
/**
* 消息拉取服务 — 对应 PHP worker.php
* - HTTP 模式：GET /worker?key=xxx
* - CLI 模式：node worker.js [--debug]
* - Cron 模式：由 node-cron 自动定时执行
*/
const axios = require('axios');
const crypto = require('crypto');
const db = require('../lib/db');
const config = require('../config');
const ILink = require('../lib/ilink');
const plugins = require('../lib/plugins');
const msgEvents = require('../lib/msg-events');
const WORKER_KEY = config.worker_key || 'ngwebot2026';
/** 日志输出 */
function log(s) {
const ts = '[' + new Date().toISOString().replace('T', ' ').slice(0, 19) + ']';
console.log(ts + ' ' + s);
}
/** 从响应中提取消息 */
function extractMessages(resp) {
if ((resp.ret ?? -1) !== 0) return null;
const buf = resp.get_updates_buf || resp.upd_buf || null;
let list = resp.msgs || resp.msg_list || resp.messages || null;
if (!Array.isArray(list) && resp.data) {
list = resp.data.msg_list || resp.data.messages || [];
}
if (!Array.isArray(list)) list = [];
return { buf, list };
}
/** 容错提取入站消息文本内容（兼容多种 iLink 返回结构） */
function extractText(m) {
if (!m || typeof m !== 'object') return '';
if (typeof m.content === 'string' && m.content) return m.content;
if (typeof m.text === 'string' && m.text) return m.text;
const msg = m.msg || m.message || {};
if (typeof msg.content === 'string' && msg.content) return msg.content;
if (typeof msg.text === 'string' && msg.text) return msg.text;
const lists = [];
if (Array.isArray(m.item_list)) lists.push(m.item_list);
if (Array.isArray(msg.item_list)) lists.push(msg.item_list);
for (const list of lists) {
for (const it of list) {
if (it && it.text_item && typeof it.text_item.text === 'string' && it.text_item.text) return it.text_item.text;
if (it && typeof it.text === 'string' && it.text) return it.text;
if (it && it.title) return it.title;
}
}
return '';
}
/** 容错提取入站消息类型 */
function extractType(m) {
if (!m || typeof m !== 'object') return 'text';
const msg = m.msg || m.message || {};
const lists = [];
if (Array.isArray(m.item_list)) lists.push(m.item_list);
if (Array.isArray(msg.item_list)) lists.push(msg.item_list);
for (const list of lists) {
for (const it of list) {
const t = it && it.type;
if (t === 2) return 'image';
if (t === 4 || t === 5) return 'file'; // type 4/5 均为文件消息
if (t === 9) return 'video';
if (t === 3) return 'voice';
}
}
return 'text';
}
/** 根据消息类型返回兜底文字，用于无文本内容的消息入库展示 */
const TYPE_LABEL = { image: '[图片]', video: '[视频]', file: '[文件]', voice: '[语音]' };
/** 提取媒体 CDN 下载引用（encrypt_query_param + aes_key），供网页端下载/展示 */
function extractMediaRef(m) {
if (!m || typeof m !== 'object') return null;
const msg = m.msg || m.message || {};
const lists = [];
if (Array.isArray(m.item_list)) lists.push(m.item_list);
if (Array.isArray(msg.item_list)) lists.push(msg.item_list);
for (const list of lists) {
for (const it of list) {
if (!it) continue;
const t = it.type;
if (t === 2 && it.image_item && it.image_item.media) {
return { ...it.image_item.media, _t: 'image' };
}
if ((t === 4 || t === 5) && it.file_item && it.file_item.media) {
return { ...it.file_item.media, file_name: it.file_item.file_name || null, _t: 'file' };
}
if (t === 9 && it.video_item && it.video_item.media) {
return { ...it.video_item.media, _t: 'video' };
}
if (t === 3 && it.voice_item && it.voice_item.media) {
return { ...it.voice_item.media, _t: 'voice' };
}
}
}
return null;
}
/** 统一解析一条入站消息 */
function parseInbound(m, debug = false) {
const rawContent = extractText(m);
const msg_type = extractType(m);
// 非文本消息（图片/视频/文件/语音）：总是提取 CDN 媒体引用，
// 文件消息常带有 title（文件名），必须同时保留文字和引用。
// text 类型但无内容的（心跳/同步事件）保持空，不污染数据库。
let content = rawContent;
if (msg_type !== 'text') {
const ref = extractMediaRef(m);
// DEBUG: 打印非文本消息的 item_list 结构，排查文件提取失败原因（仅 debug 模式，避免长期刷屏/磁盘 IO）
if (debug) {
const msg = m.msg || m.message || {};
const lists = [];
if (Array.isArray(m.item_list)) lists.push(m.item_list);
if (Array.isArray(msg.item_list)) lists.push(msg.item_list);
log(`[parseInbound] msg_type=${msg_type}, rawContent="${rawContent}", ref=${!!ref}, lists_count=${lists.length}, lists_keys=${JSON.stringify(lists.map(l=>l.map(it=>it?.type||'?')))}`);
}
if (ref) {
content = JSON.stringify({ l: TYPE_LABEL[msg_type] || ('[' + msg_type + ']'), r: ref, t: rawContent || '' });
} else if (!content) {
content = TYPE_LABEL[msg_type] || ('[' + msg_type + ']');
}
}
const peer = m.from || m.peer_id || m.user_id || m.from_user_id || null;
const ctx = m.context_token || null;
return { content, msg_type, peer, ctx };
}
/** 并发拉取所有机器人消息 */
async function pollAllBots(bots, timeout) {
const uin = Buffer.from(String(crypto.randomInt(0, 0xffffffff))).toString('base64');
const promises = bots.map(async (bot) => {
const il = new ILink(bot);
const buf = bot.upd_buf || '';
try {
const resp = await axios.post(
(bot.base_url || config.ilink_base) + '/ilink/bot/getupdates',
{
base_info: { channel_version: config.channel_version },
get_updates_buf: String(buf),
},
{
headers: {
'Content-Type': 'application/json',
'AuthorizationType': 'ilink_bot_token',
'Authorization': `Bearer ${bot.bot_token}`,
'X-WECHAT-UIN': uin,
},
timeout: (timeout + 5) * 1000,
}
);
const data = resp.data;
if (!('ret' in data) && 'get_updates_buf' in data) data.ret = 0;
return { botId: bot.id, resp: data };
} catch (err) {
log(`bot#${bot.id} API请求失败: ${err.message}`);
return { botId: bot.id, resp: { ret: -1, error: err.message } };
}
});
return Promise.all(promises);
}
/**
* 处理单个机器人的一轮消息（提取、去重、入库、欢迎语、插件钩子）
* 由 runPoll 和 runCLI 共享
*/
async function processBotMessages(bot, resp, now, opts = {}) {
const { debug = false, totalMsg = { count: 0 }, errors = [] } = opts;
const ret = resp.ret ?? -1;
if (ret === -14) {
log(`bot#${bot.id} 会话过期`);
await db.exec("UPDATE bots SET login_status='expired', updated_at=? WHERE id=?", [now, bot.id]);
errors.push(`bot#${bot.id} 会话过期`);
return { handled: false };
}
if (ret !== 0) {
errors.push(`bot#${bot.id} API返回错误: ret=${ret}`);
return { handled: false };
}
// 自愈：能成功拉到消息(ret=0)说明机器人确实已绑定。
// 若 login_status 因前端轮询未及时更新而停留在 wait，这里修正为 confirmed，
// 避免 UI 一直显示"等待扫码"、也避免其他依赖 confirmed 的判断误判。
if (bot.login_status !== 'confirmed') {
await db.exec(
"UPDATE bots SET login_status='confirmed', bind_at=COALESCE(bind_at,?) WHERE id=? AND login_status!='confirmed'",
[now, bot.id]
);
bot.login_status = 'confirmed';
}
const parsed = extractMessages(resp);
if (parsed && parsed.buf !== null) {
await db.exec("UPDATE bots SET upd_buf=?, updated_at=? WHERE id=?", [parsed.buf, now, bot.id]);
if (debug) log(`bot#${bot.id} 更新 buf`);
}
if (parsed && Array.isArray(parsed.list)) {
log(`bot#${bot.id} 收到 ${parsed.list.length} 条消息`);
for (const m of parsed.list) {
if (debug) {
const rawItemTypes = (() => {
const lists = [m.item_list, (m.msg||m.message||{}).item_list].filter(Array.isArray);
return lists.map(l => l.map(it => `${it.type}${it.file_item?'(file)':it.image_item?'(img)':'?'}`));
})();
log(`[RAW-MSG] types=${JSON.stringify(rawItemTypes)}, topKeys=${JSON.stringify(Object.keys(m))}`);
}
const { content, msg_type, peer, ctx } = parseInbound(m, debug);
if (debug) log(`[MSG-DEBUG] msg_type=${msg_type}, content="${String(content).substring(0,120)}", peer=${peer}, ctx=${!!ctx}`);
if (!content) {
if (debug) { try { log(`[RAW-PAYLOAD] ${JSON.stringify(m).substring(0,800)}`); } catch(e){} }
continue;
}
// 去重：5秒内相同内容
const dup = await db.row(
"SELECT id FROM messages WHERE bot_id=? AND direction='in' AND peer_id=? AND content=? AND created_at>=?",
[bot.id, peer, content, now - 5]
);
if (dup) continue;
// 首条消息判断（去重后）
const firstEver = !(await db.row(
"SELECT id FROM messages WHERE bot_id=? AND direction='in' AND peer_id=? LIMIT 1",
[bot.id, peer]
));
const insertInfo = await db.exec(
"INSERT INTO messages (bot_id, direction, peer_id, content, msg_type, context_token, created_at) VALUES (?,?,?,?,?,?,?)",
[bot.id, 'in', peer, content, msg_type, ctx, now]
);
const messageId = (insertInfo && typeof insertInfo.lastInsertRowid !== 'undefined') ? Number(insertInfo.lastInsertRowid) : null;
if (ctx) {
await db.exec("UPDATE bots SET context_token=? WHERE id=?", [ctx, bot.id]);
}
totalMsg.count++;
// 欢迎语（仅首次）
if (firstEver) {
await sendWelcome(bot, peer, ctx, now);
}
// 插件钩子（msg.content 保持明文，插件层无需感知加密）
msgEvents.push(bot.id, 'inbound', `收到消息: ${String(content).slice(0, 80)}`, `类型: ${msg_type}`);
try {
await plugins.onMessage(bot, { content, peer_id: peer, context_token: ctx, msg_type, message_id: messageId });
} catch (pe) {
msgEvents.push(bot.id, 'error', '消息处理异常', pe.message);
console.error(`[plugin] bot#${bot.id} onMessage 异常:`, pe.message);
}
log(`bot#${bot.id} 存储消息: from=${peer}, type=${msg_type}, content="${String(content).slice(0, 20)}"`);
}
}
return { handled: true };
}
/** 发送首条连接欢迎语 */
async function sendWelcome(bot, peer, ctx, now) {
try {
const dt = new Date();
const pad = (n) => String(n).padStart(2, '0');
const timeStr = `${dt.getFullYear()}-${pad(dt.getMonth() + 1)}-${pad(dt.getDate())} ${pad(dt.getHours())}:${pad(dt.getMinutes())}:${pad(dt.getSeconds())}`;
const botName = bot.name || bot.bot_code || '奶狗WeBot';
const il = new ILink(bot);
const resp = await il.sendMessage(peer, ctx, `您已成功连接${botName} 「${timeStr}」`);
if ((resp.ret ?? -1) === 0) {
const welcomeText = `您已成功连接${botName} 「${timeStr}」`;
await db.exec(
"INSERT INTO messages (bot_id, direction, peer_id, content, msg_type, context_token, created_at) VALUES (?,?,?,?,?,?,?)",
[bot.id, 'out', peer, welcomeText, 'text', ctx, now]
);
log(`bot#${bot.id} 发送欢迎语给 ${peer}`);
} else {
log(`bot#${bot.id} 欢迎语发送失败 ret=${resp.ret ?? -1}`);
}
} catch (we) {
console.error(`[welcome] bot#${bot.id} 发送欢迎语异常:`, we.message);
}
}
/** 执行一轮拉取 */
async function runPoll(debug = false, logs = []) {
const now = Math.floor(Date.now() / 1000);
const totalMsg = { count: 0 };
let botsProcessed = 0;
const errors = [];
let bots;
try {
// 以 bot_token 是否存在作为"已绑定"判据（token 即绑定凭证）；
// 不要求 login_status='confirmed'，否则状态滞后(wait)时轮询被整体跳过，消息收不到。
bots = await db.rows(
"SELECT * FROM bots WHERE bot_token IS NOT NULL"
);
log(`找到 ${bots.length} 个已绑定的机器人`);
if (debug) logs.push(`找到 ${bots.length} 个已绑定的机器人`);
} catch (e) {
errors.push('DB错误: ' + e.message);
return { ok: false, msg: 'DB错误', errors };
}
if (bots.length === 0) {
log('无活跃机器人');
return { ok: true, bots: 0, messages: 0 };
}
const results = await pollAllBots(bots, 40);
log('并发拉取完成');
for (const { botId, resp } of results) {
const bot = bots.find(b => b.id === botId);
if (!bot) continue;
const { handled } = await processBotMessages(bot, resp, now, { debug, totalMsg, errors });
if (handled) botsProcessed++;
}
log(`完成：处理 ${botsProcessed} 个机器人，收到 ${totalMsg.count} 条新消息`);
return { ok: true, bots: botsProcessed, messages: totalMsg.count, errors };
}
/** HTTP 路由处理函数（供 Express 使用） */
async function handleRequest(req, res) {
const key = req.query.key || req.body.key || '';
const debug = !!(req.query.debug || req.body.debug);
if (key !== WORKER_KEY) {
return res.json({ ok: false, msg: '无效的密钥' });
}
req.setTimeout(120000);
const logs = [];
const result = await runPoll(debug, logs);
result.time = new Date().toISOString().replace('T', ' ').slice(0, 19);
if (debug) {
result.logs = logs;
}
res.json(result);
}
/** CLI 模式 */
async function runCLI(debug = false) {
log('--- 首次拉取 ---');
await runPoll(debug);
log('CLI 模式，持续轮询中...');
while (true) {
await new Promise(r => setTimeout(r, 1000));
let bots;
try {
bots = await db.rows("SELECT * FROM bots WHERE bot_token IS NOT NULL");
} catch (e) {
log('DB错误: ' + e.message);
continue;
}
if (bots.length === 0) continue;
const now = Math.floor(Date.now() / 1000);
const results = await pollAllBots(bots, 40);
for (const { botId, resp } of results) {
if ((resp.ret ?? -1) !== 0) { log('[runCLI] bot=' + botId + ' poll ret=' + (resp.ret ?? -1) + '，跳过'); continue; }
const bot = bots.find(b => b.id === botId);
if (!bot) continue;
await processBotMessages(bot, resp, now);
}
}
}
// 直接 CLI 运行
if (require.main === module) {
const debug = process.argv.includes('--debug');
try {
runCLI(debug).catch(e => {
log('致命错误: ' + e.message);
process.exit(1);
});
} catch (e) {
log('启动失败: ' + e.message);
process.exit(1);
}
}
module.exports = { handleRequest, runPoll, runCLI };