myquant/djapi/docs/report_db_design.md

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）

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

解析策略：表头驱动列映射。不同日报表格列集合不同：

板块	表格列（	）	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 关键函数：

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：

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：

@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）

# ---- 日报结构化入库 (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 更新