From 2f2428aa9a8a68317df982aeba48021d14211f2e Mon Sep 17 00:00:00 2001 From: Simon Date: Wed, 5 Aug 2026 08:34:02 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=97=A5=E6=8A=A5=E6=91=98=E8=A6=81?= =?UTF-8?q?=E5=8F=AF=E9=9D=A0=E6=80=A7(=E5=8E=BB=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E5=85=9C=E5=BA=95+=E9=87=8D=E8=AF=95)=20=E4=B8=8E=E5=8F=96?= =?UTF-8?q?=E6=95=B0=E9=80=BB=E8=BE=91=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - llm/client: 移除内置默认模型兜底(deepseek-chat/qwen-plus), 模型必须显式配置否则报错 - reporter._llm_call: 指数退避重试(LLM_RETRY_TIMES 默认3 / LLM_RETRY_BACKOFF_SEC 默认2s) - pipeline report: report_date 改为当天(原昨天+回溯3天) - reporter._collect_news_events: 读当天+前一天目录, publish_time 30h 回溯(NEWS_LOOKBACK_HOURS=30), 统一时区 - reporter._collect_xwlb: 固定取 day_str 前一日(已播出联播), source_date 同步 - 公告/调研/互动保持近15日设置(CNINFO_DAYS_BACK), 不受 30h 影响 - 测试: 新增 30h回溯/带时区/重试/模型缺失/xwlb 前一日 用例 --- .env.example | 4 + continuation.md | 57 +++++++++++++-- llm/client.py | 19 ++--- scheduler/pipeline.py | 32 +------- scheduler/reporter.py | 96 +++++++++++++++++------- tests/test_llm.py | 17 ++++- tests/test_report_builder.py | 138 +++++++++++++++++++++++++++++++++++ 7 files changed, 293 insertions(+), 70 deletions(-) diff --git a/.env.example b/.env.example index 94851ca..67b7106 100644 --- a/.env.example +++ b/.env.example @@ -77,3 +77,7 @@ NEWS_DB_USER=myquant NEWS_DB_PASSWORD= # 填真实值，禁止写入源码/文档 NEWS_DB_NAME=myquant REPORT_HISTORY_DIR=data/reports_history + +# ---- LLM 日报摘要重试 ---- +LLM_RETRY_TIMES=3 # AI 摘要调用失败重试次数(默认 3) +LLM_RETRY_BACKOFF_SEC=2.0 # 指数退避基数,秒(默认 2.0: 2s,4s,8s...) diff --git a/continuation.md b/continuation.md index ad896b6..5f1c2cb 100644 --- a/continuation.md +++ b/continuation.md @@ -1,6 +1,6 @@ # continuation.md -> `checkpoint` @ 2026-08-03 21:30 +> `checkpoint` @ 2026-08-05 08:30 --- @@ -10,14 +10,61 @@ | --- | --- | | 新闻源 | 14 个（13 Web + 1 API: xwlb 新闻联播） | | Qdrant | 本地文件模式 `data/qdrant_storage/` | -| 日报 | **M10 完成: 结构化入库 MySQL（myquant 库 news_report/news_event 表），不再生成 HTML** | -| 调度器 | APScheduler，systemd `a-share-research.service` | -| LLM | `deepseek-v4-flash`（绝不允许擅自修改） | -| 服务器 | `pi@192.168.1.160`，项目 `/home/pi/news/` | +| 日报 | **M10 完成并已部署 pi5: 结构化入库 MySQL；日报按当天日期生成（新闻 30h 回溯 / xwlb 取前一日 / 公告调研近 15 日）** | +| DB 连接 | pi 上 systemd 服务 `a-share-db-tunnel` 常驻（0.0.0.0:13306 → doorcome.cn:3306）；**pi5 直连 192.168.1.10:13306** | +| 调度器 | APScheduler，systemd `a-share-research.service`（pi5）；每天 07:00 首次任务生成日报（12/18/22 点不生成） | +| LLM | `deepseek-v4-flash`（绝不允许擅自修改；模型必须显式配置，无内置兜底） | +| 服务器 | `pi@192.168.1.160`（生产）/ `pi@192.168.1.10`（DB 隧道宿主） | | 抓取方式 | js_render=false → httpx 直连；js_render=true → Playwright | --- +## 本次完成 (2026-08-05) — 日报可靠性修复与取数逻辑优化 + +**目标：** 解决日报 AI 摘要偶发失败；修正日报日期与 xwlb/新闻取数语义。 + +**1. AI 摘要可靠性（llm/client.py + scheduler/reporter.py）：** +- 去掉内置默认模型兜底（`deepseek-chat`/`qwen-plus`），模型必须显式配置（`DEEPSEEK_MODEL`/`QWEN_MODEL` → `LLM_MODEL`），缺失即报错，避免静默用错模型 +- `_llm_call` 增加指数退避重试：`_LLM_RETRY_TIMES`（默认 3 次）、`_LLM_RETRY_BACKOFF_SEC`（默认 2.0s，可 .env 覆盖），全部失败才抛异常 +- 确认 AI 摘要模型：`deepseek` + `deepseek-v4-flash`（生产实测） + +**2. 日报取数逻辑（scheduler/pipeline.py + reporter.py）：** +- pipeline report 步骤：`report_date = date.today()`（原为昨天+回溯 3 天） +- `_collect_news_events`：读当天+前一天事件目录，按 `publish_time` 过滤最近 30 小时（`_NEWS_LOOKBACK_HOURS=30`）；时区统一（naive 假定本地时区）；无时间戳事件保留 +- `_collect_xwlb`：固定查 `day_str` 前一日（《新闻联播》19:00 播出，早间日报只能取昨晚已播出的一期）；`source_date` 同步为实际来源日 +- **公告/调研/互动保持原设置：近 15 日（`STOCK_REPORT_DAYS=15`），irm 互动仍跳过**——未受 30h 改动影响 + +**验证（pi5）：** +- 单测 9 个（30h 回溯/带时区时间戳/重试/模型缺失报错/xwlb 前一日）全部通过；全量 212 passed + 3 crawler 预存在失败 +- 生产端到端 report_id=191（2026-08-05）：news 20 + cninfo 20 + xwlb 34（08-04 联播），AI 摘要 2076 字 +- 生产服务已重启生效 + +**本次代码尚未 git 提交（见待办）。** + +--- + +## 本次完成 (2026-08-03 22:00) — pi5 部署与生产测试 + +**操作：** M10 代码全量同步 pi5 + 生产环境验收（所有测试在 pi5 执行，Mac 不再作为测试环境）。 + +**文件同步：** rsync 本地 → `pi5:/home/pi/news/`（排除 .venv/data/logs/.git/.env/configs/*.yaml）；清除 pi5 根目录 6 月 17 日旧版散文件（已 tar 备份 /tmp/news_backup_20260803.tar.gz）；pi5 `uv sync` 装 pymysql。 + +**DB 隧道修复：** pi 原 autossh 参数 `-L 13306:0.0.0.0:3306` 未生效（0.0.0.0 被当远端目标），改为 `-L 0.0.0.0:13306:127.0.0.1:3306` 并持久化为 systemd 服务 `a-share-db-tunnel`（enabled + active）。 + +**测试结果（pi5）：** +- 全量 pytest：**208 passed, 3 failed**（3 个失败均为 crawler retry mock 预先存在问题，与 M10 无关） +- `report-import` 幂等：已存在文件正确 skipped +- 生产端到端：`a-share report --date 20260802/20260803` → report_id=181/182 入库成功（AI 摘要正常，57/58 事件） +- DB 总量：finance 52 + intl 128 = 180 行（历史 177 + 端到端测试 2 + 本机 1） +- 生产服务 `a-share-research` 已重启 active，22:00 定时任务起用新代码 + +**已知问题：** +- `tests/test_crawler.py` 3 个 retry 测试失败（预先存在，crawl4ai mock 行为） +- Mac 本机 anaconda/uv python 出站到 192.168.1.10 被拦截（EHOSTUNREACH；nc/bash/系统 python 正常）——仅影响本机，pi5 不受影响；Mac 本地用 `127.0.0.1` + ssh 隧道绕过 +- 生产 pi5 的 `.env` 里 `NEWS_DB_HOST=192.168.1.10`（直连）；Mac 本地 `.env` 为 `127.0.0.1`（隧道）——**两处 .env 不同，勿互相覆盖** + +--- + ## 本次完成 (2026-08-03) — M10 日报结构化入库 **目标：** 日报前后端分离的数据层——日报内容结构化存入 MySQL（`news_` 前缀表），历史 178 份日报 HTML 解析入库；本项目不做 API/前端（用户另行实现）。 diff --git a/llm/client.py b/llm/client.py index 4332e60..3396721 100644 --- a/llm/client.py +++ b/llm/client.py @@ -7,7 +7,8 @@ DeepSeek: DEEPSEEK_API_KEY / DEEPSEEK_BASE_URL / DEEPSEEK_MODEL Qwen: QWEN_API_KEY / QWEN_BASE_URL / QWEN_MODEL (QWEN_API_KEY -> DASHSCOPE_API_KEY 兜底) - LLM_MODEL (兜底) LLM_TEMPERATURE / LLM_TIMEOUT_SEC + 模型必须显式配置(provider 对应的 *_MODEL 或 LLM_MODEL),不再提供内置默认模型。 + LLM_TEMPERATURE / LLM_TIMEOUT_SEC """ from __future__ import annotations @@ -22,10 +23,6 @@ from openai import AsyncOpenAI, OpenAI _DEEPSEEK_DEFAULT_BASE = "https://api.deepseek.com" _QWEN_DEFAULT_BASE = "https://dashscope.aliyuncs.com/compatible-mode/v1" -# 默认模型 -_DEEPSEEK_DEFAULT_MODEL = "deepseek-chat" -_QWEN_DEFAULT_MODEL = "qwen-plus" - # 抽取任务默认参数 DEFAULT_TIMEOUT_SEC = 60.0 DEFAULT_TEMPERATURE = 0.1 @@ -69,13 +66,17 @@ def load_llm_config( if p == "deepseek": api_key = _read_env("DEEPSEEK_API_KEY") or "" base = _read_env("DEEPSEEK_BASE_URL", _DEEPSEEK_DEFAULT_BASE) or _DEEPSEEK_DEFAULT_BASE - # DEEPSEEK_MODEL → LLM_MODEL(兜底) → 默认 - m = model or _read_env("DEEPSEEK_MODEL") or _read_env("LLM_MODEL") or _DEEPSEEK_DEFAULT_MODEL + # DEEPSEEK_MODEL → LLM_MODEL；模型必须显式配置，不提供内置默认 + m = model or _read_env("DEEPSEEK_MODEL") or _read_env("LLM_MODEL") + if not m: + raise ValueError("未配置 LLM 模型: 请设置 DEEPSEEK_MODEL 或 LLM_MODEL") elif p in ("qwen", "dashscope"): api_key = _read_env("QWEN_API_KEY") or _read_env("DASHSCOPE_API_KEY") or "" base = _read_env("QWEN_BASE_URL", _QWEN_DEFAULT_BASE) or _QWEN_DEFAULT_BASE - # QWEN_MODEL → LLM_MODEL(兜底) → 默认 - m = model or _read_env("QWEN_MODEL") or _read_env("LLM_MODEL") or _QWEN_DEFAULT_MODEL + # QWEN_MODEL → LLM_MODEL；模型必须显式配置，不提供内置默认 + m = model or _read_env("QWEN_MODEL") or _read_env("LLM_MODEL") + if not m: + raise ValueError("未配置 LLM 模型: 请设置 QWEN_MODEL 或 LLM_MODEL") p = "qwen" # 内部统一用 qwen else: raise ValueError(f"未知 LLM provider: {p!r},仅支持 deepseek / qwen") diff --git a/scheduler/pipeline.py b/scheduler/pipeline.py index 30f99ec..c37251d 100644 --- a/scheduler/pipeline.py +++ b/scheduler/pipeline.py @@ -74,39 +74,15 @@ def run_step(name: str, date_str: str) -> StepResult: 返回: StepResult。 """ # report 步骤:内部函数,不走子进程 - # 日报默认统计"昨天"的数据（因为今天的数据由当天的定时任务处理）。 - # 如果昨天没有事件数据，向前回溯最多 3 天，取最近有数据的日期。 + # 日报按当天日期生成: 新闻由 _collect_news_events 回溯过去 30 小时, + # xwlb 由 _collect_xwlb 固定取前一日(已播出)联播。 if name == "report": started = datetime.now() try: - from datetime import timedelta # noqa: E402 - from pathlib import Path # noqa: E402 - from .reporter import generate_report # noqa: E402 - # 向前回溯找最近有事件数据的日期（最多回溯 3 天） - report_date: str | None = None - for offset in range(1, 4): - candidate = (date.today() - timedelta(days=offset)).strftime("%Y%m%d") - ev_dir = Path(f"data/events/{candidate}") - if ev_dir.is_dir() and list(ev_dir.glob("*.json")): - report_date = candidate - break - - if report_date is None: - # 没有任何事件数据，仍然尝试生成昨天日报（至少展示管道统计） - report_date = (date.today() - timedelta(days=1)).strftime("%Y%m%d") - logger.warning( - "日报: 近 3 日均无事件数据 ({} ~ {}), 日报将只含管道统计", - (date.today() - timedelta(days=3)).strftime("%Y%m%d"), - (date.today() - timedelta(days=1)).strftime("%Y%m%d"), - ) - elif report_date != (date.today() - timedelta(days=1)).strftime("%Y%m%d"): - logger.warning( - "日报: 昨天 ({}) 无事件数据, 回退使用 {}", - (date.today() - timedelta(days=1)).strftime("%Y%m%d"), - report_date, - ) + report_date = date.today().strftime("%Y%m%d") + logger.info("日报: report_date={} (新闻 30h 回溯, xwlb 前一日)", report_date) path = generate_report(report_date, upload=True) elapsed = (datetime.now() - started).total_seconds() diff --git a/scheduler/reporter.py b/scheduler/reporter.py index 26b3af9..86351bf 100644 --- a/scheduler/reporter.py +++ b/scheduler/reporter.py @@ -13,6 +13,7 @@ import json import os as _os import re as _re import subprocess +import time from collections import Counter from datetime import date, datetime, timedelta from pathlib import Path @@ -38,6 +39,13 @@ NEWS_DAYS_BACK = 1 # 新闻回溯天数 _MAX_HIGH_EVENTS = 20 +# LLM 摘要调用重试参数（环境变量可覆盖） +_LLM_RETRY_TIMES = int(_os.environ.get("LLM_RETRY_TIMES", "3")) +_LLM_RETRY_BACKOFF_SEC = float(_os.environ.get("LLM_RETRY_BACKOFF_SEC", "2.0")) + +# 日报新闻回溯窗口（小时）：07:00 生成当日日报时覆盖昨日全天至今晨的新闻 +_NEWS_LOOKBACK_HOURS = 30 + def _load_source_names() -> dict[str, str]: import yaml @@ -102,16 +110,31 @@ def _load_events_from_dir(day_str: str) -> list[dict]: def _collect_news_events(day_str: str) -> dict[str, Any]: """收集新闻事件(排除 cninfo)。 - 事件已按日期目录组织(data/events/{day_str}/), - 不再用 datetime.now() 做 24h 二次过滤, - 避免日报早上 8 点跑时前一天新闻被全部过滤掉。 + 读取 `day_str` 与前一天两个事件目录，按 publish_time 过滤最近 + `_NEWS_LOOKBACK_HOURS`（默认 30）小时内的新闻——07:00 生成当日日报时 + 可覆盖昨日全天至今晨的新闻。无 publish_time 的事件保留(容错)。 """ - all_ev = _load_events_from_dir(day_str) + day = datetime.strptime(day_str, "%Y%m%d").date() + prev_day = (day - timedelta(days=1)).strftime("%Y%m%d") + all_ev = _load_events_from_dir(day_str) + _load_events_from_dir(prev_day) + # publish_time 过滤: 最近 30 小时（时间缺失/格式异常的事件保留） + cutoff = (datetime.now() - timedelta(hours=_NEWS_LOOKBACK_HOURS)).astimezone() news_ev: list[dict] = [] for e in all_ev: if e["source_id"] == "cninfo": continue + pt = e.get("publish_time") + if pt: + try: + # naive 时间假定为本地时区, 与带时区(aware)的 cutoff 统一比较 + t = datetime.fromisoformat(pt) + if t.tzinfo is None: + t = t.astimezone() + if t < cutoff: + continue + except (ValueError, TypeError): + pass # 时间格式异常时保留 news_ev.append(e) sentiments: Counter = Counter() @@ -311,16 +334,21 @@ def _score_xwlb_importance(title: str, content: str = "") -> int: def _collect_xwlb(day_str: str) -> dict[str, Any]: """收集新闻联播要闻（从 doorcome API /api/xwlbFine/ 获取）。 + 《新闻联播》每天 19:00 播出：日报在早上生成时当日联播尚未播出， + 因此固定取 `day_str` 前一日（最近一期已播出）的联播数据。 + API 返回 AI 精编后的独立新闻条目（含标题+正文）， 跳过第 1 条"内容提要"（仅为节目开场白）。 - 返回: {"items": [event_dict, ...], "date": "MM月DD日", "source_date": "20260622"} + 返回: {"items": [event_dict, ...], "date": "MM月DD日", "source_date": "前一日"} """ import urllib.request - result: dict[str, Any] = {"items": [], "date": "", "source_date": day_str} + # 取前一晚(已播出)的联播：day_str 前一天 + prev_day = (datetime.strptime(day_str, "%Y%m%d") - timedelta(days=1)).strftime("%Y%m%d") + result: dict[str, Any] = {"items": [], "date": "", "source_date": prev_day} - api_url = f"https://api.doorcome.cn/api/xwlbFine/?start_date={day_str}&end_date={day_str}" + api_url = f"https://api.doorcome.cn/api/xwlbFine/?start_date={prev_day}&end_date={prev_day}" try: req = urllib.request.Request(api_url) with urllib.request.urlopen(req, timeout=15) as resp: @@ -338,7 +366,7 @@ def _collect_xwlb(day_str: str) -> dict[str, Any]: if dates: d = min(dates) result["date"] = f"{d[5:7]}月{d[8:10]}日" - result["source_date"] = day_str + result["source_date"] = prev_day # 转换为事件格式，跳过第 1 条（内容提要/开场白） events: list[dict] = [] @@ -595,27 +623,43 @@ def _build_prompt(lines: list[str], day_str: str) -> str: def _llm_call(client, model: str, prompt: str, max_tokens: int = 1500) -> str: - """单次 LLM 调用，返回 strip 后的文本。 + """单次 LLM 调用（带重试），返回 strip 后的文本。 + 失败按指数退避重试 `_LLM_RETRY_TIMES` 次（默认 3），全部失败则抛出最后一次异常。 若 finish_reason 为 'length' 则说明达到 max_tokens 上限被截断。 """ - resp = client.chat.completions.create( - model=model, - messages=[ - {"role": "system", "content": "你是 A 股日报撰写助手,输出简洁、有洞察的新闻摘要。"}, - {"role": "user", "content": prompt}, - ], - temperature=0.3, - max_tokens=max_tokens, - ) - content = (resp.choices[0].message.content or "").strip() - finish = getattr(resp.choices[0], "finish_reason", None) - if finish == "length": - logger.warning( - "AI 摘要可能被截断: max_tokens={} finish_reason=length 实际输出 {} 字符", - max_tokens, len(content), - ) - return content + last_exc: Exception | None = None + for attempt in range(_LLM_RETRY_TIMES): + try: + resp = client.chat.completions.create( + model=model, + messages=[ + {"role": "system", "content": "你是 A 股日报撰写助手,输出简洁、有洞察的新闻摘要。"}, + {"role": "user", "content": prompt}, + ], + temperature=0.3, + max_tokens=max_tokens, + ) + content = (resp.choices[0].message.content or "").strip() + finish = getattr(resp.choices[0], "finish_reason", None) + if finish == "length": + logger.warning( + "AI 摘要可能被截断: max_tokens={} finish_reason=length 实际输出 {} 字符", + max_tokens, len(content), + ) + return content + except Exception as e: + last_exc = e + if attempt < _LLM_RETRY_TIMES - 1: + wait = _LLM_RETRY_BACKOFF_SEC * (2 ** attempt) + logger.warning( + "AI 摘要 LLM 调用失败(第 {}/{} 次): {}; {} 秒后重试", + attempt + 1, _LLM_RETRY_TIMES, e, round(wait, 2), + ) + time.sleep(wait) + logger.error("AI 摘要 LLM 调用重试 {} 次仍失败: {}", _LLM_RETRY_TIMES, last_exc) + assert last_exc is not None + raise last_exc # --------------------------------------------------------------------------- # diff --git a/tests/test_llm.py b/tests/test_llm.py index db9d9aa..505e132 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -322,21 +322,24 @@ async def test_extract_event_async_retries(fake_config: LLMConfig) -> None: def test_load_llm_config_deepseek_from_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("LLM_PROVIDER", "deepseek") monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-test-deepseek") + monkeypatch.setenv("DEEPSEEK_MODEL", "deepseek-v4-flash") monkeypatch.delenv("LLM_MODEL", raising=False) cfg = load_llm_config() assert cfg.provider == "deepseek" assert cfg.api_key == "sk-test-deepseek" - assert cfg.model.startswith("deepseek") + assert cfg.model == "deepseek-v4-flash" assert "deepseek" in cfg.base_url def test_load_llm_config_qwen_from_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("LLM_PROVIDER", "qwen") monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test-qwen") + monkeypatch.setenv("QWEN_MODEL", "qwen-plus") monkeypatch.delenv("LLM_MODEL", raising=False) cfg = load_llm_config() assert cfg.provider == "qwen" assert cfg.api_key == "sk-test-qwen" + assert cfg.model == "qwen-plus" assert "dashscope" in cfg.base_url or "aliyuncs" in cfg.base_url @@ -347,5 +350,15 @@ def test_load_llm_config_unknown_provider_raises(monkeypatch: pytest.MonkeyPatch def test_load_llm_config_missing_key_raises(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) - with pytest.raises(ValueError): + monkeypatch.setenv("DEEPSEEK_MODEL", "deepseek-v4-flash") + with pytest.raises(ValueError, match="API key"): + load_llm_config(provider="deepseek") + + +def test_load_llm_config_missing_model_raises(monkeypatch: pytest.MonkeyPatch) -> None: + """去掉内置默认模型后：未显式配置模型必须报错（不再回退 deepseek-chat）。""" + monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-test") + monkeypatch.delenv("DEEPSEEK_MODEL", raising=False) + monkeypatch.delenv("LLM_MODEL", raising=False) + with pytest.raises(ValueError, match="模型"): load_llm_config(provider="deepseek") diff --git a/tests/test_report_builder.py b/tests/test_report_builder.py index e05b35e..3bd1d64 100644 --- a/tests/test_report_builder.py +++ b/tests/test_report_builder.py @@ -5,6 +5,8 @@ from __future__ import annotations import json from datetime import date +import pytest + from scheduler.reporter import _build_report_data @@ -85,3 +87,139 @@ class TestBuildReportData: r = _build_report_data(news, cninfo, {}, None, "20260710") assert r.events == [] assert r.ai_summary is None + + +class TestLlmCallRetry: + """_llm_call 重试逻辑（纯逻辑，mock client）。""" + + @staticmethod + def _fake_client(failures: int): + """构造 mock client：前 failures 次抛 ConnectionError，之后成功。""" + from types import SimpleNamespace + + n = {"count": 0} + + class Completions: + def create(self, **kwargs): + n["count"] += 1 + if n["count"] <= failures: + raise ConnectionError("transient") + return SimpleNamespace( + choices=[SimpleNamespace( + message=SimpleNamespace(content="今日要点摘要"), + finish_reason="stop", + )] + ) + + return SimpleNamespace(chat=SimpleNamespace(completions=Completions())), n + + def test_success_first_try(self) -> None: + from scheduler.reporter import _llm_call + client, n = self._fake_client(0) + out = _llm_call(client, "deepseek-v4-flash", "p") + assert out == "今日要点摘要" + assert n["count"] == 1 + + def test_retry_then_success(self, monkeypatch) -> None: + import scheduler.reporter as rep + monkeypatch.setattr(rep, "_LLM_RETRY_TIMES", 3) + monkeypatch.setattr(rep, "_LLM_RETRY_BACKOFF_SEC", 0.01) + client, n = self._fake_client(2) # 前 2 次失败，第 3 次成功 + out = rep._llm_call(client, "deepseek-v4-flash", "p") + assert out == "今日要点摘要" + assert n["count"] == 3 + + def test_exhausts_retries_raises(self, monkeypatch) -> None: + import scheduler.reporter as rep + monkeypatch.setattr(rep, "_LLM_RETRY_TIMES", 2) + monkeypatch.setattr(rep, "_LLM_RETRY_BACKOFF_SEC", 0.01) + client, n = self._fake_client(99) # 一直失败 + with pytest.raises(ConnectionError): + rep._llm_call(client, "deepseek-v4-flash", "p") + assert n["count"] == 2 # 重试 2 次后放弃 + + +class TestCollectXwlb: + """_collect_xwlb 取数逻辑：应查询日报前一日(已播出的联播)，并跳过内容提要。""" + + def test_queries_previous_day_and_skips_toc(self, monkeypatch) -> None: + import json as _json + import urllib.request + + captured: dict[str, str] = {} + + def fake_urlopen(req, timeout=15): # noqa: ARG001 + captured["url"] = req.full_url + + class Resp: + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def read(self): + return _json.dumps({"data": {"news": [ + {"daily_sub_id": 1, "news_title": "内容提要", "news_days": "2026-08-04", "news_improve": "开场白"}, + {"daily_sub_id": 2, "news_title": "联播要闻A", "news_days": "2026-08-04", "news_improve": "正文A"}, + ]}}).encode("utf-8") + + return Resp() + + monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen) + from scheduler.reporter import _collect_xwlb + + result = _collect_xwlb("20260805") + # 查询的是前一日(20260804)而非当日 + assert "start_date=20260804" in captured["url"] + assert "end_date=20260804" in captured["url"] + # 跳过第 1 条内容提要 + assert len(result["items"]) == 1 + assert result["items"][0]["title"] == "联播要闻A" + assert result["source_date"] == "20260804" + assert result["date"] == "08月04日" + + +class TestCollectNewsEventsLookback: + """_collect_news_events 30 小时回溯逻辑。""" + + def test_filters_30h_and_excludes_cninfo(self, monkeypatch) -> None: + from datetime import datetime, timedelta + + import scheduler.reporter as rep + + now = datetime.now().astimezone() + + def fake_load(day_str: str) -> list[dict]: # noqa: ARG001 + def ev(title: str, hours_ago: float | None, source: str = "cls", + importance: int = 5, aware: bool = False) -> dict: + pt = None + if hours_ago is not None: + t = now - timedelta(hours=hours_ago) + pt = t.isoformat() if not aware else t.astimezone().isoformat() + return { + "title": title, "url": "u", "source_id": source, + "publish_time": pt, + "event": {"importance": importance, "sentiment": "neutral", + "event_type": "其他", "summary": "s"}, + } + + return [ + ev("窗口内新闻", 10), + ev("窗口内新闻带时区", 12, aware=True), + ev("窗口外旧闻", 40), + ev("无时间戳", None), + ev("公告排除", 5, source="cninfo", importance=2), + ] + + monkeypatch.setattr(rep, "_load_events_from_dir", fake_load) + result = rep._collect_news_events("20260805") + + # 两个日期目录各返回 5 条(共 10): 旧闻×2、公告×2 被滤, 保留 6 条 + assert result["total"] == 6 + titles = {e["title"] for e in result["high"]} + assert "窗口内新闻" in titles + assert "窗口内新闻带时区" in titles + assert "无时间戳" in titles + assert "窗口外旧闻" not in titles + assert "公告排除" not in titles