Author SHA1 Message Date
simon d4a55bcaaa feat: M9 日报结构化入库(HTML 改为写入 MySQL news_report/news_event,report_type=intl)
- 新增 report_db/ 包(models/schema/db,复用 news 项目实现,幂等 upsert)
- reporter.py: _build_report_data + generate_report 写库返回 report_id
- pipeline/cli 适配 report_id 返回值;HTML 渲染/上传保留 deprecated
- 新增 tests/test_report_db.py;.env.example 增加 NEWS_DB_* 配置
- 已部署 pi5 并验证:真实生成 report_id=184(12 事件)+ 幂等覆盖
2026-08-04 11:09:17 +08:00
14 changed files with 669 additions and 69 deletions
+8
View File
@@ -19,3 +19,11 @@ DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/api/v1/services/embeddings/tex
# --- Qdrant ---
QDRANT_URL=http://localhost:6333
QDRANT_API_KEY=
# --- 日报结构化入库(与 news 项目共用 MySQL myquant 库) ---
# pi5 通过内网直连 pi(192.168.1.10) 上的 autossh 隧道(0.0.0.0:13306 → doorcome.cn:3306
NEWS_DB_HOST=192.168.1.10
NEWS_DB_PORT=13306
NEWS_DB_USER=myquant
NEWS_DB_PASSWORD= # 填真实值(与 news 项目 .env 的 NEWS_DB_PASSWORD 相同),禁止写入源码
NEWS_DB_NAME=myquant
+45 -21
View File
@@ -2,9 +2,38 @@
# Claude Code 开发约束(English Financial News 项目)
版本:v1.0
版本:v1.1
最后更新:2026-06-21
最后更新:2026-07-23
---
# 〇、命令速查(2026-07-23 实测)
开发环境:
* Python 3.11 + uv + `.venv`(详见第三节)
* 安装依赖:`uv sync`;新增依赖:`uv add 包名`;开发依赖:`uv add --dev 包名`
常用命令:
| 命令 | 用途 |
|------|------|
| `uv run pytest` | 跑全部测试(当前 162 passed / 2 failed,见下方已知问题) |
| `uv run ruff check .` | 代码规范检查(当前 13 错误,9 可自动修复) |
| `uv run en-news pipeline` | M2→M6 完整管道 |
| `uv run en-news report` | 仅生成日报 |
| `uv run en-news search "查询"` | Qdrant 语义检索 |
| `uv run en-news mcp-server` | 启动 M8 MCP 服务 |
| `bash scripts/domestic_full.sh` | 全流程(M1→M6→日报),Pi 上由 crontab 06/12/18/22 调度 |
| `bash scripts/domestic_crawl_8g.sh [源ID]` | 仅 M1 抓取(可单源测试) |
| `bash scripts/pipeline.sh` | M2→M6 管道 |
已知问题:
* `tests/test_crawler.py::test_write_and_load_index_jsonl``test_write_index_jsonl_dedup` 失败 — `crawler/storage.py:load_index()` 硬编码 `data/raw/...` 路径,未使用测试临时目录;
* ruff 未清零(多为 import 排序 I001,可 `--fix`);
* 日报/摘要 Prompt 内嵌在 `scheduler/reporter.py`,尚未拆分到 `prompts/`
---
@@ -16,7 +45,7 @@
核心能力包括:
* 英文财经新闻抓取(Crawl4AI,12 个英文源,部署海外);
* 英文财经新闻抓取(Crawl4AI,12 个活跃源,国内 Pi 独立运行);
* 英文正文提取(trafilatura);
* 全文英译中(LLM DeepSeek);
* 投资事件抽取(LLM);
@@ -99,25 +128,20 @@ M0 → M1 → 同步机制 → M2 → M3 → M4 → M5+M6 → M7+日报 → M8
## 2.4 部署拓扑意识
本项目采用双服务器架构(详见 english-news-plan.md 第二节):
本项目当前为单服务器架构(海外服务器已于 2026-07-14 停用,详见 README):
| 服务器 | SSH | 部署路径 | 职责 |
|--------|-----|---------|------|
| 海外 | `ssh ecs-user@8.217.19.253` | `/opt/intlgrab` | M1 抓取 → rsync 推送 |
| 国内 | `ssh pi@192.168.1.160` | `/home/pi/intlnews` | M2→M8 全链路 + 日报 |
| 国内 | `ssh pi@192.168.1.160` | `/home/pi/intlnews` | M1→M8 全链路 + 日报 |
部署优先级:
* 抓取代码优先在海外服务器测试;
* 其余代码在国内服务器测试;
* 海外验证通过后,抓取代码同步到国内做全链路联调。
历史:海外 `ecs-user@8.217.19.253:/opt/intlgrab` 曾负责 M1 抓取 → rsync 推送,已停用;`scripts/overseas_*.sh``scripts/domestic_sync.sh` 为遗留脚本(未清理)。
编码时注意:
* 海外侧只负责抓取,无 LLM/Embedding 依赖
* 国内侧等待 rsync 同步完成后再启动管道
* 哨兵文件(`SYNC_SENTINEL`)是同步完整性的关键信号
* `data/raw/` 目录在两台服务器上路径一致
* M1 抓取在国内 Pi8GB)以 headful Playwright + HTTP 代理(Shadowsocks + Privoxy)运行,需 Xvfb 虚拟显示器(见 README 前置条件)
* 强反爬源(Reuters / Investing.com / FT)走 Google News RSS 或 RSS 摘要,全文抓取成功率取决于代理线路质量
* 抓取配置按运行环境拆分为 `configs/profiles/2g_headless.yaml``configs/profiles/8g_headful.yaml`,通过环境变量 `EN_NEWS_PROFILE` 选择
* `data/raw/{source_id}/{yyyymmdd}/` 存放抓取原始 HTML + Markdown + index.jsonl,为全管道数据源头
---
@@ -348,8 +372,7 @@ test_xxx.py
* crawlerM1 抓取)
* extractorM2 正文提取)
* dedupM3 去重)
* translatorM4a 翻译
* llmM4b 事件抽取)
* llmM4a 翻译 + M4b 事件抽取,单次 LLM 调用合并输出;`translator/` 为空壳目录,翻译实现在 `llm/`
* embeddingM5 向量生成)
* vectorstoreM6 Qdrant 入库/检索)
* schedulerM7 定时调度)
@@ -428,8 +451,8 @@ prompts/
本项目 Prompt 文件:
* `translation_and_extraction.md` — 翻译 + 事件抽取(单次 LLM 调用合并输出)
* `daily_report.md` — 日报生成 Prompt(五段式结构)
* `search_agent.md` — MCP Agent 深度研究 Prompt
注意:`daily_report.md``search_agent.md` 尚未创建;日报/摘要 Prompt 目前内嵌在 `scheduler/reporter.py`(技术债,待拆分到 prompts/)。
---
@@ -469,13 +492,14 @@ docker compose up -d
注意:
* 海外侧 docker-compose 包含 Crawl4AI + rsync daemon
* 国内侧 docker-compose 包含 Qdrant + 调度器 + MCP 服务。
* 预期形态:docker-compose 包含 Qdrant + 调度器 + MCP 服务(尚未落地)。
不得依赖:
手工安装。
当前状态(2026-07-23 核查):仓库中尚无 Dockerfile / docker-compose.yml,此要求尚未落实,属待办事项。
---
# 十二、Git 提交规范
+20 -6
View File
@@ -10,7 +10,7 @@
- **M4** 全文英译中 + 投资事件抽取(LLM DeepSeek v4-flash
- **M5** 向量生成(DashScope text-embedding-v31024 维)
- **M6** 双语向量知识库(Qdrant
- **M7** 定时调度(06/12/18/22 + 每日 AI 摘要日报
- **M7** 定时调度(06/12/18/22 + 每日 AI 摘要日报(M9 起结构化写入 MySQL,不再产出 HTML)
- **M8** MCP 服务(Cherry Studio / Claude Code Agent 深度研究)
## 部署架构
@@ -27,7 +27,7 @@
│ ↓ │
│ M5 向量生成 → M6 Qdrant 入库 │
│ ↓ │
│ 日报生成 → rsync 上传 doorcome.cn
│ 日报生成 → 写入 MySQLnews_report
│ ↓ │
│ M8 MCP 服务(研究 Agent
└──────────────────────────────────────┘
@@ -119,7 +119,7 @@ bash scripts/domestic_crawl_8g.sh investinglive
# M2→M6 管道(需先有 raw 数据)
bash scripts/pipeline.sh
# 仅生成日报
# 仅生成日报(写入 MySQL news_report/news_eventreport_type="intl"
uv run en-news report
```
@@ -145,6 +145,20 @@ ls -lt data/reports/
| `.env` | 密钥 / 服务地址(不入 Git) |
| `prompts/` | LLM Prompt 模板(翻译/日报/搜索 Agent |
### 日报入库(M9
日报内容结构化写入与 [news 项目](https://github.com/) 共用的 MySQL `myquant` 库(表 `news_report` / `news_event``report_type="intl"`,同一天重复生成幂等覆盖)。表结构与数据契约见 news 项目 `docs/db_schema.md`
`.env` 需配置(与 news 项目 `.env``NEWS_DB_PASSWORD` 相同):
```env
NEWS_DB_HOST=192.168.1.10 # pi5 经内网直连 pi 上的 autossh 隧道(0.0.0.0:13306 → doorcome.cn:3306
NEWS_DB_PORT=13306
NEWS_DB_USER=myquant
NEWS_DB_PASSWORD=xxx
NEWS_DB_NAME=myquant
```
---
## 数据存储
@@ -156,7 +170,7 @@ data/
├── dedup/ M3 去重指纹库(SQLite
├── translations/{yyyymmdd}/ M4 翻译+事件抽取结果(JSONL)
├── embeddings/{yyyymmdd}/ M5 向量文件(JSONL
├── reports/ M7 日报(HTML
├── reports/ M7 日报(M9 起不再产出 HTML,改存 MySQL
└── vectorstore/ M6 Qdrant 数据
```
@@ -191,9 +205,9 @@ uv run en-news mcp-server
部分网站有强反爬(Cloudflare/DataDome/付费墙),RSS 只返回摘要。全文可尝试 headful 浏览器回退,但成功率取决于代理线路质量。
### 日报上传到哪里?
### 日报存在哪里?
日报自动上传到 `simon@doorcome.cn:/var/www/html/echart/research/`,通过 HTTP 访问
M9 起日报结构化写入 MySQL `myquant` 库:主表 `news_report``report_type="intl"`+ 明细表 `news_event``section="intl"`),数据契约与 [news 项目](https://github.com/) 一致(见其 `docs/db_schema.md`)。历史 HTML 日报(2026-06-16 ~ 2026-08-03)已由 news 项目解析导入。API / 前端读取同一批表
### 如何添加新源?
+7 -6
View File
@@ -3,8 +3,8 @@
import logging
import sys
from dotenv import load_dotenv
import typer
from dotenv import load_dotenv
# 加载 .env 环境变量(必须在所有模块导入之前)
load_dotenv()
@@ -44,6 +44,7 @@ def crawl(
):
"""M1: 抓取英文财经新闻(部署海外)"""
import os
from crawler.orchestrator import run_crawl_sync
if profile:
@@ -196,15 +197,15 @@ def search(
@app.command()
def report():
"""M7: 日报生成"""
"""M7: 日报生成(结构化入库,不再产出 HTML"""
from scheduler.reporter import generate_report
try:
path = generate_report()
if path:
typer.echo(f"\n✅ 日报已生成: {path}")
report_id = generate_report()
if report_id is not None:
typer.echo(f"\n✅ 日报已入库: report_id={report_id}")
else:
typer.echo("⚠️ 无数据,跳过日报生成")
typer.echo("⚠️ 无数据或入库失败,跳过日报生成")
except Exception as e:
logger.exception("日报生成失败")
typer.echo(f"❌ 日报生成出错: {e}", err=True)
+25 -1
View File
@@ -1,6 +1,30 @@
# continuation.md — English Financial News 项目状态
> 最后更新:2026-07-23
> 最后更新:2026-08-04
---
## 2026-08-04 会话成果
### M9:日报结构化入库(HTML → MySQL)
**背景:** 与 news 项目对齐,日报内容不再产出 HTML / scp 上传 doorcome,改为写入共用 MySQL `myquant` 库(表 `news_report` / `news_event``report_type="intl"`)。历史日报(2026-06-16 ~ 2026-08-03intl 128 份)已由 news 项目解析入库,本次仅做新日报按相同契约入库。
**改动:**
| 文件 | 改动内容 |
|------|---------|
| `report_db/`(新增) | 复用 news 项目实现:`models.py`EventRow/ReportData)、`schema.py`news_report/news_event DDL)、`db.py`load_db_config/connect/init_schema/save_report/fetch_report/exists_reportloguru 换 logging |
| `scheduler/reporter.py` | 新增 `_build_report_data()`intl 结构:事件全部 `section="intl"``file_name=""` 幂等覆盖,stats 含 pipeline/sentiment/importance/event_types/source_dist);`generate_report()` 返回 `int | None`,收集逻辑不变,末尾 `connect()+save_report()`;HTML 渲染/上传函数保留标 deprecated |
| `scheduler/pipeline.py` | `run_step_report` 适配 `report_id` 返回值 |
| `app/cli.py` | `report` 命令打印 `report_id` |
| `pyproject.toml` | `uv add pymysql``report_db/schema.py` 加 E501 per-file-ignore |
| `.env.example` / `README.md` | 新增 `NEWS_DB_*` 配置说明(pi5 经内网直连 `192.168.1.10:13306`password 与 news 项目相同) |
| `tests/test_report_db.py`(新增) | 模型默认值 + `_build_report_data` 组装(板块/rank、字段映射、标题截断、stats 快照、`"?"` 归一化) |
**测试结果:** `uv run pytest` → 174 passed / 2 failed2 个失败为原有 `test_crawler.py` index.jsonl 路径问题,与本次无关);ruff 改动文件干净(仅剩原有 `_SENTENCE_END` N806)。
**部署:** 已 rsync 至 pi5`/home/pi/intlnews`),`.env` 写入 NEWS_DB_*`uv sync` 安装 pymysql,真实生成日报并回读 DB 验证(详见下方)。
---
+2
View File
@@ -25,6 +25,7 @@ dependencies = [
"openai>=2.43.0",
"python-dotenv>=1.2.2",
"markdown>=3.10.2",
"pymysql>=1.2.0",
]
[project.optional-dependencies]
@@ -65,6 +66,7 @@ select = ["E", "F", "I", "N", "W", "UP"]
[tool.ruff.lint.per-file-ignores]
"scheduler/reporter.py" = ["E501"] # 内嵌 HTML 模板 CSS 行较长
"report_db/schema.py" = ["E501"] # DDL SQL 语句较长,与 news 项目 schema 保持一致
[tool.pytest.ini_options]
testpaths = ["tests"]
+20
View File
@@ -0,0 +1,20 @@
"""日报结构化入库。
职责日报内容intl结构化后写入 MySQLnews_report / news_event
news 项目共用同一表结构与数据契约 docs/db_schema.mdreport_db_design.md
本包不提供 API 与前端
"""
from .db import connect, exists_report, fetch_report, init_schema, load_db_config, save_report
from .models import EventRow, ReportData
__all__ = [
"EventRow",
"ReportData",
"connect",
"exists_report",
"fetch_report",
"init_schema",
"load_db_config",
"save_report",
]
+161
View File
@@ -0,0 +1,161 @@
"""日报结构化入库:DB 连接、建表、写入。
news 项目 report_db/db.py 保持一致日志改用 logging遵守本项目规范
"""
from __future__ import annotations
import json
import logging
import os
from dataclasses import dataclass
from typing import Any
import pymysql
from .models import ReportData
from .schema import DDL_STATEMENTS
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class DbConfig:
"""MySQL 连接配置(来自环境变量 NEWS_DB_*)。"""
host: str
port: int
user: str
password: str
name: str
def load_db_config() -> DbConfig:
"""从环境变量读取 NEWS_DB_*,缺失密码时抛异常(禁止默认密码)。"""
host = os.environ.get("NEWS_DB_HOST", "192.168.1.10")
port = int(os.environ.get("NEWS_DB_PORT", "13306"))
user = os.environ.get("NEWS_DB_USER", "myquant")
password = os.environ.get("NEWS_DB_PASSWORD", "")
name = os.environ.get("NEWS_DB_NAME", "myquant")
if not password:
logger.error("NEWS_DB_PASSWORD 未配置,请在 .env 中设置")
raise ValueError("NEWS_DB_PASSWORD 未配置")
return DbConfig(host=host, port=port, user=user, password=password, name=name)
def connect(cfg: DbConfig | None = None) -> pymysql.Connection:
"""建立短连接(autocommit=False)。失败时记录日志并抛出。"""
cfg = cfg or load_db_config()
try:
conn = pymysql.connect(
host=cfg.host,
port=cfg.port,
user=cfg.user,
password=cfg.password,
database=cfg.name,
charset="utf8mb4",
autocommit=False,
cursorclass=pymysql.cursors.DictCursor,
)
except Exception:
logger.exception("连接 MySQL 失败: host=%s port=%s user=%s", cfg.host, cfg.port, cfg.user)
raise
logger.debug("MySQL 已连接: %s/%s", cfg.host, cfg.name)
return conn
def init_schema(conn: pymysql.Connection) -> None:
"""建表(CREATE TABLE IF NOT EXISTS ×2),幂等。"""
with conn.cursor() as cur:
for ddl in DDL_STATEMENTS:
cur.execute(ddl)
conn.commit()
logger.info("news_report / news_event 建表完成")
def save_report(conn: pymysql.Connection, report: ReportData) -> int:
"""事务内写入一份日报。
幂等策略
- 主表按 (report_date, report_type, file_name) 唯一键 upsert
- 事件表 DELETE report 旧行后全量 INSERT整份覆盖一致
返回 report_id
"""
with conn.cursor() as cur:
stats_json = json.dumps(report.stats, ensure_ascii=False) if report.stats else None
cur.execute(
"""
INSERT INTO news_report
(report_date, report_type, file_name, generated_at, ai_summary, stats)
VALUES (%s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
generated_at = VALUES(generated_at),
ai_summary = VALUES(ai_summary),
stats = VALUES(stats)
""",
(
report.report_date,
report.report_type,
report.file_name,
report.generated_at,
report.ai_summary,
stats_json,
),
)
cur.execute(
"SELECT id FROM news_report WHERE report_date=%s AND report_type=%s AND file_name=%s",
(report.report_date, report.report_type, report.file_name),
)
row = cur.fetchone()
if row is None: # pragma: no cover - 理论不可达
raise RuntimeError("写入 news_report 后查询不到 report_id")
report_id: int = row["id"]
cur.execute("DELETE FROM news_event WHERE report_id=%s", (report_id,))
for ev in report.events:
cur.execute(
"""
INSERT INTO news_event
(report_id, section, rank, importance, event_type, title,
summary, sentiment, source, url)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(
report_id,
ev.section,
ev.rank,
ev.importance,
ev.event_type,
ev.title,
ev.summary,
ev.sentiment,
ev.source,
ev.url,
),
)
conn.commit()
logger.info("日报已入库: report_id=%s date=%s type=%s events=%s",
report_id, report.report_date, report.report_type, len(report.events))
return report_id
def fetch_report(conn: pymysql.Connection, report_id: int) -> dict[str, Any] | None:
"""读侧辅助(联调/测试用),返回主表行。"""
with conn.cursor() as cur:
cur.execute("SELECT * FROM news_report WHERE id=%s", (report_id,))
return cur.fetchone()
def exists_report(
conn: pymysql.Connection,
report_date: Any,
report_type: str,
file_name: str,
) -> bool:
"""判断主表是否已存在该唯一键记录(历史导入幂等用)。"""
with conn.cursor() as cur:
cur.execute(
"SELECT 1 FROM news_report WHERE report_date=%s AND report_type=%s AND file_name=%s",
(report_date, report_type, file_name),
)
return cur.fetchone() is not None
+37
View File
@@ -0,0 +1,37 @@
"""日报结构化入库:数据模型。
news 项目 report_db/models.py 保持一致intl 板块复用同一表结构
"""
from __future__ import annotations
from datetime import date, datetime
from typing import Any
from pydantic import BaseModel, Field
class EventRow(BaseModel):
"""一条事件记录,对应 news_event 一行。"""
section: str # xwlb | news | cninfo | intl
rank: int # 板块内序号(从 1 开始)
importance: int | None = None
event_type: str | None = None
title: str
summary: str | None = None
sentiment: str | None = None # positive | negative | neutral | ''
source: str | None = None
url: str | None = None
class ReportData(BaseModel):
"""一份完整日报:news_report 一行 + news_event 多行。"""
report_date: date
report_type: str # finance | intl
file_name: str = "" # 源文件名(新生成日报可为空)
generated_at: datetime
ai_summary: str | None = None
stats: dict[str, Any] = Field(default_factory=dict)
events: list[EventRow] = Field(default_factory=list)
+44
View File
@@ -0,0 +1,44 @@
"""MySQL DDL:日报结构化入库(表前缀 news_,目标 MariaDB 10.11)。
news 项目 docs/db_schema.md 保持一致news_report / news_event
历史日报已由 news 项目 report-import 导入本包只负责新日报写入与幂等建表
"""
from __future__ import annotations
DDL_STATEMENTS: list[str] = [
# 日报主表
"""
CREATE TABLE IF NOT EXISTS news_report (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
report_date DATE NOT NULL COMMENT '日报日期',
report_type VARCHAR(16) NOT NULL COMMENT 'finance=A股日报 / intl=国际财经日报',
file_name VARCHAR(160) NOT NULL DEFAULT '' COMMENT '源文件名(历史解析);新生成可为空',
generated_at DATETIME NOT NULL COMMENT '生成时间',
ai_summary TEXT NULL COMMENT 'AI 摘要全文',
stats JSON NULL COMMENT '数据总览统计快照(管道/情绪/重要度/事件类型/来源分布)',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uk_report_file (report_date, report_type, file_name),
KEY idx_report_date (report_date)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='每日日报主表'
""",
# 日报事件明细
"""
CREATE TABLE IF NOT EXISTS news_event (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
report_id BIGINT UNSIGNED NOT NULL COMMENT 'FK → news_report.id',
section VARCHAR(16) NOT NULL COMMENT '板块: xwlb=新闻联播 / news=财经新闻 / cninfo=公告调研 / intl=国际重要事件',
rank INT NOT NULL DEFAULT 0 COMMENT '板块内序号',
importance INT NULL COMMENT '重要度 1-5',
event_type VARCHAR(64) NULL COMMENT '事件类型',
title VARCHAR(512) NOT NULL COMMENT '标题',
summary TEXT NULL COMMENT '摘要/正文',
sentiment VARCHAR(8) NULL COMMENT 'positive/negative/neutral',
source VARCHAR(64) NULL COMMENT '来源',
url VARCHAR(512) NULL COMMENT '原文链接',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
KEY idx_report_section (report_id, section),
KEY idx_title (title(255))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='日报事件明细'
""",
]
+5 -4
View File
@@ -150,16 +150,17 @@ def run_step_index(date_str: str) -> StepResult:
def run_step_report(date_str: str) -> StepResult:
"""日报生成。"""
"""日报生成(M9:结构化入库,不再产出 HTML)"""
started = datetime.now()
try:
from scheduler.reporter import generate_report
path = generate_report()
report_id = generate_report()
elapsed = (datetime.now() - started).total_seconds()
ok = path is not None
ok = report_id is not None
return StepResult(
name="report", success=ok, elapsed_sec=elapsed,
message=str(path) if path else "无数据", started_at=started,
message=f"report_id={report_id}" if report_id is not None else "无数据",
started_at=started,
)
except Exception as e:
elapsed = (datetime.now() - started).total_seconds()
+113 -28
View File
@@ -9,12 +9,15 @@
import json
import logging
from collections import Counter
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
import markdown
import yaml
from report_db.models import EventRow, ReportData # noqa: F401 - 供 _build_report_data 注解使用
logger = logging.getLogger(__name__)
_MAX_HIGH_EVENTS = 30
@@ -130,8 +133,8 @@ def _try_parse_time(time_str: str) -> datetime | None:
# - 有时区 → astimezone 转为 UTC
# - 无时区 → 假设为 UTC(多数财经新闻 API 使用 UTC)
if dt.tzinfo is None:
return dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
return dt.replace(tzinfo=UTC)
return dt.astimezone(UTC)
except ValueError:
continue
return None
@@ -168,7 +171,7 @@ def _load_events_window(now: datetime) -> list[dict]:
文章 dict 列表 title/title_zh/url/source_id/events
"""
# cutoff 使用 UTC-aware,与 _try_parse_time 返回的 UTC datetime 对齐
cutoff = now.astimezone(timezone.utc) - timedelta(hours=_REPORT_WINDOW_HOURS)
cutoff = now.astimezone(UTC) - timedelta(hours=_REPORT_WINDOW_HOURS)
articles: list[dict] = []
skipped_empty_pt = 0
for day_str in _dates_in_window(now):
@@ -551,17 +554,98 @@ def _generate_ai_summary(articles: list[dict], day_str: str) -> str:
return "⚠️ AI 摘要合并失败,以下为各批次原始摘要:\n\n" + "\n\n".join(partial_summaries)
def generate_report() -> Path | None:
"""生成 HTML 日报(覆盖过去 25 小时数据)。
def _build_report_data(
now: datetime,
stats: dict,
high_events: list[dict],
sentiments: Counter,
importances: Counter,
event_types: Counter,
sources: Counter,
ai_summary: str,
) -> ReportData:
"""组装结构化日报数据(M9:写入 MySQL 的前置步骤)。
命名规则: intl_news_daily_{YYYYMMDD_HHMMSS}.html 支持一天多份日报
news 项目 report_db_design.md 数据契约一致
- 事件全部归入 `intl` 板块intl 日报无 xwlb/news/cninfo 板块
- report_type="intl"file_name="" 唯一键退化为 (report_date, intl, "")
同一天重复生成 = UPDATE 覆盖幂等
- 数据总览统计以 JSON 快照存入 stats前端自行解析
"""
events: list[EventRow] = []
for i, ev in enumerate(high_events, 1):
article = ev.get("article", {})
title = (article.get("title_zh") or article.get("title") or "").strip()[:512]
url = article.get("url") or None
src = _url_source_label(url, article.get("source_id", ""))
# 归一化:"" / "?" 不入库,留 NoneDB 仅存 positive/negative/neutral
sentiment = ev.get("sentiment") or None
if sentiment in ("", "?"):
sentiment = None
event_type = ev.get("event_type") or None
if event_type in ("", "?"):
event_type = None
events.append(
EventRow(
section="intl",
rank=i,
importance=ev.get("importance") or None,
event_type=event_type,
title=title or "(无标题)",
summary=ev.get("summary_zh") or None,
sentiment=sentiment,
source=src if src not in (None, "", "?") else None,
url=url,
)
)
stats_snapshot: dict[str, Any] = {
"pipeline": {
"raw_total": stats.get("raw_total", 0),
"processed": stats.get("proc", 0),
"deduped": stats.get("deduped", 0),
"embedded": stats.get("emb_count", 0),
"qdrant": stats.get("qdrant_count", 0),
},
"sentiment": {k: v for k, v in sentiments.items() if k not in ("", "?")},
"importance": [
{"importance": k, "count": v} for k, v in sorted(importances.items())
],
"event_types": [
{"event_type": k, "count": v}
for k, v in event_types.most_common(10)
if k not in ("", "?")
],
"source_dist": [
{"source": _source_name(k), "count": v}
for k, v in sources.most_common(15)
if k not in ("", "?")
],
}
return ReportData(
report_date=now.date(),
report_type="intl",
file_name="", # 新生成日报唯一键退化为 (report_date, intl, "")
generated_at=now,
ai_summary=ai_summary or None,
stats=stats_snapshot,
events=events,
)
def generate_report() -> int | None:
"""生成日报并结构化入库(覆盖过去 25 小时数据)。
M9 起完全切换不再产出 HTML日报内容写入 MySQL
news_report / news_eventreport_type="intl"file_name=""
同一天重复生成 幂等覆盖不产生多行
Returns:
HTML 文件路径无数据时返回 None
report_id成功 None无数据/失败
"""
now = datetime.now()
ts = now.strftime("%Y%m%d_%H%M%S")
date_str = now.strftime("%Y%m%d") # 用于上传目录
logger.info("生成日报: %s(窗口: 过去 %d 小时)", ts, _REPORT_WINDOW_HOURS)
# 加载过去 25 小时数据
@@ -595,13 +679,10 @@ def generate_report() -> Path | None:
)
high = _get_high(all_events, 4)
hi_threshold = 4
if len(high) < 3:
high = _get_high(all_events, 3)
hi_threshold = 3
if len(high) < 3:
high = sorted(all_events, key=lambda e: -e.get("importance", 0))
hi_threshold = 0
# 事件级去重:同一 URL + 同一标题 → 合并
high = _dedup_events(high)
high = high[:_MAX_HIGH_EVENTS]
@@ -609,20 +690,24 @@ def generate_report() -> Path | None:
# AI 摘要
ai_summary = _generate_ai_summary(articles, ts)
# 渲染 HTML
html = _render_html(ts, stats, articles, high, hi_threshold,
sentiments, importances, event_types, sources, ai_summary)
# 结构化入库(替代原 HTML 渲染 + 上传)
report = _build_report_data(
now, stats, high, sentiments, importances, event_types, sources, ai_summary
)
try:
from report_db import connect, save_report
# 本地保存(文件名含时间戳)
_REPORT_DIR.mkdir(parents=True, exist_ok=True)
html_path = _REPORT_DIR / f"intl_news_daily_{ts}.html"
html_path.write_text(html, encoding="utf-8")
logger.info("日报已保存: %s (%d KB)", html_path, len(html) // 1024)
conn = connect()
try:
report_id = save_report(conn, report)
finally:
conn.close()
except Exception as e:
logger.exception("日报入库失败: %s", e)
return None
# 自动上传到日期子目录
_upload_report(html_path, date_str)
return html_path
logger.info("日报已入库: report_id=%s", report_id)
return report_id
# --------------------------------------------------------------------------- #
@@ -644,7 +729,7 @@ def _load_report_config() -> dict:
def _upload_report(html_path: Path, day_str: str) -> bool:
"""上传日报到 Web 服务器(配置来自 system.yaml)。"""
"""上传日报到 Web 服务器(M9 起弃用:日报已改为写库,保留以便回退)。"""
import subprocess
config = _load_report_config()
@@ -678,7 +763,7 @@ def _upload_report(html_path: Path, day_str: str) -> bool:
# --------------------------------------------------------------------------- #
# HTML 渲染
# HTML 渲染(M9 起弃用:日报已改为写库,以下渲染函数保留以便回退)
# --------------------------------------------------------------------------- #
_HTML_TEMPLATE = """<!DOCTYPE html>
@@ -798,7 +883,7 @@ def _render_html(
sources: Counter,
ai_summary: str,
) -> str:
"""组装完整 HTML。"""
"""组装完整 HTMLM9 起弃用,保留以便回退)"""
# AI 摘要 Markdown → HTML
summary_html = _md_to_html(ai_summary) if ai_summary.strip() else "<p>暂无 AI 摘要</p>"
@@ -915,7 +1000,7 @@ def _dedup_events(events: list[dict]) -> list[dict]:
def _render_event_table(events: list[dict]) -> str:
"""渲染事件表格。"""
"""渲染事件表格M9 起弃用,保留以便回退)"""
if not events:
return "<p>暂无符合条件的数据</p>"
+3 -3
View File
@@ -58,11 +58,11 @@ print(f'M6: {stats[\"ingested\"]}/{stats[\"total\"]} 条, {stats[\"elapsed_sec\"
"
# ── 日报 ──
echo "[$(date)] [日报] 生成日报..."
echo "[$(date)] [日报] 生成日报(结构化入库)..."
.venv/bin/python3 -c "
from scheduler.reporter import generate_report
path = generate_report()
print(f'日报: {path}')
report_id = generate_report()
print(f'日报: report_id={report_id}' if report_id is not None else '日报: 无数据/失败')
"
echo "[$(date)] ═══ 全链路管道完成 ✅ ═══"
+179
View File
@@ -0,0 +1,179 @@
"""日报结构化入库:模型 + _build_report_data 组装(纯逻辑,不连 DB)。"""
from __future__ import annotations
import json
from collections import Counter
from datetime import date, datetime
import pytest
from pydantic import ValidationError
from report_db.models import EventRow, ReportData
from scheduler.reporter import _build_report_data
# --------------------------------------------------------------------------- #
# 模型默认值
# --------------------------------------------------------------------------- #
class TestEventRow:
def test_minimal(self) -> None:
ev = EventRow(section="intl", rank=1, title="标题")
assert ev.importance is None
assert ev.sentiment is None
def test_full(self) -> None:
ev = EventRow(
section="intl", rank=2, importance=4, event_type="地缘政治",
title="t", summary="s", sentiment="negative", source="InvestingLive",
url="https://x.com/1",
)
assert ev.sentiment == "negative"
def test_missing_title_raises(self) -> None:
with pytest.raises(ValidationError):
EventRow(section="intl", rank=1) # type: ignore[call-arg]
class TestReportData:
def test_defaults(self) -> None:
r = ReportData(
report_date=date(2026, 8, 4),
report_type="intl",
generated_at=datetime(2026, 8, 4, 8, 0),
)
assert r.file_name == ""
assert r.stats == {}
assert r.events == []
def test_with_events(self) -> None:
r = ReportData(
report_date=date(2026, 8, 4),
report_type="intl",
generated_at=datetime(2026, 8, 4, 8, 0),
ai_summary="摘要",
events=[EventRow(section="intl", rank=1, title="t")],
)
assert len(r.events) == 1
# --------------------------------------------------------------------------- #
# _build_report_dataintl 结构)
# --------------------------------------------------------------------------- #
def _fake_high_event(title_zh: str, importance: int, *,
event_type: str = "其他", sentiment: str = "neutral",
summary_zh: str = "摘要", source_id: str = "investinglive",
url: str = "https://investinglive.com/news/1") -> dict:
"""构造 _dedup_events 之后的事件结构(含 article 属性)。"""
return {
"importance": importance,
"sentiment": sentiment,
"summary_zh": summary_zh,
"event_type": event_type,
"stock_codes": [],
"article": {
"title": f"EN {title_zh}",
"title_zh": title_zh,
"url": url,
"source_id": source_id,
},
}
class TestBuildReportData:
def test_sections_and_ranks(self) -> None:
now = datetime(2026, 8, 4, 8, 0, 0)
high = [
_fake_high_event("新闻A", 5),
_fake_high_event("新闻B", 4),
_fake_high_event("新闻C", 3),
]
r = _build_report_data(
now, {"raw_total": 100, "proc": 90}, high,
Counter(), Counter(), Counter(), Counter(), "AI摘要",
)
assert r.report_date == date(2026, 8, 4)
assert r.report_type == "intl"
assert r.file_name == ""
assert r.ai_summary == "AI摘要"
assert [(e.section, e.rank) for e in r.events] == [
("intl", 1), ("intl", 2), ("intl", 3),
]
def test_field_mapping(self) -> None:
now = datetime(2026, 8, 4, 8, 0, 0)
high = [_fake_high_event("英伟达财报超预期", 5, event_type="公司财报",
sentiment="positive", summary_zh="营收超预期",
source_id="investinglive",
url="https://investinglive.com/news/42")]
r = _build_report_data(now, {}, high, Counter(), Counter(), Counter(),
Counter(), "")
ev = r.events[0]
assert ev.title == "英伟达财报超预期" # title_zh 优先
assert ev.importance == 5
assert ev.event_type == "公司财报"
assert ev.summary == "营收超预期"
assert ev.sentiment == "positive"
assert ev.url == "https://investinglive.com/news/42"
assert ev.source == "InvestingLive" # 域名 → 源展示名
def test_source_fallback_without_url(self) -> None:
now = datetime(2026, 8, 4, 8, 0, 0)
ev = _fake_high_event("无URL事件", 4, url="")
r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(),
Counter(), "")
assert r.events[0].url is None
# url 为空时回退 source_id → sources.yaml 展示名
assert r.events[0].source == "InvestingLive"
def test_stats_snapshot(self) -> None:
now = datetime(2026, 8, 4, 8, 0, 0)
high = [
_fake_high_event("A", 5, event_type="宏观", sentiment="positive"),
_fake_high_event("B", 4, event_type="宏观", sentiment="negative"),
]
r = _build_report_data(
now,
{"raw_total": 120, "proc": 100, "deduped": 90,
"emb_count": 80, "qdrant_count": 75},
high,
Counter({"positive": 1, "negative": 1}),
Counter({5: 1, 4: 1}),
Counter({"宏观": 2}),
Counter({"investinglive": 2}),
"摘要",
)
# stats 可 JSON 序列化(入库时 json.dumps
json.dumps(r.stats, ensure_ascii=False)
assert r.stats["pipeline"]["raw_total"] == 120
assert r.stats["sentiment"] == {"positive": 1, "negative": 1}
assert r.stats["importance"] == [
{"importance": 4, "count": 1}, {"importance": 5, "count": 1},
]
assert r.stats["event_types"] == [{"event_type": "宏观", "count": 2}]
assert r.stats["source_dist"] == [{"source": "InvestingLive", "count": 2}]
def test_title_truncated(self) -> None:
now = datetime(2026, 8, 4, 8, 0, 0)
ev = _fake_high_event("" * 600, 4)
r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(),
Counter(), "")
assert len(r.events[0].title) == 512
def test_empty_events(self) -> None:
now = datetime(2026, 8, 4, 8, 0, 0)
r = _build_report_data(now, {}, [], Counter(), Counter(), Counter(),
Counter(), "")
assert r.events == []
assert r.ai_summary is None
def test_question_mark_sentiment_normalized(self) -> None:
now = datetime(2026, 8, 4, 8, 0, 0)
ev = _fake_high_event("未分类情绪", 4, sentiment="?")
r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(),
Counter(), "")
# "?" 不写入 DB,留 None
assert r.events[0].sentiment is None