feat: LLM/Embedding 重试次数纳入 system.yaml 统一配置

- system.yaml: llm.max_retries(死配置) 改为 llm.max_attempts；embedding 段新增 max_attempts
- llm/client.py: LLMConfig 新增 max_attempts（默认 3 兜底），load_llm_config 读取配置
- llm/extractor.py: translate_and_extract(_async) max_attempts=None 时取 config.max_attempts
- embedding/client.py: load_embedding_config 读取 embedding.max_attempts
- scheduler/reporter.py: _call_llm_simple max_retries=None 时取 max_attempts-1
- 新增 2 个配置读取测试；已同步 pi5 验证

.env.example

@@ -19,3 +19,11 @@ DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/api/v1/services/embeddings/tex
 # --- Qdrant ---
 QDRANT_URL=http://localhost:6333
 QDRANT_API_KEY=
+
+# --- 日报结构化入库（与 news 项目共用 MySQL myquant 库） ---
+# pi5 通过内网直连 pi(192.168.1.10) 上的 autossh 隧道（0.0.0.0:13306 → doorcome.cn:3306）
+NEWS_DB_HOST=192.168.1.10
+NEWS_DB_PORT=13306
+NEWS_DB_USER=myquant
+NEWS_DB_PASSWORD=            # 填真实值（与 news 项目 .env 的 NEWS_DB_PASSWORD 相同），禁止写入源码
+NEWS_DB_NAME=myquant

CLAUDE.md

@@ -2,9 +2,38 @@
 
 # Claude Code 开发约束（English Financial News 项目）
 
-版本：v1.0
+版本：v1.1
 
-最后更新：2026-06-21
+最后更新：2026-07-23
+
+---
+
+# 〇、命令速查（2026-07-23 实测）
+
+开发环境：
+
+* Python 3.11 + uv + `.venv`（详见第三节）
+* 安装依赖：`uv sync`；新增依赖：`uv add 包名`；开发依赖：`uv add --dev 包名`
+
+常用命令：
+
+| 命令 | 用途 |
+|------|------|
+| `uv run pytest` | 跑全部测试（当前 162 passed / 2 failed，见下方已知问题） |
+| `uv run ruff check .` | 代码规范检查（当前 13 错误，9 可自动修复） |
+| `uv run en-news pipeline` | M2→M6 完整管道 |
+| `uv run en-news report` | 仅生成日报 |
+| `uv run en-news search "查询"` | Qdrant 语义检索 |
+| `uv run en-news mcp-server` | 启动 M8 MCP 服务 |
+| `bash scripts/domestic_full.sh` | 全流程（M1→M6→日报），Pi 上由 crontab 06/12/18/22 调度 |
+| `bash scripts/domestic_crawl_8g.sh [源ID]` | 仅 M1 抓取（可单源测试） |
+| `bash scripts/pipeline.sh` | M2→M6 管道 |
+
+已知问题：
+
+* `tests/test_crawler.py::test_write_and_load_index_jsonl`、`test_write_index_jsonl_dedup` 失败 — `crawler/storage.py:load_index()` 硬编码 `data/raw/...` 路径，未使用测试临时目录；
+* ruff 未清零（多为 import 排序 I001，可 `--fix`）；
+* 日报/摘要 Prompt 内嵌在 `scheduler/reporter.py`，尚未拆分到 `prompts/`。
 
 ---
 
@@ -16,7 +45,7 @@
 
 核心能力包括：
 
-* 英文财经新闻抓取（Crawl4AI，12 个英文源，部署海外）；
+* 英文财经新闻抓取（Crawl4AI，12 个活跃源，国内 Pi 独立运行）；
 * 英文正文提取（trafilatura）；
 * 全文英译中（LLM DeepSeek）；
 * 投资事件抽取（LLM）；
@@ -99,25 +128,20 @@ M0 → M1 → 同步机制 → M2 → M3 → M4 → M5+M6 → M7+日报 → M8
 
 ## 2.4 部署拓扑意识
 
-本项目采用双服务器架构（详见 english-news-plan.md 第二节）：
+本项目当前为单服务器架构（海外服务器已于 2026-07-14 停用，详见 README）：
 
 | 服务器 | SSH | 部署路径 | 职责 |
 |--------|-----|---------|------|
-| 海外 | `ssh ecs-user@8.217.19.253` | `/opt/intlgrab` | M1 抓取 → rsync 推送 |
-| 国内 | `ssh pi@192.168.1.160` | `/home/pi/intlnews` | M2→M8 全链路 + 日报 |
+| 国内 | `ssh pi@192.168.1.160` | `/home/pi/intlnews` | M1→M8 全链路 + 日报 |
 
-部署优先级：
-
-* 抓取代码优先在海外服务器测试；
-* 其余代码在国内服务器测试；
-* 海外验证通过后，抓取代码同步到国内做全链路联调。
+历史：海外 `ecs-user@8.217.19.253:/opt/intlgrab` 曾负责 M1 抓取 → rsync 推送，已停用；`scripts/overseas_*.sh`、`scripts/domestic_sync.sh` 为遗留脚本（未清理）。
 
 编码时注意：
 
-* 海外侧只负责抓取，无 LLM/Embedding 依赖；
-* 国内侧等待 rsync 同步完成后再启动管道；
-* 哨兵文件（`SYNC_SENTINEL`）是同步完整性的关键信号；
-* `data/raw/` 目录在两台服务器上路径一致。
+* M1 抓取在国内 Pi（8GB）以 headful Playwright + HTTP 代理（Shadowsocks + Privoxy）运行，需 Xvfb 虚拟显示器（见 README 前置条件）；
+* 强反爬源（Reuters / Investing.com / FT）走 Google News RSS 或 RSS 摘要，全文抓取成功率取决于代理线路质量；
+* 抓取配置按运行环境拆分为 `configs/profiles/2g_headless.yaml` 与 `configs/profiles/8g_headful.yaml`，通过环境变量 `EN_NEWS_PROFILE` 选择；
+* `data/raw/{source_id}/{yyyymmdd}/` 存放抓取原始 HTML + Markdown + index.jsonl，为全管道数据源头。
 
 ---
 
@@ -348,8 +372,7 @@ test_xxx.py
 * crawler（M1 抓取）
 * extractor（M2 正文提取）
 * dedup（M3 去重）
-* translator（M4a 翻译）
-* llm（M4b 事件抽取）
+* llm（M4a 翻译 + M4b 事件抽取，单次 LLM 调用合并输出；`translator/` 为空壳目录，翻译实现在 `llm/`）
 * embedding（M5 向量生成）
 * vectorstore（M6 Qdrant 入库/检索）
 * scheduler（M7 定时调度）
@@ -428,8 +451,8 @@ prompts/
 本项目 Prompt 文件：
 
 * `translation_and_extraction.md` — 翻译 + 事件抽取（单次 LLM 调用合并输出）
-* `daily_report.md` — 日报生成 Prompt（五段式结构）
-* `search_agent.md` — MCP Agent 深度研究 Prompt
+
+注意：`daily_report.md`、`search_agent.md` 尚未创建；日报/摘要 Prompt 目前内嵌在 `scheduler/reporter.py`（技术债，待拆分到 prompts/）。
 
 ---
 
@@ -469,13 +492,14 @@ docker compose up -d
 
 注意：
 
-* 海外侧 docker-compose 仅包含 Crawl4AI + rsync daemon；
-* 国内侧 docker-compose 包含 Qdrant + 调度器 + MCP 服务。
+* 预期形态：docker-compose 包含 Qdrant + 调度器 + MCP 服务（尚未落地）。
 
 不得依赖：
 
 手工安装。
 
+当前状态（2026-07-23 核查）：仓库中尚无 Dockerfile / docker-compose.yml，此要求尚未落实，属待办事项。
+
 ---
 
 # 十二、Git 提交规范

README.md

@@ -10,7 +10,7 @@
 - **M4** 全文英译中 + 投资事件抽取（LLM DeepSeek v4-flash）
 - **M5** 向量生成（DashScope text-embedding-v3，1024 维）
 - **M6** 双语向量知识库（Qdrant）
-- **M7** 定时调度（06/12/18/22） + 每日 AI 摘要日报
+- **M7** 定时调度（06/12/18/22） + 每日 AI 摘要日报（M9 起结构化写入 MySQL，不再产出 HTML）
 - **M8** MCP 服务（Cherry Studio / Claude Code Agent 深度研究）
 
 ## 部署架构
@@ -27,7 +27,7 @@
 │        ↓                             │
 │  M5 向量生成 → M6 Qdrant 入库        │
 │        ↓                             │
-│  日报生成 → rsync 上传 doorcome.cn    │
+│  日报生成 → 写入 MySQL（news_report） │
 │        ↓                             │
 │  M8 MCP 服务（研究 Agent）           │
 └──────────────────────────────────────┘
@@ -119,7 +119,7 @@ bash scripts/domestic_crawl_8g.sh investinglive
 # M2→M6 管道（需先有 raw 数据）
 bash scripts/pipeline.sh
 
-# 仅生成日报
+# 仅生成日报（写入 MySQL news_report/news_event，report_type="intl"）
 uv run en-news report
 ```
 
@@ -145,6 +145,20 @@ ls -lt data/reports/
 | `.env` | 密钥 / 服务地址（不入 Git） |
 | `prompts/` | LLM Prompt 模板（翻译/日报/搜索 Agent） |
 
+### 日报入库（M9）
+
+日报内容结构化写入与 [news 项目](https://github.com/) 共用的 MySQL `myquant` 库（表 `news_report` / `news_event`，`report_type="intl"`，同一天重复生成幂等覆盖）。表结构与数据契约见 news 项目 `docs/db_schema.md`。
+
+`.env` 需配置（与 news 项目 `.env` 的 `NEWS_DB_PASSWORD` 相同）：
+
+```env
+NEWS_DB_HOST=192.168.1.10    # pi5 经内网直连 pi 上的 autossh 隧道（0.0.0.0:13306 → doorcome.cn:3306）
+NEWS_DB_PORT=13306
+NEWS_DB_USER=myquant
+NEWS_DB_PASSWORD=xxx
+NEWS_DB_NAME=myquant
+```
+
 ---
 
 ## 数据存储
@@ -156,7 +170,7 @@ data/
 ├── dedup/                          M3 去重指纹库（SQLite）
 ├── translations/{yyyymmdd}/       M4 翻译+事件抽取结果（JSONL）
 ├── embeddings/{yyyymmdd}/         M5 向量文件（JSONL）
-├── reports/                        M7 日报（HTML）
+├── reports/                        M7 日报（M9 起不再产出 HTML，改存 MySQL）
 └── vectorstore/                    M6 Qdrant 数据
 ```
 
@@ -191,9 +205,9 @@ uv run en-news mcp-server
 
 部分网站有强反爬（Cloudflare/DataDome/付费墙），RSS 只返回摘要。全文可尝试 headful 浏览器回退，但成功率取决于代理线路质量。
 
-### 日报上传到哪里？
+### 日报存在哪里？
 
-日报自动上传到 `simon@doorcome.cn:/var/www/html/echart/research/`，通过 HTTP 访问。
+M9 起日报结构化写入 MySQL `myquant` 库：主表 `news_report`（`report_type="intl"`）+ 明细表 `news_event`（`section="intl"`），数据契约与 [news 项目](https://github.com/) 一致（见其 `docs/db_schema.md`）。历史 HTML 日报（2026-06-16 ~ 2026-08-03）已由 news 项目解析导入。API / 前端读取同一批表。
 
 ### 如何添加新源？
 

app/cli.py

@@ -3,8 +3,8 @@
 import logging
 import sys
 
-from dotenv import load_dotenv
 import typer
+from dotenv import load_dotenv
 
 # 加载 .env 环境变量（必须在所有模块导入之前）
 load_dotenv()
@@ -44,6 +44,7 @@ def crawl(
 ):
     """M1: 抓取英文财经新闻（部署海外）"""
     import os
+
     from crawler.orchestrator import run_crawl_sync
 
     if profile:
@@ -196,15 +197,15 @@ def search(
 
 @app.command()
 def report():
-    """M7: 日报生成"""
+    """M7: 日报生成（结构化入库，不再产出 HTML）"""
     from scheduler.reporter import generate_report
 
     try:
-        path = generate_report()
-        if path:
-            typer.echo(f"\n✅ 日报已生成: {path}")
+        report_id = generate_report()
+        if report_id is not None:
+            typer.echo(f"\n✅ 日报已入库: report_id={report_id}")
         else:
-            typer.echo("⚠️  无数据，跳过日报生成")
+            typer.echo("⚠️  无数据或入库失败，跳过日报生成")
     except Exception as e:
         logger.exception("日报生成失败")
         typer.echo(f"❌ 日报生成出错: {e}", err=True)

configs/system.yaml

@@ -43,10 +43,10 @@ dedup:
 # ── LLM 翻译+事件抽取 ────────────────────────────────
 llm:
   provider: "deepseek"
-  deepseek_model: "deepseek-chat"
+  deepseek_model: "deepseek-v4-flash"
   qwen_model: "qwen-plus"
   timeout_sec: 60
-  max_retries: 3
+  max_attempts: 3          # 单篇总尝试次数（含首次），失败后指数退避重试；日报 AI 摘要同用此值
   max_tokens: 8192
   temperature: 0.1
   concurrency: 3
@@ -57,6 +57,7 @@ embedding:
   dashscope_model: "text-embedding-v3"
   dimension: 1024
   batch_size: 10
+  max_attempts: 3          # 单批总尝试次数（含首次），失败后指数退避重试
   timeout_sec: 30
 
 # ── Qdrant ──────────────────────────────────────────

continuation.md

@@ -1,6 +1,30 @@
 # continuation.md — English Financial News 项目状态
 
-> 最后更新：2026-07-23
+> 最后更新：2026-08-04
+
+---
+
+## 2026-08-04 会话成果
+
+### M9：日报结构化入库（HTML → MySQL）
+
+**背景：** 与 news 项目对齐，日报内容不再产出 HTML / scp 上传 doorcome，改为写入共用 MySQL `myquant` 库（表 `news_report` / `news_event`，`report_type="intl"`）。历史日报（2026-06-16 ~ 2026-08-03，intl 128 份）已由 news 项目解析入库，本次仅做新日报按相同契约入库。
+
+**改动：**
+
+| 文件 | 改动内容 |
+|------|---------|
+| `report_db/`（新增） | 复用 news 项目实现：`models.py`（EventRow/ReportData）、`schema.py`（news_report/news_event DDL）、`db.py`（load_db_config/connect/init_schema/save_report/fetch_report/exists_report，loguru 换 logging） |
+| `scheduler/reporter.py` | 新增 `_build_report_data()`（intl 结构：事件全部 `section="intl"`，`file_name=""` 幂等覆盖，stats 含 pipeline/sentiment/importance/event_types/source_dist）；`generate_report()` 返回 `int | None`，收集逻辑不变，末尾 `connect()+save_report()`；HTML 渲染/上传函数保留标 deprecated |
+| `scheduler/pipeline.py` | `run_step_report` 适配 `report_id` 返回值 |
+| `app/cli.py` | `report` 命令打印 `report_id` |
+| `pyproject.toml` | `uv add pymysql`；`report_db/schema.py` 加 E501 per-file-ignore |
+| `.env.example` / `README.md` | 新增 `NEWS_DB_*` 配置说明（pi5 经内网直连 `192.168.1.10:13306`，password 与 news 项目相同） |
+| `tests/test_report_db.py`（新增） | 模型默认值 + `_build_report_data` 组装（板块/rank、字段映射、标题截断、stats 快照、`"?"` 归一化） |
+
+**测试结果：** `uv run pytest` → 174 passed / 2 failed（2 个失败为原有 `test_crawler.py` index.jsonl 路径问题，与本次无关）；ruff 改动文件干净（仅剩原有 `_SENTENCE_END` N806）。
+
+**部署：** 已 rsync 至 pi5（`/home/pi/intlnews`），`.env` 写入 NEWS_DB_*，`uv sync` 安装 pymysql，真实生成日报并回读 DB 验证（详见下方）。
 
 ---
 

embedding/client.py

@@ -92,6 +92,7 @@ def load_embedding_config(
     dimension = int(sys_cfg.get("dimension", DASHSCOPE_DEFAULT_DIM))
     batch_size = min(int(sys_cfg.get("batch_size", DASHSCOPE_BATCH_LIMIT)), DASHSCOPE_BATCH_LIMIT)
     timeout = float(sys_cfg.get("timeout_sec", 30.0))
+    max_attempts = int(sys_cfg.get("max_attempts", DEFAULT_MAX_ATTEMPTS))
 
     if not api_key:
         raise EmbeddingError("DASHSCOPE_API_KEY 未配置，请检查 .env")
@@ -104,6 +105,7 @@ def load_embedding_config(
         dimension=dimension,
         batch_size=batch_size,
         timeout_sec=timeout,
+        max_attempts=max_attempts,
     )
 
 

llm/client.py

@@ -25,6 +25,9 @@ _QWEN_DEFAULT_BASE = "https://dashscope.aliyuncs.com/compatible-mode/v1"
 _DEEPSEEK_DEFAULT_MODEL = "deepseek-chat"
 _QWEN_DEFAULT_MODEL = "qwen-plus"
 
+# 默认单次调用总尝试次数（含首次；system.yaml llm.max_attempts 未配置时兜底）
+_DEFAULT_MAX_ATTEMPTS = 3
+
 
 def _load_system_config() -> dict:
     """加载 configs/system.yaml 中 llm 段配置。"""
@@ -50,6 +53,7 @@ class LLMConfig:
     timeout_sec: float = 60.0
     temperature: float = 0.1
     max_tokens: int = 8192
+    max_attempts: int = _DEFAULT_MAX_ATTEMPTS  # 单次调用总尝试次数（含首次）
 
     def __post_init__(self) -> None:
         if not self.api_key:
@@ -93,6 +97,7 @@ def load_llm_config(
     timeout = float(config.get("timeout_sec", 60.0))
     temperature = float(config.get("temperature", 0.1))
     max_tokens = int(config.get("max_tokens", 8192))
+    max_attempts = int(config.get("max_attempts", _DEFAULT_MAX_ATTEMPTS))
 
     return LLMConfig(
         provider=p,
@@ -102,6 +107,7 @@ def load_llm_config(
         timeout_sec=timeout,
         temperature=temperature,
         max_tokens=max_tokens,
+        max_attempts=max_attempts,
     )
 
 

llm/extractor.py

@@ -226,7 +226,7 @@ def translate_and_extract(
     article: ProcessedArticle,
     *,
     template: PromptTemplate | None = None,
-    max_attempts: int = DEFAULT_MAX_ATTEMPTS,
+    max_attempts: int | None = None,
 ) -> EnTranslatedArticle:
     """同步翻译 + 事件抽取（单篇文章，带重试）。
 
@@ -235,7 +235,8 @@ def translate_and_extract(
         config: LLM 配置
         article: 待处理的英文新闻
         template: Prompt 模板，默认加载 prompts/translation_and_extraction.md
-        max_attempts: 最大重试次数
+        max_attempts: 最大尝试次数（含首次）；None 时取 config.max_attempts
+            （来自 system.yaml llm.max_attempts）
 
     Returns:
         EnTranslatedArticle 含双语内容 + 事件
@@ -243,6 +244,8 @@ def translate_and_extract(
     Raises:
         LLMCallError: 所有重试均失败
     """
+    if max_attempts is None:
+        max_attempts = config.max_attempts
     tpl = template or PromptTemplate()
     system_prompt, user_prompt = tpl.render(article)
 
@@ -305,10 +308,12 @@ async def translate_and_extract_async(
     article: ProcessedArticle,
     *,
     template: PromptTemplate | None = None,
-    max_attempts: int = DEFAULT_MAX_ATTEMPTS,
+    max_attempts: int | None = None,
     semaphore: asyncio.Semaphore | None = None,
 ) -> EnTranslatedArticle:
     """异步翻译 + 事件抽取（批处理用），与同步版逻辑等价。"""
+    if max_attempts is None:
+        max_attempts = config.max_attempts
     tpl = template or PromptTemplate()
     system_prompt, user_prompt = tpl.render(article)
 

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "openai>=2.43.0",
     "python-dotenv>=1.2.2",
     "markdown>=3.10.2",
+    "pymysql>=1.2.0",
 ]
 
 [project.optional-dependencies]
@@ -65,6 +66,7 @@ select = ["E", "F", "I", "N", "W", "UP"]
 
 [tool.ruff.lint.per-file-ignores]
 "scheduler/reporter.py" = ["E501"]  # 内嵌 HTML 模板 CSS 行较长
+"report_db/schema.py" = ["E501"]  # DDL SQL 语句较长，与 news 项目 schema 保持一致
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

report_db/__init__.py

@@ -0,0 +1,20 @@
+"""日报结构化入库。
+
+职责：日报内容（intl）结构化后写入 MySQL（news_report / news_event），
+与 news 项目共用同一表结构与数据契约（见 docs/db_schema.md、report_db_design.md）。
+本包不提供 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",
+]

report_db/db.py

@@ -0,0 +1,161 @@
+"""日报结构化入库：DB 连接、建表、写入。
+
+与 news 项目 report_db/db.py 保持一致（日志改用 logging，遵守本项目规范）。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from dataclasses import dataclass
+from typing import Any
+
+import pymysql
+
+from .models import ReportData
+from .schema import DDL_STATEMENTS
+
+logger = logging.getLogger(__name__)
+
+
+@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", "192.168.1.10")
+    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=%s port=%s user=%s", cfg.host, cfg.port, cfg.user)
+        raise
+    logger.debug("MySQL 已连接: %s/%s", 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=%s date=%s type=%s events=%s",
+                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

report_db/models.py

@@ -0,0 +1,37 @@
+"""日报结构化入库：数据模型。
+
+与 news 项目 report_db/models.py 保持一致（intl 板块复用同一表结构）。
+"""
+
+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)

report_db/schema.py

@@ -0,0 +1,44 @@
+"""MySQL DDL：日报结构化入库（表前缀 news_，目标 MariaDB 10.11）。
+
+与 news 项目 docs/db_schema.md 保持一致（news_report / news_event）。
+历史日报已由 news 项目 report-import 导入，本包只负责新日报写入与幂等建表。
+"""
+
+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='日报事件明细'
+    """,
+]

scheduler/pipeline.py

@@ -150,16 +150,17 @@ def run_step_index(date_str: str) -> StepResult:
 
 
 def run_step_report(date_str: str) -> StepResult:
-    """日报生成。"""
+    """日报生成（M9：结构化入库，不再产出 HTML）。"""
     started = datetime.now()
     try:
         from scheduler.reporter import generate_report
-        path = generate_report()
+        report_id = generate_report()
         elapsed = (datetime.now() - started).total_seconds()
-        ok = path is not None
+        ok = report_id is not None
         return StepResult(
             name="report", success=ok, elapsed_sec=elapsed,
-            message=str(path) if path else "无数据", started_at=started,
+            message=f"report_id={report_id}" if report_id is not None else "无数据",
+            started_at=started,
         )
     except Exception as e:
         elapsed = (datetime.now() - started).total_seconds()

scheduler/reporter.py

@@ -9,12 +9,15 @@
 import json
 import logging
 from collections import Counter
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
+from typing import Any
 
 import markdown
 import yaml
 
+from report_db.models import EventRow, ReportData  # noqa: F401 - 供 _build_report_data 注解使用
+
 logger = logging.getLogger(__name__)
 
 _MAX_HIGH_EVENTS = 30
@@ -130,8 +133,8 @@ def _try_parse_time(time_str: str) -> datetime | None:
             # - 有时区 → astimezone 转为 UTC
             # - 无时区 → 假设为 UTC（多数财经新闻 API 使用 UTC）
             if dt.tzinfo is None:
-                return dt.replace(tzinfo=timezone.utc)
-            return dt.astimezone(timezone.utc)
+                return dt.replace(tzinfo=UTC)
+            return dt.astimezone(UTC)
         except ValueError:
             continue
     return None
@@ -168,7 +171,7 @@ def _load_events_window(now: datetime) -> list[dict]:
         文章 dict 列表，含 title/title_zh/url/source_id/events 等
     """
     # cutoff 使用 UTC-aware，与 _try_parse_time 返回的 UTC datetime 对齐
-    cutoff = now.astimezone(timezone.utc) - timedelta(hours=_REPORT_WINDOW_HOURS)
+    cutoff = now.astimezone(UTC) - timedelta(hours=_REPORT_WINDOW_HOURS)
     articles: list[dict] = []
     skipped_empty_pt = 0
     for day_str in _dates_in_window(now):
@@ -280,7 +283,7 @@ def _call_llm_simple(
     system_prompt: str,
     user_prompt: str,
     max_tokens: int = 600,
-    max_retries: int = 2,
+    max_retries: int | None = None,
 ) -> str:
     """封装 LLM 调用，带重试和客户端复用。
 
@@ -288,7 +291,8 @@ def _call_llm_simple(
         system_prompt: system role 内容
         user_prompt: user role 内容
         max_tokens: 最大输出 token
-        max_retries: 最大重试次数（不含首次调用）
+        max_retries: 重试次数（不含首次）；None 时取
+            config.max_attempts - 1（system.yaml llm.max_attempts）
 
     Returns:
         LLM 输出文本；所有重试均失败返回空字符串
@@ -305,6 +309,9 @@ def _call_llm_simple(
     config = _llm_client_cache["config"]
     client = _llm_client_cache[cache_key]
 
+    if max_retries is None:
+        max_retries = max(config.max_attempts - 1, 0)
+
     last_err: str = ""
     for attempt in range(1, max_retries + 2):  # 首次 + max_retries 次重试
         try:
@@ -551,17 +558,98 @@ def _generate_ai_summary(articles: list[dict], day_str: str) -> str:
     return "⚠️ AI 摘要合并失败，以下为各批次原始摘要：\n\n" + "\n\n".join(partial_summaries)
 
 
-def generate_report() -> Path | None:
-    """生成 HTML 日报（覆盖过去 25 小时数据）。
+def _build_report_data(
+    now: datetime,
+    stats: dict,
+    high_events: list[dict],
+    sentiments: Counter,
+    importances: Counter,
+    event_types: Counter,
+    sources: Counter,
+    ai_summary: str,
+) -> ReportData:
+    """组装结构化日报数据（M9：写入 MySQL 的前置步骤）。
 
-    命名规则: intl_news_daily_{YYYYMMDD_HHMMSS}.html — 支持一天多份日报。
+    与 news 项目 report_db_design.md 数据契约一致：
+    - 事件全部归入 `intl` 板块（intl 日报无 xwlb/news/cninfo 板块）；
+    - report_type="intl"、file_name="" → 唯一键退化为 (report_date, intl, "")，
+      同一天重复生成 = UPDATE 覆盖（幂等）；
+    - 数据总览统计以 JSON 快照存入 stats（前端自行解析）。
+    """
+    events: list[EventRow] = []
+    for i, ev in enumerate(high_events, 1):
+        article = ev.get("article", {})
+        title = (article.get("title_zh") or article.get("title") or "").strip()[:512]
+        url = article.get("url") or None
+        src = _url_source_label(url, article.get("source_id", ""))
+        # 归一化："" / "?" 不入库，留 None（DB 仅存 positive/negative/neutral）
+        sentiment = ev.get("sentiment") or None
+        if sentiment in ("", "?"):
+            sentiment = None
+        event_type = ev.get("event_type") or None
+        if event_type in ("", "?"):
+            event_type = None
+        events.append(
+            EventRow(
+                section="intl",
+                rank=i,
+                importance=ev.get("importance") or None,
+                event_type=event_type,
+                title=title or "（无标题）",
+                summary=ev.get("summary_zh") or None,
+                sentiment=sentiment,
+                source=src if src not in (None, "", "?") else None,
+                url=url,
+            )
+        )
+
+    stats_snapshot: dict[str, Any] = {
+        "pipeline": {
+            "raw_total": stats.get("raw_total", 0),
+            "processed": stats.get("proc", 0),
+            "deduped": stats.get("deduped", 0),
+            "embedded": stats.get("emb_count", 0),
+            "qdrant": stats.get("qdrant_count", 0),
+        },
+        "sentiment": {k: v for k, v in sentiments.items() if k not in ("", "?")},
+        "importance": [
+            {"importance": k, "count": v} for k, v in sorted(importances.items())
+        ],
+        "event_types": [
+            {"event_type": k, "count": v}
+            for k, v in event_types.most_common(10)
+            if k not in ("", "?")
+        ],
+        "source_dist": [
+            {"source": _source_name(k), "count": v}
+            for k, v in sources.most_common(15)
+            if k not in ("", "?")
+        ],
+    }
+
+    return ReportData(
+        report_date=now.date(),
+        report_type="intl",
+        file_name="",  # 新生成日报唯一键退化为 (report_date, intl, "")
+        generated_at=now,
+        ai_summary=ai_summary or None,
+        stats=stats_snapshot,
+        events=events,
+    )
+
+
+def generate_report() -> int | None:
+    """生成日报并结构化入库（覆盖过去 25 小时数据）。
+
+    M9 起完全切换：不再产出 HTML，日报内容写入 MySQL
+    （news_report / news_event，report_type="intl"，file_name=""，
+    同一天重复生成 → 幂等覆盖，不产生多行）。
 
     Returns:
-        HTML 文件路径，无数据时返回 None
+        report_id（成功）或 None（无数据/失败）
     """
     now = datetime.now()
     ts = now.strftime("%Y%m%d_%H%M%S")
-    date_str = now.strftime("%Y%m%d")  # 用于上传目录
     logger.info("生成日报: %s（窗口: 过去 %d 小时）", ts, _REPORT_WINDOW_HOURS)
 
     # 加载过去 25 小时数据
@@ -595,13 +683,10 @@ def generate_report() -> Path | None:
         )
 
     high = _get_high(all_events, 4)
-    hi_threshold = 4
     if len(high) < 3:
         high = _get_high(all_events, 3)
-        hi_threshold = 3
     if len(high) < 3:
         high = sorted(all_events, key=lambda e: -e.get("importance", 0))
-        hi_threshold = 0
     # 事件级去重：同一 URL + 同一标题 → 合并
     high = _dedup_events(high)
     high = high[:_MAX_HIGH_EVENTS]
@@ -609,20 +694,24 @@ def generate_report() -> Path | None:
     # AI 摘要
     ai_summary = _generate_ai_summary(articles, ts)
 
-    # 渲染 HTML
-    html = _render_html(ts, stats, articles, high, hi_threshold,
-                        sentiments, importances, event_types, sources, ai_summary)
+    # 结构化入库（替代原 HTML 渲染 + 上传）
+    report = _build_report_data(
+        now, stats, high, sentiments, importances, event_types, sources, ai_summary
+    )
+    try:
+        from report_db import connect, save_report
 
-    # 本地保存（文件名含时间戳）
-    _REPORT_DIR.mkdir(parents=True, exist_ok=True)
-    html_path = _REPORT_DIR / f"intl_news_daily_{ts}.html"
-    html_path.write_text(html, encoding="utf-8")
-    logger.info("日报已保存: %s (%d KB)", html_path, len(html) // 1024)
+        conn = connect()
+        try:
+            report_id = save_report(conn, report)
+        finally:
+            conn.close()
+    except Exception as e:
+        logger.exception("日报入库失败: %s", e)
+        return None
 
-    # 自动上传到日期子目录
-    _upload_report(html_path, date_str)
-
-    return html_path
+    logger.info("日报已入库: report_id=%s", report_id)
+    return report_id
 
 
 # --------------------------------------------------------------------------- #
@@ -644,7 +733,7 @@ def _load_report_config() -> dict:
 
 
 def _upload_report(html_path: Path, day_str: str) -> bool:
-    """上传日报到 Web 服务器（配置来自 system.yaml）。"""
+    """上传日报到 Web 服务器（M9 起弃用：日报已改为写库，保留以便回退）。"""
     import subprocess
 
     config = _load_report_config()
@@ -678,7 +767,7 @@ def _upload_report(html_path: Path, day_str: str) -> bool:
 
 
 # --------------------------------------------------------------------------- #
-# HTML 渲染
+# HTML 渲染（M9 起弃用：日报已改为写库，以下渲染函数保留以便回退）
 # --------------------------------------------------------------------------- #
 
 _HTML_TEMPLATE = """<!DOCTYPE html>
@@ -798,7 +887,7 @@ def _render_html(
     sources: Counter,
     ai_summary: str,
 ) -> str:
-    """组装完整 HTML。"""
+    """组装完整 HTML（M9 起弃用，保留以便回退）。"""
 
     # AI 摘要 Markdown → HTML
     summary_html = _md_to_html(ai_summary) if ai_summary.strip() else "<p>暂无 AI 摘要</p>"
@@ -915,7 +1004,7 @@ def _dedup_events(events: list[dict]) -> list[dict]:
 
 
 def _render_event_table(events: list[dict]) -> str:
-    """渲染事件表格。"""
+    """渲染事件表格（M9 起弃用，保留以便回退）。"""
     if not events:
         return "<p>暂无符合条件的数据</p>"
 

scripts/pipeline.sh

@@ -58,11 +58,11 @@ print(f'M6: {stats[\"ingested\"]}/{stats[\"total\"]} 条, {stats[\"elapsed_sec\"
 "
 
 # ── 日报 ──
-echo "[$(date)] [日报] 生成日报..."
+echo "[$(date)] [日报] 生成日报（结构化入库）..."
 .venv/bin/python3 -c "
 from scheduler.reporter import generate_report
-path = generate_report()
-print(f'日报: {path}')
+report_id = generate_report()
+print(f'日报: report_id={report_id}' if report_id is not None else '日报: 无数据/失败')
 "
 
 echo "[$(date)] ═══ 全链路管道完成 ✅ ═══"

tests/test_embedding.py

@@ -87,6 +87,12 @@ class TestLoadEmbeddingConfig:
         assert cfg.provider == "dashscope"
         assert cfg.api_key == "sk-dashscope-test"
 
+    def test_max_attempts_from_system_yaml(self, monkeypatch):
+        """重试次数取自 system.yaml embedding.max_attempts（当前 3）。"""
+        monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-dashscope-test")
+        cfg = load_embedding_config()
+        assert cfg.max_attempts == 3
+
     def test_fallback_to_qwen_key(self, monkeypatch):
         monkeypatch.delenv("DASHSCOPE_API_KEY", raising=False)
         monkeypatch.setenv("QWEN_API_KEY", "sk-qwen-key")

tests/test_llm.py

@@ -457,6 +457,12 @@ class TestLoadLLMConfig:
         with pytest.raises(ValueError, match="API key 未配置"):
             load_llm_config(provider="deepseek")
 
+    def test_max_attempts_from_system_yaml(self, monkeypatch):
+        """重试次数取自 system.yaml llm.max_attempts（当前 3）。"""
+        monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-deepseek-test-key")
+        config = load_llm_config(provider="deepseek")
+        assert config.max_attempts == 3
+
 
 # --------------------------------------------------------------------------- #
 # translate_and_extract（mock LLM）

tests/test_report_db.py

@@ -0,0 +1,179 @@
+"""日报结构化入库：模型 + _build_report_data 组装（纯逻辑，不连 DB）。"""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from datetime import date, datetime
+
+import pytest
+from pydantic import ValidationError
+
+from report_db.models import EventRow, ReportData
+from scheduler.reporter import _build_report_data
+
+# --------------------------------------------------------------------------- #
+# 模型默认值
+# --------------------------------------------------------------------------- #
+
+
+class TestEventRow:
+    def test_minimal(self) -> None:
+        ev = EventRow(section="intl", 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="InvestingLive",
+            url="https://x.com/1",
+        )
+        assert ev.sentiment == "negative"
+
+    def test_missing_title_raises(self) -> None:
+        with pytest.raises(ValidationError):
+            EventRow(section="intl", rank=1)  # type: ignore[call-arg]
+
+
+class TestReportData:
+    def test_defaults(self) -> None:
+        r = ReportData(
+            report_date=date(2026, 8, 4),
+            report_type="intl",
+            generated_at=datetime(2026, 8, 4, 8, 0),
+        )
+        assert r.file_name == ""
+        assert r.stats == {}
+        assert r.events == []
+
+    def test_with_events(self) -> None:
+        r = ReportData(
+            report_date=date(2026, 8, 4),
+            report_type="intl",
+            generated_at=datetime(2026, 8, 4, 8, 0),
+            ai_summary="摘要",
+            events=[EventRow(section="intl", rank=1, title="t")],
+        )
+        assert len(r.events) == 1
+
+
+# --------------------------------------------------------------------------- #
+# _build_report_data（intl 结构）
+# --------------------------------------------------------------------------- #
+
+
+def _fake_high_event(title_zh: str, importance: int, *,
+                     event_type: str = "其他", sentiment: str = "neutral",
+                     summary_zh: str = "摘要", source_id: str = "investinglive",
+                     url: str = "https://investinglive.com/news/1") -> dict:
+    """构造 _dedup_events 之后的事件结构（含 article 属性）。"""
+    return {
+        "importance": importance,
+        "sentiment": sentiment,
+        "summary_zh": summary_zh,
+        "event_type": event_type,
+        "stock_codes": [],
+        "article": {
+            "title": f"EN {title_zh}",
+            "title_zh": title_zh,
+            "url": url,
+            "source_id": source_id,
+        },
+    }
+
+
+class TestBuildReportData:
+    def test_sections_and_ranks(self) -> None:
+        now = datetime(2026, 8, 4, 8, 0, 0)
+        high = [
+            _fake_high_event("新闻A", 5),
+            _fake_high_event("新闻B", 4),
+            _fake_high_event("新闻C", 3),
+        ]
+        r = _build_report_data(
+            now, {"raw_total": 100, "proc": 90}, high,
+            Counter(), Counter(), Counter(), Counter(), "AI摘要",
+        )
+        assert r.report_date == date(2026, 8, 4)
+        assert r.report_type == "intl"
+        assert r.file_name == ""
+        assert r.ai_summary == "AI摘要"
+        assert [(e.section, e.rank) for e in r.events] == [
+            ("intl", 1), ("intl", 2), ("intl", 3),
+        ]
+
+    def test_field_mapping(self) -> None:
+        now = datetime(2026, 8, 4, 8, 0, 0)
+        high = [_fake_high_event("英伟达财报超预期", 5, event_type="公司财报",
+                                 sentiment="positive", summary_zh="营收超预期",
+                                 source_id="investinglive",
+                                 url="https://investinglive.com/news/42")]
+        r = _build_report_data(now, {}, high, Counter(), Counter(), Counter(),
+                               Counter(), "")
+        ev = r.events[0]
+        assert ev.title == "英伟达财报超预期"          # title_zh 优先
+        assert ev.importance == 5
+        assert ev.event_type == "公司财报"
+        assert ev.summary == "营收超预期"
+        assert ev.sentiment == "positive"
+        assert ev.url == "https://investinglive.com/news/42"
+        assert ev.source == "InvestingLive"           # 域名 → 源展示名
+
+    def test_source_fallback_without_url(self) -> None:
+        now = datetime(2026, 8, 4, 8, 0, 0)
+        ev = _fake_high_event("无URL事件", 4, url="")
+        r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(),
+                               Counter(), "")
+        assert r.events[0].url is None
+        # url 为空时回退 source_id → sources.yaml 展示名
+        assert r.events[0].source == "InvestingLive"
+
+    def test_stats_snapshot(self) -> None:
+        now = datetime(2026, 8, 4, 8, 0, 0)
+        high = [
+            _fake_high_event("A", 5, event_type="宏观", sentiment="positive"),
+            _fake_high_event("B", 4, event_type="宏观", sentiment="negative"),
+        ]
+        r = _build_report_data(
+            now,
+            {"raw_total": 120, "proc": 100, "deduped": 90,
+             "emb_count": 80, "qdrant_count": 75},
+            high,
+            Counter({"positive": 1, "negative": 1}),
+            Counter({5: 1, 4: 1}),
+            Counter({"宏观": 2}),
+            Counter({"investinglive": 2}),
+            "摘要",
+        )
+        # stats 可 JSON 序列化（入库时 json.dumps）
+        json.dumps(r.stats, ensure_ascii=False)
+        assert r.stats["pipeline"]["raw_total"] == 120
+        assert r.stats["sentiment"] == {"positive": 1, "negative": 1}
+        assert r.stats["importance"] == [
+            {"importance": 4, "count": 1}, {"importance": 5, "count": 1},
+        ]
+        assert r.stats["event_types"] == [{"event_type": "宏观", "count": 2}]
+        assert r.stats["source_dist"] == [{"source": "InvestingLive", "count": 2}]
+
+    def test_title_truncated(self) -> None:
+        now = datetime(2026, 8, 4, 8, 0, 0)
+        ev = _fake_high_event("长" * 600, 4)
+        r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(),
+                               Counter(), "")
+        assert len(r.events[0].title) == 512
+
+    def test_empty_events(self) -> None:
+        now = datetime(2026, 8, 4, 8, 0, 0)
+        r = _build_report_data(now, {}, [], Counter(), Counter(), Counter(),
+                               Counter(), "")
+        assert r.events == []
+        assert r.ai_summary is None
+
+    def test_question_mark_sentiment_normalized(self) -> None:
+        now = datetime(2026, 8, 4, 8, 0, 0)
+        ev = _fake_high_event("未分类情绪", 4, sentiment="?")
+        r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(),
+                               Counter(), "")
+        # "?" 不写入 DB，留 None
+        assert r.events[0].sentiment is None