fix: 回填脚本改为定向回填多源 point(本地模式全量不可行)

- 本地模式 Qdrant 写入逐点 0.4~0.65s,全量 3.3 万条需 4~6h 且占单进程锁
- 改为仅回填指纹库 227 条多源记录(单源点展示层兜底 source_id):1m57s 完成
- 测试改为 _load_multi_source 语义(NULL/损坏/单源排除)

continuation.md

@@ -19,17 +19,19 @@
 - B4 `vectorstore/client.py`: 检索读取 payload.sources
 - A1 `scheduler/reporter.py`: 日报 HTML 多源时显示「财联社 / 新浪 📰」
 - A2 `a_share_cli/main.py` + `mcp_server/tools.py`: 检索展示「cecn / cscn [多源]」,MCP 返回 sources 数组
-- 存量回填 `scripts/backfill_qdrant_sources.py`(新):以指纹库 source_ids 为权威源,scroll+批量 upsert 回写 3.3 万条 payload
+- 存量回填 `scripts/backfill_qdrant_sources.py`(新):以指纹库 source_ids 为权威源回填 Qdrant payload
-  - ⚠️ 坑:本地文件模式 Qdrant 的 `set_payload` 逐点极慢(0.65s/点,33k 条约 6h),改为与 ingest 相同的 scroll+upsert 快速路径;且本地模式单进程锁,必须避开调度窗口
+  - ⚠️ 坑:本地文件模式 Qdrant 写入逐点 ≈0.4~0.65s,**全量 3.3 万条回填需 4~6 小时**且占用单进程锁,不可行
+  - 改为**定向回填**:只有多源 point 需要 sources(单源点展示层兜底 source_id),仅回填指纹库 227 条多源 → **1m57s 完成 226 条**(1 条不在集合)
+  - 执行时机:避开定时调度窗口(22:00 pipeline 结束后 22:34 执行)
 
 **验证**:
 - 新增 `tests/test_multisource.py` 10 个(validator/`_result_to_point` 三优先级/文本构造透传/回填加载)
 - 全量 **276 passed**(3 个 crawler 基线失败与本次无关);ruff 干净(reporter 5 个 N806/SIM115 为既有问题未动)
 - 今日 M5/M6 重跑:313 条全含 sources(11 条多源);CLI 检索显示「来源: cecn / cscn [多源]」✓,MCP 返回 `sources: ['cecn','cscn']` ✓
-- 存量回填:待执行(等 22:00 pipeline 结束后运行,避开 Qdrant 单进程锁)
+- 存量回填:226 条多源 point payload 已补 sources;CLI 检索「碧根果反倾销」显示「来源: cnstock / cscn [多源]」✓;日报 HTML 冒烟测试多源渲染「财联社 / 新浪财经 / 东方财富 📰」✓
 
 **待办/遗留**:
-- 回填脚本执行 + 回填后检索验证
+- 单源存量 point 无 sources 字段(展示层兜底 source_id,行为不变,如需统一可后续补)
 - P1-2(补跑 steps 含 cninfo)、P1-3(时区一致性),待用户决策
 
 ---

scripts/backfill_qdrant_sources.py

@@ -1,22 +1,18 @@
-"""一次性回填脚本:为 Qdrant 存量 point 的 payload 补充 sources 多源字段。
+"""一次性回填脚本:为 Qdrant 存量**多源** point 的 payload 补充 sources 字段。
 
 背景:
     M5/M6 旧链路未把 M3 多源记录写入 EmbeddingResult / Qdrant payload,
     导致知识库检索只能看到单源。新链路修复后,此脚本为存量数据补齐。
 
-数据来源(按优先级):
+范围(关键设计):
-    1. 指纹库 data/dedup/fingerprints.sqlite3 的 source_ids 列(权威多源记录);
+    本地文件模式 Qdrant 写入逐点成本 ≈0.4~0.65s,全量 3.3 万条回填需数小时,
-    2. source_ids 为 NULL 的旧行回退 [source_id];
+    且本地模式为单进程锁,长时间占用会阻塞后续定时调度。
-    3. 指纹库查不到的 point 回退 payload.source_id。
+    但**只有多源 point 需要 sources 字段**——单源 point 的展示层
+    (CLI/MCP/日报)会在 sources 为空时兜底显示 source_id。
+    因此本脚本仅回填指纹库中 source_ids 长度 > 1 的多源记录,分钟级完成。
 
-实现:
+数据来源:
-    本地文件模式 Qdrant 的 set_payload 逐点极慢(≈0.65s/点,33k 条约 6 小时),
+    指纹库 data/dedup/fingerprints.sqlite3 的 source_ids 列(权威多源记录)。
-    故采用与正常 ingest 相同的快速路径:
-        scroll 分页取全部 point(含向量)→ 合并 sources → 批量 upsert 回写。
-
-注意:
-    Qdrant 为本地文件模式,同一时刻仅允许一个进程打开;
-    执行前请确认无 pipeline 子进程正在运行(避开调度窗口)。
 
 用法:
     uv run python -m scripts.backfill_qdrant_sources          # 实际执行
@@ -29,15 +25,13 @@ import argparse
 import json
 import sqlite3
 import sys
+from collections import defaultdict
 from pathlib import Path
 
 from loguru import logger
-from qdrant_client.http.models import PointStruct
 
 from vectorstore import DEFAULT_COLLECTION, make_qdrant_client
-
+from vectorstore.client import url_hash_to_uuid
-_SCROLL_LIMIT = 1000
-_UPSERT_BATCH = 100
 
 
 def _setup_logger(level: str) -> None:
@@ -49,8 +43,8 @@ def _setup_logger(level: str) -> None:
     )
 
 
-def _load_fingerprint_sources(db_path: Path) -> dict[str, list[str]]:
+def _load_multi_source(db_path: Path) -> dict[str, list[str]]:
-    """从指纹库读 url_hash -> sources(source_ids 为 NULL 时回退 [source_id])。"""
+    """从指纹库读多源记录: url_hash -> sources(仅 source_ids 长度 > 1)。"""
     conn = sqlite3.connect(db_path)
     conn.row_factory = sqlite3.Row
     rows = conn.execute(
@@ -60,47 +54,26 @@ def _load_fingerprint_sources(db_path: Path) -> dict[str, list[str]]:
 
     result: dict[str, list[str]] = {}
     for r in rows:
-        sources: list[str] = []
+        if not r["source_ids"]:
-        if r["source_ids"]:
+            continue
-            try:
+        try:
-                val = json.loads(r["source_ids"])
+            val = json.loads(r["source_ids"])
-                if isinstance(val, list):
+        except (TypeError, ValueError):
-                    sources = [s for s in val if s]
+            continue
-            except (TypeError, ValueError):
+        if not isinstance(val, list):
-                sources = []
+            continue
-        if not sources:
+        sources = [s for s in val if s]
-            sources = [r["source_id"]] if r["source_id"] else []
+        if len(sources) <= 1:
+            continue
         # 保证主源居首
         if r["source_id"] and r["source_id"] in sources:
             sources = [r["source_id"], *[s for s in sources if s != r["source_id"]]]
-        if sources:
+        result[r["url_hash"]] = sources
-            result[r["url_hash"]] = sources
     return result
 
 
-def _build_points(
-    records: list,
-    fp_sources: dict[str, list[str]],
-) -> list[PointStruct]:
-    """把 scroll 记录转回 PointStruct,payload 补入 sources 字段。"""
-    points: list[PointStruct] = []
-    for rec in records:
-        payload = dict(rec.payload or {})
-        url_hash = payload.get("url_hash") or ""
-        source_id = payload.get("source_id") or ""
-        sources = fp_sources.get(url_hash) or ([source_id] if source_id else [])
-        if sources:
-            payload["sources"] = sources
-        points.append(PointStruct(
-            id=rec.id,
-            vector=rec.vector,
-            payload=payload,
-        ))
-    return points
-
-
 def main() -> int:
-    parser = argparse.ArgumentParser(description="Qdrant 存量 payload 回填 sources 多源字段")
+    parser = argparse.ArgumentParser(description="Qdrant 存量多源 payload 回填 sources 字段")
     parser.add_argument("--db", default="data/dedup/fingerprints.sqlite3",
                         help="指纹库路径")
     parser.add_argument("--collection", default=DEFAULT_COLLECTION)
@@ -110,56 +83,53 @@ def main() -> int:
 
     _setup_logger(args.log_level)
 
-    fp_sources = _load_fingerprint_sources(Path(args.db))
+    multi = _load_multi_source(Path(args.db))
-    logger.info("指纹库加载完成: {} 条,其中多源 {} 条",
+    logger.info("指纹库多源记录: {} 条", len(multi))
-                len(fp_sources), sum(1 for v in fp_sources.values() if len(v) > 1))
+    if not multi:
-
+        logger.info("无多源记录,退出")
-    client = make_qdrant_client()
-
-    # scroll 分页 + (非 dry-run) 批量 upsert 回写
-    total = 0
-    multi_cnt = 0
-    updated = 0
-    offset = None
-    while True:
-        records, offset = client.scroll(
-            collection_name=args.collection,
-            limit=_SCROLL_LIMIT,
-            offset=offset,
-            with_payload=True,
-            with_vectors=True,
-        )
-        if not records:
-            if offset is None:
-                break
-            continue
-        total += len(records)
-        multi_cnt += sum(
-            1 for r in records
-            if len((r.payload or {}).get("sources") or []) > 1
-        )
-        if not args.dry_run:
-            points = _build_points(records, fp_sources)
-            for i in range(0, len(points), _UPSERT_BATCH):
-                client.upsert(
-                    collection_name=args.collection,
-                    points=points[i : i + _UPSERT_BATCH],
-                )
-            updated += len(points)
-        if offset is None:
-            break
-
-    logger.info("Qdrant 现存 point: {} 条(其中 payload 已含多源 {} 条)", total, multi_cnt)
-    if args.dry_run:
-        logger.info("--dry-run: 不执行写入")
-        client.close()
         return 0
 
-    logger.info("回填完成: 重新 upsert {} 条 point(payload 已补 sources)", updated)
+    # 按 sources 值分组(同值一次 set_payload)
+    groups: dict[str, list[str]] = defaultdict(list)
+    for url_hash, sources in multi.items():
+        groups[json.dumps(sources, ensure_ascii=False)].append(url_hash)
+
+    if args.dry_run:
+        logger.info("待回填: {} 组 / {} 条多源 point (--dry-run 不写入)",
+                    len(groups), len(multi))
+        return 0
+
+    client = make_qdrant_client()
+    updated = 0
+    missing = 0
+    for sources_json, url_hashes in groups.items():
+        payload = {"sources": json.loads(sources_json)}
+        ids = [url_hash_to_uuid(h) for h in url_hashes]
+        try:
+            client.set_payload(
+                collection_name=args.collection,
+                payload=payload,
+                points=ids,
+            )
+            updated += len(ids)
+        except KeyError:
+            # 部分 point 已不在集合中(如指纹库保留但向量库已清理),逐点重试
+            for uid in ids:
+                try:
+                    client.set_payload(
+                        collection_name=args.collection,
+                        payload=payload,
+                        points=[uid],
+                    )
+                    updated += 1
+                except KeyError:
+                    missing += 1
+    logger.info("回填完成: 更新 {} 条多源 point,{} 条不存在于集合",
+                updated, missing)
+
     client.close()
     return 0
 
 
 if __name__ == "__main__":
     raise SystemExit(main())
-

tests/test_multisource.py

@@ -166,7 +166,7 @@ def test_build_text_from_article_reads_sources(tmp_path: Path) -> None:
 
 
 # --------------------------------------------------------------------------- #
-# 回填脚本 _load_fingerprint_sources
+# 回填脚本 _load_multi_source
 # --------------------------------------------------------------------------- #
 
 def _make_fp_db(tmp_path: Path) -> Path:
@@ -183,23 +183,30 @@ def _make_fp_db(tmp_path: Path) -> Path:
     conn.execute(
         "INSERT INTO fingerprints VALUES (?, ?, ?)", ("h2", "zqrb", None)
     )
+    conn.execute(
+        "INSERT INTO fingerprints VALUES (?, ?, ?)", ("h3", "cls", "not-json")
+    )
+    conn.execute(
+        "INSERT INTO fingerprints VALUES (?, ?, ?)",
+        ("h4", "cls", json.dumps(["cls"])),
+    )
     conn.commit()
     conn.close()
     return db
 
 
-def test_load_fingerprint_sources_parses_and_falls_back(tmp_path: Path) -> None:
+def test_load_multi_source_only_multi(tmp_path: Path) -> None:
-    from scripts.backfill_qdrant_sources import _load_fingerprint_sources
+    """仅返回多源记录;NULL/损坏/单源均排除。"""
+    from scripts.backfill_qdrant_sources import _load_multi_source
 
     db = _make_fp_db(tmp_path)
-    res = _load_fingerprint_sources(db)
+    res = _load_multi_source(db)
-    assert res["h1"] == ["cls", "sina"]
+    assert res == {"h1": ["cls", "sina"]}
-    assert res["h2"] == ["zqrb"]  # source_ids NULL → 回退 [主源]
 
 
-def test_load_fingerprint_sources_main_first(tmp_path: Path) -> None:
+def test_load_multi_source_main_first(tmp_path: Path) -> None:
     """source_ids 顺序异常时保证主源居首。"""
-    from scripts.backfill_qdrant_sources import _load_fingerprint_sources
+    from scripts.backfill_qdrant_sources import _load_multi_source
 
     db = tmp_path / "fp.sqlite3"
     conn = sqlite3.connect(db)
@@ -213,5 +220,5 @@ def test_load_fingerprint_sources_main_first(tmp_path: Path) -> None:
     )
     conn.commit()
     conn.close()
-    res = _load_fingerprint_sources(db)
+    res = _load_multi_source(db)
     assert res["h1"] == ["cls", "sina"]