simon
366e60e8a9
- 新增 report_db 包: MySQL 连接/建表/幂等写入 (news_report/news_event, myquant 库) - 新增 report_import 包: 历史 178 份日报 HTML 解析入库, 表头驱动列映射 - reporter.py 完全切换: generate_report 结构化入库, 不再生成/上传 HTML - CLI: 新增 report-import 子命令 - 依赖: uv add pymysql; 配置: NEWS_DB_* / REPORT_HISTORY_DIR - 文档: docs/report_db_design.md(实现逻辑), docs/db_schema.md(表结构供 API/前端) - 测试: 24 个单测通过 (parser/builder/models/importer)
331 lines
15 KiB
Markdown
331 lines
15 KiB
Markdown
# 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 更新
|