init

src/domains/demand_search_article/_mapper.py

@@ -90,16 +90,16 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
 
         items 每条须包含:
           search_key, key_type, experiment_id, search_cursor,
-          url, wx_sn, title, cover_url, publish_time
+          url, wx_sn, biz, title, cover_url, publish_time
         """
         if not items:
             return 0
         query = f"""
             INSERT IGNORE INTO {self.RELATION_TABLE}
                 (search_key, key_type, experiment_id, search_cursor,
-                 url, wx_sn, title, cover_url, publish_time)
+                 url, wx_sn, biz, title, cover_url, publish_time)
             VALUES
-                (%s, %s, %s, %s, %s, %s, %s, %s, %s)
+                (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
         """
         params = [
             (
@@ -109,6 +109,7 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
                 it["search_cursor"],
                 it["url"],
                 it["wx_sn"],
+                it["biz"],
                 it["title"],
                 it["cover_url"],
                 it["publish_time"],
@@ -123,7 +124,6 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
 
     async def fetch_pending(
         self,
-        experiment_id: str,
         *,
         limit: int = 500,
     ) -> List[Dict]:
@@ -131,13 +131,11 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         query = f"""
             SELECT *
             FROM {self.RELATION_TABLE}
-            WHERE experiment_id = %s AND status = %s
+            WHERE status = %s
             LIMIT %s
         """
-        return await self.pool.fetch(
-            query=query,
-            params=(experiment_id, self.SearchStatus.INIT, limit),
-        )
+        params = (self.SearchStatus.INIT, limit)
+        return await self.pool.fetch(query=query, params=params)
 
     async def count_by_status(
         self,

src/domains/demand_search_article/_utils.py

@@ -5,16 +5,29 @@ from typing import Dict
 from urllib.parse import parse_qs, urlparse
 
 
+def _extract_query_param(content_url: str | None, param: str) -> str | None:
+    """从 URL query 中提取指定参数值"""
+    if not content_url:
+        return None
+    query = urlparse(content_url).query
+    return parse_qs(query).get(param, [None])[0]
+
+
 def extract_wx_sn(content_url: str | None) -> str | None:
     """从文章 URL 中提取 sn 参数作为 wx_sn
 
     文章链接格式: http://mp.weixin.qq.com/s?__biz=...&sn=xxx...
     sn 参数值即为微信文章链接 ID。
     """
-    if not content_url:
-        return None
-    query = urlparse(content_url).query
-    return parse_qs(query).get("sn", [None])[0]
+    return _extract_query_param(content_url, "sn")
+
+
+def extract_biz(content_url: str | None) -> str | None:
+    """从文章 URL 中提取 __biz 参数，标识公众号主体
+
+    文章链接格式: http://mp.weixin.qq.com/s?__biz=MzIxNDE5NDA2OQ==&sn=xxx...
+    """
+    return _extract_query_param(content_url, "__biz")
 
 
 # 无效搜索词集合 — 仅包含无意义的占位符
@@ -51,6 +64,7 @@ def parse_search_result(
         "search_cursor": page,
         "url": article_url,
         "wx_sn": extract_wx_sn(article_url) or "",
+        "biz": extract_biz(article_url) or "",
         "title": article.get("title", ""),
         "cover_url": article.get("cover_url", ""),
         "publish_time": publish_time,
@@ -86,11 +100,10 @@ def parse_detail(detail: Dict) -> Dict:
         "video_url_list": detail.get("video_url_list") or [],
         "is_cache": 1 if detail.get("is_cache") else 0,
         "extra": {
-            "biz_info": detail.get("biz_info"),
             "item_index": detail.get("item_index"),
             "channel_account_name": detail.get("channel_account_name"),
         },
     }
 
 
-__all__ = ["extract_wx_sn", "is_valid_keyword", "parse_search_result", "parse_detail"]
+__all__ = ["extract_wx_sn", "extract_biz", "is_valid_keyword", "parse_search_result", "parse_detail"]

src/domains/demand_search_article/article_fetch_detail.py

@@ -31,7 +31,6 @@ class ArticleFetchDetail(DemandSearchArticleConst):
 
     async def deal(
         self,
-        experiment_id: str,
         *,
         limit: int = 500,
     ) -> dict:
@@ -40,9 +39,9 @@ class ArticleFetchDetail(DemandSearchArticleConst):
         Returns:
             {"pending": int, "success": int, "failed": int, "skipped": int}
         """
-        items = await self.relation_mapper.fetch_pending(experiment_id, limit=limit)
+        items = await self.relation_mapper.fetch_pending(limit=limit)
         if not items:
-            logger.info("无待拉详情项, experiment_id=%s", experiment_id)
+            logger.info("无待拉详情项")
             return {"pending": 0, "success": 0, "failed": 0, "skipped": 0}
 
         ids = [it["id"] for it in items]
@@ -72,8 +71,8 @@ class ArticleFetchDetail(DemandSearchArticleConst):
                 skipped += 1
 
         logger.info(
-            "详情拉取完成: experiment_id=%s, total=%d, success=%d, failed=%d, skipped=%d",
-            experiment_id, len(items), success, failed, skipped,
+            "详情拉取完成: total=%d, success=%d, failed=%d, skipped=%d",
+            len(items), success, failed, skipped,
         )
         return {
             "pending": len(items),

src/handlers/article_fetch_detail.py

@@ -0,0 +1,59 @@
+"""文章详情拉取 XXL-JOB Handler —— relation(status=0) → 微信详情 API → detail 表
+
+XXL-JOB Admin 配置:
+    执行器: long-articles-agentic-src
+    处理器: articleFetchDetail
+
+设计:
+    - Admin 调度的 HTTP 请求秒级返回，实际拉取在后台异步执行
+    - limit 取自 DemandSearchArticleConst.DEAL_LIMIT，全表取 status=0 的行
+    - 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 ArticleFetchDetail, 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
+
+
+async def _do_fetch_detail() -> None:
+    """后台执行: relation(status=0) 取待拉项 → 微信详情 API → detail 表写入"""
+    limit = DemandSearchArticleConst.DEAL_LIMIT
+    logger.info("articleFetchDetail background task started, limit=%d", limit)
+
+    try:
+        mysql = _db_ref.mysql_manager("long_articles")
+        fetcher = ArticleFetchDetail(mysql)
+        result = await fetcher.deal(limit=limit)
+    except Exception:
+        logger.exception("文章详情拉取失败")
+        return
+
+    logger.info(
+        "articleFetchDetail done: pending=%d, success=%d, failed=%d, skipped=%d",
+        result["pending"], result["success"], result["failed"], result["skipped"],
+    )
+
+
+@xxl_job("articleFetchDetail")
+async def article_fetch_detail(param: str) -> dict:
+    """文章详情拉取 handler — 秒级返回，实际工作在后台执行"""
+    asyncio.create_task(_do_fetch_detail())
+    logger.info("articleFetchDetail accepted (background task spawned)")
+    return {"code": 200, "msg": "accepted"}

src/server/app.py

@@ -14,6 +14,7 @@ from src.handlers import discover_and_register
 from src.server.xxl_endpoints import set_xxl_executor
 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
+from src.handlers.article_fetch_detail import set_db as set_db_article_fetch_detail
 
 logger = logging.getLogger(__name__)
 
@@ -46,6 +47,7 @@ def create_app(global_config: GlobalConfig = None) -> Quart:
     set_xxl_executor(xxl_executor)
     set_db_demand_enqueue(db)
     set_db_article_search(db)
+    set_db_article_fetch_detail(db)
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")