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

暂无 AI 摘要

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

暂无符合条件的数据

" diff --git a/scripts/pipeline.sh b/scripts/pipeline.sh index 9cf3fa3..ffbd2a0 100755 --- a/scripts/pipeline.sh +++ b/scripts/pipeline.sh @@ -58,11 +58,11 @@ print(f'M6: {stats[\"ingested\"]}/{stats[\"total\"]} 条, {stats[\"elapsed_sec\" " # ── 日报 ── -echo "[$(date)] [日报] 生成日报..." +echo "[$(date)] [日报] 生成日报(结构化入库)..." .venv/bin/python3 -c " from scheduler.reporter import generate_report -path = generate_report() -print(f'日报: {path}') +report_id = generate_report() +print(f'日报: report_id={report_id}' if report_id is not None else '日报: 无数据/失败') " echo "[$(date)] ═══ 全链路管道完成 ✅ ═══" diff --git a/tests/test_report_db.py b/tests/test_report_db.py new file mode 100644 index 0000000..4bd29f6 --- /dev/null +++ b/tests/test_report_db.py @@ -0,0 +1,179 @@ +"""日报结构化入库:模型 + _build_report_data 组装(纯逻辑,不连 DB)。""" + +from __future__ import annotations + +import json +from collections import Counter +from datetime import date, datetime + +import pytest +from pydantic import ValidationError + +from report_db.models import EventRow, ReportData +from scheduler.reporter import _build_report_data + +# --------------------------------------------------------------------------- # +# 模型默认值 +# --------------------------------------------------------------------------- # + + +class TestEventRow: + def test_minimal(self) -> None: + ev = EventRow(section="intl", rank=1, title="标题") + assert ev.importance is None + assert ev.sentiment is None + + def test_full(self) -> None: + ev = EventRow( + section="intl", rank=2, importance=4, event_type="地缘政治", + title="t", summary="s", sentiment="negative", source="InvestingLive", + url="https://x.com/1", + ) + assert ev.sentiment == "negative" + + def test_missing_title_raises(self) -> None: + with pytest.raises(ValidationError): + EventRow(section="intl", rank=1) # type: ignore[call-arg] + + +class TestReportData: + def test_defaults(self) -> None: + r = ReportData( + report_date=date(2026, 8, 4), + report_type="intl", + generated_at=datetime(2026, 8, 4, 8, 0), + ) + assert r.file_name == "" + assert r.stats == {} + assert r.events == [] + + def test_with_events(self) -> None: + r = ReportData( + report_date=date(2026, 8, 4), + report_type="intl", + generated_at=datetime(2026, 8, 4, 8, 0), + ai_summary="摘要", + events=[EventRow(section="intl", rank=1, title="t")], + ) + assert len(r.events) == 1 + + +# --------------------------------------------------------------------------- # +# _build_report_data(intl 结构) +# --------------------------------------------------------------------------- # + + +def _fake_high_event(title_zh: str, importance: int, *, + event_type: str = "其他", sentiment: str = "neutral", + summary_zh: str = "摘要", source_id: str = "investinglive", + url: str = "https://investinglive.com/news/1") -> dict: + """构造 _dedup_events 之后的事件结构(含 article 属性)。""" + return { + "importance": importance, + "sentiment": sentiment, + "summary_zh": summary_zh, + "event_type": event_type, + "stock_codes": [], + "article": { + "title": f"EN {title_zh}", + "title_zh": title_zh, + "url": url, + "source_id": source_id, + }, + } + + +class TestBuildReportData: + def test_sections_and_ranks(self) -> None: + now = datetime(2026, 8, 4, 8, 0, 0) + high = [ + _fake_high_event("新闻A", 5), + _fake_high_event("新闻B", 4), + _fake_high_event("新闻C", 3), + ] + r = _build_report_data( + now, {"raw_total": 100, "proc": 90}, high, + Counter(), Counter(), Counter(), Counter(), "AI摘要", + ) + assert r.report_date == date(2026, 8, 4) + assert r.report_type == "intl" + assert r.file_name == "" + assert r.ai_summary == "AI摘要" + assert [(e.section, e.rank) for e in r.events] == [ + ("intl", 1), ("intl", 2), ("intl", 3), + ] + + def test_field_mapping(self) -> None: + now = datetime(2026, 8, 4, 8, 0, 0) + high = [_fake_high_event("英伟达财报超预期", 5, event_type="公司财报", + sentiment="positive", summary_zh="营收超预期", + source_id="investinglive", + url="https://investinglive.com/news/42")] + r = _build_report_data(now, {}, high, Counter(), Counter(), Counter(), + Counter(), "") + ev = r.events[0] + assert ev.title == "英伟达财报超预期" # title_zh 优先 + assert ev.importance == 5 + assert ev.event_type == "公司财报" + assert ev.summary == "营收超预期" + assert ev.sentiment == "positive" + assert ev.url == "https://investinglive.com/news/42" + assert ev.source == "InvestingLive" # 域名 → 源展示名 + + def test_source_fallback_without_url(self) -> None: + now = datetime(2026, 8, 4, 8, 0, 0) + ev = _fake_high_event("无URL事件", 4, url="") + r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(), + Counter(), "") + assert r.events[0].url is None + # url 为空时回退 source_id → sources.yaml 展示名 + assert r.events[0].source == "InvestingLive" + + def test_stats_snapshot(self) -> None: + now = datetime(2026, 8, 4, 8, 0, 0) + high = [ + _fake_high_event("A", 5, event_type="宏观", sentiment="positive"), + _fake_high_event("B", 4, event_type="宏观", sentiment="negative"), + ] + r = _build_report_data( + now, + {"raw_total": 120, "proc": 100, "deduped": 90, + "emb_count": 80, "qdrant_count": 75}, + high, + Counter({"positive": 1, "negative": 1}), + Counter({5: 1, 4: 1}), + Counter({"宏观": 2}), + Counter({"investinglive": 2}), + "摘要", + ) + # stats 可 JSON 序列化(入库时 json.dumps) + json.dumps(r.stats, ensure_ascii=False) + assert r.stats["pipeline"]["raw_total"] == 120 + assert r.stats["sentiment"] == {"positive": 1, "negative": 1} + assert r.stats["importance"] == [ + {"importance": 4, "count": 1}, {"importance": 5, "count": 1}, + ] + assert r.stats["event_types"] == [{"event_type": "宏观", "count": 2}] + assert r.stats["source_dist"] == [{"source": "InvestingLive", "count": 2}] + + def test_title_truncated(self) -> None: + now = datetime(2026, 8, 4, 8, 0, 0) + ev = _fake_high_event("长" * 600, 4) + r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(), + Counter(), "") + assert len(r.events[0].title) == 512 + + def test_empty_events(self) -> None: + now = datetime(2026, 8, 4, 8, 0, 0) + r = _build_report_data(now, {}, [], Counter(), Counter(), Counter(), + Counter(), "") + assert r.events == [] + assert r.ai_summary is None + + def test_question_mark_sentiment_normalized(self) -> None: + now = datetime(2026, 8, 4, 8, 0, 0) + ev = _fake_high_event("未分类情绪", 4, sentiment="?") + r = _build_report_data(now, {}, [ev], Counter(), Counter(), Counter(), + Counter(), "") + # "?" 不写入 DB,留 None + assert r.events[0].sentiment is None