docs: 多源入库与生产迁移说明

- README: 多源记录补充 M4/日报/DB(news_event.sources)透传说明
- continuation.md: 记录多源入库、阶段/AI 显性输出、AI 摘要崩溃修复与生产迁移

README.md

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

continuation.md

@@ -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` 兜底；模型必须显式配置，无内置兜底 |
+| LLM | 场景化配置 `configs/llm_models.yaml`（4 场景）；YAML 优先、`.env` 兜底 |
-| 去重 | 多源记录：指纹库 `source_ids` 列 + uniques JSON `sources` 字段 + `data/deduped/{day}/sources.json` |
+| 去重 | 多源记录：指纹库 `source_ids` 列 + uniques `sources` 字段 + `sources.json`；**M4 事件 / 日报 / DB `news_event.sources` 全链路透传** |
-| 增量/断点 | M2/M4/M5 产物存在即跳过（`--force` 全量）；`pipeline --once --resume` 断点续跑（状态 `data/pipeline/state.json`） |
+| 增量/断点 | 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` 全量）：**
+**2. pipeline 阶段与 AI 显性输出（scheduler/pipeline.py）：**
-- M2 `run_extractor.py`：输出目录已有 `{url_hash}.json` 即跳过提取，仅回补 index 行；`--force` 重建；成功率统计含跳过项（修复全跳过时误报 rc=1）
+- 每步前打印 `阶段 i/n: 中文名 [步骤] 日期` 分隔标题
-- M4 `run_event_extraction.py`：`data/events/{day}/{url_hash}.json` 已存在即跳过（**不重复调用 LLM API**）；`--force` 全量；failed.jsonl 只保留本次失败、index 累积追加
+- llm/report/embedding 步骤打印 `🤖 AI 大模型: provider=..., model=...`（按场景配置解析，缺 key 也可展示）
-- M5 `run_embedding.py`：`data/embeddings/{day}/{url_hash}.json` 已存在即跳过（**不重复调用 embed API**）；`--force` 全量
-- M1（seen_urls 增量）/ M3（指纹库判重）/ M6（upsert 幂等）为既有能力，README 汇总成表
 
-**2. pipeline 断点续跑（scheduler/pipeline.py + run_scheduler.py）：**
+**3. 修复日报 AI 摘要崩溃（'str' object has no attribute 'model'）：**
-- 新增 `data/pipeline/state.json`（按日期隔离，记录每步骤 ok/failed + 退出码 + 耗时），原子写
+- `_llm_summarize` 内 3 处 `_llm_call(client, model, ...)` 改为传 `config`（上轮改签名遗漏）
-- `run_pipeline(resume=True)` 跳过连续成功前缀，从首个失败/未执行步骤继续执行到结尾
+- 新增回归测试（单块/多分块全链路）
-- `pipeline --once --resume`（默认全量不变；`--resume` 与 `--steps` 互斥报错）；定时守护模式不受影响
 
-**验证（Mac 本地，20260616 数据 100 篇）：**
+**验证（pi5 生产）：**
-- pytest **225 passed**（新增 tests/test_incremental.py 10 个：M2/M4/M5 跳过、状态记录、resume 续跑、resume 全完成 noop、--resume+--steps 互斥）；crawler 3 个基线失败仍与本次无关
+- 本地 pytest **231 passed**（新增 ExtractedEvent.sources×2、EventRow.sources×2、_load_events_from_dir×1）；ruff 基线 9 零新增
-- ruff 零新增（9 个基线错误不变）
+- pi5：`init_schema` 迁移成功（news_event 新增 sources 列）；日报生成入库正常（report_id=225, events=60）
-- 端到端：M2 增量重跑 140 条跳过 126 条，2.5s 完成、rc=0（修复前误报失败）；state.json 正确记录 extractor ok
+- 部署前已修复并在生产实测 AI 摘要恢复
-
-**效果评估（100 篇规模中断重跑场景）：**
-- M4 减少约 100 次 LLM API 调用、M5 减少约 100 次 embedding API 调用 → 中断恢复不再重复计费，耗时从分钟级降至秒级
-- M2 重跑从全量 GNE 提取（分钟级）降至约 2.5s
-- 断点恢复操作：中断后直接重跑同一条 `pipeline --once --resume` 命令即可
 
 **待办：**
-- 同步 pi5（代码 + 文档），重启 `a-share-research`；首次同步后 pi5 的 `data/pipeline/state.json` 不存在 → resume 按全量处理，行为安全
+- 下次全链路运行后抽查 `news_event.sources` 多源值（当前生产 events 为旧产物，sources=NULL）
 - git 提交（本次改动尚未提交）
 
 ---

llm/extractor.py

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

llm/models.py

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

report_db/db.py

@@ -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)
+                     summary, sentiment, source, sources, url)
-                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+                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,
                 ),
             )

report_db/models.py

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

report_db/schema.py

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

scheduler/reporter.py

@@ -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,
                 )
             )

scripts/run_event_extraction.py

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

tests/test_llm.py

@@ -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 提取与解析
 # --------------------------------------------------------------------------- #

tests/test_report_builder.py

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

tests/test_report_db.py

@@ -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):