oil-price-monitoring/docs/23.后端定时任务详细设计.md

第三部分：后端定时任务详细设计

调度器：robfig/cron/v3，时区 Asia/Shanghai。 所有任务内部通过 Go context.Context 控制超时（单次任务上限 10 分钟）；任务间通过内存事件总线（简单 channel / 直接函数调用）衔接，不依赖消息队列。

---

T1 · 国际原油价格定时采集

Cron：0 5 * * *（每天 05:00 北京时间；布伦特/WTI 已收盘）

输入：无（外部数据源：新浪财经）

处理流程

1. 构建请求：GET https://hq.sinajs.cn/list=hf_CL,hf_OIL
   Header: Referer: https://finance.sina.com.cn
2. 响应体为 GBK → 转 UTF-8（golang.org/x/text/encoding/simplifiedchinese）
3. 解析两个报价块（形如 var hq_str_hf_CL="..."，逗号分隔）：
   字段索引：
     [0] 最新价   [7] 昨结算  [12] 日期(YYYY-MM-DD)  [13] 名称
4. 日期字段作为 quote_date（防跨日脏写）
5. 校验数值非零 → 构造 crude_quotes 记录
6. 幂等写入（quote_date 冲突 → UPDATE）
7. 调用 T4 预测计算（同进程内）

DB 操作（SQL）

INSERT INTO crude_quotes (quote_date, brent_price, wti_price, updated_at)
VALUES (?, ?, ?, datetime('now','localtime'))
ON CONFLICT(quote_date) DO UPDATE SET
  brent_price = excluded.brent_price,
  wti_price   = excluded.wti_price,
  updated_at  = excluded.updated_at;

异常处理

• 网络失败：指数退避重试 3 次（1/2/4 分钟），仍失败 → 写 notify_logs（告警）+ 日志 error，本轮结束（次日 cron 自动补）。
• 解析失败（页面结构变化）：日志 error + 保留原始响应到 logs/raw/ 便于排查。
• 周末/节假日：新浪接口仍返回最近交易日数据，以响应日期为准，不产生脏数据。

任务执行日志：所有定时任务经统一包装器 runTask 注册（internal/cron/scheduler.go），记录执行生命周期——

• 开始：level=DEBUG msg=定时任务开始 task=<任务名>（上线 info 静默）
• 正常结束：level=INFO msg=定时任务完成 task=<任务名> dur_ms=<耗时>
• 业务失败：level=ERROR msg=定时任务失败 task=<任务名> dur_ms=<耗时> err=<原因>
• 异常中断：level=ERROR msg=定时任务异常 task=<任务名> dur_ms=<耗时> panic=<内容>（panic 被 recover，不连累调度器）

衔接：成功后 → T4；写 server_settings.last_crawl_crude。

---

T2 · 发改委调价公告轮询采集

触发：内部调度器"高频监测模式"——非固定 cron，由代码控制：

- 平时：每 30 分钟 检查一次"是否需要进入高频监测"
- 判定：从最近一次 price_adjustments.effective_at 推算下一调价窗口日 D；
        若今天 ∈ [D-1, D]，则 16:30~18:30 每 2 分钟执行一次采集
- 采到新公告 → 立即退出高频模式（当日不再轮询）

处理流程

1. 抓列表页：GET https://www.ndrc.gov.cn/xwdt/xwfb/
2. goquery 解析文章条目（<a> 标题 + 日期）
3. 过滤：标题 contains "成品油价格按机制调整" 且 发布日期 == 今天
4. 未命中 → 返回"未发布"，等待下一 tick
5. 命中 → 抓详情页
6. 正文正则提取（正文被 HTML <span> 切碎，正则需容忍空白/标签）：

   re := `每吨(?:分别)?(?:降低|下调|上调|提高)\s*(\d+)\s*元和\s*(\d+)\s*元`
   group[1] = 汽油元/吨   group[2] = 柴油元/吨

   方向：含"降低/下调"→ down；含"上调/提高"→ up

   生效时间：`自\s*(\d+)月(\d+)日24时起` → 当年该日 24:00

7. 幂等：SELECT id FROM price_adjustments WHERE source_url = ? → 已存在则跳过
8. 落库 price_adjustments
9. 依次触发：T3 省份价格同步 → T6 官方锤定推送

DB 操作（SQL）

-- 幂等检查
SELECT id FROM price_adjustments WHERE source_url = ?;

-- 落库
INSERT INTO price_adjustments
(announcement_title, announced_at, effective_at,
 gas_delta, diesel_delta, direction, source_url, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now','localtime'));

异常处理

• 列表页解析失败：告警（notify_logs）+ 降级数据源（探数/接口盒子当日是否调价，人工核对）；页面改版时保留原始 HTML 到 logs/raw/。
• 正则不匹配：日志 error + 原文快照保留，人工介入；不产生错误入库。
• 频繁空转（已发布但未识别）：连续 6 次（12 分钟）未命中后暂停本轮，日志 warning。

衔接：成功后 → T3、T6；写 server_settings.last_crawl_adjustment。

---

T3 · 各省零售价同步

Cron：30 5 * * *（每日 05:30 兜底）+ 被 T2 成功触发

处理流程

1. 判定是否需要：读 server_settings.last_province_sync_adjustment_id；
   若 最新 price_adjustments.id > 该值 → 需要同步
2. 抓取 http://www.qiyoujiage.com（PC 首页表格）
3. goquery 解析表格：行=省份，列=92/95/98/0
   - 校验列头含"92号""95号""98号""0号"
   - 省份名与 31 省名单比对，非法行跳过
4. 事务写入：先 DELETE 该 adjustment_id 旧行，再批量 INSERT（防重跑重复）
5. 更新 server_settings.last_province_sync_adjustment_id

DB 操作（SQL）

BEGIN;
DELETE FROM province_prices WHERE adjustment_id = ?;

INSERT INTO province_prices
    (adjustment_id, province, price_92, price_95, price_98, price_0, effective_date)
VALUES (?, ?, ?, ?, ?, ?, ?);   -- 每省一行，共 31 行

UPDATE server_settings SET value = ? WHERE key = 'last_province_sync_adjustment_id';
COMMIT;

异常处理

• 抓取/解析失败：指数退避重试 3 次；仍失败 → 告警，保留下次兜底（05:30）再试；last_province_sync_adjustment_id 不更新。
• 某省缺列/空值：该行跳过并在日志记录，不影响其他省。

衔接：写 server_settings.last_crawl_province_prices。

---

T4 · 调价预测计算与置信度更新

Cron：5 5 * * *（每日 05:05，紧跟 T1）

处理流程

1. 取最近调价 effective_at 作为"周期起点 S"
2. 推算 target_date = S 之后第 10 个工作日（工作日函数见附录）
3. 取原油数据：
   avg_now   = AVG(brent_price)  WHERE quote_date ∈ (S, 今日]
   avg_prev  = AVG(brent_price)  WHERE quote_date ∈ (S-10工作日, S]
4. 变化率 R = (avg_now - avg_prev) / avg_prev × 100
5. 校准系数 K：
   取最近 3 次已发生的调价记录，K = AVG(实际 gas_delta / 当时估算变化率)
   （一期简化：K 取常量 6.0 元/吨·百分点，后续用历史数据自动校准）
6. delta_per_ton = R × K（取整）
7. 判定：
   delta_per_ton ≥ 50   → direction=up
   delta_per_ton ≤ -50  → direction=down
   else                 → direction=stalled
8. delta_per_liter = delta_per_ton / 1350（round 2 位）
9. 置信度（按剩余工作日 n = S→target_date 之间剩余工作日）：
   n ≤ 2  → high
   3 ≤ n ≤ 7 → medium
   n ≥ 8  → low
10. 落库 predictions（每日一条，保留历史）

DB 操作（SQL）

-- ① 周期均价
SELECT AVG(brent_price) FROM crude_quotes WHERE quote_date BETWEEN ? AND ?;

-- ② 近 3 次调价（校准用）
SELECT gas_delta FROM price_adjustments
ORDER BY announced_at DESC LIMIT 3;

-- ③ 落库
INSERT INTO predictions
(target_date, direction, delta_per_ton, delta_per_liter,
 confidence, reason, created_at)
VALUES (?, ?, ?, ?, ?, ?, datetime('now','localtime'));

异常处理：原油数据不足（<5 个交易日）→ 本轮预测落 confidence=low 且 reason 标注"数据不足"；无任何调价历史 → 跳过预测（predictions 为空，前端显示"暂无预测"）。

---

T5 · T-1 日预测预警推送

Cron：0 20 * * *（每天 20:00）

处理流程

1. 计算下一个调价窗口日 target_date（同 T4 推算）
2. 若 target_date != 明天（或最近一个工作日）→ 本轮结束
3. 取最新 predictions（target_date 匹配）
4. 读全局阈值 server_settings.threshold_per_liter（默认 0.05）
5. 若 |delta_per_liter| < 全局阈值 或 direction=stalled → 记录"未达阈值"，不推送
6. 否则：遍历所有"已配置推送渠道"的登录用户：
   a. 读该用户 user_settings（个人渠道 / 个人阈值 / 默认省）
   b. 按用户个人阈值二次判断（个人可覆盖全局阈值）；未达则跳过该用户
   c. 读该用户 vehicles → 计算每车成本（复用 A8 计算逻辑）
   d. 组装个性化预警消息（含 ta 的车辆与省份，模板见 10.概要设计.md 第九章）
   e. 调用 T7 通用发送器（传 user_id）→ 写 notify_logs

DB 操作（SQL）

-- ① 全局阈值
SELECT value FROM server_settings WHERE key = 'threshold_per_liter';

-- ② 全部已配置渠道的用户（user_settings 中存在 push_webhook_* 或 receiver_email）
SELECT DISTINCT user_id FROM user_settings
WHERE key IN ('push_webhook_feishu','push_webhook_wecom','push_webhook_dingtalk','receiver_email')
  AND value != '';

-- ③ 某用户偏好 + 车辆（按 user 循环执行）
SELECT key, value FROM user_settings WHERE user_id = ?;
SELECT id, name, province, fuel_type, tank_capacity, fuel_consumption, monthly_mileage
FROM vehicles WHERE user_id = ?;

-- ④ 推送日志（带 user_id）
INSERT INTO notify_logs(user_id, channel, notify_type, title, content, status, error, created_at)
VALUES (?, ?, 'pre_alert', ?, ?, ?, ?, datetime('now','localtime'));

---

T6 · 调价日官方锤定推送

触发：T2 落库成功后（非 cron，事件驱动）

处理流程

1. 读取本次调价（gas/diesel_delta、direction、effective_at）
2. 遍历所有"已配置推送渠道"的登录用户（SQL 同 T5 ②）：
   a. 读该用户 vehicles + user_settings.default_province
   b. 读该用户默认省最新价格（无则用全国行情）
   c. 计算该用户每车一箱油/月度成本变化（复用 A8 逻辑）
   d. 组装个性化"锤定通知"（方向/幅度/每升/每车建议）
   e. 调用 T7 通用发送器（传 user_id）→ 写 notify_logs（逐渠道一条）
3. 无任何已配置用户 → 仅写系统日志

DB 操作（SQL）

-- ① 本次调价
SELECT id, announced_at, effective_at, gas_delta, diesel_delta, direction
FROM price_adjustments ORDER BY announced_at DESC LIMIT 1;

-- ② 已配置渠道的用户（同 T5 ②）
-- ③ 按用户循环：user_settings + vehicles + 该省最新价格（SQL 同 A1/A8）
-- ④ 推送日志（带 user_id）
INSERT INTO notify_logs(user_id, channel, notify_type, title, content, status, error, created_at)
VALUES (?, ?, 'announce', ?, ?, ?, ?, datetime('now','localtime'));

---

T7 · 通用发送器（被 T5/T6/A11 复用）

入参：user_id（可空，测试/匿名时为空）、channel(feishu/wecom/dingtalk/smtp)、title、content_md

流程

1. 读渠道配置：
   - 飞书/企微/钉钉 webhook：来自 user_settings（user_id 指定）或 server_settings（匿名/系统告警用兜底）
   - SMTP 服务器：smtp_host/port/user/pass 来自 server_settings（全局）；收件人 receiver_email 来自 user_settings
2. 渠道未配置 → 返回 error("channel_not_configured")
3. 发送：
   - Webhook 类：POST {url}，Body:
       飞书: {"msg_type":"text","content":{"text": title+"\n"+content_md}}
       企微: {"msgtype":"markdown","markdown":{"content": ...}}
       钉钉: {"msgtype":"markdown","markdown":{"title":title,"text":...}}
   - SMTP：构建 text/plain + text/html 双 MIME 邮件发送
4. 重试：失败指数退避 3 次（1/2/4 分钟）
5. 返回 (status, detail)

DB 操作（SQL）

-- 用户渠道（有 user_id 时）
SELECT key, value FROM user_settings WHERE user_id = ? AND key = ?;
-- 全局配置（SMTP 服务器 / 系统告警兜底渠道）
SELECT value FROM server_settings WHERE key = ?;

异常处理：3 次重试仍失败 → 返回失败；调用方决定是否写 notify_logs(status='failed', error=...)。