init

src/domains/demand_search_article/_const.py

@@ -57,8 +57,6 @@ class DemandSearchArticleConst:
     # ═══════════════════════════════════════════════════════════════
 
     MAX_SEARCH_PAGES = 3        # 每个搜索词最多翻多少页
-    SEARCH_PAGE_SIZE = 10       # 每页结果数（微信 API 默认）
-    BATCH_INSERT_SIZE = 100     # 批量写入每批行数
     CACHE_TTL_DAYS = 10         # 搜索缓存有效期（天）
     SEARCH_INTERVAL_SEC = 5     # 搜索页间延迟（秒）
     DETAIL_INTERVAL_SEC = 3     # 详情拉取条间延迟（秒）
@@ -67,7 +65,8 @@ class DemandSearchArticleConst:
     # XXL-JOB 参数
     # ═══════════════════════════════════════════════════════════════
 
-    DEAL_LIMIT = 2000           # 单次搜索任务取队列条数上限
+    SEARCH_DEAL_LIMIT = 2000    # Phase 1 搜索：单次取队列条数上限（批量写入，可以大）
+    DETAIL_DEAL_LIMIT = 100     # Phase 2 详情：单次拉取条数上限（串行 3s/条，小批次高频）
 
 
 __all__ = ["DemandSearchArticleConst"]

src/domains/demand_search_article/_mapper.py

@@ -44,41 +44,42 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         return await self.pool.fetch(query=query, params=params)
 
     async def queue_mark_processing(self, ids: List[int]) -> int:
-        """批量标记队列项为处理中"""
+        """INIT → PROCESSING，CAS 防重复加锁"""
         if not ids:
             return 0
         placeholders = ",".join(["%s"] * len(ids))
         query = f"""
             UPDATE {self.QUEUE_TABLE}
             SET status = %s
-            WHERE id IN ({placeholders})
+            WHERE id IN ({placeholders}) AND status = %s
         """
         return await self.pool.save(
             query=query,
-            params=(self.QueueStatus.PROCESSING, *ids),
+            params=(self.QueueStatus.PROCESSING, *ids, self.QueueStatus.INIT),
         )
 
     async def queue_mark_done(self, item_id: int, remark: str = "") -> int:
-        """标记队列项为成功"""
+        """PROCESSING → SUCCESS，CAS 防误改已完成项"""
         query = f"""
             UPDATE {self.QUEUE_TABLE}
             SET status = %s, remark = %s
-            WHERE id = %s
+            WHERE id = %s AND status = %s
         """
         return await self.pool.save(
-            query=query, params=(self.QueueStatus.SUCCESS, remark, item_id),
+            query=query,
+            params=(self.QueueStatus.SUCCESS, remark, item_id, self.QueueStatus.PROCESSING),
         )
 
     async def queue_mark_failed(self, item_id: int, reason: str = "") -> int:
-        """标记队列项为失败"""
+        """PROCESSING → FAIL，CAS 防误改已完成项"""
         query = f"""
             UPDATE {self.QUEUE_TABLE}
             SET status = %s, remark = %s
-            WHERE id = %s
+            WHERE id = %s AND status = %s
         """
         return await self.pool.save(
             query=query,
-            params=(self.QueueStatus.FAIL, reason[:512], item_id),
+            params=(self.QueueStatus.FAIL, reason[:512], item_id, self.QueueStatus.PROCESSING),
         )
 
     # ═══════════════════════════════════════════════════════════════
@@ -132,6 +133,7 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
             SELECT *
             FROM {self.RELATION_TABLE}
             WHERE status = %s
+            ORDER BY id ASC
             LIMIT %s
         """
         params = (self.SearchStatus.INIT, limit)
@@ -139,16 +141,26 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
 
     async def count_by_status(
         self,
-        experiment_id: str,
         status: int,
+        *,
+        experiment_id: str | None = None,
     ) -> int:
-        """统计指定状态的条目数"""
-        query = f"""
-            SELECT COUNT(*) AS cnt
-            FROM {self.RELATION_TABLE}
-            WHERE experiment_id = %s AND status = %s
-        """
-        row = await self.pool.fetch_one(query=query, params=(experiment_id, status))
+        """统计指定状态的条目数。experiment_id=None 则不按实验过滤"""
+        if experiment_id is not None:
+            query = f"""
+                SELECT COUNT(*) AS cnt
+                FROM {self.RELATION_TABLE}
+                WHERE experiment_id = %s AND status = %s
+            """
+            params = (experiment_id, status)
+        else:
+            query = f"""
+                SELECT COUNT(*) AS cnt
+                FROM {self.RELATION_TABLE}
+                WHERE status = %s
+            """
+            params = (status,)
+        row = await self.pool.fetch_one(query=query, params=params)
         return row["cnt"] if row else 0
 
     # ═══════════════════════════════════════════════════════════════
@@ -183,18 +195,18 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         )
 
     async def mark_failed(self, ids: List[int]) -> int:
-        """批量标记为失败 (→99)"""
+        """PROCESSING → FAIL，CAS 防覆盖 SUCCESS 状态"""
         if not ids:
             return 0
         placeholders = ",".join(["%s"] * len(ids))
         query = f"""
             UPDATE {self.RELATION_TABLE}
             SET status = %s
-            WHERE id IN ({placeholders})
+            WHERE id IN ({placeholders}) AND status = %s
         """
         return await self.pool.save(
             query=query,
-            params=(self.SearchStatus.FAIL, *ids),
+            params=(self.SearchStatus.FAIL, *ids, self.SearchStatus.PROCESSING),
         )
 
 

src/domains/demand_search_article/article_fetch_detail.py

@@ -37,24 +37,21 @@ class ArticleFetchDetail(DemandSearchArticleConst):
         """串行处理待拉详情项
 
         Returns:
-            {"pending": int, "success": int, "failed": int, "skipped": int}
+            {"pending": int, "success": int, "failed": int}
         """
         items = await self.relation_mapper.fetch_pending(limit=limit)
         if not items:
             logger.info("无待拉详情项")
-            return {"pending": 0, "success": 0, "failed": 0, "skipped": 0}
-
-        ids = [it["id"] for it in items]
-        await self.relation_mapper.mark_processing(ids)
+            return {"pending": 0, "success": 0, "failed": 0}
 
         success = 0
         failed = 0
-        skipped = 0
         for item in items:
+            await self.relation_mapper.mark_processing([item["id"]])
             if success + failed > 0:  # 非首条，等待后执行
                 await asyncio.sleep(self.DETAIL_INTERVAL_SEC)
             try:
-                outcome, _ = await self._fetch_detail_and_save(item)
+                outcome = await self._fetch_detail_and_save(item)
             except Exception:
                 logger.exception(
                     "详情拉取异常: relation_id=%s", item["id"],
@@ -65,37 +62,34 @@ class ArticleFetchDetail(DemandSearchArticleConst):
 
             if outcome == "success":
                 success += 1
-            elif outcome == "failed":
-                failed += 1
             else:
-                skipped += 1
+                failed += 1
 
         logger.info(
-            "详情拉取完成: total=%d, success=%d, failed=%d, skipped=%d",
-            len(items), success, failed, skipped,
+            "详情拉取完成: total=%d, success=%d, failed=%d",
+            len(items), success, failed,
         )
         return {
             "pending": len(items),
             "success": success,
             "failed": failed,
-            "skipped": skipped,
         }
 
     # ═══════════════════════════════════════════════════════════════
     # 单条处理
     # ═══════════════════════════════════════════════════════════════
 
-    async def _fetch_detail_and_save(self, item: Dict) -> tuple[str, int | None]:
+    async def _fetch_detail_and_save(self, item: Dict) -> str:
         """拉取详情 → 写 detail 表 → 更新 relation 状态
 
         Returns:
-            ("success"/"failed"/"skipped", relation_id)
+            "success" 或 "failed"
         """
         url = item.get("url", "")
         if not url:
             logger.warning("空 URL, 跳过 relation_id=%s", item["id"])
             await self.relation_mapper.mark_failed([item["id"]])
-            return ("failed", item["id"])
+            return "failed"
 
         try:
             resp = await get_article_detail(url)
@@ -104,7 +98,7 @@ class ArticleFetchDetail(DemandSearchArticleConst):
                 "get_article_detail 调用异常: url=%s, relation_id=%s", url, item["id"],
             )
             await self.relation_mapper.mark_failed([item["id"]])
-            return ("failed", item["id"])
+            return "failed"
 
         if not resp or resp.get("code") != 0:
             logger.warning(
@@ -112,7 +106,7 @@ class ArticleFetchDetail(DemandSearchArticleConst):
                 url, item["id"], resp.get("code") if resp else "None",
             )
             await self.relation_mapper.mark_failed([item["id"]])
-            return ("failed", item["id"])
+            return "failed"
 
         detail_data = (resp.get("data") or {}).get("data")
         if not detail_data:
@@ -120,7 +114,7 @@ class ArticleFetchDetail(DemandSearchArticleConst):
                 "详情数据为空: url=%s, relation_id=%s", url, item["id"],
             )
             await self.relation_mapper.mark_failed([item["id"]])
-            return ("failed", item["id"])
+            return "failed"
 
         channel_content_id = detail_data.get("channel_content_id", "")
         if not channel_content_id:
@@ -128,7 +122,7 @@ class ArticleFetchDetail(DemandSearchArticleConst):
                 "channel_content_id 为空: url=%s, relation_id=%s", url, item["id"],
             )
             await self.relation_mapper.mark_failed([item["id"]])
-            return ("failed", item["id"])
+            return "failed"
 
         detail_row = parse_detail(detail_data)
         await self.detail_mapper.insert_one(detail_row)
@@ -140,7 +134,7 @@ class ArticleFetchDetail(DemandSearchArticleConst):
             "详情拉取成功: relation_id=%s, channel_content_id=%s, title=%s",
             item["id"], channel_content_id, detail_row.get("title", "")[:40],
         )
-        return ("success", item["id"])
+        return "success"
 
 
 __all__ = ["ArticleFetchDetail"]

src/domains/demand_search_article/article_search.py

@@ -45,12 +45,10 @@ class DemandSearchArticle(DemandSearchArticleConst):
             logger.info("无待处理队列项")
             return {"queued": 0, "processed": 0, "results_written": 0}
 
-        ids = [it["id"] for it in items]
-        await self.relation_mapper.queue_mark_processing(ids)
-
         total_written = 0
         processed = 0
         for item in items:
+            await self.relation_mapper.queue_mark_processing([item["id"]])
             try:
                 n, from_cache_only = await self._search_and_write(item)
             except Exception as e:

src/handlers/article_fetch_detail.py

@@ -6,7 +6,7 @@ XXL-JOB Admin 配置:
 
 设计:
     - Admin 调度的 HTTP 请求秒级返回，实际拉取在后台异步执行
-    - limit 取自 DemandSearchArticleConst.DEAL_LIMIT，全表取 status=0 的行
+    - limit 取自 DemandSearchArticleConst.DETAIL_DEAL_LIMIT，全表取 status=0 的行
     - MySQL 复用 app 级别的连接池（set_db() 注入），不每次自建
 """
 
@@ -34,7 +34,7 @@ def set_db(db: "DatabaseManager") -> None:
 
 async def _do_fetch_detail() -> None:
     """后台执行: relation(status=0) 取待拉项 → 微信详情 API → detail 表写入"""
-    limit = DemandSearchArticleConst.DEAL_LIMIT
+    limit = DemandSearchArticleConst.DETAIL_DEAL_LIMIT
     logger.info("articleFetchDetail background task started, limit=%d", limit)
 
     try:
@@ -46,8 +46,8 @@ async def _do_fetch_detail() -> None:
         return
 
     logger.info(
-        "articleFetchDetail done: pending=%d, success=%d, failed=%d, skipped=%d",
-        result["pending"], result["success"], result["failed"], result["skipped"],
+        "articleFetchDetail done: pending=%d, success=%d, failed=%d",
+        result["pending"], result["success"], result["failed"],
     )
 
 

src/handlers/article_search.py

@@ -7,7 +7,7 @@ XXL-JOB Admin 配置:
 
 设计:
     - Admin 调度的 HTTP 请求秒级返回，实际搜索在后台异步执行
-    - limit 取自 DemandSearchArticleConst.DEAL_LIMIT，按 id ASC FIFO
+    - limit 取自 DemandSearchArticleConst.SEARCH_DEAL_LIMIT，按 id ASC FIFO
     - MySQL 复用 app 级别的连接池（set_db() 注入），不每次自建
 """
 
@@ -45,7 +45,7 @@ def _parse_dt(param: str) -> str | None:
 
 async def _do_search(dt: str | None) -> None:
     """后台执行: 队列取搜索词 → 微信搜索 → relation 写入"""
-    limit = DemandSearchArticleConst.DEAL_LIMIT
+    limit = DemandSearchArticleConst.SEARCH_DEAL_LIMIT
     logger.info("articleSearch background task started, dt=%s, limit=%d", dt, limit)
 
     try: