init

src/domains/fetch_demand/_mapper.py

@@ -1,6 +1,6 @@
 """需求搜索队列 MySQL 读写 + 视频搜索词 PG 查询 —— 仅负责 DB I/O，不含业务逻辑"""
 
-from typing import Dict, List, Optional
+from typing import Dict, List
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.database.postgresql.manager import PgManager
@@ -155,28 +155,29 @@ class VideoDeconstructMapper:
     def __init__(self, pg: PgManager):
         self.pg = pg
 
-    async def find_by_video_id(self, video_id: int) -> Optional[list[Dict]]:
-        """按 video_id 查询所有搜索词
+    async def find_by_video_ids(self, video_ids: list[int]) -> dict[int, list[Dict]]:
+        """批量查询 video_vectors，一次网络往返
 
         Returns:
-            [
-                {"config_code": "VIDEO_KEYPOINT", "text": "关键点文本"},
-                {"config_code": "VIDEO_INSPIRATION", "text": "灵感点文本"},
-                ...
-            ]
-            无结果返回空 list（非 None，统一语义）。
+            {video_id: [{"config_code": "VIDEO_KEYPOINT", "text": "xx"}, ...], ...}
+            不存在的 video_id 不会出现在返回 key 中。
         """
+        if not video_ids:
+            return {}
         cols = ", ".join(self.COLUMNS)
         rows = await self.pg.fetch(
-            f"SELECT {cols} FROM video_vectors WHERE video_id = $1",
-            params=(video_id,),
+            f"SELECT {cols} FROM video_vectors WHERE video_id = ANY($1)",
+            params=(video_ids,),
         )
-        result: list[Dict] = []
+        result: dict[int, list[Dict]] = {}
         for row in rows:
+            vid = int(row["video_id"])
             config_code = (row.get("config_code") or "").strip()
             text_val = (row.get("text") or "").strip()
             if config_code and text_val:
-                result.append({"config_code": config_code, "text": text_val})
+                if vid not in result:
+                    result[vid] = []
+                result[vid].append({"config_code": config_code, "text": text_val})
         return result
 
 

src/domains/fetch_demand/fetch_demands.py

@@ -11,6 +11,8 @@ import asyncio
 import logging
 from typing import Dict, List, Set
 
+from tqdm import tqdm
+
 from src.config.odps import OdpsConfig
 from src.infra.external.odps import fetch_from_odps
 from src.infra.database.postgresql.manager import PgManager
@@ -167,25 +169,59 @@ class FetchDemands(DemandConst):
         return result
 
     # ═══════════════════════════════════════════════════════════════
-    # 5. 视频解构 + 搜索词展开
+    # 5. 视频搜索词批查询 + 展开
     # ═══════════════════════════════════════════════════════════════
 
+    # PG 批量查询每批最多 video_id 数量
+    _VIDEO_BATCH_SIZE = 500
+
     async def _explode_search_items(self, rows: List[Dict]) -> List[Dict]:
-        """对每条去重后的视频记录，获取解构结果并展开为搜索队列条目
+        """收集所有 video_id → 分页 PG 批量查询 → 按 id 分发展开
 
-        一条视频可能产生多条队列项（每个搜索词一条）。
+        分页避免一次 ANY($1) 塞入数万 id 导致 PG 压力过大。
         """
-        items: List[Dict] = []
+        # 收集去重后的 video_id 列表（保持插入顺序去重）
+        video_id_to_rows: dict[int, List[Dict]] = {}
+        ordered_ids: list[int] = []
+        seen: set[int] = set()
         for r in rows:
-            video_id = int(r.get("match_video_id") or 0)
-            if not video_id:
+            vid = int(r.get("match_video_id") or 0)
+            if not vid:
                 continue
-            decomp = await self._get_video_decomposition(video_id)
-            if not decomp:
-                logger.warning("视频 %d 解构结果为空，跳过", video_id)
+            if vid not in seen:
+                seen.add(vid)
+                ordered_ids.append(vid)
+            if vid not in video_id_to_rows:
+                video_id_to_rows[vid] = []
+            video_id_to_rows[vid].append(r)
+
+        if not ordered_ids:
+            return []
+
+        # 分页批量查询（带进度条）
+        total_batches = (len(ordered_ids) + self._VIDEO_BATCH_SIZE - 1) // self._VIDEO_BATCH_SIZE
+        vectors_map: dict[int, list[Dict]] = {}
+        with tqdm(total=len(ordered_ids), desc="video_vectors", unit="vid") as pbar:
+            for i in range(0, len(ordered_ids), self._VIDEO_BATCH_SIZE):
+                batch = ordered_ids[i:i + self._VIDEO_BATCH_SIZE]
+                chunk = await self._deconstruct_mapper.find_by_video_ids(batch)
+                vectors_map.update(chunk)
+                pbar.update(len(batch))
+                pbar.set_postfix({"hit": len(vectors_map), "batch": f"{i // self._VIDEO_BATCH_SIZE + 1}/{total_batches}"})
+
+        logger.info(
+            "video_vectors 批量查询完成: %d 个 video_id → %d 个有结果",
+            len(ordered_ids), len(vectors_map),
+        )
+
+        # 带回分发到每条 ODPS 行
+        items: List[Dict] = []
+        for vid, sub_rows in video_id_to_rows.items():
+            vectors = vectors_map.get(vid)
+            if not vectors:
                 continue
-            exploded = self._build_items_from_decomp(r, decomp)
-            items.extend(exploded)
+            for row in sub_rows:
+                items.extend(self._build_items_from_decomp(row, vectors))
         return items
 
     @staticmethod
@@ -213,10 +249,21 @@ class FetchDemands(DemandConst):
             "drive_type": DemandConst.DRIVE_TYPE_VIDEO,
         }
 
+        # 仅保留四种有效 key_type，过滤 video_vectors 中其他 config_code
+        _VALID_KEY_TYPES = frozenset({
+            DemandConst.KeyType.INSPIRATION_SUBSTANCE,
+            DemandConst.KeyType.VIDEO_INSPIRATION,
+            DemandConst.KeyType.VIDEO_KEYPOINT,
+            DemandConst.KeyType.VIDEO_TITLE,
+        })
+
         items: List[Dict] = []
         seen: Set[str] = set()
 
         for v in vectors:
+            key_type = v.get("config_code") or ""
+            if key_type not in _VALID_KEY_TYPES:
+                continue
             text_val = (v.get("text") or "").strip()
             if not text_val:
                 continue
@@ -227,22 +274,11 @@ class FetchDemands(DemandConst):
             items.append({
                 **base,
                 "search_key": text_val,
-                "key_type": v.get("config_code") or "",
+                "key_type": key_type,
             })
 
         return items
 
-    # ═══════════════════════════════════════════════════════════════
-    # 视频搜索词（PG: video_vectors）
-    # ═══════════════════════════════════════════════════════════════
-
-    async def _get_video_decomposition(self, video_id: int) -> list[Dict]:
-        """从 PostgreSQL video_vectors 表查询搜索词，返回 [{config_code, text}, ...]"""
-        vectors = await self._deconstruct_mapper.find_by_video_id(video_id)
-        if not vectors:
-            logger.debug("视频 %d 在 video_vectors 中无记录", video_id)
-        return vectors
-
     # ═══════════════════════════════════════════════════════════════
     # 工具
     # ═══════════════════════════════════════════════════════════════