ObsidianHome/小鱼/闲鱼消息通知/watch.js

// 闲鱼卖家中心 未读消息监听
// 用法:
//   node watch.js            正常轮询
//   node watch.js --test     只发一条测试通知，验证 Toast / webhook 是否通
//   node watch.js --dump     每次轮询打印看到的未读元素（调试用）
const fs = require('fs');
const path = require('path');
const { CDP, listPages, EXPR } = require('./cdp');
const { notify, sendWebhook } = require('./notify');

const DIR = __dirname;
const CONFIG_PATH = path.join(DIR, 'config.json');
const LOG_PATH = path.join(DIR, 'watch.log');
const STATE_PATH = path.join(DIR, 'state.json');

const cfg = JSON.parse(fs.readFileSync(CONFIG_PATH, 'utf8').replace(/^\uFEFF/, ''));
const ARGS = process.argv.slice(2);
const TEST = ARGS.includes('--test');
const DUMP = ARGS.includes('--dump');

function log(...parts) {
  const line = `[${new Date().toLocaleString('zh-CN', { hour12: false })}] ${parts.join(' ')}`;
  console.log(line);
  try {
    if (fs.existsSync(LOG_PATH) && fs.statSync(LOG_PATH).size > 2 * 1024 * 1024) {
      fs.renameSync(LOG_PATH, LOG_PATH + '.old');
    }
    if (!fs.existsSync(LOG_PATH)) fs.writeFileSync(LOG_PATH, '\uFEFF', 'utf8'); // BOM: 记事本/编辑器不乱码
    fs.appendFileSync(LOG_PATH, line + '\r\n', 'utf8');
  } catch { }
}

const sleep = (ms) => new Promise(r => setTimeout(r, ms));

const conns = new Map(); // targetId -> CDP

async function pollPages() {
  const pages = await listPages(cfg.debugPort, cfg.urlMatch);
  const states = [];
  const seen = new Set();

  for (const p of pages) {
    seen.add(p.id);
    let cdp = conns.get(p.id);
    if (!cdp || cdp.closed) {
      cdp = new CDP(p.webSocketDebuggerUrl);
      try {
        await cdp.connect();
        conns.set(p.id, cdp);
      } catch (e) {
        continue;
      }
    }
    try {
      const st = await cdp.eval(EXPR);
      if (st) states.push(st);
    } catch (e) {
      log('读取页面失败（可能正在刷新）:', e.message);
      try { cdp.close(); } catch { }
      conns.delete(p.id);
    }
  }
  for (const id of [...conns.keys()]) {
    if (!seen.has(id)) { try { conns.get(id).close(); } catch { } conns.delete(id); }
  }
  return states;
}

function pathAllowed(p, filter) {
  if (!filter) return true;
  if (filter.includePathRegex) {
    if (!new RegExp(filter.includePathRegex).test(p)) return false;
  }
  if (filter.excludePathRegex) {
    if (new RegExp(filter.excludePathRegex).test(p)) return false;
  }
  return true;
}

// 把页面状态压平成 { key: 未读数 }
function extractCounters(states) {
  const map = new Map();
  const detail = [];
  for (const st of states) {
    const page = (st.url || '').split('?')[0];

    for (const b of st.badges || []) {
      if (cfg.badgeFilter?.requireRed && !b.red) continue;
      if (!pathAllowed(b.p, cfg.badgeFilter)) continue;
      const n = parseInt(String(b.t).replace('+', ''), 10);
      if (!Number.isFinite(n) || n <= 0) continue;
      const key = `${page}||${b.p}`;
      map.set(key, Math.max(n, map.get(key) || 0));
      detail.push({ key, n, t: b.t, p: b.p, page, parentText: b.parentText });
    }

    if (cfg.titleRegex) {
      try {
        const m = (st.title || '').match(new RegExp(cfg.titleRegex));
        if (m) {
          const n = parseInt(m[1], 10);
          if (Number.isFinite(n) && n > 0) map.set(`title||${page}`, n);
        }
      } catch { }
    }
  }
  return { map, detail };
}

function loadPrev() {
  try { return new Map(Object.entries(JSON.parse(fs.readFileSync(STATE_PATH, 'utf8')).counters || {})); }
  catch { return null; }
}

function savePrev(map) {
  try {
    fs.writeFileSync(STATE_PATH, JSON.stringify({ savedAt: new Date().toISOString(), counters: Object.fromEntries(map) }, null, 2), 'utf8');
  } catch { }
}

// ---------- webhook 节流 + 合并（防刷屏、防企微 20条/分钟限流） ----------
const pushLog = [];      // 已推送时间戳
let pendingPush = [];    // 被节流时暂存的提醒，攒批后一次发出

function rateState() {
  const rl = cfg.rateLimit || {};
  const minI = Math.max(0, Number(rl.minIntervalSeconds ?? 60)) * 1000;
  const maxH = Number(rl.maxPerHour ?? 30);
  const now = Date.now();
  while (pushLog.length && now - pushLog[0] > 3600000) pushLog.shift();
  if (maxH > 0 && pushLog.length >= maxH) {
    return { ok: false, reason: `已达每小时上限 ${maxH} 条` };
  }
  const last = pushLog[pushLog.length - 1];
  if (last && now - last < minI) {
    return { ok: false, reason: `距上条推送 ${Math.round((now - last) / 1000)}s（下限 ${minI / 1000}s）` };
  }
  return { ok: true };
}

async function queuePush(lines, pageUrl) {
  const hooks = cfg.webhooks || [];
  if (!hooks.length) return;
  pendingPush.push({ lines, pageUrl, at: new Date().toLocaleTimeString('zh-CN', { hour12: false }) });
  const st = rateState();
  if (!st.ok) {
    log(`webhook 节流：${st.reason}，已并入待发队列（${pendingPush.length} 条）`);
    return;
  }
  await flushPush();
}

async function flushPush() {
  if (!pendingPush.length) return;
  const hooks = cfg.webhooks || [];
  if (!hooks.length) { pendingPush = []; return; }
  const batch = pendingPush.slice();
  pendingPush = [];
  const lastUrl = batch[batch.length - 1].pageUrl;
  const body = batch.map(b => `[${b.at}] ${b.lines.join(' / ')}`).join('\n') + `\n${lastUrl}`;
  const title = batch.length > 1 ? `闲鱼新消息（合并 ${batch.length} 条提醒）` : '闲鱼有新消息';
  pushLog.push(Date.now());
  await Promise.all(hooks.map(async (h) => {
    try {
      const r = await sendWebhook(h, title, body, { url: lastUrl });
      log(`[webhook] ${h.type} -> ${r.status}`);
    } catch (e) {
      log(`[webhook] ${h.type} 失败: ${e.message}`);
    }
  }));
}

function inQuietHours() {
  const q = cfg.quietHours;
  if (!q || !q.from || !q.to) return false;
  const now = new Date();
  const mins = now.getHours() * 60 + now.getMinutes();
  const [fh, fm] = String(q.from).split(':').map(Number);
  const [th, tm] = String(q.to).split(':').map(Number);
  const from = fh * 60 + fm, to = th * 60 + tm;
  return from <= to ? (mins >= from && mins < to) : (mins >= from || mins < to);
}

async function tick(prev) {
  let states;
  try {
    states = await pollPages();
  } catch (e) {
    log(`调试端口 ${cfg.debugPort} 连不上: ${e.message}（Edge 关了吗？）`);
    return prev;
  }

  if (!states.length) {
    log(`没找到 goofish 标签页。请在调试端口的 Edge 里打开 seller.goofish.com 消息页。`);
    return prev;
  }

  const { map, detail } = extractCounters(states);

  if (DUMP) {
    log(`页面 ${states.length} 个 | 命中计数器 ${map.size} 个`);
    detail.forEach(d => log(`  ● ${d.t}  ${d.p}  parent="${d.parentText}"`));
    if (!map.size) {
      log('  （没有命中的未读元素，用 node probe.js 看候选）');
    }
  }

  if (!prev) {
    log(`已建立基线：当前未读计数器 ${map.size} 个 -> ${JSON.stringify(Object.fromEntries(map))}`);
    savePrev(map);
    return map;
  }

  const events = [];
  for (const [k, v] of map) {
    const before = prev.get(k) || 0;
    if (v > before) {
      const d = detail.find(x => x.key === k);
      events.push({ key: k, from: before, to: v, page: k.split('||')[0], p: d ? d.p : '', parentText: d ? d.parentText : '' });
    }
  }
  if (cfg.notifyOnDecrease) {
    for (const [k, v] of prev) {
      if (!map.has(k) && v > 0) events.push({ key: k, from: v, to: 0, page: k.split('||')[0], p: k.split('||')[1] || '', parentText: '' });
    }
  }

  if (events.length) {
    const total = [...map.values()].reduce((a, b) => a + b, 0);
    const head = events[0];
    const title = `闲鱼有新消息（未读约 ${total}）`;
    const lines = events.map(e => {
      const where = e.p ? e.p.split(' > ').slice(-1)[0] : '页面';
      return `· ${where}：${e.from} → ${e.to}`;
    });
    const text = `${lines.join('\n')}\n${head.page}`;
    if (inQuietHours()) {
      log('静默时段，跳过通知:', JSON.stringify(events));
    } else {
      log('检测到新消息:', JSON.stringify(events));
      // 本地弹窗不节流（这才是工具的意义）；webhook 走节流 + 合并
      await notify({ ...cfg, webhooks: [] }, title, text, { url: head.page }, log);
      await queuePush(lines, head.page);
    }
  }

  // 有积压且节流窗口已过，补发
  if (pendingPush.length) {
    const st = rateState();
    if (st.ok) await flushPush();
  }

  savePrev(map);
  return map;
}

(async () => {
  if (TEST) {
    log('发送测试通知…');
    await notify(cfg, '闲鱼消息提醒 · 测试', `这是一条测试通知。\n本地 Toast / 声音 / webhook 配置检查。\n${new Date().toLocaleString('zh-CN')}`, {}, log);
    log('测试通知已发出。若无弹窗，请检查 Windows 通知设置里是否允许 PowerShell/脚本宿主通知。');
    return;
  }

  log(`启动监听 | 端口 ${cfg.debugPort} | 页面匹配 "${cfg.urlMatch}" | 轮询 ${cfg.pollSeconds}s | webhook ${(cfg.webhooks || []).length} 个`);

  let prev = loadPrev();
  if (prev) log(`已载入上次状态（${prev.size} 个计数器）`);

  let lastHeartbeat = Date.now();
  process.on('SIGINT', () => { log('收到 Ctrl+C，退出'); process.exit(0); });
  process.on('uncaughtException', (e) => log('未捕获异常:', e.message));

  while (true) {
    try {
      prev = await tick(prev);
    } catch (e) {
      log('轮询异常:', e.message);
    }
    if (cfg.heartbeatMinutes > 0 && Date.now() - lastHeartbeat > cfg.heartbeatMinutes * 60000) {
      lastHeartbeat = Date.now();
      const s = prev ? [...prev.values()].reduce((a, b) => a + b, 0) : 0;
      log(`心跳：运行中，当前未读合计 ${s}`);
    }
    await sleep(Math.max(2, cfg.pollSeconds) * 1000);
  }
})();