init

src/domains/demand_search_article/_const.py

@@ -24,18 +24,23 @@ class DemandSearchArticleConst:
         FAIL = 99
 
     # ═══════════════════════════════════════════════════════════════
-    # 搜索词类型（来源列）
+    # 搜索词类型（video 解构驱动，config_code → key_type）
     # ═══════════════════════════════════════════════════════════════
 
     class KeyType:
-        """搜索词来源类型，对应 demand_search_queue 的三列"""
-        STANDARD_ELEMENT = "standard_element"
-        MATCH_TEXT = "match_text"
-        MATCH_GENERALIZED_ELEMENT = "match_generalized_element"
+        INSPIRATION_SUBSTANCE = "INSPIRATION_SUBSTANCE"
+        VIDEO_INSPIRATION = "VIDEO_INSPIRATION"
+        VIDEO_KEYPOINT = "VIDEO_KEYPOINT"
+        VIDEO_TITLE = "VIDEO_TITLE"
 
         @classmethod
         def all_types(cls) -> tuple:
-            return cls.STANDARD_ELEMENT, cls.MATCH_TEXT, cls.MATCH_GENERALIZED_ELEMENT
+            return (
+                cls.INSPIRATION_SUBSTANCE,
+                cls.VIDEO_INSPIRATION,
+                cls.VIDEO_KEYPOINT,
+                cls.VIDEO_TITLE,
+            )
 
     # ═══════════════════════════════════════════════════════════════
     # 搜索状态: 0→1→2 / 0→1→99
@@ -58,5 +63,11 @@ class DemandSearchArticleConst:
     SEARCH_INTERVAL_SEC = 5     # 搜索页间延迟（秒）
     DETAIL_INTERVAL_SEC = 3     # 详情拉取条间延迟（秒）
 
+    # ═══════════════════════════════════════════════════════════════
+    # XXL-JOB 参数
+    # ═══════════════════════════════════════════════════════════════
+
+    DEAL_LIMIT = 2000           # 单次搜索任务取队列条数上限
+
 
 __all__ = ["DemandSearchArticleConst"]

src/domains/demand_search_article/_mapper.py

@@ -18,19 +18,30 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
     # demand_search_queue 读取 & 状态更新
     # ═══════════════════════════════════════════════════════════════
 
-    async def fetch_queue_pending(self, *, limit: int = 500) -> List[Dict]:
-        """取待搜索的队列项，按 match_score 降序"""
-        query = f"""
-            SELECT *
-            FROM {self.QUEUE_TABLE}
-            WHERE status = %s
-            ORDER BY match_score DESC
-            LIMIT %s
+    async def fetch_queue_pending(self, dt: str | None = None, *, limit: int = 500) -> List[Dict]:
+        """取待搜索的队列项（id 升序，FIFO）
+
+        传 dt 则按分区过滤，不传则全表扫描。
         """
-        return await self.pool.fetch(
-            query=query,
-            params=(self.QueueStatus.INIT, limit),
-        )
+        if dt is not None:
+            query = f"""
+                SELECT *
+                FROM {self.QUEUE_TABLE}
+                WHERE dt = %s AND status = %s
+                ORDER BY id ASC
+                LIMIT %s
+            """
+            params = (dt, self.QueueStatus.INIT, limit)
+        else:
+            query = f"""
+                SELECT *
+                FROM {self.QUEUE_TABLE}
+                WHERE status = %s
+                ORDER BY id ASC
+                LIMIT %s
+            """
+            params = (self.QueueStatus.INIT, limit)
+        return await self.pool.fetch(query=query, params=params)
 
     async def queue_mark_processing(self, ids: List[int]) -> int:
         """批量标记队列项为处理中"""
@@ -47,22 +58,22 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
             params=(self.QueueStatus.PROCESSING, *ids),
         )
 
-    async def queue_mark_done(self, item_id: int) -> int:
+    async def queue_mark_done(self, item_id: int, remark: str = "") -> int:
         """标记队列项为成功"""
         query = f"""
             UPDATE {self.QUEUE_TABLE}
-            SET status = %s
+            SET status = %s, remark = %s
             WHERE id = %s
         """
         return await self.pool.save(
-            query=query, params=(self.QueueStatus.SUCCESS, item_id),
+            query=query, params=(self.QueueStatus.SUCCESS, remark, item_id),
         )
 
     async def queue_mark_failed(self, item_id: int, reason: str = "") -> int:
         """标记队列项为失败"""
         query = f"""
             UPDATE {self.QUEUE_TABLE}
-            SET status = %s, fail_reason = %s
+            SET status = %s, remark = %s
             WHERE id = %s
         """
         return await self.pool.save(

src/domains/demand_search_article/article_search.py

@@ -1,8 +1,8 @@
 """Phase 1: 从 demand_search_queue 取搜索词 → 微信搜索 → 写入 demand_search_article_relation
 
 流程:
-  ArticleSearchRelationMapper.fetch_queue_pending() → 微信搜索 → RelationMapper.insert_batch()
-  experiment_id 从上到下透传，不做过滤条件。
+  ArticleSearchRelationMapper.fetch_queue_pending(dt) → 微信搜索 → RelationMapper.insert_batch()
+  每个队列项包含单条 search_key + key_type（由 video_vectors 解构驱动），experiment_id 透传。
 """
 
 import asyncio
@@ -18,18 +18,11 @@ from ._utils import parse_search_result, is_valid_keyword
 
 logger = logging.getLogger(__name__)
 
-# demand_search_queue 列名 → key_type 映射
-_SEARCH_KEY_COLUMNS = [
-    ("standard_element", DemandSearchArticleConst.KeyType.STANDARD_ELEMENT),
-    ("match_text", DemandSearchArticleConst.KeyType.MATCH_TEXT),
-    ("match_generalized_element", DemandSearchArticleConst.KeyType.MATCH_GENERALIZED_ELEMENT),
-]
-
 
 class DemandSearchArticle(DemandSearchArticleConst):
     """从需求队列取搜索词 → 微信搜索 → 写入关系表
 
-    experiment_id 只透传不过滤——取 match_score 最高的待处理项。
+    按 dt 分区读取，experiment_id 只透传不过滤。
     """
 
     def __init__(self, pool: MysqlManager):
@@ -39,13 +32,15 @@ class DemandSearchArticle(DemandSearchArticleConst):
     # 入口
     # ═══════════════════════════════════════════════════════════════
 
-    async def deal(self, *, limit: int = 500) -> dict:
-        """按 match_score 降序取待处理项 → 微信搜索 → 写入 relation
+    async def deal(self, dt: str | None = None, *, limit: int = 500) -> dict:
+        """取待处理项（id ASC FIFO）→ 微信搜索 → 写入 relation
+
+        传 dt 则按分区过滤，不传则全表扫描。
 
         Returns:
             {"queued": int, "processed": int, "results_written": int}
         """
-        items = await self.relation_mapper.fetch_queue_pending(limit=limit)
+        items = await self.relation_mapper.fetch_queue_pending(dt, limit=limit)
         if not items:
             logger.info("无待处理队列项")
             return {"queued": 0, "processed": 0, "results_written": 0}
@@ -56,9 +51,8 @@ class DemandSearchArticle(DemandSearchArticleConst):
         total_written = 0
         processed = 0
         for item in items:
-            n = 0
             try:
-                n = await self._search_and_write(item)
+                n, from_cache_only = await self._search_and_write(item)
             except Exception as e:
                 logger.exception(
                     "搜索失败: queue_id=%s, experiment_id=%s",
@@ -72,10 +66,13 @@ class DemandSearchArticle(DemandSearchArticleConst):
             if n > 0:
                 total_written += n
                 processed += 1
-                await self.relation_mapper.queue_mark_done(item["id"])
+                await self.relation_mapper.queue_mark_done(item["id"], remark="搜索成功")
+            elif from_cache_only:
+                processed += 1
+                await self.relation_mapper.queue_mark_done(item["id"], remark="命中缓存")
             else:
                 await self.relation_mapper.queue_mark_failed(
-                    item["id"], reason="所有搜索词均未返回结果",
+                    item["id"], reason="搜索词未返回结果",
                 )
 
         logger.info(
@@ -88,39 +85,48 @@ class DemandSearchArticle(DemandSearchArticleConst):
             "results_written": total_written,
         }
 
-    async def _search_and_write(self, item: Dict) -> int:
-        """对单个队列项的每个搜索词列执行搜索，累计写入行数"""
-        total = 0
-        for col, key_type in _SEARCH_KEY_COLUMNS:
-            keyword = (item.get(col) or "").strip()
-            if not is_valid_keyword(keyword):
-                continue
-            n = await self._search_one_keyword(
-                keyword=keyword,
-                key_type=key_type,
-                experiment_id=item["experiment_id"],
-            )
-            total += n
-        return total
+    async def _search_and_write(self, item: Dict) -> tuple[int, bool]:
+        """对单个队列项的 search_key 执行搜索
+
+        Returns:
+            (newly_written, from_cache_only) — from_cache_only 表示有数据但全部来自缓存命中。
+        """
+        keyword = (item.get("search_key") or "").strip()
+        if not is_valid_keyword(keyword):
+            return 0, False
+        return await self._search_one_keyword(
+            keyword=keyword,
+            key_type=item["key_type"],
+            experiment_id=item["experiment_id"],
+        )
 
     async def _search_one_keyword(
         self,
         keyword: str,
         key_type: str,
         experiment_id: str,
-    ) -> int:
-        """单个搜索词: 翻页搜索 → 批量写入 relation 表"""
-        total = 0
+    ) -> tuple[int, bool]:
+        """单个搜索词: 翻页搜索 → 批量写入 relation 表
+
+        Returns:
+            (newly_written, from_cache_only) — from_cache_only 为 True 表示
+            至少有一页返回了文章，且没有新写入（全部来自缓存命中）。
+        """
+        newly_written = 0
+        had_data = False
         cursor = "0"
         for page in range(self.MAX_SEARCH_PAGES):
             result, from_cache = await self._search_with_cache(keyword, page, cursor)
             if result is None:
                 break
 
+            data = result.get("data") or {}
+            articles = data.get("data") or []
+            if articles:
+                had_data = True
+
             # 非缓存命中时才写 relation + 等待（缓存命中的数据已经写过了）
             if not from_cache:
-                data = result.get("data", {})
-                articles = data.get("data", [])
                 if not articles:
                     break
 
@@ -129,22 +135,21 @@ class DemandSearchArticle(DemandSearchArticleConst):
                     for a in articles
                 ]
                 n = await self.relation_mapper.insert_batch(items)
-                total += n
+                newly_written += n
                 logger.info(
                     "搜索写入: keyword=%s, page=%d, 返回 %d 条, 写入 %d 条",
                     keyword, page, len(articles), n,
                 )
                 await asyncio.sleep(self.SEARCH_INTERVAL_SEC)
 
-            # 从响应中取翻页游标（缓存和非缓存都要翻页）
-            data = result.get("data", {})
             has_more = data.get("has_more", False)
             next_cursor = data.get("next_cursor")
             if not has_more or not next_cursor:
                 break
             cursor = str(next_cursor)
 
-        return total
+        from_cache_only = had_data and newly_written == 0
+        return newly_written, from_cache_only
 
     async def _search_with_cache(
         self, keyword: str, page: int, cursor: str,

src/domains/fetch_demand/_const.py

@@ -21,6 +21,9 @@ class DemandConst:
     # 匹配视频来源表
     MATCH_RESULT_TABLE = "public_channel_demand_match_result"
 
+    # 匹配分数阈值（只保留 score >= 此值的匹配结果）
+    MIN_MATCH_SCORE = 0.5
+
     # ═══════════════════════════════════════════════════════════════
     # 搜索词类型（video 解构驱动）
     # ═══════════════════════════════════════════════════════════════

src/domains/fetch_demand/_mapper.py

@@ -34,11 +34,11 @@ class DemandQueueMapper(DemandConst):
         query = f"""
             INSERT IGNORE INTO {self.QUEUE_TABLE}
                 (account, search_key, dt, video_id,
-                 channel_name, key_type, experiment_id,
+                 channel_name, category, key_type, experiment_id,
                  status, drive_type)
             VALUES
                 (%s, %s, %s, %s,
-                 %s, %s, %s,
+                 %s, %s, %s, %s,
                  %s, %s)
         """
         params = [
@@ -48,6 +48,7 @@ class DemandQueueMapper(DemandConst):
                 it["dt"],
                 it["video_id"],
                 it["channel_name"],
+                it.get("category", ""),
                 it["key_type"],
                 it["experiment_id"],
                 self.QueueStatus.INIT,

src/domains/fetch_demand/fetch_demands.py

@@ -110,7 +110,7 @@ class FetchDemands(DemandConst):
             f"FROM {self.MATCH_RESULT_TABLE} "
             f"WHERE dt = '{dt}' "
             f"  AND channel_name IN {self._format_tuple(self.CHANNEL_NAMES)} "
-            f"  AND match_score > 0.1 "
+            f"  AND match_score >= {self.MIN_MATCH_SCORE} "
             f"ORDER BY match_score DESC"
         )
         logger.info("ODPS 查询匹配视频, dt=%s", dt)
@@ -245,6 +245,7 @@ class FetchDemands(DemandConst):
             "dt": row.get("dt") or "",
             "video_id": int(row.get("match_video_id") or 0),
             "channel_name": row.get("channel_name") or "",
+            "category": row.get("category_name") or "",
             "experiment_id": row.get("experiment_id") or "",
             "drive_type": DemandConst.DRIVE_TYPE_VIDEO,
         }

src/domains/fetch_demand_videos/__init__.py

@@ -1,6 +0,0 @@
-"""
-基于人群需求匹配到的视频
-取 video_id、 video_title
-取 解构结果（关键点、灵感点）
-存 账号信息、 dt、 搜索词、 video_id、 experiment_id 、config_code
-"""

src/handlers/article_search.py

@@ -0,0 +1,71 @@
+"""文章搜索 XXL-JOB Handler —— 从队列取搜索词 → 微信搜索 → 写入 relation
+
+XXL-JOB Admin 配置:
+    执行器: long-articles-agentic-src
+    处理器: articleSearch
+    参数:   dt=yyyyMMdd (可选，不传则全表扫描)
+
+设计:
+    - Admin 调度的 HTTP 请求秒级返回，实际搜索在后台异步执行
+    - limit 取自 DemandSearchArticleConst.DEAL_LIMIT，按 id ASC FIFO
+    - MySQL 复用 app 级别的连接池（set_db() 注入），不每次自建
+"""
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING
+
+from src.infra.xxl_jobs import xxl_job
+from src.domains.demand_search_article import DemandSearchArticle, DemandSearchArticleConst
+
+if TYPE_CHECKING:
+    from src.infra.database import DatabaseManager
+
+logger = logging.getLogger(__name__)
+
+# 由 create_app() 在启动时注入
+_db_ref: "DatabaseManager | None" = None
+
+
+def set_db(db: "DatabaseManager") -> None:
+    """注入 app 级别的 DatabaseManager，供 handler 复用已有连接池"""
+    global _db_ref
+    _db_ref = db
+
+
+def _parse_dt(param: str) -> str | None:
+    """从 XXL-JOB 参数中解析 dt，格式 dt=yyyyMMdd；不传返回 None（全表扫描）"""
+    if param and param.strip():
+        for part in param.strip().split():
+            if part.startswith("dt="):
+                return part[3:]
+        return param.strip()
+    return None
+
+
+async def _do_search(dt: str | None) -> None:
+    """后台执行: 队列取搜索词 → 微信搜索 → relation 写入"""
+    limit = DemandSearchArticleConst.DEAL_LIMIT
+    logger.info("articleSearch background task started, dt=%s, limit=%d", dt, limit)
+
+    try:
+        mysql = _db_ref.mysql_manager("long_articles")
+        searcher = DemandSearchArticle(mysql)
+        result = await searcher.deal(dt, limit=limit)
+    except Exception:
+        logger.exception("文章搜索失败, dt=%s", dt)
+        return
+
+    logger.info(
+        "articleSearch done: dt=%s, queued=%d, processed=%d, written=%d",
+        dt, result["queued"], result["processed"], result["results_written"],
+    )
+
+
+@xxl_job("articleSearch")
+async def article_search(param: str) -> dict:
+    """文章搜索 handler — 秒级返回，实际工作在后台执行"""
+    dt = _parse_dt(param)
+    asyncio.create_task(_do_search(dt))
+    logger.info("articleSearch accepted, dt=%s (background task spawned)", dt)
+    return {"code": 200, "msg": f"accepted, dt={dt}"}

src/server/app.py

@@ -12,7 +12,8 @@ from src.infra.observability import LogService
 from src.infra.xxl_jobs import XxlJobExecutor
 from src.handlers import discover_and_register
 from src.server.xxl_endpoints import set_xxl_executor
-from src.handlers.demand_enqueue import set_db
+from src.handlers.demand_enqueue import set_db as set_db_demand_enqueue
+from src.handlers.article_search import set_db as set_db_article_search
 
 logger = logging.getLogger(__name__)
 
@@ -43,7 +44,8 @@ def create_app(global_config: GlobalConfig = None) -> Quart:
     xxl_executor = XxlJobExecutor(global_config.xxl_job)
 
     set_xxl_executor(xxl_executor)
-    set_db(db)
+    set_db_demand_enqueue(db)
+    set_db_article_search(db)
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")