Compare commits
2
Commits
182f788857
...
8bbf5104a7
| Author | SHA1 | Date | |
|---|---|---|---|
 simon
|
8bbf5104a7 | ||
|
simon
|
794ce6c853 |
@@ -217,7 +217,8 @@ uv run python -m scripts.run_dedup --reset
|
||||
**多源记录**:同一内容被多个新闻源发布时,去重后只保留一条唯一新闻,但会记录全部来源
|
||||
(主源居首)。指纹库 `source_ids` 列为跨日累积的权威记录;`uniques/{url_hash}.json` 的
|
||||
`sources` 字段与 `data/deduped/{YYYYMMDD}/sources.json`(本次去重涉及内容组的源汇总,
|
||||
含跨日命中)为当日产物,展示/下游可按需读取多源列表。
|
||||
含跨日命中)为当日产物;M4 事件、日报与 DB 入库(`news_event.sources` 列)均透传多源,
|
||||
展示/下游可按需读取多源列表。
|
||||
|
||||
日志:`logs/dedup.log`
|
||||
|
||||
|
||||
+23
-25
@@ -1,6 +1,6 @@
|
||||
# continuation.md
|
||||
|
||||
> `checkpoint` @ 2026-08-12 10:30
|
||||
> `checkpoint` @ 2026-08-12 13:00
|
||||
|
||||
---
|
||||
|
||||
@@ -13,41 +13,39 @@
|
||||
| 日报 | **M10 完成并已部署 pi5: 结构化入库 MySQL；日报按当天日期生成（新闻 30h 回溯 / xwlb 取前一日 / 公告调研近 15 日）** |
|
||||
| DB 连接 | pi 上 systemd 服务 `a-share-db-tunnel` 常驻（0.0.0.0:13306 → doorcome.cn:3306）；**pi5 直连 192.168.1.10:13306** |
|
||||
| 调度器 | APScheduler，systemd `a-share-research.service`（pi5）；每天 07:00 首次任务生成日报（12/18/22 点不生成） |
|
||||
| LLM | 场景化配置 `configs/llm_models.yaml`（4 场景: event_extraction/daily_report/stock_report/embedding）；YAML 优先、`.env` 兜底；模型必须显式配置，无内置兜底 |
|
||||
| 去重 | 多源记录：指纹库 `source_ids` 列 + uniques JSON `sources` 字段 + `data/deduped/{day}/sources.json` |
|
||||
| 增量/断点 | M2/M4/M5 产物存在即跳过（`--force` 全量）；`pipeline --once --resume` 断点续跑（状态 `data/pipeline/state.json`） |
|
||||
| LLM | 场景化配置 `configs/llm_models.yaml`（4 场景）；YAML 优先、`.env` 兜底 |
|
||||
| 去重 | 多源记录：指纹库 `source_ids` 列 + uniques `sources` 字段 + `sources.json`；**M4 事件 / 日报 / DB `news_event.sources` 全链路透传** |
|
||||
| 增量/断点 | M2/M4/M5 产物存在即跳过（`--force` 全量）；`pipeline --once --resume` 断点续跑 |
|
||||
| 服务器 | `pi@192.168.1.160`（生产）/ `pi@192.168.1.10`（DB 隧道宿主） |
|
||||
| 抓取方式 | js_render=false → httpx 直连；js_render=true → Playwright |
|
||||
|
||||
---
|
||||
|
||||
## 本次完成 (2026-08-12) — 增量处理与 pipeline 断点续跑
|
||||
## 本次完成 (2026-08-12 13:00) — 日报事件多新闻源入库 + 阶段/AI 显性输出 + 摘要崩溃修复
|
||||
|
||||
**目标：** ① 全链路中断后可从断点恢复；② 各子任务排除已处理文件，避免全量重跑与重复 API 计费。
|
||||
**1. 多新闻源贯通至 DB（方案 A：新增 sources 列）：**
|
||||
- `llm/models.py`：`ExtractedEvent` 增加 `sources`（validator 保主源居首、去重；旧产物兜底 `[source_id]`）
|
||||
- `llm/extractor.py`：`extract_event(_async)` 增加 `sources` 参数透传
|
||||
- `scripts/run_event_extraction.py`：读取 deduped uniques 的 `sources` 字段传入抽取（`_load_sources`）
|
||||
- `scheduler/reporter.py`：`_load_events_from_dir` 读取 `sources`（兜底 `[source_id]`）→ `EventRow.sources`
|
||||
- `report_db`：`news_event` 建表加 `sources TEXT` 列；`init_schema` 幂等迁移（ALTER，Duplicate column 捕获）；`save_report` INSERT 写 JSON
|
||||
- 生产 pi5：`init_schema` 已执行，`news_event.sources` 列就绪（text, NULL）；旧数据为 NULL，下次全链路 M4 后新事件带多源
|
||||
|
||||
**1. 各步骤增量处理（产物存在即跳过，`--force` 全量）：**
|
||||
- M2 `run_extractor.py`：输出目录已有 `{url_hash}.json` 即跳过提取，仅回补 index 行；`--force` 重建；成功率统计含跳过项（修复全跳过时误报 rc=1）
|
||||
- M4 `run_event_extraction.py`：`data/events/{day}/{url_hash}.json` 已存在即跳过（**不重复调用 LLM API**）；`--force` 全量；failed.jsonl 只保留本次失败、index 累积追加
|
||||
- M5 `run_embedding.py`：`data/embeddings/{day}/{url_hash}.json` 已存在即跳过（**不重复调用 embed API**）；`--force` 全量
|
||||
- M1（seen_urls 增量）/ M3（指纹库判重）/ M6（upsert 幂等）为既有能力，README 汇总成表
|
||||
**2. pipeline 阶段与 AI 显性输出（scheduler/pipeline.py）：**
|
||||
- 每步前打印 `阶段 i/n: 中文名 [步骤] 日期` 分隔标题
|
||||
- llm/report/embedding 步骤打印 `🤖 AI 大模型: provider=..., model=...`（按场景配置解析，缺 key 也可展示）
|
||||
|
||||
**2. pipeline 断点续跑（scheduler/pipeline.py + run_scheduler.py）：**
|
||||
- 新增 `data/pipeline/state.json`（按日期隔离，记录每步骤 ok/failed + 退出码 + 耗时），原子写
|
||||
- `run_pipeline(resume=True)` 跳过连续成功前缀，从首个失败/未执行步骤继续执行到结尾
|
||||
- `pipeline --once --resume`（默认全量不变；`--resume` 与 `--steps` 互斥报错）；定时守护模式不受影响
|
||||
**3. 修复日报 AI 摘要崩溃（'str' object has no attribute 'model'）：**
|
||||
- `_llm_summarize` 内 3 处 `_llm_call(client, model, ...)` 改为传 `config`（上轮改签名遗漏）
|
||||
- 新增回归测试（单块/多分块全链路）
|
||||
|
||||
**验证（Mac 本地，20260616 数据 100 篇）：**
|
||||
- pytest **225 passed**（新增 tests/test_incremental.py 10 个：M2/M4/M5 跳过、状态记录、resume 续跑、resume 全完成 noop、--resume+--steps 互斥）；crawler 3 个基线失败仍与本次无关
|
||||
- ruff 零新增（9 个基线错误不变）
|
||||
- 端到端：M2 增量重跑 140 条跳过 126 条，2.5s 完成、rc=0（修复前误报失败）；state.json 正确记录 extractor ok
|
||||
|
||||
**效果评估（100 篇规模中断重跑场景）：**
|
||||
- M4 减少约 100 次 LLM API 调用、M5 减少约 100 次 embedding API 调用 → 中断恢复不再重复计费，耗时从分钟级降至秒级
|
||||
- M2 重跑从全量 GNE 提取（分钟级）降至约 2.5s
|
||||
- 断点恢复操作：中断后直接重跑同一条 `pipeline --once --resume` 命令即可
|
||||
**验证（pi5 生产）：**
|
||||
- 本地 pytest **231 passed**（新增 ExtractedEvent.sources×2、EventRow.sources×2、_load_events_from_dir×1）；ruff 基线 9 零新增
|
||||
- pi5：`init_schema` 迁移成功（news_event 新增 sources 列）；日报生成入库正常（report_id=225, events=60）
|
||||
- 部署前已修复并在生产实测 AI 摘要恢复
|
||||
|
||||
**待办：**
|
||||
- 同步 pi5（代码 + 文档），重启 `a-share-research`；首次同步后 pi5 的 `data/pipeline/state.json` 不存在 → resume 按全量处理，行为安全
|
||||
- 下次全链路运行后抽查 `news_event.sources` 多源值（当前生产 events 为旧产物，sources=NULL）
|
||||
- git 提交（本次改动尚未提交）
|
||||
|
||||
---
|
||||
|
||||
@@ -188,10 +188,12 @@ def extract_event(
|
||||
*,
|
||||
template: PromptTemplate | None = None,
|
||||
max_attempts: int | None = None,
|
||||
sources: list[str] | None = None,
|
||||
) -> ExtractedEvent:
|
||||
"""同步抽取单篇文章的事件(带重试)。
|
||||
|
||||
max_attempts 为 None 时使用 config.max_attempts(来自 YAML/环境变量配置)。
|
||||
sources 为该新闻全部来源(来自去重层多源记录);None 时兜底 [article.source_id]。
|
||||
"""
|
||||
tpl = template or PromptTemplate()
|
||||
prompt = tpl.render(article)
|
||||
@@ -208,6 +210,7 @@ def extract_event(
|
||||
url_hash=article.url_hash,
|
||||
title=article.title,
|
||||
publish_time=article.publish_time,
|
||||
sources=sources or [article.source_id],
|
||||
event=event,
|
||||
provider=config.provider,
|
||||
model=config.model,
|
||||
@@ -246,11 +249,13 @@ async def extract_event_async(
|
||||
*,
|
||||
template: PromptTemplate | None = None,
|
||||
max_attempts: int | None = None,
|
||||
sources: list[str] | None = None,
|
||||
semaphore: asyncio.Semaphore | None = None,
|
||||
) -> ExtractedEvent:
|
||||
"""异步抽取(批处理用),与同步版逻辑等价。
|
||||
|
||||
max_attempts 为 None 时使用 config.max_attempts。
|
||||
sources 为该新闻全部来源;None 时兜底 [article.source_id]。
|
||||
"""
|
||||
tpl = template or PromptTemplate()
|
||||
prompt = tpl.render(article)
|
||||
@@ -268,6 +273,7 @@ async def extract_event_async(
|
||||
url_hash=article.url_hash,
|
||||
title=article.title,
|
||||
publish_time=article.publish_time,
|
||||
sources=sources or [article.source_id],
|
||||
event=event,
|
||||
provider=config.provider,
|
||||
model=config.model,
|
||||
|
||||
+20
-1
@@ -130,7 +130,12 @@ class EventExtraction(BaseModel):
|
||||
|
||||
|
||||
class ExtractedEvent(BaseModel):
|
||||
"""落盘格式:文章元数据 + LLM 抽取结果 + 调用元信息。"""
|
||||
"""落盘格式:文章元数据 + LLM 抽取结果 + 调用元信息。
|
||||
|
||||
sources: 该唯一新闻的全部来源(主源 source_id 居首)。
|
||||
来自去重层多源记录(M3 sources.json / uniques JSON 的 sources 字段),
|
||||
旧产物无此字段时兜底为 [source_id]。
|
||||
"""
|
||||
|
||||
# ---- 来源标识 ----
|
||||
source_id: str
|
||||
@@ -138,6 +143,10 @@ class ExtractedEvent(BaseModel):
|
||||
url_hash: str
|
||||
title: str
|
||||
publish_time: datetime | None = None
|
||||
sources: list[str] = Field(
|
||||
default_factory=list,
|
||||
description="全部来源(主源居首,去重保序);旧产物无字段时兜底为 [source_id]",
|
||||
)
|
||||
|
||||
# ---- 抽取结果 ----
|
||||
event: EventExtraction
|
||||
@@ -150,6 +159,16 @@ class ExtractedEvent(BaseModel):
|
||||
prompt_tokens: int | None = None
|
||||
completion_tokens: int | None = None
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _ensure_sources(self) -> Self:
|
||||
"""保证 sources 非空、去重且以主源 source_id 开头。"""
|
||||
seen: list[str] = []
|
||||
for s in [self.source_id, *self.sources]:
|
||||
if s and s not in seen:
|
||||
seen.append(s)
|
||||
self.sources = seen
|
||||
return self
|
||||
|
||||
def short_summary(self) -> str:
|
||||
ev = self.event
|
||||
codes = ",".join(ev.stock_codes) or "-"
|
||||
|
||||
+14
-4
@@ -11,7 +11,7 @@ import pymysql
|
||||
from loguru import logger
|
||||
|
||||
from .models import ReportData
|
||||
from .schema import DDL_STATEMENTS
|
||||
from .schema import _ALTER_ADD_SOURCES, DDL_STATEMENTS
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -60,10 +60,19 @@ def connect(cfg: DbConfig | None = None) -> pymysql.Connection:
|
||||
|
||||
|
||||
def init_schema(conn: pymysql.Connection) -> None:
|
||||
"""建表（CREATE TABLE IF NOT EXISTS ×2），幂等。"""
|
||||
"""建表（CREATE TABLE IF NOT EXISTS ×2）+ 旧表 sources 列迁移，幂等。"""
|
||||
with conn.cursor() as cur:
|
||||
for ddl in DDL_STATEMENTS:
|
||||
cur.execute(ddl)
|
||||
# 旧库兼容:news_event 已存在但缺 sources 列时补列
|
||||
try:
|
||||
cur.execute(_ALTER_ADD_SOURCES)
|
||||
logger.info("news_event 迁移:新增 sources 多源列")
|
||||
except pymysql.err.OperationalError as e:
|
||||
if e.args and "Duplicate column" in str(e.args[0]):
|
||||
logger.debug("news_event.sources 列已存在,跳过迁移")
|
||||
else:
|
||||
raise
|
||||
conn.commit()
|
||||
logger.info("news_report / news_event 建表完成")
|
||||
|
||||
@@ -112,8 +121,8 @@ def save_report(conn: pymysql.Connection, report: ReportData) -> int:
|
||||
"""
|
||||
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)
|
||||
summary, sentiment, source, sources, url)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
""",
|
||||
(
|
||||
report_id,
|
||||
@@ -125,6 +134,7 @@ def save_report(conn: pymysql.Connection, report: ReportData) -> int:
|
||||
ev.summary,
|
||||
ev.sentiment,
|
||||
ev.source,
|
||||
json.dumps(ev.sources, ensure_ascii=False) if ev.sources else None,
|
||||
ev.url,
|
||||
),
|
||||
)
|
||||
|
||||
+2
-1
@@ -18,7 +18,8 @@ class EventRow(BaseModel):
|
||||
title: str
|
||||
summary: str | None = None
|
||||
sentiment: str | None = None # positive | negative | neutral | ''
|
||||
source: str | None = None
|
||||
source: str | None = None # 主源
|
||||
sources: list[str] | None = None # 全部来源(主源居首),多源新闻记录
|
||||
url: str | None = None
|
||||
|
||||
|
||||
|
||||
+8
-1
@@ -33,7 +33,8 @@ DDL_STATEMENTS: list[str] = [
|
||||
title VARCHAR(512) NOT NULL COMMENT '标题',
|
||||
summary TEXT NULL COMMENT '摘要/正文',
|
||||
sentiment VARCHAR(8) NULL COMMENT 'positive/negative/neutral',
|
||||
source VARCHAR(64) NULL COMMENT '来源',
|
||||
source VARCHAR(64) NULL COMMENT '主来源',
|
||||
sources TEXT NULL COMMENT '全部来源 JSON 数组(主源居首),多源新闻记录',
|
||||
url VARCHAR(512) NULL COMMENT '原文链接',
|
||||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
KEY idx_report_section (report_id, section),
|
||||
@@ -41,3 +42,9 @@ DDL_STATEMENTS: list[str] = [
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='日报事件明细'
|
||||
""",
|
||||
]
|
||||
|
||||
# 旧库迁移:为已存在的 news_event 表补充 sources 列(MySQL 无 ADD COLUMN IF NOT EXISTS)
|
||||
_ALTER_ADD_SOURCES = (
|
||||
"ALTER TABLE news_event ADD COLUMN sources TEXT NULL "
|
||||
"COMMENT '全部来源 JSON 数组(主源居首),多源新闻记录' AFTER source"
|
||||
)
|
||||
|
||||
@@ -102,6 +102,7 @@ def _load_events_from_dir(day_str: str) -> list[dict]:
|
||||
"title": obj.get("title", ""),
|
||||
"url": obj.get("url", ""),
|
||||
"source_id": obj.get("source_id", ""),
|
||||
"sources": obj.get("sources") or [obj.get("source_id", "")],
|
||||
"publish_time": obj.get("publish_time"),
|
||||
"event": ev,
|
||||
})
|
||||
@@ -1004,6 +1005,7 @@ def _build_report_data(news: dict, cninfo: dict, pipeline: dict,
|
||||
summary=(ev.get("summary") or None),
|
||||
sentiment=ev.get("sentiment") or None,
|
||||
source=e.get("source_id") or None,
|
||||
sources=e.get("sources") or None,
|
||||
url=e.get("url") or None,
|
||||
)
|
||||
)
|
||||
|
||||
@@ -109,6 +109,21 @@ def _load_article(p: Path) -> Article | None:
|
||||
return None
|
||||
|
||||
|
||||
def _load_sources(p: Path) -> list[str] | None:
|
||||
"""从输入 JSON 读取 sources 多源字段(deduped uniques 才有);无则返回 None。
|
||||
|
||||
None 表示无多源记录,由 extract_event 兜底为 [source_id]。
|
||||
"""
|
||||
try:
|
||||
data = json.loads(p.read_text(encoding="utf-8"))
|
||||
srcs = data.get("sources")
|
||||
if isinstance(srcs, list) and srcs:
|
||||
return [s for s in srcs if s]
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
async def _run(args: argparse.Namespace) -> int:
|
||||
load_dotenv() # 读 .env 到 os.environ
|
||||
# scene=event_extraction: 读取 configs/llm_models.yaml 场景 1 配置,未配置字段回退 .env
|
||||
@@ -173,6 +188,7 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
article = _load_article(fp)
|
||||
if article is None:
|
||||
return fp, None, "无法解析输入"
|
||||
sources = _load_sources(fp) # 多源记录(去重层),无则 None
|
||||
try:
|
||||
event = await extract_event_async(
|
||||
client=client,
|
||||
@@ -180,6 +196,7 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
article=article,
|
||||
template=template,
|
||||
max_attempts=args.max_attempts,
|
||||
sources=sources,
|
||||
semaphore=semaphore,
|
||||
)
|
||||
return fp, event, None
|
||||
|
||||
@@ -121,6 +121,31 @@ def test_event_types_constant_includes_common() -> None:
|
||||
assert must in EVENT_TYPES
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# ExtractedEvent.sources 多源字段
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def test_extracted_event_sources_defaults_to_main_source() -> None:
|
||||
"""未提供 sources 时兜底为 [source_id](兼容旧产物)。"""
|
||||
ev = ExtractedEvent(
|
||||
source_id="cls", url="https://x/1", url_hash="h1", title="t",
|
||||
event=EventExtraction(sentiment="positive", importance=3, event_type="重大合同"),
|
||||
provider="deepseek", model="m",
|
||||
)
|
||||
assert ev.sources == ["cls"]
|
||||
|
||||
|
||||
def test_extracted_event_sources_keeps_main_first_and_dedup() -> None:
|
||||
"""sources 保主源居首、去重保序。"""
|
||||
ev = ExtractedEvent(
|
||||
source_id="cls", url="https://x/1", url_hash="h1", title="t",
|
||||
sources=["sina", "cls", "eastmoney", "sina"],
|
||||
event=EventExtraction(sentiment="neutral", importance=2, event_type="其他"),
|
||||
provider="deepseek", model="m",
|
||||
)
|
||||
assert ev.sources == ["cls", "sina", "eastmoney"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# JSON 提取与解析
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@@ -255,3 +255,31 @@ class TestCollectNewsEventsLookback:
|
||||
assert "无时间戳" in titles
|
||||
assert "窗口外旧闻" not in titles
|
||||
assert "公告排除" not in titles
|
||||
|
||||
|
||||
class TestLoadEventsSources:
|
||||
"""_load_events_from_dir 读取多源(sources)字段。"""
|
||||
|
||||
def test_loads_sources_with_fallback(self, monkeypatch, tmp_path) -> None:
|
||||
import json as _json
|
||||
|
||||
import scheduler.reporter as rep
|
||||
|
||||
ev_dir = tmp_path / "data" / "events" / "20260812"
|
||||
ev_dir.mkdir(parents=True)
|
||||
(ev_dir / "aaa.json").write_text(_json.dumps({
|
||||
"source_id": "cls", "sources": ["cls", "sina", "eastmoney"],
|
||||
"title": "多源新闻", "url": "https://x/1", "event": {},
|
||||
}, ensure_ascii=False), encoding="utf-8")
|
||||
(ev_dir / "bbb.json").write_text(_json.dumps({
|
||||
"source_id": "cls", "title": "旧产物无 sources", "url": "https://x/2",
|
||||
"event": {},
|
||||
}, ensure_ascii=False), encoding="utf-8")
|
||||
|
||||
monkeypatch.chdir(tmp_path)
|
||||
events = rep._load_events_from_dir("20260812")
|
||||
by_url = {e["url"]: e for e in events}
|
||||
# 新产物:多源完整透传
|
||||
assert by_url["https://x/1"]["sources"] == ["cls", "sina", "eastmoney"]
|
||||
# 旧产物:兜底 [source_id]
|
||||
assert by_url["https://x/2"]["sources"] == ["cls"]
|
||||
|
||||
@@ -20,9 +20,16 @@ class TestEventRow:
|
||||
ev = EventRow(
|
||||
section="intl", rank=2, importance=4, event_type="地缘政治",
|
||||
title="t", summary="s", sentiment="negative", source="ForexLive",
|
||||
sources=["ForexLive", "新浪财经"],
|
||||
url="https://x.com/1",
|
||||
)
|
||||
assert ev.sentiment == "negative"
|
||||
assert ev.sources == ["ForexLive", "新浪财经"]
|
||||
|
||||
def test_sources_optional(self) -> None:
|
||||
"""sources 为可选项(旧数据无多源记录)。"""
|
||||
ev = EventRow(section="news", rank=1, title="t")
|
||||
assert ev.sources is None
|
||||
|
||||
def test_missing_title_raises(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
|
||||
Reference in New Issue
Block a user