智叟云集定时采集与消息推送架构——实现可恢复的订阅通知流程
一、为什
么采集成功不等于服务完成
智叟云集系统面向招标、政策和行业资讯扩展时，业务流程需要从“采集到数据”继续延伸到“把相关信息送到正确客户”。
完整流程包括：
定时执行
→ 获取内容
→ 判断新增或变化
→ 匹配客户订阅
→ 生成消息
→ 邮件或企业微信投递
→ 保存结果
任何一个阶段出错，都可能导致漏通知、重复通知或错误发送。
因此，需要分别管理采集任务、内容事件和通知任务。
二、整体模块框架
前端管理台
├─ 数据源配置
├─ 订阅条件
├─ 执行时间
└─ 消息记录
↓
业务 API 与租户权限
↓
定时调度器
↓
采集任务队列
↓
网站适配器 / 浏览器采集器
↓
内容版本与事件
↓
订阅匹配
↓
通知发件箱
↓
邮件适配器 / 企业微信适配器
建议代码结构：
src/modules/
sources/
subscriptions/
schedules/
content-events/
notifications/
src/workers/
collection-worker.js
analysis-worker.js
notification-worker.js
src/infrastructure/
database/
browser/
mail/
wecom/
前端配置规则，后端执行规则。不能依靠用户一直打开网页，才能让定时任务继续运行。
三、第一步：定义清楚调度语义
“每 30 分钟执行一次”至少需要明确：
- 使用哪个时区；
- 是否限定执行时段；
- 起始时间如何对齐；
- 上一次还未结束时怎么办；
- 服务停机后是否补跑。
例如：
{
"timezone": "Asia/Shanghai",
"startTime": "08:00",
"endTime": "20:00",
"intervalMinutes": 30,
"overlapPolicy": "skip",
"misfirePolicy": "run_once"
}
这里的字段是扩展设计示例。
建议把结束时刻设为明确的排除边界。例如 [08:00, 20:00)，最后一个执行时刻为 19:30，而不是再执行一次 20:00。
本次本地调度示例采用固定 UTC+8、同日时间窗口，不覆盖夏令时和跨午夜时段。生产系统需要选择明确支持时区的调度实现。
四、第二步：防止任务重叠与重启丢失
每个计划执行时刻生成唯一任务标识：
调度编号 + 计划执行时间
数据库增加唯一约束，避免调度器重启或多个实例同时扫描时重复创建任务。
Worker 领取任务时，需要记录：
- 领取者；
- 租约截止时间；
- 执行开始时间；
- 心跳；
- 尝试次数。
任务状态可以设计为：
queued → running → completed
├─ retry_wait
├─ waiting_review
└─ failed
验证码、登录失效等进入人工处理，不进行无限重试。
对于 SQLite，可以使用短事务组织状态变更，但仍需要处理写入竞争。BEGIN IMMEDIATE 在已有写事务时可能返回 SQLITE_BUSY，不能认为使用事务后就不存在并发问题。SQLite 事务说明
五、第三步：让浏览器采集稳定执行
浏览器自动化应围绕明确的页面条件执行：
- 打开目标页面。
- 确认关键区域加载。
- 输入查询条件。
- 等待列表更新。
- 读取公告链接。
- 进入详情。
- 获取正文和附件。
- 保存执行证据。
不要用大量固定延时替代页面条件检查。
Playwright 提供元素操作前的自动等待与可操作性检查，可以减少因为元素尚未就绪导致的失败，但业务上仍需判断列表是否真正更新。Playwright 自动等待说明
每个适配器还应有明确的成功条件。例如“正文长度非空且标题存在”，而不是页面返回 HTTP 200 就认为采集成功。
六、第四步：将内容保存与消息事件放进同一个事务
常见故障是：
内容保存成功
↓
进程崩溃
↓
通知事件没有创建
为避免这个缺口，可以采用事务发件箱：
同一数据库事务
├─ 保存内容新版本
└─ 保存待处理事件
↓
后台读取事件
↓
匹配订阅并创建通知
发送消息不要占用数据库事务。网络可能缓慢或失败，应该先提交本地事务，再由通知 Worker 调用外部通道。
七、第五步：通知去重必须覆盖收件人和渠道
通知唯一标识应包括：
租户
订阅编号与版本
内容编号与版本
通知渠道
接收目标
这样能够区分：
- 同一客户的邮件和企业微信；
- 同一内容的初次通知和变更通知；
- 不同接收人的通知；
- 同一任务的重复处理。
示例：
// notifications/delivery-key.mjs
import {
createHash } from "node:crypto";
export function deliveryKey(input) {
return createHash("sha256")
.update(JSON.stringify([
input.tenantId,
input.subscriptionId,
input.subscriptionVersion,
input.documentId,
input.documentVersion,
input.channel,
input.destinationId
]))
.digest("hex");
}
本地去重可以阻止重复入队，但无法天然保证外部通道“绝对只收到一次”。
如果发送成功后程序崩溃、尚未记录成功，再次重试可能重复发送。支持幂等请求的通道可利用其幂等能力；不支持的通道需要记录请求结果、消息标识和异常对账策略。
八、第六步：区分发送状态与阅读状态
建议通知状态为：
| 状态 | 含义 |
|---|---|
queued |
已进入发送队列 |
sending |
正在提交 |
accepted |
通道已接受 |
retry_wait |
等待重试 |
failed |
发送失败 |
unknown |
结果不明确，需要核对 |
邮件服务接受请求，不等于邮件进入收件箱；企业微信接口接受消息，也不等于成员已经阅读。
页面应展示能被实际确认的状态，不把“接受成功”直接标注成“客户已读”。
还需要分别处理：
- 临时网络故障；
- 通道限流；
- 收件地址错误；
- 凭据失效；
- 消息内容不符合通道要求；
- 超时但结果不明确。
具体返回字段、发送限制和重试规则，应按接入时的通道文档实现。
九、第七步：支持即时提醒和每日摘要
两种通知模式适合不同需求：
| 模式 | 适用信息 |
|---|---|
| 即时提醒 | 关键截止时间变化、重要项目更新 |
| 每日摘要 | 新发现招标、政策和行业资讯 |
每日摘要需要保存时间窗口。例如今天 18:00 汇总前一窗口中的新增记录，使用左闭右开区间，避免窗口边界重复。
内容变更和新内容也应分开显示：
今日新增项目 6 项
重要变更 2 项
相关政策 3 条
如果某次摘要失败，恢复时应重新发送同一摘要任务，而不是重新选取一个不断变化的数据集合。
十、第八步：本地部署与安全边界
该扩展可以随智叟云集部署在客户自己的服务器中：
- 业务数据和内容版本存储在本地；
- 网站登录状态按租户隔离；
- 邮件与企业微信配置由客户管理；
- 内部业务系统通过 API 查询；
- 外部访问通过 HTTPS 和权限控制。
本地部署不等于完全离线。新增采集和外部通知仍需要连接目标网站或消息服务。
还应注意：
- 来源 URL 和回调地址限制，防止访问内部敏感地址；
- 附件下载大小和类型检查；
- 解析器隔离，避免不可信附件影响主服务；
- 密钥加密或受限保存；
- 日志隐藏 Cookie、Token 和完整收件人信息；
- 订阅、接收目标和内容归属都检查租户权限。
十一、本次已完成哪些验证
本次附带的验证代码分为：
verification/
content.mjs
delivery.mjs
verification.test.mjs
运行说明.txt
无需安装第三方依赖，在 verification 目录执行：
node --test verification.test.mjs
实际结果：
tests 13
pass 13
fail 0
覆盖内容：
| 验证项 | 结果 |
|---|---|
| 重复内容不生成新版本 | 通过 |
| 截止时间变化生成新版本 | 通过 |
| 附件变化识别 | 通过 |
| 不同租户记录隔离 | 通过 |
| 地区、类型、关键词匹配 | 通过 |
| 跨租户订阅拒绝 | 通过 |
| 证据片段与原文核对 | 通过 |
| 同渠道通知重复入队拦截 | 通过 |
| 不同渠道独立通知 | 通过 |
| 内容更新可产生新通知 | 通过 |
| 跨租户通知拒绝 | 通过 |
| UTC+8 执行窗口边界 | 通过 |
| 非法调度参数拒绝 | 通过 |
这些测试使用内存状态和本地样本，验证的是规则逻辑。生产上线仍需要补充数据库持久化、并发、任务恢复、网站适配和真实通知通道的验收。
通过这种分层实现方式，智叟云集可以逐步从品牌监测扩展到行业信息服务：既能发现信息，也能解释来源、跟踪变化，并将相关内容可靠地交付给客户。
