fix: 日报摘要可靠性(去模型兜底+重试) 与取数逻辑优化

- 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 前一日用例
feat: 日报结构化入库（M10 前后端分离数据层）
2026-08-05 08:34:02 +08:00 · 2026-08-03 21:32:07 +08:00
28 changed files with 2225 additions and 101 deletions
@@ -69,3 +69,15 @@ PIPELINE_STEP_TIMEOUT=1800
 # ---- cninfo 公告抓取 ----
 CNINFO_PDF_BASE=http://static.cninfo.com.cn
 # ---- 日报结构化入库 (M10) ----
 NEWS_DB_HOST=127.0.0.1        # 开发走 ssh 隧道: ssh -L 13306:127.0.0.1:13306 pi
 NEWS_DB_PORT=13306
 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...)
@@ -57,3 +57,6 @@ qdrant_storage/
 # 调试输出
 debug/
 tmp/
 # M10: 历史日报源文件副本(可从 doorcome 重新拉取)
 data/reports_history/
@@ -229,7 +229,7 @@ MAX_IMPORTANCE = 5
 统一使用：
-logging
+loguru（现有代码全部使用 loguru，禁止混用 stdlib logging）
 禁止：
@@ -623,5 +623,40 @@ Claude Code 完成任务时必须输出：
 “小步迭代、稳定演进、长期维护”。
 # 二十一、项目现状速查（2026-07 校验）
 ## 入口与常用命令
 - 统一 CLI：`uv run a-share <子命令>`，入口 `a_share_cli/main.py`
 - 子命令：`crawl` / `extract` / `dedup` / `events` / `embed` / `ingest` / `pipeline --once` / `search` / `report` / `status` / `discover`
 - 测试：`uv run pytest`（tests/，asyncio_mode=auto；integration 标记默认跳过）
 - 静态检查：`uv run ruff check .`、`uv run mypy`（pyproject.toml 已配置）
 - 环境：`uv sync`；本地 BGE-M3 需 `uv sync --extra local-embedding`
 ## 架构（9 个包）
 | 包 | 职责 |
 | --- | --- |
 | crawler | Crawl4AI 抓取；js_render=false 走 httpx 静态直连，否则 Playwright；含 cninfo 公告 |
 | extractor | GNE 中文正文提取 |
 | dedup | 新闻去重（SimHash） |
 | llm | DeepSeek/Qwen 投资事件抽取 |
 | embedding | 远程 DashScope / 本地 BGE-M3 |
 | vectorstore | Qdrant；默认本地文件模式 data/qdrant_storage/ |
 | scheduler | APScheduler 调度 + pipeline + 日报 reporter |
 | mcp_server | MCP 工具（供 Cherry Studio） |
 | api | 占位（空包） |
 scripts/ 为独立运行脚本 + systemd 服务文件 `scripts/a-share-research.service`。
 ## 关键约束与事实
 - LLM 模型 `deepseek-v4-flash` 绝不允许修改（记忆：never-change-llm-model）
 - 生产部署：Pi `pi@192.168.1.160:/home/pi/news/`，systemd 服务 `a-share-research`；改动后需 scp 同步
 - 改 sources.yaml 前先从 Pi 拉取、改完推回（记忆：sync-sources-yaml）
 - 配置在 configs/sources.yaml、configs/watchlist.yaml、.env；API Key 只放 .env
 - Prompt 在 prompts/*.md，禁止写死在代码中
 - 会话恢复上下文以 continuation.md 为准
 —— CLAUDE.md 结束 ——
@@ -26,9 +26,9 @@
 ## 当前状态
-| 已完成 | M0~M8、**M9 投研 Agent Prompt** |
+| 已完成 | M0~M8、M9 投研 Agent Prompt、**M10 日报结构化入库** |
 | --- | --- |
-| 进行中 | 等待 M9 验收 |
+| 进行中 | 等待 M10 验收 |
 | 下一步 | 项目整合与优化 |
 详见 `continuation.md`。开发遵循 `project_plan.md` 9 个 Milestone,分阶段交付。
@@ -110,9 +110,10 @@ uv run a-share search "宁德时代固态电池"
 uv run a-share search "政策" --source cls --sentiment positive
 uv run a-share search "风险" --stock 001212 --min-importance 3
-# 日报(生成 HTML 报告并上传)
+# 日报(结构化入库, 不再生成 HTML; API/前端另行实现, 读取 MySQL news_ 表)
-uv run a-share report                     # 生成今日日报
+uv run a-share report                     # 生成今日日报并入 myquant 库
 uv run a-share report --date 20260616     # 指定日期
 uv run a-share report-import              # 历史日报 HTML 解析入库(幂等)
 uv run a-share pipeline --once --report   # 全链路末尾自动生成日报
 # 状态总览
@@ -18,6 +18,7 @@ from __future__ import annotations
 import argparse
 import json
 import os
 import subprocess
 import sys
 from collections import Counter
@@ -394,24 +395,34 @@ def cmd_status(args: argparse.Namespace) -> int:  # noqa: ARG001
 # --------------------------------------------------------------------------- #
 def cmd_report(args: argparse.Namespace) -> int:
-    """生成 HTML 日报并上传到 Web 服务器。"""
+    """生成日报并结构化入库（M10：不再生成 HTML/上传）。"""
    load_dotenv()
    from scheduler.reporter import generate_report
    day_str = args.date or _today()
    print(f"📊 生成日报: {day_str}")
-    path = generate_report(day_str, upload=not args.no_upload)
+    report_id = generate_report(day_str, upload=not args.no_upload)
-    if path is None:
+    if report_id is None:
        print("⚠️  无数据或生成失败")
        return 1
-    print(f"✅ 日报已保存: {path}")
+    print(f"✅ 日报已入库: report_id={report_id}")
    if not args.no_upload:
        from datetime import date as dt_date
        today_str = dt_date.today().strftime("%Y%m%d")
        print(f"✅ 已上传: http://doorcome.cn/echart/research/{today_str}/")
    return 0
 def cmd_report_import(args: argparse.Namespace) -> int:
    """历史日报 HTML 解析入库（M10）。"""
    load_dotenv()
    from report_import.importer import import_history
    report_dir = args.dir or os.environ.get("REPORT_HISTORY_DIR", "data/reports_history")
    print(f"📥 导入历史日报: {report_dir} (date={args.date or '全部'} type={args.type or '全部'})")
    stats = import_history(report_dir, date_str=args.date, report_type=args.type, force=args.force)
    print(f"✅ 扫描 {stats.scanned} | 新导入 {stats.imported} | 跳过 {stats.skipped} | 失败 {stats.failed}")
    for err in stats.errors[:20]:
        print(f"   ❌ {err}")
    return 0 if stats.failed == 0 else 1
 # --------------------------------------------------------------------------- #
 # discover — 自动分析站点,生成 sources.yaml 配置建议
 # --------------------------------------------------------------------------- #
@@ -917,6 +928,14 @@ def main() -> int:
    pl = sp.add_parser("list", help="查看关注列表")
    pl.set_defaults(func=cmd_watchlist_list)
    # ---- report-import ----
    p = sub.add_parser("report-import", help="历史日报 HTML 解析入库 (M10)")
    p.add_argument("--dir", default=None, help="日报目录(默认 REPORT_HISTORY_DIR)")
    p.add_argument("--date", default=None, help="仅导入指定日期 YYYYMMDD")
    p.add_argument("--type", default=None, choices=["finance", "intl"], help="仅导入指定类型")
    p.add_argument("--force", action="store_true", help="已存在也覆盖重导")
    p.set_defaults(func=cmd_report_import)
    # ---- stock-report ----
    p = sub.add_parser("stock-report", help="生成关注列表个股日报")
    p.add_argument("--no-upload", action="store_true", help="仅生成不上传")
@@ -1,6 +1,6 @@
 # continuation.md
-> `checkpoint` @ 2026-07-17 07:51
+> `checkpoint` @ 2026-08-05 08:30
 ---
@@ -10,15 +10,95 @@
 | --- | --- |
 | 新闻源 | 14 个（13 Web + 1 API: xwlb 新闻联播） |
 | Qdrant | 本地文件模式 `data/qdrant_storage/` |
-| 日报 | 5 板块: AI摘要 / 新闻联播 / 财经新闻 / 公告调研 / 数据总览 |
+| 日报 | **M10 完成并已部署 pi5: 结构化入库 MySQL；日报按当天日期生成（新闻 30h 回溯 / xwlb 取前一日 / 公告调研近 15 日）** |
-| 调度器 | APScheduler，systemd `a-share-research.service` |
+| DB 连接 | pi 上 systemd 服务 `a-share-db-tunnel` 常驻（0.0.0.0:13306 → doorcome.cn:3306）；**pi5 直连 192.168.1.10:13306** |
-| LLM | `deepseek-v4-flash`（绝不允许擅自修改） |
+| 调度器 | APScheduler，systemd `a-share-research.service`（pi5）；每天 07:00 首次任务生成日报（12/18/22 点不生成） |
-| 服务器 | `pi@192.168.1.160`，项目 `/home/pi/news/` |
+| LLM | `deepseek-v4-flash`（绝不允许擅自修改；模型必须显式配置，无内置兜底） |
 | 服务器 | `pi@192.168.1.160`（生产）/ `pi@192.168.1.10`（DB 隧道宿主） |
 | 抓取方式 | js_render=false → httpx 直连；js_render=true → Playwright |
 ---
-## 本次完成 (2026-07-17) — 禁用个股日报
+## 本次完成 (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/前端（用户另行实现）。
 **表结构（myquant 库，见 docs/db_schema.md）：**
 - `news_report`：主表，唯一键 (report_date, report_type, file_name)；`file_name=''` 表示新生成日报（每天每类型一行，重复生成覆盖）
 - `news_event`：事件明细，section ∈ xwlb/news/cninfo/intl
 **新增/修改文件：**
 - `report_db/`（models/schema/db）：连接、建表、幂等 save_report
 - `report_import/`（parser/importer）：历史 HTML 解析（表头驱动列映射）+ 批量导入
 - `scheduler/reporter.py`：`generate_report()` 完全切换为结构化入库；`_render_html/_upload` 标记废弃保留
 - `a_share_cli/main.py`：新增 `report-import` 子命令；`report` 命令改为入库提示
 - `pyproject.toml`：`uv add pymysql`（唯一新增依赖）
 - `.env`：新增 `NEWS_DB_*`（连接 127.0.0.1:13306，需 ssh 隧道）、`REPORT_HISTORY_DIR`
 - `docs/report_db_design.md`（实现逻辑）、`docs/db_schema.md`（表结构，供 API/前端）
 **验证结果：**
 - 历史 178 份全部入库：scanned=178 imported=7(首轮)+170 skipped=171 failed=0；DB 177 行（finance 49 + intl 128，1 个跨目录同名文件被幂等合并）+ 事件 4222 条
 - 端到端：`a-share report --date 20260616` → report_id=180 入库成功，无 HTML 产出
 - 单测 24 个通过（parser 19 + builder 4 + models/import 补充）
 **命令：**
 ```bash
 ssh -L 13306:127.0.0.1:13306 pi        # Mac 开发连库隧道
 uv run a-share report --date YYYYMMDD # 生成日报入库
 uv run a-share report-import          # 历史导入(幂等)
 ```
 **注意：** 本机 .env 无 LLM API key，AI 摘要会降级 WARNING（不影响入库）；生产 pi5 需配置 NEWS_DB_PASSWORD 且解决 13306 隧道可达性（见待确认项）。
 ---
 ## 历史 (2026-07-17) — 禁用个股日报
 **操作：** `STOCK_REPORT_TIME=` 设为空，`run_scheduler.py` 加空值守卫。
@@ -0,0 +1,111 @@
 # 日报结构化入库：数据库表结构与数据契约
 > 版本：v1.0 | 2026-08-03
 > 用途：供 API / 前端对接读取日报数据。表位于 MySQL `myquant` 库，表前缀 `news_`。
 > 连接：`192.168.1.10:13306`（pi 上 autossh 隧道 → doorcome.cn:3306 MariaDB 10.11），用户 `myquant`（密码在服务器 `.env` 的 `NEWS_DB_PASSWORD`）。
 ---
 ## 1. 表结构
 ### 1.1 news_report（日报主表，一行 = 一份日报）
 | 字段 | 类型 | 说明 |
 | --- | --- | --- |
 | id | BIGINT UNSIGNED PK | 自增主键 |
 | report_date | DATE | 日报日期 |
 | report_type | VARCHAR(16) | `finance`=A 股日报 / `intl`=国际财经日报 |
 | file_name | VARCHAR(160) | 历史文件源文件名；**新生成日报为空字符串 `""`** |
 | generated_at | DATETIME | 生成时间 |
 | ai_summary | TEXT | AI 摘要全文（含换行，按条目分行） |
 | stats | JSON | 数据总览统计快照（见第 3 节），可为 NULL |
 | created_at | DATETIME | 入库时间 |
 唯一键：`(report_date, report_type, file_name)` —— 历史同一天多次生成（intl 一日 3 次）保留多行；新生成日报 `file_name=''` 每天每类型仅一行，重复生成覆盖。
 ### 1.2 news_event（日报事件明细，一行 = 一条事件）
 | 字段 | 类型 | 说明 |
 | --- | --- | --- |
 | id | BIGINT UNSIGNED PK | 自增主键 |
 | report_id | BIGINT UNSIGNED | FK → news_report.id |
 | section | VARCHAR(16) | 板块：`xwlb`=新闻联播 / `news`=财经新闻 / `cninfo`=公告调研 / `intl`=国际重要事件 |
 | rank | INT | 板块内序号（1 起） |
 | importance | INT NULL | 重要度 1-5 |
 | event_type | VARCHAR(64) NULL | 事件类型（如 宏观经济/地缘政治/新闻联播/公告） |
 | title | VARCHAR(512) | 标题 |
 | summary | TEXT NULL | 摘要/正文 |
 | sentiment | VARCHAR(8) NULL | `positive` / `negative` / `neutral` |
 | source | VARCHAR(64) NULL | 来源（如 `cls`、`investinglive.com`） |
 | url | VARCHAR(512) NULL | 原文链接（新闻联播为空） |
 | created_at | DATETIME | 入库时间 |
 索引：`idx_report_section (report_id, section)`。
 ---
 ## 2. 数据契约
 - **幂等语义**：同一 `(report_date, report_type, file_name)` 重复写入会覆盖主表并全量替换事件（DELETE + INSERT），不会产生重复行。
 - **取最新**：同一天存在多份时（历史 intl 一日 3 次），前端按 `generated_at` 取最新；新日报 `file_name=''` 每天唯一。
 - **板块差异**：finance 日报含 `xwlb`+`news`+`cninfo` 三板块；intl 日报仅 `intl` 板块。前端按 `section` 过滤展示。
 - **历史覆盖范围**：2026-06-16 ~ 2026-08-03，共 177 行（finance 49 + intl 128；finance 少 1 因为两个目录存在同名文件被幂等合并）。事件总计 4222 条。
 ---
 ## 3. stats JSON 结构
 `news_report.stats` 为数据总览快照，前端自行解析。finance 与 intl 的 key 集合不同：
 | key | finance | intl | 内容 |
 | --- | --- | --- | --- |
 | `pipeline` | ✅ | ✅ | M1→M6 管道各环节数量：`{label: 数量}` |
 | `sources` | ✅ | — | 各新闻源文章数：`{源名: 数量}` |
 | `news` | ✅ | — | 新闻统计：`{total, hi_threshold, sentiments, importances, event_types}` |
 | `cninfo` | ✅ | — | 公告调研统计：`{total, hi_threshold, by_day, announcement, research, irm}` |
 | `xwlb` | ✅ | — | 联播统计：`{total, date}`（有数据时才有） |
 | `sentiment` | ✅ | ✅ | 情绪分布（历史文件为图例文本列表；新生成在 `news.sentiments`） |
 | `importance` | ✅ | ✅ | 重要度分布：`[{重要度, 数量}, ...]` |
 | `event_types` | ✅ | ✅ | 事件类型 TOP：`[{事件类型, 数量}, ...]` |
 | `source_dist` | — | ✅ | 文章来源分布：`[{来源, 文章数}, ...]` |
 > 历史文件与新生成日报的 stats 结构存在差异（历史为 HTML 解析快照，新生成为结构化组装），前端建议按 key 防御性读取。
 ---
 ## 4. 常用查询示例（API 实现参考）
 ```sql
 -- 某类型日报列表（取每天最新一份）
 SELECT r.* FROM news_report r
 JOIN (
    SELECT report_date, report_type, MAX(generated_at) AS g
    FROM news_report GROUP BY report_date, report_type
 ) t ON r.report_date = t.report_date AND r.report_type = t.report_type
   AND r.generated_at = t.g
 WHERE r.report_type = 'finance' AND r.report_date >= '2026-07-01'
 ORDER BY r.report_date DESC;
 -- 某日报的全部事件（按板块）
 SELECT section, rank, importance, event_type, title, summary, sentiment, source, url
 FROM news_event WHERE report_id = ? ORDER BY section, rank;
 -- 最近 N 天重要事件聚合（跨日报检索）
 SELECT e.* FROM news_event e
 JOIN news_report r ON r.id = e.report_id
 WHERE r.report_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
  AND e.importance >= 4
 ORDER BY e.importance DESC, r.report_date DESC;
 ```
 ---
 ## 5. 相关命令（数据生产侧）
 ```bash
 uv run a-share report --date YYYYMMDD     # 生成当日日报并入库（finance）
 uv run a-share report-import              # 历史 HTML 全量解析入库（幂等）
 uv run a-share report-import --date YYYYMMDD --type intl
 ```
 代码：`report_db/`（连接/写入）、`report_import/`（历史解析/导入）、`scheduler/reporter.py`（日报生成）。
@@ -0,0 +1,330 @@
 # Milestone 10 后端实现逻辑：日报结构化入库
 > 版本：v0.1（设计稿） | 2026-07
 > 对应 project_plan.md「十八、Milestone 10」
 > **范围**：本项目侧"后端"= 数据生产层（日报内容生成 + 结构化写入 MySQL）。
 > 不包含 API 服务与前端页面（由用户另行实现），但表结构与数据契约以本文档为准，供 API/前端对接。
 ---
 ## 1. 定位
 现有链路：`reporter.py` 收集数据 → `_render_html()` 渲染 HTML → scp 上传 doorcome。
 改造后：`reporter.py` 收集数据 → 组装结构化 `ReportData` → 写入 MySQL（`news_report` / `news_event`），不再产出 HTML。
 另需：把 doorcome 上 178 份历史日报 HTML（`finance_news_daily_*` ×50、`intl_news_daily_*` ×128）解析成同一 `ReportData` 结构入库。
 ---
 ## 2. 数据流总览
 ```
 [历史 HTML ×178]                                   [每日 pipeline]
  doorcome:/var/www/html/echart/research/            crawler→extractor→dedup→llm→embed→qdrant
  （一次性 scp 到 data/reports_history/）                      │
        │                                                      ▼
        ▼                                             reporter.generate_report()
  report_import/parser.py                                     │
        │            （BeautifulSoup 解析）                    ▼
        ▼              组装 ReportData                组装 ReportData
  report_import/importer.py                                  │
        │                （幂等 upsert）                       ▼
        ▼                                                     │
  ┌────────────────────── MySQL (myquant 库) ──────────────────────┐
  │  news_report（主表）   news_event（事件明细）                    │
  └────────────────────────────────────────────────────────────────┘
                          ▲
                 API / 前端（用户另行实现，只读）
 ```
 ---
 ## 3. 数据模型（Pydantic，`report_db/models.py`）
 ```python
 class EventRow(BaseModel):
    """一条事件记录，对应 news_event 一行。"""
    section: str          # xwlb | news | cninfo | intl
    rank: int             # 板块内序号（从 1 开始）
    importance: int | None = None
    event_type: str | None = None
    title: str
    summary: str | None = None
    sentiment: str | None = None   # positive | negative | neutral | ''
    source: str | None = None      # 来源（如 cls / ForexLive）
    url: str | None = None
 class ReportData(BaseModel):
    """一份完整日报，对应 news_report 一行 + news_event 多行。"""
    report_date: date               # 日报日期（YYYY-MM-DD）
    report_type: str                # finance | intl
    file_name: str                  # 源文件名（新生成时可为 ""）
    generated_at: datetime          # 生成时间
    ai_summary: str | None = None
    stats: dict[str, Any] = Field(default_factory=dict)   # 数据总览统计快照 → JSON 列
    events: list[EventRow] = Field(default_factory=list)
 ```
 ---
 ## 4. 字段映射（核心契约）
 ### 4.1 事件 JSON（data/events/）→ news_event
 现有事件文件结构与 news_event 字段对应关系（reporter 收集时直接转换）：
 | news_event 字段 | 事件 JSON 来源 |
 | --- | --- |
 | section | 来源判定：`source_id=="cninfo"` → `cninfo`；`source_id=="xwlb"` → `xwlb`；否则 `news`；intl 解析固定 `intl` |
 | importance | `event.importance` |
 | event_type | `event.event_type` |
 | title | `title` |
 | summary | `event.summary` |
 | sentiment | `event.sentiment` |
 | source | `source_id` |
 | url | `url`（xwlb 为空） |
 ### 4.2 历史 HTML → ReportData
 解析策略：**表头驱动列映射**。不同日报表格列集合不同：
 | 板块 | 表格列（<th>） | section |
 | --- | --- | --- |
 | 新闻联播（finance） | `# / (空) / 标题 / 重要度 / 事件类型` | xwlb |
 | 重要事件：新闻（finance） | `# / (空) / 标题 / 源 / 重要度 / 事件类型 / 摘要` | news |
 | 重要事件：公告调研（finance） | 同上 | cninfo |
 | 重要事件（intl） | `# / (空) / 标题 / 重要度 / 事件类型 / 摘要` | intl |
 要点：
 - 以表头文本定位列索引（"标题""重要度""事件类型""摘要""源"），空 `<th>` 为情绪图标列（⚪/🔴/🟢 → neutral/negative/positive），**不要依赖列位置**。
 - 情绪图标仅存在于有图标列的表；intl 表情绪列存在，finance 表情绪列存在（空 th 首列后）。
 - intl 无"源"列时，尝试从标题尾部 `[来源]` 或摘要尾部提取，提取不到则 `source=None`。
 - 标题中的股票代码标注 `(600519, ...)` 与 ⭐（自选股标记）需剥除，只保留纯标题。
 - AI 摘要：取 `h2`（"一、AI 摘要"）之后紧随的 `.ai-summary` 区块纯文本（保留换行）。
 - 数据总览 → `stats` JSON：按 `h3` 标题映射 key（见 4.3），解析该 h3 后的首个 `<table>`，缺失的板块跳过、不报错。
 - 容错：任一板块解析失败 → 记 WARNING 日志，该板块置空，不影响整份入库；整份文件解析失败 → 抛 `ReportParseError`（由 importer 捕获计数）。
 ### 4.3 数据总览 → stats JSON
 | h3 标题（含板块名） | stats key |
 | --- | --- |
 | M1→M6 管道 / 管道 | `pipeline`（保留原始行） |
 | 各源数据 | `sources` |
 | 情绪分布 | `sentiment` |
 | 重要度分布 | `importance` |
 | 事件类型（TOP 10 / 分布） | `event_types` |
 | 文章来源分布 | `source_dist` |
 `stats` 存 MySQL `JSON` 列，前端自行解析展示。历史文件与未来新日报的 stats 结构可能不同（finance 与 intl 板块不同），一律按快照存储，不做跨版本规范化。
 ---
 ## 5. 模块设计
 ### 5.1 新包 `report_db/`（DB 层）
 ```
 report_db/
 ├── __init__.py      # 导出 connect / init_schema / save_report
 ├── models.py        # EventRow / ReportData（Pydantic）
 ├── schema.py        # DDL 常量（news_report / news_event，见 project_plan.md 十八）
 └── db.py            # 连接、事务、写入
 ```
 `db.py` 关键函数：
 ```python
 def load_db_config() -> DbConfig:
    """从环境变量读取 NEWS_DB_HOST/PORT/USER/PASSWORD/NAME。
    缺失 PASSWORD 时记 ERROR 并 raise，禁止默认密码。"""
 def connect(cfg: DbConfig) -> Connection:
    """pymysql.connect(autocommit=False, charset="utf8mb4", cursorclass=DictCursor)。
    失败时 logger.exception + raise。"""
 def init_schema(conn: Connection) -> None:
    """执行 schema.py 中的 CREATE TABLE IF NOT EXISTS ×2。"""
 def save_report(conn: Connection, report: ReportData) -> int:
    """事务内：
    1. INSERT INTO news_report (...) VALUES (...) 或按 (report_date, report_type, file_name)
       唯一键命中时 UPDATE（新生成日报重复执行 = 覆盖同 file_name/同日期，幂等）；
    2. 取 report_id，DELETE 旧事件后批量 INSERT news_event（保证整份覆盖一致）。
    返回 report_id。"""
 def transaction(conn: Connection) -> contextmanager:
    """提交/回滚上下文管理器。"""
 def fetch_report(conn: Connection, report_id: int) -> dict | None:
    """读侧辅助（联调/测试用），API 侧由用户自行实现。"""
 ```
 要点：
 - 所有 SQL 为 MySQL/MariaDB 方言（`JSON` 列、`ENGINE=InnoDB`、`COMMENT`），**不依赖 ORM**。
 - 连接生命周期：每次 `save_report` 短连接（report 一天跑几次，量小，无需连接池；如未来加大再换）。
 - 字符集 utf8mb4，`SET NAMES utf8mb4` 由 pymysql charset 参数处理。
 ### 5.2 新包 `report_import/`（历史解析）
 ```
 report_import/
 ├── __init__.py
 ├── parser.py       # parse_finance_report / parse_intl_report（BeautifulSoup）
 └── importer.py     # import_history(dir, date=None, type=None) -> ImportStats
 ```
 `parser.py`：
 ```python
 class ReportParseError(Exception): ...
 def parse_finance_report(html: str, file_name: str) -> ReportData: ...
 def parse_intl_report(html: str, file_name: str) -> ReportData: ...
 def parse_report(html: str, file_name: str) -> ReportData:
    """按文件名前缀分流：finance_news_daily_* / intl_news_daily_*。"""
 ```
 - 依赖复用现有 `beautifulsoup4`（已在 pyproject 依赖），**不新增解析库**。
 - `report_date` 从文件名解析（`*_daily_{YYYYMMDD}_*.html`），不信任目录名。
 - `generated_at` 从文件名时间（`{HHMMSS}`）或 `<header>` 中"生成于"文本解析，解析不到用文件 mtime。
 `importer.py`：
 ```python
@dataclass
 class ImportStats:
    scanned: int = 0      # 扫描到的日报文件数
    imported: int = 0     # 新入库
    skipped: int = 0      # 已存在（幂等跳过）
    failed: int = 0       # 解析失败
    errors: list[str] = field(default_factory=list)
 def import_history(report_dir: Path, date: str | None = None,
                   report_type: str | None = None) -> ImportStats:
    """遍历 {report_dir}/{YYYYMMDD}/*_news_daily_*.html，
    过滤 date / type，逐个 parse → save_report。"""
 ```
 ### 5.3 `scheduler/reporter.py` 改造（完全切换）
 - 新增 `_build_report_data(news, cninfo, pipeline, ai_summary, day_str, xwlb) -> ReportData`：
  - 事件转换：`news["high"]` → `EventRow(section="news", ...)`；`cninfo["high"]` → `section="cninfo"`；`xwlb["items"]` → `section="xwlb"`；
  - `stats` 组装：`{"pipeline": pipeline, "sources": {...}, "sentiment": news["sentiments"], "importance": news["importances"], "event_types": news["event_types"], "cninfo": {...}}`；
  - 事件 `rank` 按板块内顺序编号。
 - `generate_report(day_str, *, upload=True)` 改为：收集（逻辑不变）→ `_build_report_data` → `connect()` + `save_report()`；删除 `_render_html`/`_upload` 调用。
 - `_render_*` 函数**保留但标记 deprecated**（注释说明"完全切换后不再调用"），不删除，保证最小改动、可回退。
 - 返回值由 `Path | None` 改为 `report_id: int | None`；`scheduler/pipeline.py` 中 report 步骤仅判断非 None（实施时核实该处调用，保持兼容）。
 - `stock_reporter.py` **不改动**（个股日报不在本期范围）。
 ### 5.4 `a_share_cli/main.py` 新增子命令
 ```
 uv run a-share report-import [--dir data/reports_history] [--date YYYYMMDD] [--type finance|intl]
 ```
 - 默认全量扫描 `REPORT_HISTORY_DIR`（.env 可配，默认 `data/reports_history/`）。
 - 输出 ImportStats 汇总（扫描/导入/跳过/失败）。
 ---
 ## 6. 关键流程
 ### 6.1 历史导入（一次执行，可重复）
 ```
 1. scp -r doorcome:/var/www/html/echart/research/2026*  →  data/reports_history/
   （一次手工操作，不进代码）
 2. uv run a-share report-import
   for each {date}/{file}:
       report_type = 文件名前缀（finance|intl）
       ReportData = parse_report(html, file_name)
       try: save_report(conn, ReportData) → imported += 1
       except DuplicateKey: skipped += 1      # 已导入过
       except ReportParseError as e: failed += 1; errors.append(str(e))
 3. 校验: SELECT report_type, COUNT(*) FROM news_report GROUP BY report_type
       期望 50 / 128
 ```
 ### 6.2 每日日报生成（pipeline 07:00 步骤）
 ```
 generate_report(day_str):
    news   = _collect_news_events(day_str)        # 不变
    cninfo = _collect_cninfo_events(day_str)      # 不变
    xwlb   = _collect_xwlb(day_str)               # 不变
    pipeline = _collect_pipeline_stats(day_str)   # 不变
    ai_summary = _generate_ai_summary(...)        # 不变
    report = _build_report_data(...)              # 新增
    save_report(connect(), report)                # 新增（替代渲染+上传）
 ```
 ### 6.3 幂等策略
 - 唯一键 `(report_date, report_type, file_name)`：
  - 历史导入：命中 → 跳过（或 `--force` 覆盖）；
  - 新日报：`file_name=""` 时唯一键退化为 `(report_date, report_type, "")`，同一天重复跑 → UPDATE 覆盖，事件表 DELETE+INSERT 全量替换，**不产生历史残留**。
 ---
 ## 7. 配置项（.env / .env.example）
 ```env
 # ---- 日报结构化入库 (M10) ----
 NEWS_DB_HOST=127.0.0.1        # 开发走 ssh 隧道: ssh -L 13306:127.0.0.1:13306 pi
 NEWS_DB_PORT=13306
 NEWS_DB_USER=myquant
 NEWS_DB_PASSWORD=             # 填真实值，禁止写入源码/文档
 NEWS_DB_NAME=myquant
 REPORT_HISTORY_DIR=data/reports_history
 ```
 ---
 ## 8. 错误处理
 | 场景 | 行为 |
 | --- | --- |
 | DB 不可达/凭据错误 | `connect()` 抛异常 → `generate_report` 记 ERROR 并返回 None（pipeline 该步骤失败，其余步骤不受影响） |
 | 单份历史文件解析失败 | 记 WARNING，`failed += 1`，继续下一份；结束输出失败清单 |
 | 事件字段缺失（如无摘要列） | 对应字段留 None，不抛错 |
 | 全部失败 | `report-import` 返回非 0 退出码，便于排查 |
 ---
 ## 9. 测试策略（tests/）
 | 文件 | 内容 |
 | --- | --- |
 | `tests/test_report_parser.py` | 用 fixtures（从 178 份中拷贝 finance/intl 各 1 份真实样例到 `tests/fixtures/`）断言：板块数、事件行数、字段映射、标题净化、幂等文件日期解析 |
 | `tests/test_report_db.py` | 纯逻辑：`_build_report_data` 组装正确；SQL 层用 sqlite3 内存库建同构（简化 DDL）验证 upsert/覆盖语义 |
 | `tests/test_report_import.py` | 临时目录构造 2-3 份假 HTML → 全流程导入 → 断言 ImportStats 计数与幂等 |
 | 集成（`@pytest.mark.integration`，默认跳过） | 连真实 MySQL：init_schema + save_report + 查询回读 |
 新增 pytest marker 说明：真实 DB 连接一律走 integration，**单元测试不得依赖生产库**。
 ---
 ## 10. 依赖变更
 - `uv add pymysql`（纯 Python 驱动，唯一新增依赖）
 - 解析复用现有 `beautifulsoup4`，不新增
 ---
 ## 11. 开放问题（沿自 project_plan.md 十八，不阻塞开发）
 1. 生产连接：pi5 无法直连 `192.168.1.10:13306`（隧道仅绑 loopback）——需决定改 pi 的 autossh 绑定 / pi5 自建隧道。
 2. intl 日报生成方不在本项目，未来 intl 新日报需按同一表结构写入（本项目仅负责解析历史 + finance 新日报）。
 3. 个股日报（research 根目录文件）本期不处理。
 ---
 ## 12. 实施顺序（供开发排期）
 1. `report_db/`（models/schema/db）+ `.env` 配置 + 建表验证
 2. `report_import/parser.py` + fixtures + 单测
 3. `report_import/importer.py` + CLI `report-import` + 178 份全量导入验收
 4. `reporter.py` 改造（_build_report_data + save_report）+ pipeline 兼容性验证
 5. docs/db_schema.md 定稿（给 API/前端）、README / continuation.md 更新
@@ -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(兜底) → 默认
+        # DEEPSEEK_MODEL → LLM_MODEL；模型必须显式配置，不提供内置默认
-        m = model or _read_env("DEEPSEEK_MODEL") or _read_env("LLM_MODEL") or _DEEPSEEK_DEFAULT_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(兜底) → 默认
+        # QWEN_MODEL → LLM_MODEL；模型必须显式配置，不提供内置默认
-        m = model or _read_env("QWEN_MODEL") or _read_env("LLM_MODEL") or _QWEN_DEFAULT_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")
@@ -602,5 +602,99 @@ Claude Code 必须严格遵守：
 系统自动检索知识库、分析新闻事件，并生成完整研究报告。
 ---
 # 十八、Milestone 10：日报结构化入库（前后端分离数据层）
 > 状态：✅ 已实施（2026-08-03，等待人工验收）
 > 范围：本项目只负责「日报内容生成 + 结构化存入 MySQL」，**不实现 API 与前端**（由用户另行实现）。
 ## 背景与决策
 | 决策点 | 结论 |
 | --- | --- |
 | DB | 192.168.1.10:13306（pi 上 autossh 隧道 → doorcome.cn:3306 MariaDB 10.11.18），业务库 `myquant`，表前缀 `news_` |
 | API / 前端 | 用户另行实现，本项目只保证数据完整、表结构文档清晰 |
 | 日报生成 | 完全切换：只存 DB + 前端渲染，不再生成 HTML 静态文件 |
 | 历史数据 | doorcome `/var/www/html/echart/research/{YYYYMMDD}/` 下 178 份 `*_news_daily_*.html`（finance×50 + intl×128，日期 20260608~20260803），解析入库 |
 ## 表结构设计
 ### news_report（日报主表）
 ```sql
 CREATE TABLE IF NOT EXISTS news_report (
    id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    report_date DATE NOT NULL COMMENT '日报日期',
    report_type VARCHAR(16) NOT NULL COMMENT 'finance=A股日报 / intl=国际财经日报',
    file_name VARCHAR(160) NOT NULL DEFAULT '' COMMENT '源文件名（历史解析）；新生成可为空',
    generated_at DATETIME NOT NULL COMMENT '生成时间',
    ai_summary TEXT NULL COMMENT 'AI 摘要全文',
    stats JSON NULL COMMENT '数据总览统计快照（管道/情绪/重要度/事件类型/来源分布）',
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_report_file (report_date, report_type, file_name),
    KEY idx_report_date (report_date)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='每日日报主表';
 ```
 ### news_event（日报事件明细，新闻联播/财经新闻/公告调研/intl 事件统一入此表）
 ```sql
 CREATE TABLE IF NOT EXISTS news_event (
    id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    report_id BIGINT UNSIGNED NOT NULL COMMENT 'FK → news_report.id',
    section VARCHAR(16) NOT NULL COMMENT '板块: xwlb=新闻联播 / news=财经新闻 / cninfo=公告调研 / intl=国际重要事件',
    rank INT NOT NULL DEFAULT 0 COMMENT '板块内序号',
    importance INT NULL COMMENT '重要度 1-5',
    event_type VARCHAR(64) NULL COMMENT '事件类型',
    title VARCHAR(512) NOT NULL COMMENT '标题',
    summary TEXT NULL COMMENT '摘要/正文',
    sentiment VARCHAR(8) NULL COMMENT 'positive/negative/neutral',
    source VARCHAR(64) NULL COMMENT '来源',
    url VARCHAR(512) NULL COMMENT '原文链接',
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    KEY idx_report_section (report_id, section)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='日报事件明细';
 ```
 设计说明：
 - 数据总览统计（M1→M6 管道、各源数据、情绪/重要度/事件类型/来源分布）以 JSON 快照存 `news_report.stats`，前端自行解析。
 - 幂等：导入/生成按 `(report_date, report_type, file_name)` 唯一键，重复执行跳过或覆盖，不产生重复行。
 - 历史同一天多次生成（intl 一日 3 次）保留多行，前端取最新。
 ## 阶段划分（每阶段验收后进入下一阶段）
 ### M10-A：DB 连接层与建表
 - `uv add pymysql`（纯 Python 驱动，无编译依赖）
 - 新包 `report_db/`：`db.py`（连接、事务、建表）、`schema.py`（DDL）
 - `.env` 新增 `NEWS_DB_HOST / NEWS_DB_PORT / NEWS_DB_USER / NEWS_DB_PASSWORD / NEWS_DB_NAME`（默认 myquant），`.env.example` 同步
 - 验收：连接成功、`news_report` / `news_event` 建表成功；`tests/test_report_db.py` 通过（测试用 sqlite3 内存模拟同构 SQL，真实 MySQL 走 integration 标记）
 ### M10-B：历史日报解析入库
 - 一次性将 doorcome 178 份 HTML 拉到 `data/reports_history/`
 - 新包 `report_import/`：`parser.py`（finance/intl 两类 HTML → 结构化 dict，容错缺板块）、`importer.py`（幂等入库）
 - CLI 子命令：`a-share report-import [--date YYYYMMDD] [--type finance|intl]`（默认全量）
 - 验收：178 份全部入库，`SELECT COUNT(*)` 与文件数一致；抽查 finance/intl 各 2 份字段正确；重复执行不产生重复行
 ### M10-C：日报生成改造（完全切换）
 - `scheduler/reporter.py`：`generate_report()` 改为「收集结构化数据 → 写入 news_report/news_event」，移除 `_render_html` / `_upload` 调用（函数可保留但不再触发）
 - `scheduler/pipeline.py` 调用签名保持兼容（仍调 `generate_report(day_str)`）
 - 验收：`uv run a-share report --date YYYYMMDD` 后 DB 出现当日记录且无 HTML 文件产出；AI 摘要、事件行、stats JSON 完整
 ### M10-D：文档与收尾
 - `docs/db_schema.md`：完整表结构 + 字段说明 + 示例数据（供用户实现 API/前端）
 - 更新 README、continuation.md；提交信息 `feat: 日报结构化入库`
 ## 待确认项（不阻塞开发，部署前需用户决策）
 1. **生产连接**：pi5（192.168.1.160）无法直连 `192.168.1.10:13306`（隧道只绑 loopback）。可选：a) pi 的 autossh 改为绑定 0.0.0.0（需改 pi 系统配置）b) pi5 自建隧道 c) 其他
 2. **intl 日报生成方**：项目代码中无 intl 生成逻辑（doorcome 上另有来源）。历史解析照做；未来 intl 新日报如需入库，由生成方按同一表结构写入
 3. **个股日报**（002714.SZ_0724.html 等 research 根目录文件）：本期不处理，如需请另行提出
 —— project_plan.md 结束 ——
@@ -37,6 +37,7 @@ dependencies = [
    "mcp>=1.0",
    "pypdf>=6.13.3",
    "markitdown[all]>=0.1.5",
    "pymysql>=1.2.0",
 ]
 [project.optional-dependencies]
@@ -0,0 +1,19 @@
 """日报结构化入库（Milestone 10）。
 职责：日报内容（finance/intl）结构化后写入 MySQL（news_report / news_event），
 供用户另行实现的 API/前端读取。本项目不提供 API 与前端。
 """
 from .db import connect, exists_report, fetch_report, init_schema, load_db_config, save_report
 from .models import EventRow, ReportData
 __all__ = [
    "EventRow",
    "ReportData",
    "connect",
    "exists_report",
    "fetch_report",
    "init_schema",
    "load_db_config",
    "save_report",
 ]
@@ -0,0 +1,156 @@
 """日报结构化入库：DB 连接、建表、写入。"""
 from __future__ import annotations
 import json
 import os
 from dataclasses import dataclass
 from typing import Any
 import pymysql
 from loguru import logger
 from .models import ReportData
 from .schema import DDL_STATEMENTS
@dataclass(frozen=True)
 class DbConfig:
    """MySQL 连接配置（来自环境变量 NEWS_DB_*）。"""
    host: str
    port: int
    user: str
    password: str
    name: str
 def load_db_config() -> DbConfig:
    """从环境变量读取 NEWS_DB_*，缺失密码时抛异常（禁止默认密码）。"""
    host = os.environ.get("NEWS_DB_HOST", "127.0.0.1")
    port = int(os.environ.get("NEWS_DB_PORT", "13306"))
    user = os.environ.get("NEWS_DB_USER", "myquant")
    password = os.environ.get("NEWS_DB_PASSWORD", "")
    name = os.environ.get("NEWS_DB_NAME", "myquant")
    if not password:
        logger.error("NEWS_DB_PASSWORD 未配置，请在 .env 中设置")
        raise ValueError("NEWS_DB_PASSWORD 未配置")
    return DbConfig(host=host, port=port, user=user, password=password, name=name)
 def connect(cfg: DbConfig | None = None) -> pymysql.Connection:
    """建立短连接（autocommit=False）。失败时记录日志并抛出。"""
    cfg = cfg or load_db_config()
    try:
        conn = pymysql.connect(
            host=cfg.host,
            port=cfg.port,
            user=cfg.user,
            password=cfg.password,
            database=cfg.name,
            charset="utf8mb4",
            autocommit=False,
            cursorclass=pymysql.cursors.DictCursor,
        )
    except Exception:
        logger.exception("连接 MySQL 失败: host={} port={} user={}", cfg.host, cfg.port, cfg.user)
        raise
    logger.debug("MySQL 已连接: {}/{}", cfg.host, cfg.name)
    return conn
 def init_schema(conn: pymysql.Connection) -> None:
    """建表（CREATE TABLE IF NOT EXISTS ×2），幂等。"""
    with conn.cursor() as cur:
        for ddl in DDL_STATEMENTS:
            cur.execute(ddl)
    conn.commit()
    logger.info("news_report / news_event 建表完成")
 def save_report(conn: pymysql.Connection, report: ReportData) -> int:
    """事务内写入一份日报。
    幂等策略：
    - 主表按 (report_date, report_type, file_name) 唯一键 upsert；
    - 事件表 DELETE 该 report 旧行后全量 INSERT（整份覆盖一致）。
    返回 report_id。
    """
    with conn.cursor() as cur:
        stats_json = json.dumps(report.stats, ensure_ascii=False) if report.stats else None
        cur.execute(
            """
            INSERT INTO news_report
                (report_date, report_type, file_name, generated_at, ai_summary, stats)
            VALUES (%s, %s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
                generated_at = VALUES(generated_at),
                ai_summary = VALUES(ai_summary),
                stats = VALUES(stats)
            """,
            (
                report.report_date,
                report.report_type,
                report.file_name,
                report.generated_at,
                report.ai_summary,
                stats_json,
            ),
        )
        cur.execute(
            "SELECT id FROM news_report WHERE report_date=%s AND report_type=%s AND file_name=%s",
            (report.report_date, report.report_type, report.file_name),
        )
        row = cur.fetchone()
        if row is None:  # pragma: no cover - 理论不可达
            raise RuntimeError("写入 news_report 后查询不到 report_id")
        report_id: int = row["id"]
        cur.execute("DELETE FROM news_event WHERE report_id=%s", (report_id,))
        for ev in report.events:
            cur.execute(
                """
                INSERT INTO news_event
                    (report_id, section, rank, importance, event_type, title,
                     summary, sentiment, source, url)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                """,
                (
                    report_id,
                    ev.section,
                    ev.rank,
                    ev.importance,
                    ev.event_type,
                    ev.title,
                    ev.summary,
                    ev.sentiment,
                    ev.source,
                    ev.url,
                ),
            )
    conn.commit()
    logger.info("日报已入库: report_id={} date={} type={} events={}",
                report_id, report.report_date, report.report_type, len(report.events))
    return report_id
 def fetch_report(conn: pymysql.Connection, report_id: int) -> dict[str, Any] | None:
    """读侧辅助（联调/测试用），返回主表行。"""
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM news_report WHERE id=%s", (report_id,))
        return cur.fetchone()
 def exists_report(
    conn: pymysql.Connection,
    report_date: Any,
    report_type: str,
    file_name: str,
 ) -> bool:
    """判断主表是否已存在该唯一键记录（历史导入幂等用）。"""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT 1 FROM news_report WHERE report_date=%s AND report_type=%s AND file_name=%s",
            (report_date, report_type, file_name),
        )
        return cur.fetchone() is not None
@@ -0,0 +1,34 @@
 """日报结构化入库：数据模型。"""
 from __future__ import annotations
 from datetime import date, datetime
 from typing import Any
 from pydantic import BaseModel, Field
 class EventRow(BaseModel):
    """一条事件记录，对应 news_event 一行。"""
    section: str  # xwlb | news | cninfo | intl
    rank: int  # 板块内序号（从 1 开始）
    importance: int | None = None
    event_type: str | None = None
    title: str
    summary: str | None = None
    sentiment: str | None = None  # positive | negative | neutral | ''
    source: str | None = None
    url: str | None = None
 class ReportData(BaseModel):
    """一份完整日报：news_report 一行 + news_event 多行。"""
    report_date: date
    report_type: str  # finance | intl
    file_name: str = ""  # 源文件名（新生成日报可为空）
    generated_at: datetime
    ai_summary: str | None = None
    stats: dict[str, Any] = Field(default_factory=dict)
    events: list[EventRow] = Field(default_factory=list)
@@ -0,0 +1,43 @@
 """MySQL DDL：日报结构化入库（表前缀 news_，目标 MariaDB 10.11）。
 与 project_plan.md「十八、Milestone 10」及 docs/report_db_design.md 保持一致。
 """
 from __future__ import annotations
 DDL_STATEMENTS: list[str] = [
    # 日报主表
    """
    CREATE TABLE IF NOT EXISTS news_report (
        id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
        report_date DATE NOT NULL COMMENT '日报日期',
        report_type VARCHAR(16) NOT NULL COMMENT 'finance=A股日报 / intl=国际财经日报',
        file_name VARCHAR(160) NOT NULL DEFAULT '' COMMENT '源文件名（历史解析）；新生成可为空',
        generated_at DATETIME NOT NULL COMMENT '生成时间',
        ai_summary TEXT NULL COMMENT 'AI 摘要全文',
        stats JSON NULL COMMENT '数据总览统计快照（管道/情绪/重要度/事件类型/来源分布）',
        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        UNIQUE KEY uk_report_file (report_date, report_type, file_name),
        KEY idx_report_date (report_date)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='每日日报主表'
    """,
    # 日报事件明细
    """
    CREATE TABLE IF NOT EXISTS news_event (
        id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
        report_id BIGINT UNSIGNED NOT NULL COMMENT 'FK → news_report.id',
        section VARCHAR(16) NOT NULL COMMENT '板块: xwlb=新闻联播 / news=财经新闻 / cninfo=公告调研 / intl=国际重要事件',
        rank INT NOT NULL DEFAULT 0 COMMENT '板块内序号',
        importance INT NULL COMMENT '重要度 1-5',
        event_type VARCHAR(64) NULL COMMENT '事件类型',
        title VARCHAR(512) NOT NULL COMMENT '标题',
        summary TEXT NULL COMMENT '摘要/正文',
        sentiment VARCHAR(8) NULL COMMENT 'positive/negative/neutral',
        source VARCHAR(64) NULL COMMENT '来源',
        url VARCHAR(512) NULL COMMENT '原文链接',
        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
        KEY idx_report_section (report_id, section),
        KEY idx_title (title(255))
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='日报事件明细'
    """,
 ]
@@ -0,0 +1,21 @@
 """历史日报解析与批量导入（Milestone 10）。
 将 doorcome 历史日报 HTML（finance/intl）解析为结构化数据并写入 MySQL。
 """
 from .importer import ImportStats, import_history
 from .parser import (
    ReportParseError,
    parse_finance_report,
    parse_intl_report,
    parse_report,
 )
 __all__ = [
    "ImportStats",
    "ReportParseError",
    "import_history",
    "parse_finance_report",
    "parse_intl_report",
    "parse_report",
 ]
@@ -0,0 +1,89 @@
 """历史日报批量导入（解析 → 入库，幂等）。"""
 from __future__ import annotations
 import re
 from dataclasses import dataclass, field
 from pathlib import Path
 from loguru import logger
 from report_db import connect, exists_report, save_report
 from report_import.parser import ReportParseError, parse_report
 _REPORT_FILE_RE = re.compile(r"([a-z]+)_news_daily_\d{8}(?:_\d{4,6})?\.html$")
@dataclass
 class ImportStats:
    """一次导入的统计结果。"""
    scanned: int = 0  # 扫描到的日报文件数
    imported: int = 0  # 新入库
    skipped: int = 0  # 已存在（幂等跳过）
    failed: int = 0  # 解析失败
    errors: list[str] = field(default_factory=list)
 def _match_file(path: Path, date_str: str | None, report_type: str | None) -> bool:
    """按文件名判断是否属于本次导入范围。"""
    m = _REPORT_FILE_RE.search(path.name)
    if not m:
        return False
    if report_type is not None and m.group(1) != report_type:
        return False
    return date_str is None or date_str in path.name
 def import_history(
    report_dir: str | Path,
    date_str: str | None = None,
    report_type: str | None = None,
    *,
    force: bool = False,
 ) -> ImportStats:
    """扫描 {report_dir}/{YYYYMMDD}/ 下全部 `*_news_daily_*.html` 并入库。
    - 幂等：主表唯一键 (report_date, report_type, file_name) 已存在则跳过；
    - `force=True` 时跳过存在性检查，直接覆盖重导；
    - 单文件解析失败不影响其他文件。
    """
    stats = ImportStats()
    root = Path(report_dir)
    if not root.is_dir():
        logger.error("日报目录不存在: {}", root)
        raise FileNotFoundError(f"日报目录不存在: {root}")
    files = sorted(p for p in root.glob("*/[a-z]*_news_daily_*.html") if _match_file(p, date_str, report_type))
    stats.scanned = len(files)
    logger.info("扫描到日报文件 {} 份: {}", stats.scanned, root)
    conn = connect()
    try:
        for path in files:
            try:
                html = path.read_text(encoding="utf-8")
                report = parse_report(html, path.name)
            except ReportParseError as e:
                stats.failed += 1
                stats.errors.append(f"{path.name}: {e}")
                logger.warning("解析失败: {} ({})", path.name, e)
                continue
            except Exception as e:  # 防御未知异常，不中断批量
                stats.failed += 1
                stats.errors.append(f"{path.name}: {type(e).__name__}: {e}")
                logger.exception("读取/解析异常: {}", path.name)
                continue
            if not force and exists_report(conn, report.report_date, report.report_type, report.file_name):
                stats.skipped += 1
                logger.debug("已存在, 跳过: {}", path.name)
                continue
            save_report(conn, report)
            stats.imported += 1
    finally:
        conn.close()
    logger.info("导入完成: scanned={} imported={} skipped={} failed={}",
                stats.scanned, stats.imported, stats.skipped, stats.failed)
    return stats
@@ -0,0 +1,298 @@
 """历史日报 HTML 解析器（finance / intl）。
 策略：表头驱动列映射，不依赖列位置；板块按 h2 标题识别；
 数据总览按 h3 标题归类为 stats JSON 快照（前端自行解析）。
 """
 from __future__ import annotations
 import re
 from datetime import date, datetime
 from bs4 import BeautifulSoup, Tag
 from report_db.models import EventRow, ReportData
 # 情绪图标 → sentiment 值
 _SENTIMENT_ICON: dict[str, str] = {"⚪": "neutral", "🔴": "negative", "🟢": "positive"}
 # 数据总览 h3 标题关键词 → stats key
 _STATS_SECTION_KEYS: list[tuple[str, str]] = [
    ("管道", "pipeline"),
    ("各源", "sources"),
    ("情绪", "sentiment"),
    ("重要度", "importance"),
    ("事件类型", "event_types"),
    ("来源", "source_dist"),
 ]
 class ReportParseError(Exception):
    """整份文件解析失败。"""
 # --------------------------------------------------------------------------- #
 # 文件名 / 时间解析
 # --------------------------------------------------------------------------- #
 def _parse_datetime_from_filename(file_name: str) -> tuple[date, datetime] | None:
    """从文件名解析日报日期与生成时间。
    支持 `{type}_news_daily_{YYYYMMDD}.html` 与带时间戳的
    `{type}_news_daily_{YYYYMMDD}_{HHMMSS}.html` / `..._{HHMM}.html`。
    """
    m = re.search(r"_daily_(\d{8})(?:_(\d{4})(\d{2})?)?", file_name)
    if not m:
        return None
    day = date(int(m.group(1)[:4]), int(m.group(1)[4:6]), int(m.group(1)[6:8]))
    hh = mm = ss = 0
    if m.group(2):
        hh, mm = int(m.group(2)[:2]), int(m.group(2)[2:4])
        ss = int(m.group(3) or 0)
    return day, datetime(day.year, day.month, day.day, hh, mm, ss)
 def _parse_header_generated_at(soup: BeautifulSoup, fallback: datetime) -> datetime:
    """从 <header> 中"生成于 YYYY-MM-DD HH:MM:SS"解析生成时间。"""
    p = soup.select_one("header p")
    if p:
        m = re.search(r"生成于 (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", p.get_text())
        if m:
            return datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S")
    return fallback
 # --------------------------------------------------------------------------- #
 # AI 摘要
 # --------------------------------------------------------------------------- #
 def _parse_ai_summary(soup: BeautifulSoup) -> str | None:
    """AI 摘要：<div class="ai-summary">，li 逐行输出。"""
    div = soup.select_one("div.ai-summary")
    if div is None:
        return None
    items = [li.get_text(strip=True) for li in div.find_all("li") if li.get_text(strip=True)]
    if items:
        return "\n".join(items)
    text = re.sub(r"\s+", " ", div.get_text(strip=True))
    return text or None
 # --------------------------------------------------------------------------- #
 # 事件表解析
 # --------------------------------------------------------------------------- #
 def _to_int(text: str) -> int | None:
    m = re.search(r"\d+", text or "")
    return int(m.group(0)) if m else None
 def _parse_summary(cell: Tag) -> tuple[str | None, str | None]:
    """摘要列：剥除 <small>[来源]</small>，返回 (摘要, 来源)。"""
    source = None
    small = cell.find("small")
    if small:
        m = re.search(r"\[([^\]]+)\]", small.get_text())
        if m:
            source = m.group(1)
        small.decompose()
    text = re.sub(r"\s+", " ", cell.get_text(strip=True))
    return (text or None, source)
 def _clean_title(cell: Tag) -> str:
    """标题列：剥除 <small> 股票代码标注等，返回纯标题。"""
    for small in cell.find_all("small"):
        small.decompose()
    return re.sub(r"\s+", " ", cell.get_text(strip=True))
 def _parse_event_table(table: Tag, section: str) -> list[EventRow]:
    """事件表 → EventRow 列表。表头驱动列映射，兼容 xwlb/news/cninfo/intl 四类表。"""
    rows = table.find_all("tr")
    if len(rows) < 2:
        return []
    header_cells = rows[0].find_all(["th", "td"])
    col_index: dict[str, int] = {}
    icon_col: int | None = None
    for i, cell in enumerate(header_cells):
        text = cell.get_text(strip=True)
        if text:
            col_index[text] = i
        elif icon_col is None:
            icon_col = i
    out: list[EventRow] = []
    for row in rows[1:]:
        tds = row.find_all("td")
        if not tds:
            continue
        def col(name: str, tds: list[Tag] = tds) -> Tag | None:  # noqa: B008 - 绑定循环变量
            idx = col_index.get(name)
            return tds[idx] if idx is not None and idx < len(tds) else None
        title_cell = col("标题")
        if title_cell is None:
            continue
        a = title_cell.find("a")
        url = a.get("href") if a else None
        summary_cell = col("摘要")
        summary: str | None = None
        source: str | None = None
        if summary_cell is not None:
            summary, source = _parse_summary(summary_cell)
        if source is None:
            src_cell = col("源")
            if src_cell is not None and src_cell.get_text(strip=True):
                source = src_cell.get_text(strip=True)
        sentiment = None
        if icon_col is not None and icon_col < len(tds):
            sentiment = _SENTIMENT_ICON.get(tds[icon_col].get_text(strip=True))
        imp_cell = col("重要度")
        type_cell = col("事件类型")
        out.append(
            EventRow(
                section=section,
                rank=_to_int(tds[0].get_text(strip=True)) or len(out) + 1,
                importance=_to_int(imp_cell.get_text(strip=True)) if imp_cell else None,
                event_type=type_cell.get_text(strip=True) if type_cell else None,
                title=_clean_title(title_cell),
                summary=summary,
                sentiment=sentiment,
                source=source,
                url=url,
            )
        )
    return out
 def _collect_events(soup: BeautifulSoup, report_type: str) -> list[EventRow]:
    """按 h2 板块标题收集各事件表。"""
    events: list[EventRow] = []
    for h2 in soup.find_all("h2"):
        title = h2.get_text()
        table = h2.find_next_sibling("table")
        if table is None:
            continue
        section: str | None = None
        if "新闻联播" in title:
            section = "xwlb"
        elif "公告" in title or "调研" in title:
            section = "cninfo"
        elif "重要事件" in title:
            section = "intl" if report_type == "intl" else "news"
        if section is not None:
            events.extend(_parse_event_table(table, section))
    return events
 # --------------------------------------------------------------------------- #
 # 数据总览 stats
 # --------------------------------------------------------------------------- #
 def _table_to_rows(table: Tag) -> list[dict[str, str]]:
    """表格 → [{表头: 值, ...}, ...]（首行为表头）。"""
    rows: list[list[str]] = []
    for tr in table.find_all("tr"):
        cells = [re.sub(r"\s+", " ", c.get_text(strip=True)) for c in tr.find_all(["th", "td"])]
        if cells:
            rows.append(cells)
    if not rows:
        return []
    header = rows[0]
    return [dict(zip(header, r, strict=False)) for r in rows[1:]]
 def _parse_stats_block(h3: Tag) -> tuple[str, object] | None:
    """h3 数据总览区块 → (stats_key, value)。缺失/未知板块返回 None。"""
    title = h3.get_text()
    key = next((k for kw, k in _STATS_SECTION_KEYS if kw in title), None)
    if key is None:
        return None
    block = h3.find_next_sibling()
    if block is None:
        return key, {}
    if block.name == "table":
        return key, _table_to_rows(block)
    classes = block.get("class", []) if isinstance(block.get("class"), list) else []
    if "stats-grid" in classes:
        cards: dict[str, str | int] = {}
        for card in block.find_all("div", class_="stat-card"):
            label = card.select_one(".label")
            num = card.select_one(".num")
            if label is not None:
                num_text = num.get_text(strip=True) if num else ""
                cards[label.get_text(strip=True)] = _to_int(num_text) if _to_int(num_text) is not None else num_text
        return key, cards
    if "source-grid" in classes:
        items: dict[str, str | int] = {}
        for item in block.find_all("div", class_="source-item"):
            name = item.select_one(".s-name")
            count = item.select_one(".s-count")
            if name is not None:
                count_text = count.get_text(strip=True) if count else ""
                items[name.get_text(strip=True)] = _to_int(count_text) or count_text
        return key, items
    if "sentiment-bar" in classes:
        legend = block.find_next_sibling("div", class_="sentiment-legend")
        spans = legend.find_all("span") if legend else []
        return key, [re.sub(r"\s+", " ", s.get_text(strip=True)) for s in spans]
    text = re.sub(r"\s+", " ", block.get_text(strip=True))
    return key, text[:500]
 def _collect_stats(soup: BeautifulSoup) -> dict[str, object]:
    stats: dict[str, object] = {}
    for h3 in soup.find_all("h3"):
        parsed = _parse_stats_block(h3)
        if parsed is not None:
            stats[parsed[0]] = parsed[1]
    return stats
 # --------------------------------------------------------------------------- #
 # 主入口
 # --------------------------------------------------------------------------- #
 def _parse(html: str, file_name: str, report_type: str) -> ReportData:
    parsed = _parse_datetime_from_filename(file_name)
    if parsed is None:
        raise ReportParseError(f"无法从文件名解析日报日期: {file_name}")
    day, gen_from_file = parsed
    soup = BeautifulSoup(html, "html.parser")
    generated_at = _parse_header_generated_at(soup, gen_from_file)
    return ReportData(
        report_date=day,
        report_type=report_type,
        file_name=file_name,
        generated_at=generated_at,
        ai_summary=_parse_ai_summary(soup),
        stats=_collect_stats(soup),
        events=_collect_events(soup, report_type),
    )
 def parse_finance_report(html: str, file_name: str) -> ReportData:
    """解析 A 股日报 finance_news_daily_*.html。"""
    return _parse(html, file_name, "finance")
 def parse_intl_report(html: str, file_name: str) -> ReportData:
    """解析国际财经日报 intl_news_daily_*.html。"""
    return _parse(html, file_name, "intl")
 def parse_report(html: str, file_name: str) -> ReportData:
    """按文件名前缀自动分流 finance / intl。"""
    if "intl_news_daily" in file_name:
        return parse_intl_report(html, file_name)
    return parse_finance_report(html, file_name)
@@ -74,39 +74,15 @@ def run_step(name: str, date_str: str) -> StepResult:
    返回: StepResult。
    """
    # report 步骤:内部函数,不走子进程
-    # 日报默认统计"昨天"的数据（因为今天的数据由当天的定时任务处理）。
+    # 日报按当天日期生成: 新闻由 _collect_news_events 回溯过去 30 小时,
-    # 如果昨天没有事件数据，向前回溯最多 3 天，取最近有数据的日期。
+    # 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 = date.today().strftime("%Y%m%d")
-            report_date: str | None = None
+            logger.info("日报: report_date={} (新闻 30h 回溯, xwlb 前一日)", report_date)
            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,
                )
            path = generate_report(report_date, upload=True)
            elapsed = (datetime.now() - started).total_seconds()
@@ -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
@@ -21,6 +22,8 @@ from typing import Any
 from dotenv import load_dotenv
 from loguru import logger
 from report_db.models import EventRow, ReportData  # noqa: F401 - 供 _build_report_data 注解使用
 # 确保 .env 已加载（模块级常量依赖环境变量）
 load_dotenv()
@@ -36,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
@@ -100,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}/),
+    读取 `day_str` 与前一天两个事件目录，按 publish_time 过滤最近
-    不再用 datetime.now() 做 24h 二次过滤,
+    `_NEWS_LOOKBACK_HOURS`（默认 30）小时内的新闻——07:00 生成当日日报时
-    避免日报早上 8 点跑时前一天新闻被全部过滤掉。
+    可覆盖昨日全天至今晨的新闻。无 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()
@@ -309,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:
@@ -336,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] = []
@@ -593,10 +623,14 @@ 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 上限被截断。
    """
    last_exc: Exception | None = None
    for attempt in range(_LLM_RETRY_TIMES):
        try:
            resp = client.chat.completions.create(
                model=model,
                messages=[
@@ -614,6 +648,18 @@ def _llm_call(client, model: str, prompt: str, max_tokens: int = 1500) -> str:
                    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
 # --------------------------------------------------------------------------- #
@@ -843,7 +889,7 @@ def _render_xwlb_section(xwlb: dict | None) -> str:
 def _render_html(news: dict, cninfo: dict, pipeline: dict,
                 ai_summary: str, day_str: str,
                 xwlb: dict | None = None) -> str:
-    """组装完整 HTML。"""
+    """组装完整 HTML（M10 起废弃：日报已改为结构化入库，此函数不再被调用，保留以便回退）。"""
    # AI 摘要 → HTML
    summary_html = _re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", ai_summary)
@@ -941,8 +987,76 @@ def _render_html(news: dict, cninfo: dict, pipeline: dict,
 # 生成 + 上传
 # --------------------------------------------------------------------------- #
-def generate_report(day_str: str | None = None, *, upload: bool = True) -> Path | None:
+def _build_report_data(news: dict, cninfo: dict, pipeline: dict,
-    """生成每日摘要 HTML 报告,可选上传到 Web 服务器。"""
+                       ai_summary: str, day_str: str,
                       xwlb: dict | None = None) -> ReportData:
    """组装结构化日报数据（M10：写入 MySQL 的前置步骤）。
    事件板块映射：news["high"]→news / cninfo["high"]→cninfo / xwlb["items"]→xwlb。
    数据总览统计以 JSON 快照存入 stats（前端自行解析）。
    """
    events: list[EventRow] = []
    def _rows(items: list[dict], section: str) -> None:
        for i, e in enumerate(items, 1):
            ev = e.get("event", {})
            events.append(
                EventRow(
                    section=section,
                    rank=i,
                    importance=ev.get("importance"),
                    event_type=ev.get("event_type"),
                    title=str(e.get("title", ""))[:512],
                    summary=(ev.get("summary") or None),
                    sentiment=ev.get("sentiment") or None,
                    source=e.get("source_id") or None,
                    url=e.get("url") or None,
                )
            )
    _rows(news.get("high", []), "news")
    _rows(cninfo.get("high", []), "cninfo")
    if xwlb:
        _rows(xwlb.get("items", []), "xwlb")
    stats: dict[str, Any] = {
        "pipeline": pipeline,
        "news": {
            "total": news.get("total", 0),
            "hi_threshold": news.get("hi_threshold"),
            "sentiments": news.get("sentiments", {}),
            "importances": news.get("importances", {}),
            "event_types": news.get("event_types", {}),
        },
        "cninfo": {
            "total": cninfo.get("total", 0),
            "hi_threshold": cninfo.get("hi_threshold"),
            "by_day": cninfo.get("by_day", {}),
            "announcement": cninfo.get("announcement", 0),
            "research": cninfo.get("research", 0),
            "irm": cninfo.get("irm", 0),
        },
    }
    if xwlb:
        stats["xwlb"] = {"total": len(xwlb.get("items", [])), "date": xwlb.get("date", "")}
    return ReportData(
        report_date=datetime.strptime(day_str, "%Y%m%d").date(),
        report_type="finance",
        file_name="",  # 新生成日报唯一键退化为 (report_date, finance, "")
        generated_at=datetime.now(),
        ai_summary=ai_summary or None,
        stats=stats,
        events=events,
    )
 def generate_report(day_str: str | None = None, *, upload: bool = True) -> int | None:
    """生成每日日报并结构化入库（M10 完全切换，不再生成 HTML）。
    `upload` 参数保留以兼容 scheduler/pipeline.py 调用，已无实际作用。
    返回 report_id（成功）或 None（无数据/失败）。
    """
    day_str = day_str or date.today().strftime("%Y%m%d")
    logger.info("生成日报: {}", day_str)
@@ -963,25 +1077,26 @@ def generate_report(day_str: str | None = None, *, upload: bool = True) -> Path
    # AI 摘要(新闻联播 + 新闻 + cninfo)
    ai_summary = _generate_ai_summary(news, cninfo, day_str, xwlb=xwlb)
-    # 渲染
+    # 结构化入库（替代原 HTML 渲染 + 上传）
-    html = _render_html(news, cninfo, pipeline, ai_summary, day_str, xwlb=xwlb)
+    report = _build_report_data(news, cninfo, pipeline, ai_summary, day_str, xwlb=xwlb)
    try:
        from report_db import connect, save_report
-    # 保存（文件名含时间，支持一天多次生成）
+        conn = connect()
-    out_dir = Path("data/reports")
+        try:
-    out_dir.mkdir(parents=True, exist_ok=True)
+            report_id = save_report(conn, report)
-    file_tag = f"{day_str}_{datetime.now():%H%M}"
+        finally:
-    html_path = out_dir / f"finance_news_daily_{file_tag}.html"
+            conn.close()
-    html_path.write_text(html, encoding="utf-8")
+    except Exception as e:
-    logger.info("日报已保存: {} ({} KB)", html_path, len(html) // 1024)
+        logger.exception("日报入库失败: {}", e)
        return None
-    if upload:
+    logger.info("日报已入库: report_id={}", report_id)
-        _upload(html_path, file_tag)
+    return report_id
    return html_path
 def _upload(html_path: Path, file_tag: str) -> bool:
-    """上传 HTML 报告到 Web 服务器。"""
+    """上传 HTML 报告到 Web 服务器（M10 起废弃：不再被调用，保留以便回退）。"""
    today_str = date.today().strftime("%Y%m%d")
    remote_dir = f"{UPLOAD_BASE}/{today_str}/"
    logger.info("上传日报到 {}:{}", UPLOAD_HOST, remote_dir)
@@ -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")
@@ -0,0 +1,225 @@
 """日报结构化组装单元测试（_build_report_data，纯逻辑）。"""
 from __future__ import annotations
 import json
 from datetime import date
 import pytest
 from scheduler.reporter import _build_report_data
 def _fake_event(title: str, importance: int, event_type: str = "其他",
                sentiment: str = "neutral", source_id: str = "cls",
                url: str = "https://x.com/1") -> dict:
    return {
        "title": title,
        "url": url,
        "source_id": source_id,
        "event": {
            "stock_codes": [],
            "company_names": [],
            "industries": [],
            "sentiment": sentiment,
            "importance": importance,
            "event_type": event_type,
            "summary": f"{title}的摘要",
        },
    }
 class TestBuildReportData:
    def test_sections_and_ranks(self) -> None:
        news = {
            "total": 2, "hi_threshold": 4,
            "high": [_fake_event("新闻A", 5), _fake_event("新闻B", 4)],
            "sentiments": {"neutral": 2}, "importances": {5: 1, 4: 1},
            "event_types": {"其他": 2},
        }
        cninfo = {
            "total": 1, "hi_threshold": 2,
            "high": [_fake_event("公告C", 3, event_type="公告")],
            "by_day": {"07月10日": 1}, "announcement": 1, "research": 0, "irm": 0,
        }
        pipeline = {"raw_total": 100, "proc": 90}
        xwlb = {"items": [_fake_event("联播D", 4, event_type="新闻联播", source_id="xwlb")],
                "date": "07月10日"}
        r = _build_report_data(news, cninfo, pipeline, "AI摘要", "20260710", xwlb=xwlb)
        assert r.report_date == date(2026, 7, 10)
        assert r.report_type == "finance"
        assert r.file_name == ""
        assert r.ai_summary == "AI摘要"
        assert [(e.section, e.rank) for e in r.events] == [
            ("news", 1), ("news", 2), ("cninfo", 1), ("xwlb", 1),
        ]
        assert r.events[0].source == "cls"
        assert r.events[3].source == "xwlb"
    def test_stats_snapshot(self) -> None:
        news = {"total": 1, "hi_threshold": 4, "high": [], "sentiments": {},
                "importances": {}, "event_types": {}}
        cninfo = {"total": 0, "hi_threshold": 0, "high": [], "by_day": {},
                  "announcement": 0, "research": 0, "irm": 0}
        r = _build_report_data(news, cninfo, {"raw_total": 100}, "s", "20260710")
        # stats 可 JSON 序列化（入库时 json.dumps）
        json.dumps(r.stats, ensure_ascii=False)
        assert r.stats["pipeline"] == {"raw_total": 100}
        assert r.stats["news"]["total"] == 1
        assert "xwlb" not in r.stats
    def test_title_truncated(self) -> None:
        news = {"total": 1, "hi_threshold": 4,
                "high": [_fake_event("长" * 600, 4)], "sentiments": {},
                "importances": {}, "event_types": {}}
        cninfo = {"total": 0, "hi_threshold": 0, "high": [], "by_day": {},
                  "announcement": 0, "research": 0, "irm": 0}
        r = _build_report_data(news, cninfo, {}, "s", "20260710")
        assert len(r.events[0].title) == 512
    def test_empty_events(self) -> None:
        news = {"total": 0, "hi_threshold": 0, "high": [], "sentiments": {},
                "importances": {}, "event_types": {}}
        cninfo = {"total": 0, "hi_threshold": 0, "high": [], "by_day": {},
                  "announcement": 0, "research": 0, "irm": 0}
        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
@@ -0,0 +1,51 @@
 """日报数据模型单元测试。"""
 from __future__ import annotations
 from datetime import date, datetime
 import pytest
 from pydantic import ValidationError
 from report_db.models import EventRow, ReportData
 class TestEventRow:
    def test_minimal(self) -> None:
        ev = EventRow(section="news", rank=1, title="标题")
        assert ev.importance is None
        assert ev.sentiment is None
    def test_full(self) -> None:
        ev = EventRow(
            section="intl", rank=2, importance=4, event_type="地缘政治",
            title="t", summary="s", sentiment="negative", source="ForexLive",
            url="https://x.com/1",
        )
        assert ev.sentiment == "negative"
    def test_missing_title_raises(self) -> None:
        with pytest.raises(ValidationError):
            EventRow(section="news", rank=1)  # type: ignore[call-arg]
 class TestReportData:
    def test_defaults(self) -> None:
        r = ReportData(
            report_date=date(2026, 7, 11),
            report_type="finance",
            generated_at=datetime(2026, 7, 11, 7, 0),
        )
        assert r.file_name == ""
        assert r.stats == {}
        assert r.events == []
    def test_with_events(self) -> None:
        r = ReportData(
            report_date=date(2026, 7, 11),
            report_type="finance",
            generated_at=datetime(2026, 7, 11, 7, 0),
            ai_summary="摘要",
            events=[EventRow(section="xwlb", rank=1, title="t")],
        )
        assert len(r.events) == 1
@@ -0,0 +1,32 @@
 """历史导入器单元测试（文件匹配/扫描逻辑，不依赖真实 DB）。"""
 from __future__ import annotations
 from pathlib import Path
 from report_import.importer import _match_file
 class TestMatchFile:
    def test_finance(self) -> None:
        p = Path("20260711/finance_news_daily_20260710_0720.html")
        assert _match_file(p, None, None)
        assert _match_file(p, "20260710", None)
        assert _match_file(p, None, "finance")
        assert not _match_file(p, "20260711", None)  # 文件名日期不含 20260711
        assert not _match_file(p, None, "intl")
    def test_no_timestamp_suffix(self) -> None:
        # 早期文件无时间戳后缀，也应匹配
        p = Path("20260616/finance_news_daily_20260616.html")
        assert _match_file(p, None, None)
        assert _match_file(p, "20260616", "finance")
    def test_intl(self) -> None:
        p = Path("20260711/intl_news_daily_20260711_070304.html")
        assert _match_file(p, "20260711", "intl")
        assert not _match_file(p, None, "finance")
    def test_non_report_ignored(self) -> None:
        assert not _match_file(Path("20260711/002714.SZ_0724.html"), None, None)
        assert not _match_file(Path("20260711/readme.md"), None, None)
@@ -0,0 +1,98 @@
 """历史日报解析器单元测试（基于真实样例 HTML）。"""
 from __future__ import annotations
 from datetime import date, datetime
 from pathlib import Path
 import pytest
 from report_import.parser import (
    ReportParseError,
    parse_finance_report,
    parse_intl_report,
    parse_report,
 )
 FIXTURES = Path(__file__).parent / "fixtures"
 FINANCE_HTML = (FIXTURES / "finance_news_daily_20260710_0720.html").read_text(encoding="utf-8")
 INTL_HTML = (FIXTURES / "intl_news_daily_20260711_070304.html").read_text(encoding="utf-8")
 class TestFinanceParse:
    def test_metadata(self) -> None:
        r = parse_finance_report(FINANCE_HTML, "finance_news_daily_20260710_0720.html")
        assert r.report_date == date(2026, 7, 10)
        assert r.report_type == "finance"
        assert r.file_name == "finance_news_daily_20260710_0720.html"
        assert r.generated_at == datetime(2026, 7, 11, 7, 20, 27)  # header"生成于"优先
    def test_ai_summary_lines(self) -> None:
        r = parse_finance_report(FINANCE_HTML, "finance_news_daily_20260710_0720.html")
        assert r.ai_summary is not None
        assert len(r.ai_summary.splitlines()) >= 5
        assert "碳达峰" in r.ai_summary
    def test_sections_and_ranks(self) -> None:
        r = parse_finance_report(FINANCE_HTML, "finance_news_daily_20260710_0720.html")
        sections = {e.section for e in r.events}
        assert sections == {"xwlb", "news", "cninfo"}
        assert sum(1 for e in r.events if e.section == "xwlb") == 16
        assert sum(1 for e in r.events if e.section == "news") == 20
        assert sum(1 for e in r.events if e.section == "cninfo") == 20
    def test_event_fields(self) -> None:
        r = parse_finance_report(FINANCE_HTML, "finance_news_daily_20260710_0720.html")
        xwlb = next(e for e in r.events if e.section == "xwlb")
        assert xwlb.importance == 4
        assert xwlb.event_type == "新闻联播"
        assert xwlb.sentiment == "neutral"
        assert "张国清" in xwlb.title
        news = next(e for e in r.events if e.section == "news" and e.source)
        assert news.source  # 新闻板块带来源
    def test_stats_keys(self) -> None:
        r = parse_finance_report(FINANCE_HTML, "finance_news_daily_20260710_0720.html")
        for key in ("pipeline", "sources", "sentiment", "importance", "event_types"):
            assert key in r.stats, f"缺少 stats.{key}"
        assert r.stats["pipeline"]["M1 原始文章"] == 758
 class TestIntlParse:
    def test_metadata(self) -> None:
        r = parse_intl_report(INTL_HTML, "intl_news_daily_20260711_070304.html")
        assert r.report_date == date(2026, 7, 11)
        assert r.report_type == "intl"
        assert r.generated_at == datetime(2026, 7, 11, 7, 3, 30)
    def test_events_and_source_from_small(self) -> None:
        r = parse_intl_report(INTL_HTML, "intl_news_daily_20260711_070304.html")
        assert len(r.events) == 19
        first = r.events[0]
        assert first.section == "intl"
        assert first.source == "investinglive.com"  # 从摘要 <small>[来源]</small> 提取
        assert first.url.startswith("https://investinglive.com/")
        assert first.importance == 4
        assert first.event_type == "地缘政治"
        # 摘要中不应残留 [来源] 标记
        assert first.summary is not None and "[investinglive.com]" not in first.summary
    def test_stats_keys(self) -> None:
        r = parse_intl_report(INTL_HTML, "intl_news_daily_20260711_070304.html")
        for key in ("pipeline", "sentiment", "importance", "event_types", "source_dist"):
            assert key in r.stats
        assert r.stats["source_dist"][0]["来源"] == "ForexLive"
 class TestParseReportDispatch:
    def test_dispatch_finance(self) -> None:
        r = parse_report(FINANCE_HTML, "finance_news_daily_20260710_0720.html")
        assert r.report_type == "finance"
    def test_dispatch_intl(self) -> None:
        r = parse_report(INTL_HTML, "intl_news_daily_20260711_070304.html")
        assert r.report_type == "intl"
    def test_bad_filename_raises(self) -> None:
        with pytest.raises(ReportParseError):
            parse_report("<html></html>", "not_a_report.html")
@@ -23,6 +23,7 @@ dependencies = [
    { name = "numpy" },
    { name = "openai" },
    { name = "pydantic" },
    { name = "pymysql" },
    { name = "pypdf" },
    { name = "python-dateutil" },
    { name = "python-dotenv" },
@@ -60,6 +61,7 @@ requires-dist = [
    { name = "numpy", specifier = ">=1.26" },
    { name = "openai", specifier = ">=1.40" },
    { name = "pydantic", specifier = ">=2.7" },
    { name = "pymysql", specifier = ">=1.2.0" },
    { name = "pypdf", specifier = ">=6.13.3" },
    { name = "python-dateutil", specifier = ">=2.9" },
    { name = "python-dotenv", specifier = ">=1.0" },
@@ -1983,6 +1985,15 @@ crypto = [
    { name = "cryptography" },
 ]
 [[package]]
 name = "pymysql"
 version = "1.2.0"
 source = { registry = "https://mirrors.aliyun.com/pypi/simple/" }
 sdist = { url = "https://mirrors.aliyun.com/pypi/packages/c9/bc/1c6a92f385940f727daeecf3bacaf186e03875dff57197801046c583bcf0/pymysql-1.2.0.tar.gz", hash = "sha256:6c7b17ca686988104d7426c27895b455cdeea3e9d3ceb1270f0c3704fead8c33" }
 wheels = [
    { url = "https://mirrors.aliyun.com/pypi/packages/c4/bd/2534e130295c8cfd4f0a2e31623baab7502278f1e97bcfe61db75656a77f/pymysql-1.2.0-py3-none-any.whl", hash = "sha256:62169ce6d5510f08e140c5e7990ee884a9764024e4a9a27b2cc11f1099322ae0" },
 ]
 [[package]]
 name = "pyopenssl"
 version = "26.3.0"