Compare commits

..
2 Commits
Author SHA1 Message Date
simon 2f2428aa9a 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 前一日 用例
2026-08-05 08:34:02 +08:00
simon 366e60e8a9 feat: 日报结构化入库(M10 前后端分离数据层)
- 新增 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)
2026-08-03 21:32:07 +08:00
28 changed files with 2225 additions and 101 deletions
+12
View File
@@ -69,3 +69,15 @@ PIPELINE_STEP_TIMEOUT=1800
# ---- cninfo 公告抓取 ---- # ---- cninfo 公告抓取 ----
CNINFO_PDF_BASE=http://static.cninfo.com.cn 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...)
+3
View File
@@ -57,3 +57,6 @@ qdrant_storage/
# 调试输出 # 调试输出
debug/ debug/
tmp/ tmp/
# M10: 历史日报源文件副本(可从 doorcome 重新拉取)
data/reports_history/
+36 -1
View File
@@ -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=autointegration 标记默认跳过)
- 静态检查:`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、.envAPI Key 只放 .env
- Prompt 在 prompts/*.md,禁止写死在代码中
- 会话恢复上下文以 continuation.md 为准
—— CLAUDE.md 结束 —— —— CLAUDE.md 结束 ——
+5 -4
View File
@@ -26,9 +26,9 @@
## 当前状态 ## 当前状态
| 已完成 | M0~M8、**M9 投研 Agent Prompt** | | 已完成 | M0~M8、M9 投研 Agent Prompt、**M10 日报结构化入库** |
| --- | --- | | --- | --- |
| 进行中 | 等待 M9 验收 | | 进行中 | 等待 M10 验收 |
| 下一步 | 项目整合与优化 | | 下一步 | 项目整合与优化 |
详见 `continuation.md`。开发遵循 `project_plan.md` 9 个 Milestone,分阶段交付。 详见 `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 "政策" --source cls --sentiment positive
uv run a-share search "风险" --stock 001212 --min-importance 3 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 --date 20260616 # 指定日期
uv run a-share report-import # 历史日报 HTML 解析入库(幂等)
uv run a-share pipeline --once --report # 全链路末尾自动生成日报 uv run a-share pipeline --once --report # 全链路末尾自动生成日报
# 状态总览 # 状态总览
+27 -8
View File
@@ -18,6 +18,7 @@ from __future__ import annotations
import argparse import argparse
import json import json
import os
import subprocess import subprocess
import sys import sys
from collections import Counter from collections import Counter
@@ -394,24 +395,34 @@ def cmd_status(args: argparse.Namespace) -> int: # noqa: ARG001
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
def cmd_report(args: argparse.Namespace) -> int: def cmd_report(args: argparse.Namespace) -> int:
"""生成 HTML 日报并上传到 Web 服务器""" """生成日报并结构化入库(M10:不再生成 HTML/上传)"""
load_dotenv() load_dotenv()
from scheduler.reporter import generate_report from scheduler.reporter import generate_report
day_str = args.date or _today() day_str = args.date or _today()
print(f"📊 生成日报: {day_str}") 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("⚠️ 无数据或生成失败") print("⚠️ 无数据或生成失败")
return 1 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 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 配置建议 # discover — 自动分析站点,生成 sources.yaml 配置建议
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
@@ -917,6 +928,14 @@ def main() -> int:
pl = sp.add_parser("list", help="查看关注列表") pl = sp.add_parser("list", help="查看关注列表")
pl.set_defaults(func=cmd_watchlist_list) 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 ---- # ---- stock-report ----
p = sub.add_parser("stock-report", help="生成关注列表个股日报") p = sub.add_parser("stock-report", help="生成关注列表个股日报")
p.add_argument("--no-upload", action="store_true", help="仅生成不上传") p.add_argument("--no-upload", action="store_true", help="仅生成不上传")
+86 -6
View File
@@ -1,6 +1,6 @@
# continuation.md # continuation.md
> `checkpoint` @ 2026-07-17 07:51 > `checkpoint` @ 2026-08-05 08:30
--- ---
@@ -10,15 +10,95 @@
| --- | --- | | --- | --- |
| 新闻源 | 14 个(13 Web + 1 API: xwlb 新闻联播) | | 新闻源 | 14 个(13 Web + 1 API: xwlb 新闻联播) |
| Qdrant | 本地文件模式 `data/qdrant_storage/` | | Qdrant | 本地文件模式 `data/qdrant_storage/` |
| 日报 | 5 板块: AI摘要 / 新闻联播 / 财经新闻 / 公告调研 / 数据总览 | | 日报 | **M10 完成并已部署 pi5: 结构化入库 MySQL;日报按当天日期生成(新闻 30h 回溯 / xwlb 取前一日 / 公告调研近 15 日)** |
| 调度器 | APSchedulersystemd `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`(绝不允许擅自修改 | | 调度器 | APSchedulersystemd `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 | | 抓取方式 | 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=1912026-08-05):news 20 + cninfo 20 + xwlb 3408-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 被拦截(EHOSTUNREACHnc/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=0DB 177 行(finance 49 + intl 1281 个跨目录同名文件被幂等合并)+ 事件 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 keyAI 摘要会降级 WARNING(不影响入库);生产 pi5 需配置 NEWS_DB_PASSWORD 且解决 13306 隧道可达性(见待确认项)。
---
## 历史 (2026-07-17) — 禁用个股日报
**操作:** `STOCK_REPORT_TIME=` 设为空,`run_scheduler.py` 加空值守卫。 **操作:** `STOCK_REPORT_TIME=` 设为空,`run_scheduler.py` 加空值守卫。
+111
View File
@@ -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 128finance 少 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`(日报生成)。
+330
View File
@@ -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 事件 JSONdata/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 / ReportDataPydantic
├── 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_idDELETE 旧事件后批量 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_reportBeautifulSoup
└── 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 并返回 Nonepipeline 该步骤失败,其余步骤不受影响) |
| 单份历史文件解析失败 | 记 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`,默认跳过) | 连真实 MySQLinit_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 更新
+10 -9
View File
@@ -7,7 +7,8 @@
DeepSeek: DEEPSEEK_API_KEY / DEEPSEEK_BASE_URL / DEEPSEEK_MODEL DeepSeek: DEEPSEEK_API_KEY / DEEPSEEK_BASE_URL / DEEPSEEK_MODEL
Qwen: QWEN_API_KEY / QWEN_BASE_URL / QWEN_MODEL Qwen: QWEN_API_KEY / QWEN_BASE_URL / QWEN_MODEL
(QWEN_API_KEY -> DASHSCOPE_API_KEY 兜底) (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 from __future__ import annotations
@@ -22,10 +23,6 @@ from openai import AsyncOpenAI, OpenAI
_DEEPSEEK_DEFAULT_BASE = "https://api.deepseek.com" _DEEPSEEK_DEFAULT_BASE = "https://api.deepseek.com"
_QWEN_DEFAULT_BASE = "https://dashscope.aliyuncs.com/compatible-mode/v1" _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_TIMEOUT_SEC = 60.0
DEFAULT_TEMPERATURE = 0.1 DEFAULT_TEMPERATURE = 0.1
@@ -69,13 +66,17 @@ def load_llm_config(
if p == "deepseek": if p == "deepseek":
api_key = _read_env("DEEPSEEK_API_KEY") or "" api_key = _read_env("DEEPSEEK_API_KEY") or ""
base = _read_env("DEEPSEEK_BASE_URL", _DEEPSEEK_DEFAULT_BASE) or _DEEPSEEK_DEFAULT_BASE 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"): elif p in ("qwen", "dashscope"):
api_key = _read_env("QWEN_API_KEY") or _read_env("DASHSCOPE_API_KEY") or "" 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 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 p = "qwen" # 内部统一用 qwen
else: else:
raise ValueError(f"未知 LLM provider: {p!r},仅支持 deepseek / qwen") raise ValueError(f"未知 LLM provider: {p!r},仅支持 deepseek / qwen")
+94
View File
@@ -602,5 +602,99 @@ Claude Code 必须严格遵守:
系统自动检索知识库、分析新闻事件,并生成完整研究报告。 系统自动检索知识库、分析新闻事件,并生成完整研究报告。
---
# 十八、Milestone 10:日报结构化入库(前后端分离数据层)
> 状态:✅ 已实施(2026-08-03,等待人工验收)
> 范围:本项目只负责「日报内容生成 + 结构化存入 MySQL」,**不实现 API 与前端**(由用户另行实现)。
## 背景与决策
| 决策点 | 结论 |
| --- | --- |
| DB | 192.168.1.10:13306pi 上 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-ADB 连接层与建表
- `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. **生产连接**pi5192.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 结束 —— —— project_plan.md 结束 ——
+1
View File
@@ -37,6 +37,7 @@ dependencies = [
"mcp>=1.0", "mcp>=1.0",
"pypdf>=6.13.3", "pypdf>=6.13.3",
"markitdown[all]>=0.1.5", "markitdown[all]>=0.1.5",
"pymysql>=1.2.0",
] ]
[project.optional-dependencies] [project.optional-dependencies]
+19
View File
@@ -0,0 +1,19 @@
"""日报结构化入库(Milestone 10)。
职责日报内容finance/intl结构化后写入 MySQLnews_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",
]
+156
View File
@@ -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
+34
View File
@@ -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)
+43
View File
@@ -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='日报事件明细'
""",
]
+21
View File
@@ -0,0 +1,21 @@
"""历史日报解析与批量导入(Milestone 10)。
doorcome 历史日报 HTMLfinance/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",
]
+89
View File
@@ -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
+298
View File
@@ -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)
+4 -28
View File
@@ -74,39 +74,15 @@ def run_step(name: str, date_str: str) -> StepResult:
返回: StepResult 返回: StepResult
""" """
# report 步骤:内部函数,不走子进程 # report 步骤:内部函数,不走子进程
# 日报默认统计"昨天"的数据(因为今天的数据由当天的定时任务处理)。 # 日报按当天日期生成: 新闻由 _collect_news_events 回溯过去 30 小时,
# 如果昨天没有事件数据,向前回溯最多 3 天,取最近有数据的日期 # xwlb 由 _collect_xwlb 固定取前一日(已播出)联播
if name == "report": if name == "report":
started = datetime.now() started = datetime.now()
try: try:
from datetime import timedelta # noqa: E402
from pathlib import Path # noqa: E402
from .reporter import generate_report # 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) path = generate_report(report_date, upload=True)
elapsed = (datetime.now() - started).total_seconds() elapsed = (datetime.now() - started).total_seconds()
+158 -43
View File
@@ -13,6 +13,7 @@ import json
import os as _os import os as _os
import re as _re import re as _re
import subprocess import subprocess
import time
from collections import Counter from collections import Counter
from datetime import date, datetime, timedelta from datetime import date, datetime, timedelta
from pathlib import Path from pathlib import Path
@@ -21,6 +22,8 @@ from typing import Any
from dotenv import load_dotenv from dotenv import load_dotenv
from loguru import logger from loguru import logger
from report_db.models import EventRow, ReportData # noqa: F401 - 供 _build_report_data 注解使用
# 确保 .env 已加载(模块级常量依赖环境变量) # 确保 .env 已加载(模块级常量依赖环境变量)
load_dotenv() load_dotenv()
@@ -36,6 +39,13 @@ NEWS_DAYS_BACK = 1 # 新闻回溯天数
_MAX_HIGH_EVENTS = 20 _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]: def _load_source_names() -> dict[str, str]:
import yaml 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]: def _collect_news_events(day_str: str) -> dict[str, Any]:
"""收集新闻事件(排除 cninfo)。 """收集新闻事件(排除 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] = [] news_ev: list[dict] = []
for e in all_ev: for e in all_ev:
if e["source_id"] == "cninfo": if e["source_id"] == "cninfo":
continue 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) news_ev.append(e)
sentiments: Counter = Counter() 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]: def _collect_xwlb(day_str: str) -> dict[str, Any]:
"""收集新闻联播要闻(从 doorcome API /api/xwlbFine/ 获取)。 """收集新闻联播要闻(从 doorcome API /api/xwlbFine/ 获取)。
新闻联播每天 19:00 播出日报在早上生成时当日联播尚未播出
因此固定取 `day_str` 前一日最近一期已播出的联播数据
API 返回 AI 精编后的独立新闻条目含标题+正文 API 返回 AI 精编后的独立新闻条目含标题+正文
跳过第 1 "内容提要"仅为节目开场白 跳过第 1 "内容提要"仅为节目开场白
返回: {"items": [event_dict, ...], "date": "MM月DD日", "source_date": "20260622"} 返回: {"items": [event_dict, ...], "date": "MM月DD日", "source_date": "前一日"}
""" """
import urllib.request 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: try:
req = urllib.request.Request(api_url) req = urllib.request.Request(api_url)
with urllib.request.urlopen(req, timeout=15) as resp: with urllib.request.urlopen(req, timeout=15) as resp:
@@ -336,7 +366,7 @@ def _collect_xwlb(day_str: str) -> dict[str, Any]:
if dates: if dates:
d = min(dates) d = min(dates)
result["date"] = f"{d[5:7]}{d[8:10]}" result["date"] = f"{d[5:7]}{d[8:10]}"
result["source_date"] = day_str result["source_date"] = prev_day
# 转换为事件格式,跳过第 1 条(内容提要/开场白) # 转换为事件格式,跳过第 1 条(内容提要/开场白)
events: list[dict] = [] events: list[dict] = []
@@ -593,27 +623,43 @@ def _build_prompt(lines: list[str], day_str: str) -> str:
def _llm_call(client, model: str, prompt: str, max_tokens: int = 1500) -> str: 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 上限被截断 finish_reason 'length' 则说明达到 max_tokens 上限被截断
""" """
resp = client.chat.completions.create( last_exc: Exception | None = None
model=model, for attempt in range(_LLM_RETRY_TIMES):
messages=[ try:
{"role": "system", "content": "你是 A 股日报撰写助手,输出简洁、有洞察的新闻摘要。"}, resp = client.chat.completions.create(
{"role": "user", "content": prompt}, model=model,
], messages=[
temperature=0.3, {"role": "system", "content": "你是 A 股日报撰写助手,输出简洁、有洞察的新闻摘要。"},
max_tokens=max_tokens, {"role": "user", "content": prompt},
) ],
content = (resp.choices[0].message.content or "").strip() temperature=0.3,
finish = getattr(resp.choices[0], "finish_reason", None) max_tokens=max_tokens,
if finish == "length": )
logger.warning( content = (resp.choices[0].message.content or "").strip()
"AI 摘要可能被截断: max_tokens={} finish_reason=length 实际输出 {} 字符", finish = getattr(resp.choices[0], "finish_reason", None)
max_tokens, len(content), if finish == "length":
) logger.warning(
return content "AI 摘要可能被截断: max_tokens={} finish_reason=length 实际输出 {} 字符",
max_tokens, len(content),
)
return content
except Exception as e:
last_exc = e
if attempt < _LLM_RETRY_TIMES - 1:
wait = _LLM_RETRY_BACKOFF_SEC * (2 ** attempt)
logger.warning(
"AI 摘要 LLM 调用失败(第 {}/{} 次): {}; {} 秒后重试",
attempt + 1, _LLM_RETRY_TIMES, e, round(wait, 2),
)
time.sleep(wait)
logger.error("AI 摘要 LLM 调用重试 {} 次仍失败: {}", _LLM_RETRY_TIMES, last_exc)
assert last_exc is not None
raise last_exc
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
@@ -843,7 +889,7 @@ def _render_xwlb_section(xwlb: dict | None) -> str:
def _render_html(news: dict, cninfo: dict, pipeline: dict, def _render_html(news: dict, cninfo: dict, pipeline: dict,
ai_summary: str, day_str: str, ai_summary: str, day_str: str,
xwlb: dict | None = None) -> str: xwlb: dict | None = None) -> str:
"""组装完整 HTML。""" """组装完整 HTML(M10 起废弃:日报已改为结构化入库,此函数不再被调用,保留以便回退)"""
# AI 摘要 → HTML # AI 摘要 → HTML
summary_html = _re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", ai_summary) 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") day_str = day_str or date.today().strftime("%Y%m%d")
logger.info("生成日报: {}", day_str) logger.info("生成日报: {}", day_str)
@@ -963,25 +1077,26 @@ def generate_report(day_str: str | None = None, *, upload: bool = True) -> Path
# AI 摘要(新闻联播 + 新闻 + cninfo) # AI 摘要(新闻联播 + 新闻 + cninfo)
ai_summary = _generate_ai_summary(news, cninfo, day_str, xwlb=xwlb) 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: def _upload(html_path: Path, file_tag: str) -> bool:
"""上传 HTML 报告到 Web 服务器。""" """上传 HTML 报告到 Web 服务器(M10 起废弃:不再被调用,保留以便回退)"""
today_str = date.today().strftime("%Y%m%d") today_str = date.today().strftime("%Y%m%d")
remote_dir = f"{UPLOAD_BASE}/{today_str}/" remote_dir = f"{UPLOAD_BASE}/{today_str}/"
logger.info("上传日报到 {}:{}", UPLOAD_HOST, remote_dir) logger.info("上传日报到 {}:{}", UPLOAD_HOST, remote_dir)
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+15 -2
View File
@@ -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: def test_load_llm_config_deepseek_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LLM_PROVIDER", "deepseek") monkeypatch.setenv("LLM_PROVIDER", "deepseek")
monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-test-deepseek") monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-test-deepseek")
monkeypatch.setenv("DEEPSEEK_MODEL", "deepseek-v4-flash")
monkeypatch.delenv("LLM_MODEL", raising=False) monkeypatch.delenv("LLM_MODEL", raising=False)
cfg = load_llm_config() cfg = load_llm_config()
assert cfg.provider == "deepseek" assert cfg.provider == "deepseek"
assert cfg.api_key == "sk-test-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 assert "deepseek" in cfg.base_url
def test_load_llm_config_qwen_from_env(monkeypatch: pytest.MonkeyPatch) -> None: def test_load_llm_config_qwen_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LLM_PROVIDER", "qwen") monkeypatch.setenv("LLM_PROVIDER", "qwen")
monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test-qwen") monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test-qwen")
monkeypatch.setenv("QWEN_MODEL", "qwen-plus")
monkeypatch.delenv("LLM_MODEL", raising=False) monkeypatch.delenv("LLM_MODEL", raising=False)
cfg = load_llm_config() cfg = load_llm_config()
assert cfg.provider == "qwen" assert cfg.provider == "qwen"
assert cfg.api_key == "sk-test-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 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: def test_load_llm_config_missing_key_raises(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) 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") load_llm_config(provider="deepseek")
+225
View File
@@ -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
+51
View File
@@ -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
+32
View File
@@ -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)
+98
View File
@@ -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")
Generated
+11
View File
@@ -23,6 +23,7 @@ dependencies = [
{ name = "numpy" }, { name = "numpy" },
{ name = "openai" }, { name = "openai" },
{ name = "pydantic" }, { name = "pydantic" },
{ name = "pymysql" },
{ name = "pypdf" }, { name = "pypdf" },
{ name = "python-dateutil" }, { name = "python-dateutil" },
{ name = "python-dotenv" }, { name = "python-dotenv" },
@@ -60,6 +61,7 @@ requires-dist = [
{ name = "numpy", specifier = ">=1.26" }, { name = "numpy", specifier = ">=1.26" },
{ name = "openai", specifier = ">=1.40" }, { name = "openai", specifier = ">=1.40" },
{ name = "pydantic", specifier = ">=2.7" }, { name = "pydantic", specifier = ">=2.7" },
{ name = "pymysql", specifier = ">=1.2.0" },
{ name = "pypdf", specifier = ">=6.13.3" }, { name = "pypdf", specifier = ">=6.13.3" },
{ name = "python-dateutil", specifier = ">=2.9" }, { name = "python-dateutil", specifier = ">=2.9" },
{ name = "python-dotenv", specifier = ">=1.0" }, { name = "python-dotenv", specifier = ">=1.0" },
@@ -1983,6 +1985,15 @@ crypto = [
{ name = "cryptography" }, { 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]] [[package]]
name = "pyopenssl" name = "pyopenssl"
version = "26.3.0" version = "26.3.0"