优化 xxljob 参数配置

优化 limit 透传

src/config/apollo.py

@@ -7,14 +7,14 @@ class ApolloConfig(BaseSettings):
     """Apollo 配置中心配置"""
 
     app_id: str = Field(
-        default="long-articles-agentic-supply", description="Apollo 应用 ID"
+        default="long-articles-agentic-content_supply", description="Apollo 应用 ID"
     )
     env: str = Field(default="prod", description="Apollo 环境: dev/pre/prod")
 
     # apollo 配置中心地址映射
     apollo_map: Dict[str, Dict[str, str]] = Field(
         default_factory=lambda: {
-            "long-articles-agentic-supply": {
+            "long-articles-agentic-content_supply": {
                 "pre": "http://preapolloconfig-internal.piaoquantv.com/",
                 "dev": "https://devapolloconfig-internal.piaoquantv.com/",
                 "prod": "https://apolloconfig-internal.piaoquantv.com/",

src/domains/__init__.py

@@ -5,32 +5,36 @@
   domains: 本项目专属的业务逻辑和数据模型，依赖 core 和 infra
 
 子域:
-  fetch_demand   — 需求获取与管理（外部需求源 → 统一需求模型）
-  demand_search_article  — 文章抓取与处理（需求驱动 → 文章采集 → 内容结构化）
-  demand_search_account  — 账号识别与管理（文章反推账号 → 账号画像 → 采集策略）
+  demand_enqueue_task — 需求获取与管理（外部需求源 → 统一需求模型）
+  content_supply      — 供给域 :: 需求驱动的文章搜索、账号识别与内容供给（Phase 1-5）
 
-数据流: fetch_demand → demand_search_article → demand_search_account（下游消费上游产出）
+数据流: demand_enqueue_task → content_supply
 """
 
-from src.domains.fetch_demand import (
+from src.domains.demand_enqueue_task import (
     FetchDemands,
     EnqueueDemands,
     DemandQueueMapper,
     VideoDeconstructMapper,
 )
-from src.domains.demand_search_article import DemandSearchArticle, ArticleFetchDetail
-from src.domains.demand_search_account import AccountFetch, AccountArticleFetch
+from src.domains.content_supply import (
+    DemandSearchArticle,
+    ArticleFetchDetail,
+    AccountFetch,
+    AccountArticleFetch,
+    AccountArticleUpdate,
+)
 
 __all__ = [
-    # fetch_demand
+    # demand_enqueue_task
     "FetchDemands",
     "EnqueueDemands",
     "DemandQueueMapper",
     "VideoDeconstructMapper",
-    # demand_search_article
+    # content_supply
     "DemandSearchArticle",
     "ArticleFetchDetail",
-    # demand_search_account
     "AccountFetch",
     "AccountArticleFetch",
+    "AccountArticleUpdate",
 ]

src/domains/content_supply/__init__.py

@@ -0,0 +1,40 @@
+"""供给领域 —— 需求驱动的文章搜索、账号识别与内容供给
+
+Phase 1-2 文章搜索与详情:
+  DemandSearchArticle — 从队列取搜索词 → 微信搜索 → relation 表
+  ArticleFetchDetail  — relation → 微信详情 API → detail 表
+
+Phase 3-5 账号识别与文章供给:
+  AccountFetch         — relation biz → 账号 API → accounts 表
+  AccountArticleFetch  — accounts gh_id → 文章列表 API → articles 表（深翻/增量）
+  AccountArticleUpdate — crawl_deep_done=1 → 第一页 → UPDATE 指标
+
+数据流: demand_enqueue_task → content_supply
+"""
+
+from src.domains.content_supply._const import ContentSupplyConst
+from src.domains.content_supply._mapper import (
+    ArticleSearchRelationMapper,
+    ArticleDetailMapper,
+    AccountMapper,
+)
+from src.domains.content_supply.article_search import DemandSearchArticle
+from src.domains.content_supply.article_fetch_detail import ArticleFetchDetail
+from src.domains.content_supply.account_fetch import AccountFetch
+from src.domains.content_supply.account_article_fetch import AccountArticleFetch
+from src.domains.content_supply.account_article_update import AccountArticleUpdate
+
+__all__ = [
+    # const
+    "ContentSupplyConst",
+    # mapper
+    "ArticleSearchRelationMapper",
+    "ArticleDetailMapper",
+    "AccountMapper",
+    # service
+    "DemandSearchArticle",
+    "ArticleFetchDetail",
+    "AccountFetch",
+    "AccountArticleFetch",
+    "AccountArticleUpdate",
+]

src/domains/demand_search_article/_const.py → src/domains/content_supply/_const.py

@@ -1,8 +1,8 @@
-"""文章域常量"""
+"""供给域常量 —— 文章搜索 + 账号管理的统一常量聚合根"""
 
 
-class DemandSearchArticleConst:
-    """文章域所有常量的聚合根，服务类继承此类即可通过 self. 访问"""
+class ContentSupplyConst:
+    """供给域所有常量的聚合根，服务类继承此类即可通过 self. 访问"""
 
     # ═══════════════════════════════════════════════════════════════
     # 表名
@@ -12,9 +12,11 @@ class DemandSearchArticleConst:
     RELATION_TABLE = "demand_search_article_relation"
     DETAIL_TABLE = "demand_search_article_detail"
     CACHE_TABLE = "demand_search_cache"
+    ACCOUNT_TABLE = "demand_search_accounts"
+    ARTICLES_TABLE = "demand_account_articles"
 
     # ═══════════════════════════════════════════════════════════════
-    # 队列状态
+    # 队列状态（search_article Phase 1）
     # ═══════════════════════════════════════════════════════════════
 
     class QueueStatus:
@@ -43,14 +45,24 @@ class DemandSearchArticleConst:
             )
 
     # ═══════════════════════════════════════════════════════════════
-    # 搜索状态: 0→1→2 / 0→1→99
+    # 搜索状态: 0→1→2 / 0→1→99（search_article Phase 2）
     # ═══════════════════════════════════════════════════════════════
 
     class SearchStatus:
-        INIT = 0  # 待拉取详情
-        PROCESSING = 1  # 拉取详情中
-        SUCCESS = 2  # 拉取详情成功
-        FAIL = 99  # 详情拉取失败
+        INIT = 0
+        PROCESSING = 1
+        SUCCESS = 2
+        FAIL = 99
+
+    # ═══════════════════════════════════════════════════════════════
+    # 账号保存状态（Phase 3）
+    # ═══════════════════════════════════════════════════════════════
+
+    class AccountSaveStatus:
+        INIT = 0
+        PROCESSING = 1
+        SUCCESS = 2
+        FAIL = 99
 
     # ═══════════════════════════════════════════════════════════════
     # 搜索参数
@@ -61,12 +73,28 @@ class DemandSearchArticleConst:
     SEARCH_INTERVAL_SEC = 15  # 搜索页间延迟（秒）
     DETAIL_INTERVAL_SEC = 3  # 详情拉取条间延迟（秒）
 
+    # ═══════════════════════════════════════════════════════════════
+    # 账号参数
+    # ═══════════════════════════════════════════════════════════════
+
+    ACCOUNT_DEAL_LIMIT = 100  # 单次取待抓账号条数
+    ACCOUNT_INTERVAL_SEC = 1  # API 调用间隔（秒）
+
+    # ═══════════════════════════════════════════════════════════════
+    # 文章抓取参数
+    # ═══════════════════════════════════════════════════════════════
+
+    ARTICLE_DEAL_LIMIT = 10  # 单次处理账号数
+    ARTICLE_PAGE_INTERVAL_SEC = 2  # 翻页间隔（秒）
+    ARTICLE_BATCH_INSERT = 200  # 文章批量写入每批条数
+    ONE_YEAR_SECONDS = 365 * 24 * 3600
+
     # ═══════════════════════════════════════════════════════════════
     # XXL-JOB 参数
     # ═══════════════════════════════════════════════════════════════
 
-    SEARCH_DEAL_LIMIT = 2000  # Phase 1 搜索：单次取队列条数上限（批量写入，可以大）
-    DETAIL_DEAL_LIMIT = 100  # Phase 2 详情：单次拉取条数上限（串行 3s/条，小批次高频）
+    SEARCH_DEAL_LIMIT = 2000  # Phase 1 搜索：单次取队列条数上限
+    DETAIL_DEAL_LIMIT = 100  # Phase 2 详情：单次拉取条数上限
 
 
-__all__ = ["DemandSearchArticleConst"]
+__all__ = ["ContentSupplyConst"]

src/domains/demand_search_article/_mapper.py → src/domains/content_supply/_mapper.py

@@ -1,15 +1,15 @@
-"""文章域 DB 读写 —— demand_search_article_relation + demand_search_article_detail"""
+"""供给域 DB 读写 —— relation + detail + queue + accounts + articles 跨表访问"""
 
 import json
 from typing import Dict, List, Optional
 
 from src.infra.database.mysql.manager import MysqlManager
 
-from ._const import DemandSearchArticleConst
+from ._const import ContentSupplyConst
 
 
-class ArticleSearchRelationMapper(DemandSearchArticleConst):
-    """demand_search_article_relation + demand_search_queue 的读写"""
+class ArticleSearchRelationMapper(ContentSupplyConst):
+    """demand_search_article_relation + demand_search_queue + demand_search_cache 的读写"""
 
     def __init__(self, pool: MysqlManager):
         self.pool = pool
@@ -132,7 +132,7 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         return await self.pool.save(query=query, params=params, batch=True)
 
     # ═══════════════════════════════════════════════════════════════
-    # 读取
+    # relation 表 — 读取
     # ═══════════════════════════════════════════════════════════════
 
     async def fetch_pending(
@@ -176,7 +176,7 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         return row["cnt"] if row else 0
 
     # ═══════════════════════════════════════════════════════════════
-    # 状态更新
+    # relation 表 — 状态更新
     # ═══════════════════════════════════════════════════════════════
 
     async def mark_processing(self, ids: List[int]) -> int:
@@ -270,16 +270,12 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         )
 
 
-class ArticleDetailMapper(DemandSearchArticleConst):
+class ArticleDetailMapper(ContentSupplyConst):
     """demand_search_article_detail 表的数据访问层"""
 
     def __init__(self, pool: MysqlManager):
         self.pool = pool
 
-    # ═══════════════════════════════════════════════════════════════
-    # 写入
-    # ═══════════════════════════════════════════════════════════════
-
     async def insert_one(self, detail: Dict) -> int:
         """写入单篇详情，channel_content_id 冲突则跳过
 
@@ -325,4 +321,258 @@ class ArticleDetailMapper(DemandSearchArticleConst):
         return await self.pool.save(query=query, params=params)
 
 
-__all__ = ["ArticleSearchRelationMapper", "ArticleDetailMapper"]
+class AccountMapper(ContentSupplyConst):
+    """demand_search_accounts 表 + relation.save_account_status + demand_account_articles 的读写"""
+
+    def __init__(self, pool: MysqlManager):
+        self.pool = pool
+
+    # ═══════════════════════════════════════════════════════════════
+    # relation 表 — 读取待抓账号
+    # ═══════════════════════════════════════════════════════════════
+
+    async def fetch_pending_relations(self, *, limit: int = 100) -> List[Dict]:
+        """取 save_account_status=0 的行（DISTINCT biz，按 id ASC）"""
+        query = f"""
+            SELECT MIN(id) AS id, biz, MIN(url) AS url
+            FROM {self.RELATION_TABLE}
+            WHERE save_account_status = %s AND biz IS NOT NULL AND biz != ''
+            GROUP BY biz
+            ORDER BY id ASC
+            LIMIT %s
+        """
+        return await self.pool.fetch(
+            query=query,
+            params=(self.AccountSaveStatus.INIT, limit),
+        )
+
+    async def fetch_relation_ids_by_biz(self, biz: str) -> List[int]:
+        """取某个 biz 下所有 save_account_status=0 的 relation id"""
+        rows = await self.pool.fetch(
+            query=f"SELECT id FROM {self.RELATION_TABLE} WHERE biz = %s AND save_account_status = %s",
+            params=(biz, self.AccountSaveStatus.INIT),
+        )
+        return [r["id"] for r in rows]
+
+    async def fetch_relation_ids_by_biz_any_status(self, biz: str) -> List[int]:
+        """取某个 biz 下未完成（INIT/PROCESSING）的 relation id，异常恢复用"""
+        rows = await self.pool.fetch(
+            query=f"SELECT id FROM {self.RELATION_TABLE} WHERE biz = %s AND save_account_status IN (%s, %s)",
+            params=(
+                biz,
+                self.AccountSaveStatus.INIT,
+                self.AccountSaveStatus.PROCESSING,
+            ),
+        )
+        return [r["id"] for r in rows]
+
+    # ═══════════════════════════════════════════════════════════════
+    # relation 表 — save_account_status 状态更新
+    # ═══════════════════════════════════════════════════════════════
+
+    async def _mark_batch(
+        self, target_status: int, ids: List[int], from_statuses: List[int]
+    ) -> int:
+        """通用：将 ids 从 from_statuses 任一改为 target_status"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        status_placeholders = ",".join(["%s"] * len(from_statuses))
+        query = f"""
+            UPDATE {self.RELATION_TABLE}
+            SET save_account_status = %s
+            WHERE id IN ({placeholders}) AND save_account_status IN ({status_placeholders})
+        """
+        return await self.pool.save(
+            query=query,
+            params=(target_status, *ids, *from_statuses),
+        )
+
+    async def relation_mark_processing(self, ids: List[int]) -> int:
+        """INIT → PROCESSING，CAS 防重复"""
+        return await self._mark_batch(
+            self.AccountSaveStatus.PROCESSING, ids, [self.AccountSaveStatus.INIT]
+        )
+
+    async def relation_mark_success(self, ids: List[int]) -> int:
+        """PROCESSING → SUCCESS"""
+        return await self._mark_batch(
+            self.AccountSaveStatus.SUCCESS, ids, [self.AccountSaveStatus.PROCESSING]
+        )
+
+    async def relation_mark_success_direct(self, ids: List[int]) -> int:
+        """INIT → SUCCESS，跳过 PROCESSING（查重命中时用）"""
+        return await self._mark_batch(
+            self.AccountSaveStatus.SUCCESS, ids, [self.AccountSaveStatus.INIT]
+        )
+
+    async def relation_mark_failed(self, ids: List[int]) -> int:
+        """INIT/PROCESSING → FAIL，异常恢复用"""
+        return await self._mark_batch(
+            self.AccountSaveStatus.FAIL,
+            ids,
+            [self.AccountSaveStatus.INIT, self.AccountSaveStatus.PROCESSING],
+        )
+
+    # ═══════════════════════════════════════════════════════════════
+    # demand_search_accounts 表
+    # ═══════════════════════════════════════════════════════════════
+
+    async def find_by_biz(self, biz: str) -> Optional[Dict]:
+        """按 biz 查账号是否已存在"""
+        return await self.pool.fetch_one(
+            query=f"SELECT * FROM {self.ACCOUNT_TABLE} WHERE biz = %s LIMIT 1",
+            params=(biz,),
+        )
+
+    async def insert_account(self, account: Dict) -> int:
+        """写入账号，channel_account_id 冲突则跳过"""
+        query = f"""
+            INSERT IGNORE INTO {self.ACCOUNT_TABLE}
+                (channel_account_id, gh_id, account_name, biz, biz_info, description)
+            VALUES (%s, %s, %s, %s, %s, %s)
+        """
+        return await self.pool.save(
+            query=query,
+            params=(
+                account["channel_account_id"],
+                account["gh_id"],
+                account["account_name"],
+                account["biz"],
+                account["biz_info"],
+                account["description"],
+            ),
+        )
+
+    async def list_accounts_for_article_crawl(self, *, limit: int) -> List[Dict]:
+        """列出有 gh_id 的账号，用于文章抓取（crawl_deep_done=0 优先）"""
+        query = f"""
+            SELECT channel_account_id, gh_id, crawl_cursor, crawl_deep_done
+            FROM {self.ACCOUNT_TABLE}
+            WHERE gh_id IS NOT NULL AND gh_id != ''
+            ORDER BY crawl_deep_done ASC, gh_id ASC
+            LIMIT %s
+        """
+        return await self.pool.fetch(query=query, params=(limit,))
+
+    async def update_crawl_status(
+        self,
+        gh_id: str,
+        *,
+        cursor: str = "",
+        deep_done: int = 0,
+    ) -> int:
+        """更新账号的爬取游标和深翻状态"""
+        query = f"""
+            UPDATE {self.ACCOUNT_TABLE}
+            SET crawl_cursor = %s, crawl_deep_done = %s, last_crawl_at = NOW()
+            WHERE gh_id = %s
+        """
+        return await self.pool.save(query=query, params=(cursor, deep_done, gh_id))
+
+    # ═══════════════════════════════════════════════════════════════
+    # demand_account_articles 表
+    # ═══════════════════════════════════════════════════════════════
+
+    async def insert_articles_batch(self, articles: List[Dict]) -> int:
+        """批量写入文章，wx_sn 冲突则跳过"""
+        if not articles:
+            return 0
+        query = f"""
+            INSERT IGNORE INTO {self.ARTICLES_TABLE}
+                (gh_id, app_msg_id, item_index, title, content_url, wx_sn,
+                 digest, is_original, item_show_type, source_url, cover_img_url, show_desc,
+                 show_view_count, show_like_count, show_pay_count, show_zs_count,
+                 type, send_time, create_time, update_time, date_time, base_info)
+            VALUES
+                (%s, %s, %s, %s, %s, %s,
+                 %s, %s, %s, %s, %s, %s,
+                 %s, %s, %s, %s,
+                 %s, %s, %s, %s, %s, %s)
+        """
+        params = [
+            (
+                a["gh_id"],
+                a["app_msg_id"],
+                a["item_index"],
+                a["title"],
+                a["content_url"],
+                a["wx_sn"],
+                a["digest"],
+                a["is_original"],
+                a["item_show_type"],
+                a["source_url"],
+                a["cover_img_url"],
+                a["show_desc"],
+                a["show_view_count"],
+                a["show_like_count"],
+                a["show_pay_count"],
+                a["show_zs_count"],
+                a["type"],
+                a["send_time"],
+                a["create_time"],
+                a["update_time"],
+                a["date_time"],
+                json.dumps(a["base_info"], ensure_ascii=False),
+            )
+            for a in articles
+        ]
+        return await self.pool.save(query=query, params=params, batch=True)
+
+    async def count_articles_by_gh_id(self, gh_id: str) -> int:
+        """统计某账号已抓文章数"""
+        row = await self.pool.fetch_one(
+            query=f"SELECT COUNT(*) AS cnt FROM {self.ARTICLES_TABLE} WHERE gh_id = %s",
+            params=(gh_id,),
+        )
+        return row["cnt"] if row else 0
+
+    # ═══════════════════════════════════════════════════════════════
+    # demand_account_articles 表 — 指标更新
+    # ═══════════════════════════════════════════════════════════════
+
+    async def list_accounts_for_article_update(self, *, limit: int) -> List[Dict]:
+        """列出已完成深翻的账号，按最久未更新排序，用于定期刷新文章指标"""
+        query = f"""
+            SELECT channel_account_id, gh_id, last_crawl_at
+            FROM {self.ACCOUNT_TABLE}
+            WHERE gh_id IS NOT NULL AND gh_id != '' AND crawl_deep_done = 1
+            ORDER BY last_crawl_at ASC
+            LIMIT %s
+        """
+        return await self.pool.fetch(query=query, params=(limit,))
+
+    async def update_article_metrics_batch(self, updates: List[Dict]) -> int:
+        """批量更新文章数值指标，按 wx_sn 匹配"""
+        if not updates:
+            return 0
+        query = f"""
+            UPDATE {self.ARTICLES_TABLE}
+            SET show_view_count = %s, show_like_count = %s, show_pay_count = %s,
+                show_zs_count = %s, show_desc = %s
+            WHERE wx_sn = %s
+        """
+        params = [
+            (
+                u["show_view_count"],
+                u["show_like_count"],
+                u["show_pay_count"],
+                u["show_zs_count"],
+                u["show_desc"],
+                u["wx_sn"],
+            )
+            for u in updates
+        ]
+        return await self.pool.save(query=query, params=params, batch=True)
+
+    async def touch_last_crawl_at(self, gh_id: str) -> int:
+        """更新 last_crawl_at = NOW()，不改变 cursor 和 deep_done 状态"""
+        query = f"""
+            UPDATE {self.ACCOUNT_TABLE}
+            SET last_crawl_at = NOW()
+            WHERE gh_id = %s
+        """
+        return await self.pool.save(query=query, params=(gh_id,))
+
+
+__all__ = ["ArticleSearchRelationMapper", "ArticleDetailMapper", "AccountMapper"]

src/domains/content_supply/_utils.py

@@ -0,0 +1,338 @@
+"""供给域工具函数 —— API 响应解析、数据转换、搜索/账号/文章通用工具"""
+
+import logging
+import re
+from datetime import datetime
+from typing import Dict, List, Tuple
+from urllib.parse import parse_qs, urlparse
+
+from src.infra.spider.wechat.gzh import get_article_list_from_account
+
+logger = logging.getLogger(__name__)
+
+
+# ═══════════════════════════════════════════════════════════════
+# URL 参数提取（通用）
+# ═══════════════════════════════════════════════════════════════
+
+
+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。
+    """
+    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")
+
+
+# ═══════════════════════════════════════════════════════════════
+# 搜索词校验
+# ═══════════════════════════════════════════════════════════════
+
+_INVALID_KEYWORDS = frozenset({"-", "--", "---", "/", ".", "。。。"})
+
+
+def is_valid_keyword(keyword: str) -> bool:
+    """搜索词是否有效：非空、非纯占位符"""
+    kw = keyword.strip()
+    if not kw:
+        return False
+    if kw in _INVALID_KEYWORDS:
+        return False
+    return True
+
+
+# ═══════════════════════════════════════════════════════════════
+# 搜索结果解析（Phase 1）
+# ═══════════════════════════════════════════════════════════════
+
+
+def parse_search_result(
+    article: Dict,
+    keyword: str,
+    key_type: str,
+    experiment_id: str,
+    page: int,
+) -> Dict:
+    """微信搜索 API 返回的单篇文章 → relation 表 INSERT dict"""
+    ts = article.get("time")
+    publish_time = datetime.fromtimestamp(ts) if ts and ts > 0 else None
+    article_url = article.get("url", "")
+    return {
+        "search_key": keyword,
+        "key_type": key_type,
+        "experiment_id": experiment_id,
+        "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,
+    }
+
+
+# ═══════════════════════════════════════════════════════════════
+# 详情解析（Phase 2）
+# ═══════════════════════════════════════════════════════════════
+
+
+def parse_detail(detail: Dict) -> Dict:
+    """微信详情 API 返回的 data.data → detail 表 INSERT dict"""
+    ts = detail.get("publish_timestamp")
+    publish_at = datetime.fromtimestamp(ts / 1000) if ts and ts > 0 else None
+    content_url = detail.get("content_link", "")
+    wx_sn = detail.get("wx_sn", "") or extract_wx_sn(content_url) or ""
+    return {
+        "channel_content_id": detail.get("channel_content_id", ""),
+        "wx_sn": wx_sn,
+        "url": content_url,
+        "title": detail.get("title", ""),
+        "body_text": detail.get("body_text", ""),
+        "content_type": detail.get("content_type", ""),
+        "channel_account_id": detail.get("channel_account_id", ""),
+        "publish_timestamp": ts or 0,
+        "publish_at": publish_at,
+        "is_original": 1 if detail.get("is_original") else 0,
+        "view_count": detail.get("view_count") or 0,
+        "like_count": detail.get("like_count") or 0,
+        "looking_count": detail.get("looking_count") or 0,
+        "comment_count": detail.get("comment_count") or 0,
+        "share_count": detail.get("share_count") or 0,
+        "image_url_list": detail.get("image_url_list") or [],
+        "video_url_list": detail.get("video_url_list") or [],
+        "is_cache": 1 if detail.get("is_cache") else 0,
+        "extra": {
+            "item_index": detail.get("item_index"),
+            "channel_account_name": detail.get("channel_account_name"),
+        },
+    }
+
+
+# ═══════════════════════════════════════════════════════════════
+# 账号解析（Phase 3）
+# ═══════════════════════════════════════════════════════════════
+
+
+def parse_account_response(api_data: Dict, biz: str = "") -> Dict:
+    """get_account_from_url 返回的 data.data → accounts 表 INSERT dict"""
+    return {
+        "channel_account_id": api_data.get("channel_account_id", ""),
+        "gh_id": api_data.get("wx_gh", "") or "",
+        "account_name": api_data.get("account_name", "") or "",
+        "biz": biz or api_data.get("biz_info", "") or "",
+        "biz_info": api_data.get("biz_info", "") or "",
+        "description": api_data.get("description", "") or "",
+    }
+
+
+# ═══════════════════════════════════════════════════════════════
+# ShowDesc 解析：阅读/赞/付费/赞赏 → 数字
+# ═══════════════════════════════════════════════════════════════
+
+
+def show_desc_to_sta(show_desc: str) -> Dict[str, int]:
+    """解析 ShowDesc 中的阅读量、点赞量等
+
+    ShowDesc 格式示例:
+        "阅读 80  赞 3  "
+        "阅读 1  "
+        "阅读 121  赞 5  "
+
+    分隔符: \\u2006 分隔 key-value, \\u2004/\\u2005 分隔 group。
+    """
+
+    def _decode_value(raw: str) -> int:
+        if not raw:
+            return 0
+        raw = raw.strip().lower().replace(",", ".")
+        m = re.search(r"(\d+(?:\.\d+)?)([a-z一-龥]*)", raw)
+        if not m:
+            return 0
+        num = float(m.group(1))
+        unit = m.group(2)
+        if "亿" in unit:
+            num *= 1e8
+        elif "万" in unit:
+            num *= 1e4
+        elif "千" in unit:
+            num *= 1e3
+        elif unit.startswith("k"):
+            num *= 1e3
+        elif unit.startswith("m"):
+            num *= 1e6
+        elif unit.startswith("b"):
+            num *= 1e9
+        return int(num)
+
+    def _decode_key(raw: str) -> str:
+        if not raw:
+            return "show_unknown"
+        raw = raw.strip().lower()
+        mapping = {
+            "阅读": "show_view_count",
+            "看过": "show_view_count",
+            "观看": "show_view_count",
+            "reads": "show_view_count",
+            "views": "show_view_count",
+            "view": "show_view_count",
+            "赞": "show_like_count",
+            "点赞": "show_like_count",
+            "likes": "show_like_count",
+            "like": "show_like_count",
+            "付费": "show_pay_count",
+            "payments": "show_pay_count",
+            "paid": "show_pay_count",
+            "赞赏": "show_zs_count",
+        }
+        return mapping.get(raw, "show_unknown")
+
+    if not show_desc:
+        return {
+            "show_view_count": 0,
+            "show_like_count": 0,
+            "show_pay_count": 0,
+            "show_zs_count": 0,
+        }
+
+    show_desc = show_desc.replace("+", "")
+    sta: Dict[str, int] = {}
+    groups = re.split(r"[  ]+", show_desc)
+
+    for group in groups:
+        group = group.strip()
+        if not group:
+            continue
+        parts = group.split(" ")
+        if len(parts) != 2:
+            continue
+        a, b = parts
+        if re.search(r"\d", a):
+            v_raw, k_raw = a, b
+        else:
+            k_raw, v_raw = a, b
+        key = _decode_key(k_raw)
+        val = _decode_value(v_raw)
+        if key != "show_unknown":
+            sta[key] = val
+
+    return {
+        "show_view_count": sta.get("show_view_count", 0),
+        "show_like_count": sta.get("show_like_count", 0),
+        "show_pay_count": sta.get("show_pay_count", 0),
+        "show_zs_count": sta.get("show_zs_count", 0),
+    }
+
+
+# ═══════════════════════════════════════════════════════════════
+# 文章列表响应解析
+# ═══════════════════════════════════════════════════════════════
+
+
+def parse_article_list_response(data_list: List[Dict], gh_id: str) -> List[Dict]:
+    """get_article_list_from_account 返回的 data.data[] → articles 表 INSERT dict 列表"""
+    rows = []
+    for group in data_list or []:
+        app_msg = group.get("AppMsg") or {}
+        base_info = app_msg.get("BaseInfo") or {}
+        detail_list = app_msg.get("DetailInfo") or []
+        top_base = group.get("BaseInfo") or {}
+
+        for detail in detail_list:
+            url = detail.get("ContentUrl", "") or ""
+            show_desc = detail.get("ShowDesc", "") or ""
+            counts = show_desc_to_sta(show_desc)
+
+            rows.append(
+                {
+                    "gh_id": gh_id,
+                    "app_msg_id": base_info.get("AppMsgId", 0),
+                    "item_index": detail.get("ItemIndex", 0),
+                    "title": detail.get("Title", "") or "",
+                    "content_url": url,
+                    "wx_sn": extract_wx_sn(url) or "",
+                    "digest": detail.get("Digest", "") or "",
+                    "is_original": 1 if detail.get("IsOriginal") else 0,
+                    "item_show_type": detail.get("ItemShowType", 0) or 0,
+                    "source_url": detail.get("SourceUrl", "") or "",
+                    "cover_img_url": detail.get("CoverImgUrl", "") or "",
+                    "show_desc": show_desc,
+                    "show_view_count": counts["show_view_count"],
+                    "show_like_count": counts["show_like_count"],
+                    "show_pay_count": counts["show_pay_count"],
+                    "show_zs_count": counts["show_zs_count"],
+                    "type": base_info.get("Type", 0) or 0,
+                    "send_time": detail.get("send_time", 0) or 0,
+                    "create_time": base_info.get("CreateTime", 0) or 0,
+                    "update_time": base_info.get("UpdateTime", 0) or 0,
+                    "date_time": top_base.get("DateTime", 0) or 0,
+                    "base_info": top_base,
+                }
+            )
+    return rows
+
+
+# ═══════════════════════════════════════════════════════════════
+# API 调用封装
+# ═══════════════════════════════════════════════════════════════
+
+
+async def fetch_article_page(
+    gh_id: str,
+    cursor: str | None,
+) -> Tuple[List[Dict], str | None, bool, int | None]:
+    """调用 get_article_list_from_account 获取一页文章
+
+    Returns:
+        (articles, next_cursor, has_more, oldest_send_time)
+    """
+    resp = await get_article_list_from_account(gh_id, index=cursor)
+    if not resp or resp.get("code") != 0:
+        logger.warning(
+            "get_article_list_from_account 返回异常: gh_id=%s, cursor=%s, code=%s",
+            gh_id,
+            cursor,
+            resp.get("code") if resp else "None",
+        )
+        return [], None, False, None
+
+    data_wrapper = resp.get("data") or {}
+    data_list = data_wrapper.get("data") or []
+    has_more = data_wrapper.get("has_more", False)
+    next_cursor = data_wrapper.get("next_cursor")
+
+    if not data_list:
+        return [], None, False, None
+
+    articles = parse_article_list_response(data_list, gh_id)
+    oldest = min((a["send_time"] for a in articles if a["send_time"] > 0), default=None)
+    return articles, next_cursor, has_more, oldest
+
+
+__all__ = [
+    "extract_wx_sn",
+    "extract_biz",
+    "is_valid_keyword",
+    "parse_search_result",
+    "parse_detail",
+    "parse_account_response",
+    "parse_article_list_response",
+    "show_desc_to_sta",
+    "fetch_article_page",
+]

src/domains/demand_search_account/account_article_fetch.py → src/domains/content_supply/account_article_fetch.py

@@ -11,19 +11,17 @@
 import asyncio
 import logging
 import time
-from typing import List
 
 from src.infra.database.mysql.manager import MysqlManager
-from src.infra.spider.wechat.gzh import get_article_list_from_account
 
-from ._const import DemandSearchAccountConst
+from ._const import ContentSupplyConst
 from ._mapper import AccountMapper
-from ._utils import parse_article_list_response
+from ._utils import fetch_article_page
 
 logger = logging.getLogger(__name__)
 
 
-class AccountArticleFetch(DemandSearchAccountConst):
+class AccountArticleFetch(ContentSupplyConst):
     """从 accounts 取 gh_id → 调文章列表 API → 写入 articles 表"""
 
     def __init__(self, pool: MysqlManager):
@@ -35,12 +33,10 @@ class AccountArticleFetch(DemandSearchAccountConst):
         Returns:
             {"accounts": int, "articles_written": int, "errors": int}
         """
-        accounts = await self.mapper.list_accounts_for_article_crawl()
+        accounts = await self.mapper.list_accounts_for_article_crawl(limit=limit)
         if not accounts:
             logger.info("无待抓取文章的账号")
             return {"accounts": 0, "articles_written": 0, "errors": 0}
-
-        accounts = accounts[:limit]
         total_articles = 0
         errors = 0
 
@@ -95,9 +91,12 @@ class AccountArticleFetch(DemandSearchAccountConst):
         cutoff = int(time.time()) - self.ONE_YEAR_SECONDS
 
         while True:
-            articles, next_cursor, has_more, oldest_send_time = await self._fetch_page(
-                gh_id, cursor
-            )
+            (
+                articles,
+                next_cursor,
+                has_more,
+                oldest_send_time,
+            ) = await fetch_article_page(gh_id, cursor)
 
             if articles:
                 # 筛选一年内的文章，过滤一年前的
@@ -127,42 +126,11 @@ class AccountArticleFetch(DemandSearchAccountConst):
 
     async def _incremental_crawl(self, gh_id: str, _cursor: str | None = None) -> int:
         """增量更新：只取第一页即停"""
-        articles, _, _, _ = await self._fetch_page(gh_id, None)
+        articles, _, _, _ = await fetch_article_page(gh_id, None)
         if not articles:
             return 0
         n = await self.mapper.insert_articles_batch(articles)
         return n
 
-    async def _fetch_page(
-        self,
-        gh_id: str,
-        cursor: str | None,
-    ) -> tuple[List[dict], str | None, bool, int | None]:
-        """调用 API 获取一页文章，返回 (articles, next_cursor, has_more, oldest_send_time)"""
-        resp = await get_article_list_from_account(gh_id, index=cursor)
-        if not resp or resp.get("code") != 0:
-            logger.warning(
-                "get_article_list_from_account 返回异常: gh_id=%s, cursor=%s, code=%s",
-                gh_id,
-                cursor,
-                resp.get("code") if resp else "None",
-            )
-            return [], None, False, None
-
-        data_wrapper = resp.get("data") or {}
-        data_list = data_wrapper.get("data") or []
-        has_more = data_wrapper.get("has_more", False)
-        next_cursor = data_wrapper.get("next_cursor")
-
-        if not data_list:
-            return [], None, False, None
-
-        articles = parse_article_list_response(data_list, gh_id)
-        # 本页最老的 send_time
-        oldest = min(
-            (a["send_time"] for a in articles if a["send_time"] > 0), default=None
-        )
-        return articles, next_cursor, has_more, oldest
-
 
 __all__ = ["AccountArticleFetch"]

src/domains/content_supply/account_article_update.py

@@ -0,0 +1,101 @@
+"""Phase 5: 对已完成深翻的账号，定期拉取第一页文章，更新数值指标
+
+流程:
+  1. 取 accounts 表中 crawl_deep_done=1 的账号（按 last_crawl_at 升序，最久未更新优先）
+  2. 每个账号调 API 取第一页文章（不翻页）
+  3. 按 wx_sn 批量 UPDATE 文章的数值指标（阅读/赞/付费/赞赏）
+  4. 只更新不插入，wx_sn 不存在于 articles 表的直接忽略
+"""
+
+import asyncio
+import logging
+
+from src.infra.database.mysql.manager import MysqlManager
+
+from ._const import ContentSupplyConst
+from ._mapper import AccountMapper
+from ._utils import fetch_article_page
+
+logger = logging.getLogger(__name__)
+
+
+class AccountArticleUpdate(ContentSupplyConst):
+    """取已完成深翻的账号 → 拉第一页文章 → 按 wx_sn 更新指标"""
+
+    def __init__(self, pool: MysqlManager):
+        self.mapper = AccountMapper(pool)
+
+    async def deal(self, *, limit: int = 10) -> dict:
+        """处理 N 个账号的文章指标更新
+
+        Returns:
+            {"accounts": int, "articles_updated": int, "errors": int}
+        """
+        accounts = await self.mapper.list_accounts_for_article_update(limit=limit)
+        if not accounts:
+            logger.info("无待更新文章指标的账号")
+            return {"accounts": 0, "articles_updated": 0, "errors": 0}
+
+        total_updated = 0
+        errors = 0
+
+        for i, acct in enumerate(accounts):
+            gh_id = acct["gh_id"]
+
+            if i > 0:
+                await asyncio.sleep(self.ARTICLE_PAGE_INTERVAL_SEC)
+
+            try:
+                n = await self._update_first_page(gh_id)
+            except Exception:
+                logger.exception("文章指标更新异常: gh_id=%s", gh_id)
+                errors += 1
+                continue
+
+            total_updated += n
+            logger.info("gh_id=%s 指标更新完成: 更新 %d 条", gh_id, n)
+
+        logger.info(
+            "指标更新完成: 账号 %d 个, 更新 %d 条, 异常 %d",
+            len(accounts),
+            total_updated,
+            errors,
+        )
+        return {
+            "accounts": len(accounts),
+            "articles_updated": total_updated,
+            "errors": errors,
+        }
+
+    async def _update_first_page(self, gh_id: str) -> int:
+        """拉取第一页文章，按 wx_sn 更新指标"""
+        articles, _, _, _ = await fetch_article_page(gh_id, None)
+        if not articles:
+            # 即使没有文章也刷新 last_crawl_at，避免下次仍然排在前面
+            await self.mapper.touch_last_crawl_at(gh_id)
+            return 0
+
+        # 只提取更新所需的字段，过滤掉没有 wx_sn 的文章
+        updates = [
+            {
+                "wx_sn": a["wx_sn"],
+                "show_view_count": a["show_view_count"],
+                "show_like_count": a["show_like_count"],
+                "show_pay_count": a["show_pay_count"],
+                "show_zs_count": a["show_zs_count"],
+                "show_desc": a["show_desc"],
+            }
+            for a in articles
+            if a.get("wx_sn")
+        ]
+        if not updates:
+            await self.mapper.touch_last_crawl_at(gh_id)
+            return 0
+
+        n = await self.mapper.update_article_metrics_batch(updates)
+        # 指标更新完成后刷新 last_crawl_at，推动轮转
+        await self.mapper.touch_last_crawl_at(gh_id)
+        return n
+
+
+__all__ = ["AccountArticleUpdate"]

src/domains/demand_search_account/account_fetch.py → src/domains/content_supply/account_fetch.py

@@ -12,14 +12,14 @@ import logging
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.spider.wechat.gzh import get_account_from_url
 
-from ._const import DemandSearchAccountConst
+from ._const import ContentSupplyConst
 from ._mapper import AccountMapper
 from ._utils import parse_account_response
 
 logger = logging.getLogger(__name__)
 
 
-class AccountFetch(DemandSearchAccountConst):
+class AccountFetch(ContentSupplyConst):
     """从 relation 取待抓 biz → 查重 → 调账号 API → 写入 accounts 表"""
 
     def __init__(self, pool: MysqlManager):

src/domains/demand_search_article/article_fetch_detail.py → src/domains/content_supply/article_fetch_detail.py

@@ -11,14 +11,14 @@ from typing import Dict
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.spider.wechat.gzh import get_article_detail
 
-from ._const import DemandSearchArticleConst
+from ._const import ContentSupplyConst
 from ._mapper import ArticleSearchRelationMapper, ArticleDetailMapper
 from ._utils import parse_detail
 
 logger = logging.getLogger(__name__)
 
 
-class ArticleFetchDetail(DemandSearchArticleConst):
+class ArticleFetchDetail(ContentSupplyConst):
     """从 relation 表取待拉项 → 调详情 API → 写入 detail 表"""
 
     def __init__(self, pool: MysqlManager):
@@ -32,13 +32,14 @@ class ArticleFetchDetail(DemandSearchArticleConst):
     async def deal(
         self,
         *,
-        limit: int = 500,
+        limit: int | None = None,
     ) -> dict:
         """串行处理待拉详情项
 
-        Returns:
-            {"pending": int, "success": int, "failed": int}
+        limit 为 None 时使用 DEFAULT_DETAIL_DEAL_LIMIT。
         """
+        if limit is None:
+            limit = self.DETAIL_DEAL_LIMIT
         items = await self.relation_mapper.fetch_pending(limit=limit)
         if not items:
             logger.info("无待拉详情项")

src/domains/demand_search_article/article_search.py → src/domains/content_supply/article_search.py

@@ -12,14 +12,14 @@ from typing import Dict
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.spider.wechat.gzh import weixin_search
 
-from ._const import DemandSearchArticleConst
+from ._const import ContentSupplyConst
 from ._mapper import ArticleSearchRelationMapper
 from ._utils import parse_search_result, is_valid_keyword
 
 logger = logging.getLogger(__name__)
 
 
-class DemandSearchArticle(DemandSearchArticleConst):
+class DemandSearchArticle(ContentSupplyConst):
     """从需求队列取搜索词 → 微信搜索 → 写入关系表
 
     按 dt 分区读取，experiment_id 只透传不过滤。

src/domains/fetch_demand/__init__.py → src/domains/demand_enqueue_task/__init__.py

@@ -1,9 +1,9 @@
-"""需求域 —— 从 ODPS 提取白名单账号匹配视频 → 解构 → 搜索词入队
+"""需求域 —— 从 ODPS 提取白名单账号匹配视频 → 获取解构结果 → 搜索词入队
 
 核心流程:
   1. FetchDemands     — 白名单加载 → 匹配视频查询 → rov 去重 → 视频解构 → 搜索词展开
   2. EnqueueDemands   — 队列条目批量写入 MySQL demand_search_queue
-  3. DemandQueueMapper — 队列表纯 DB 读写（供 demand_search_article 域消费时复用）
+  3. DemandQueueMapper — 队列表纯 DB 读写（供 search_article 域消费时复用）
 """
 
 from ._const import DemandConst

src/domains/fetch_demand/enqueue_demands.py → src/domains/demand_enqueue_task/enqueue_demands.py

@@ -12,7 +12,7 @@ logger = logging.getLogger(__name__)
 
 
 class EnqueueDemands(DemandConst):
-    """将 fetch_demand 产出的队列条目批量写入 demand_search_queue
+    """将 demand_enqueue_task 产出的队列条目批量写入 demand_search_queue
 
     使用方式:
         mysql = db.mysql_manager("long_articles")

src/domains/demand_search_account/__init__.py

@@ -1,26 +0,0 @@
-"""账号领域 —— 从 demand_search_article_relation 中提取公众号信息
-
-Phase 3 — 账号信息抓取:
-  1. 从 relation 取 DISTINCT biz（save_account_status=0）
-  2. biz 在 accounts 表中查重，已存在则跳过
-  3. 不存在则调 get_account_from_url API 获取账号信息
-  4. 写入 demand_search_accounts，回填 relation.save_account_status
-
-Phase 4 — 账号文章抓取:
-  1. 取 accounts 表中所有 gh_id
-  2. crawl_deep_done=0 → 深翻翻页到一年前
-  3. crawl_deep_done=1 → 只取第一页增量
-  4. 写入 demand_account_articles（wx_sn 去重）
-"""
-
-from src.domains.demand_search_account._const import DemandSearchAccountConst
-from src.domains.demand_search_account._mapper import AccountMapper
-from src.domains.demand_search_account.account_fetch import AccountFetch
-from src.domains.demand_search_account.account_article_fetch import AccountArticleFetch
-
-__all__ = [
-    "DemandSearchAccountConst",
-    "AccountMapper",
-    "AccountFetch",
-    "AccountArticleFetch",
-]

src/domains/demand_search_account/_const.py

@@ -1,26 +0,0 @@
-"""账号域常量"""
-
-
-class DemandSearchAccountConst:
-    """账号域所有常量的聚合根"""
-
-    ACCOUNT_TABLE = "demand_search_accounts"
-    ARTICLES_TABLE = "demand_account_articles"
-    RELATION_TABLE = "demand_search_article_relation"
-
-    class AccountSaveStatus:
-        INIT = 0
-        PROCESSING = 1
-        SUCCESS = 2
-        FAIL = 99
-
-    ACCOUNT_DEAL_LIMIT = 100  # 单次取待抓账号条数
-    ACCOUNT_INTERVAL_SEC = 1  # API 调用间隔（秒）
-
-    ARTICLE_DEAL_LIMIT = 10  # 单次处理账号数
-    ARTICLE_PAGE_INTERVAL_SEC = 2  # 翻页间隔（秒）
-    ARTICLE_BATCH_INSERT = 200  # 文章批量写入每批条数
-    ONE_YEAR_SECONDS = 365 * 24 * 3600
-
-
-__all__ = ["DemandSearchAccountConst"]

src/domains/demand_search_account/_mapper.py

@@ -1,217 +0,0 @@
-"""账号域 DB 读写 —— demand_search_accounts + demand_search_article_relation + demand_account_articles"""
-
-import json
-from typing import Dict, List, Optional
-
-from src.infra.database.mysql.manager import MysqlManager
-
-from ._const import DemandSearchAccountConst
-
-
-class AccountMapper(DemandSearchAccountConst):
-    """demand_search_accounts 表 + relation.save_account_status 的读写"""
-
-    def __init__(self, pool: MysqlManager):
-        self.pool = pool
-
-    # ═══════════════════════════════════════════════════════════════
-    # relation 表 — 读取待抓账号
-    # ═══════════════════════════════════════════════════════════════
-
-    async def fetch_pending_relations(self, *, limit: int = 100) -> List[Dict]:
-        """取 save_account_status=0 的行（DISTINCT biz，按 id ASC）"""
-        query = f"""
-            SELECT MIN(id) AS id, biz, MIN(url) AS url
-            FROM {self.RELATION_TABLE}
-            WHERE save_account_status = %s AND biz IS NOT NULL AND biz != ''
-            GROUP BY biz
-            ORDER BY id ASC
-            LIMIT %s
-        """
-        return await self.pool.fetch(
-            query=query,
-            params=(self.AccountSaveStatus.INIT, limit),
-        )
-
-    async def fetch_relation_ids_by_biz(self, biz: str) -> List[int]:
-        """取某个 biz 下所有 save_account_status=0 的 relation id"""
-        rows = await self.pool.fetch(
-            query=f"SELECT id FROM {self.RELATION_TABLE} WHERE biz = %s AND save_account_status = %s",
-            params=(biz, self.AccountSaveStatus.INIT),
-        )
-        return [r["id"] for r in rows]
-
-    async def fetch_relation_ids_by_biz_any_status(self, biz: str) -> List[int]:
-        """取某个 biz 下未完成（INIT/PROCESSING）的 relation id，异常恢复用"""
-        rows = await self.pool.fetch(
-            query=f"SELECT id FROM {self.RELATION_TABLE} WHERE biz = %s AND save_account_status IN (%s, %s)",
-            params=(
-                biz,
-                self.AccountSaveStatus.INIT,
-                self.AccountSaveStatus.PROCESSING,
-            ),
-        )
-        return [r["id"] for r in rows]
-
-    # ═══════════════════════════════════════════════════════════════
-    # relation 表 — save_account_status 状态更新
-    # ═══════════════════════════════════════════════════════════════
-
-    async def _mark_batch(
-        self, target_status: int, ids: List[int], from_statuses: List[int]
-    ) -> int:
-        """通用：将 ids 从 from_statuses 任一改为 target_status"""
-        if not ids:
-            return 0
-        placeholders = ",".join(["%s"] * len(ids))
-        status_placeholders = ",".join(["%s"] * len(from_statuses))
-        query = f"""
-            UPDATE {self.RELATION_TABLE}
-            SET save_account_status = %s
-            WHERE id IN ({placeholders}) AND save_account_status IN ({status_placeholders})
-        """
-        return await self.pool.save(
-            query=query,
-            params=(target_status, *ids, *from_statuses),
-        )
-
-    async def relation_mark_processing(self, ids: List[int]) -> int:
-        """INIT → PROCESSING，CAS 防重复"""
-        return await self._mark_batch(
-            self.AccountSaveStatus.PROCESSING, ids, [self.AccountSaveStatus.INIT]
-        )
-
-    async def relation_mark_success(self, ids: List[int]) -> int:
-        """PROCESSING → SUCCESS"""
-        return await self._mark_batch(
-            self.AccountSaveStatus.SUCCESS, ids, [self.AccountSaveStatus.PROCESSING]
-        )
-
-    async def relation_mark_success_direct(self, ids: List[int]) -> int:
-        """INIT → SUCCESS，跳过 PROCESSING（查重命中时用）"""
-        return await self._mark_batch(
-            self.AccountSaveStatus.SUCCESS, ids, [self.AccountSaveStatus.INIT]
-        )
-
-    async def relation_mark_failed(self, ids: List[int]) -> int:
-        """INIT/PROCESSING → FAIL，异常恢复用"""
-        return await self._mark_batch(
-            self.AccountSaveStatus.FAIL,
-            ids,
-            [self.AccountSaveStatus.INIT, self.AccountSaveStatus.PROCESSING],
-        )
-
-    # ═══════════════════════════════════════════════════════════════
-    # demand_search_accounts 表
-    # ═══════════════════════════════════════════════════════════════
-
-    async def find_by_biz(self, biz: str) -> Optional[Dict]:
-        """按 biz 查账号是否已存在"""
-        return await self.pool.fetch_one(
-            query=f"SELECT * FROM {self.ACCOUNT_TABLE} WHERE biz = %s LIMIT 1",
-            params=(biz,),
-        )
-
-    async def insert_account(self, account: Dict) -> int:
-        """写入账号，channel_account_id 冲突则跳过"""
-        query = f"""
-            INSERT IGNORE INTO {self.ACCOUNT_TABLE}
-                (channel_account_id, gh_id, account_name, biz, biz_info, description)
-            VALUES (%s, %s, %s, %s, %s, %s)
-        """
-        return await self.pool.save(
-            query=query,
-            params=(
-                account["channel_account_id"],
-                account["gh_id"],
-                account["account_name"],
-                account["biz"],
-                account["biz_info"],
-                account["description"],
-            ),
-        )
-
-    async def list_accounts_for_article_crawl(self) -> List[Dict]:
-        """列出所有有 gh_id 的账号，用于文章抓取"""
-        query = f"""
-            SELECT channel_account_id, gh_id, crawl_cursor, crawl_deep_done
-            FROM {self.ACCOUNT_TABLE}
-            WHERE gh_id IS NOT NULL AND gh_id != ''
-            ORDER BY crawl_deep_done ASC, gh_id ASC
-        """
-        return await self.pool.fetch(query=query)
-
-    async def update_crawl_status(
-        self,
-        gh_id: str,
-        *,
-        cursor: str = "",
-        deep_done: int = 0,
-    ) -> int:
-        """更新账号的爬取游标和深翻状态"""
-        query = f"""
-            UPDATE {self.ACCOUNT_TABLE}
-            SET crawl_cursor = %s, crawl_deep_done = %s, last_crawl_at = NOW()
-            WHERE gh_id = %s
-        """
-        return await self.pool.save(query=query, params=(cursor, deep_done, gh_id))
-
-    # ═══════════════════════════════════════════════════════════════
-    # demand_account_articles 表
-    # ═══════════════════════════════════════════════════════════════
-
-    async def insert_articles_batch(self, articles: List[Dict]) -> int:
-        """批量写入文章，wx_sn 冲突则跳过"""
-        if not articles:
-            return 0
-        query = f"""
-            INSERT IGNORE INTO {self.ARTICLES_TABLE}
-                (gh_id, app_msg_id, item_index, title, content_url, wx_sn,
-                 digest, is_original, item_show_type, source_url, cover_img_url, show_desc,
-                 show_view_count, show_like_count, show_pay_count, show_zs_count,
-                 type, send_time, create_time, update_time, date_time, base_info)
-            VALUES
-                (%s, %s, %s, %s, %s, %s,
-                 %s, %s, %s, %s, %s, %s,
-                 %s, %s, %s, %s,
-                 %s, %s, %s, %s, %s, %s)
-        """
-        params = [
-            (
-                a["gh_id"],
-                a["app_msg_id"],
-                a["item_index"],
-                a["title"],
-                a["content_url"],
-                a["wx_sn"],
-                a["digest"],
-                a["is_original"],
-                a["item_show_type"],
-                a["source_url"],
-                a["cover_img_url"],
-                a["show_desc"],
-                a["show_view_count"],
-                a["show_like_count"],
-                a["show_pay_count"],
-                a["show_zs_count"],
-                a["type"],
-                a["send_time"],
-                a["create_time"],
-                a["update_time"],
-                a["date_time"],
-                json.dumps(a["base_info"], ensure_ascii=False),
-            )
-            for a in articles
-        ]
-        return await self.pool.save(query=query, params=params, batch=True)
-
-    async def count_articles_by_gh_id(self, gh_id: str) -> int:
-        """统计某账号已抓文章数"""
-        row = await self.pool.fetch_one(
-            query=f"SELECT COUNT(*) AS cnt FROM {self.ARTICLES_TABLE} WHERE gh_id = %s",
-            params=(gh_id,),
-        )
-        return row["cnt"] if row else 0
-
-
-__all__ = ["AccountMapper"]

src/domains/demand_search_account/_utils.py

@@ -1,167 +0,0 @@
-"""账号域工具函数 —— API 响应解析"""
-
-import re
-from typing import Dict, List
-
-from src.domains.demand_search_article._utils import extract_wx_sn
-
-
-def parse_account_response(api_data: Dict, biz: str = "") -> Dict:
-    """get_account_from_url 返回的 data.data → accounts 表 INSERT dict"""
-    return {
-        "channel_account_id": api_data.get("channel_account_id", ""),
-        "gh_id": api_data.get("wx_gh", "") or "",
-        "account_name": api_data.get("account_name", "") or "",
-        "biz": biz or api_data.get("biz_info", "") or "",
-        "biz_info": api_data.get("biz_info", "") or "",
-        "description": api_data.get("description", "") or "",
-    }
-
-
-# ═══════════════════════════════════════════════════════════════
-# ShowDesc 解析：阅读/赞/付费/赞赏 → 数字
-# ═══════════════════════════════════════════════════════════════
-
-
-def show_desc_to_sta(show_desc: str) -> Dict[str, int]:
-    """解析 ShowDesc 中的阅读量、点赞量等
-
-    ShowDesc 格式示例:
-        "阅读 80  赞 3  "
-        "阅读 1  "
-        "阅读 121  赞 5  "
-
-    分隔符: \\u2006 分隔 key-value, \\u2004/\\u2005 分隔 group。
-    """
-
-    def _decode_value(raw: str) -> int:
-        if not raw:
-            return 0
-        raw = raw.strip().lower().replace(",", ".")
-        m = re.search(r"(\d+(?:\.\d+)?)([a-z一-龥]*)", raw)
-        if not m:
-            return 0
-        num = float(m.group(1))
-        unit = m.group(2)
-        if "亿" in unit:
-            num *= 1e8
-        elif "万" in unit:
-            num *= 1e4
-        elif "千" in unit:
-            num *= 1e3
-        elif unit.startswith("k"):
-            num *= 1e3
-        elif unit.startswith("m"):
-            num *= 1e6
-        elif unit.startswith("b"):
-            num *= 1e9
-        return int(num)
-
-    def _decode_key(raw: str) -> str:
-        if not raw:
-            return "show_unknown"
-        raw = raw.strip().lower()
-        mapping = {
-            "阅读": "show_view_count",
-            "看过": "show_view_count",
-            "观看": "show_view_count",
-            "reads": "show_view_count",
-            "views": "show_view_count",
-            "view": "show_view_count",
-            "赞": "show_like_count",
-            "点赞": "show_like_count",
-            "likes": "show_like_count",
-            "like": "show_like_count",
-            "付费": "show_pay_count",
-            "payments": "show_pay_count",
-            "paid": "show_pay_count",
-            "赞赏": "show_zs_count",
-        }
-        return mapping.get(raw, "show_unknown")
-
-    if not show_desc:
-        return {
-            "show_view_count": 0,
-            "show_like_count": 0,
-            "show_pay_count": 0,
-            "show_zs_count": 0,
-        }
-
-    show_desc = show_desc.replace("+", "")
-    sta: Dict[str, int] = {}
-    groups = re.split(r"[  ]+", show_desc)
-
-    for group in groups:
-        group = group.strip()
-        if not group:
-            continue
-        parts = group.split(" ")
-        if len(parts) != 2:
-            continue
-        a, b = parts
-        if re.search(r"\d", a):
-            v_raw, k_raw = a, b
-        else:
-            k_raw, v_raw = a, b
-        key = _decode_key(k_raw)
-        val = _decode_value(v_raw)
-        if key != "show_unknown":
-            sta[key] = val
-
-    return {
-        "show_view_count": sta.get("show_view_count", 0),
-        "show_like_count": sta.get("show_like_count", 0),
-        "show_pay_count": sta.get("show_pay_count", 0),
-        "show_zs_count": sta.get("show_zs_count", 0),
-    }
-
-
-# ═══════════════════════════════════════════════════════════════
-# 文章列表响应解析
-# ═══════════════════════════════════════════════════════════════
-
-
-def parse_article_list_response(data_list: List[Dict], gh_id: str) -> List[Dict]:
-    """get_article_list_from_account 返回的 data.data[] → articles 表 INSERT dict 列表"""
-    rows = []
-    for group in data_list or []:
-        app_msg = group.get("AppMsg") or {}
-        base_info = app_msg.get("BaseInfo") or {}
-        detail_list = app_msg.get("DetailInfo") or []
-        top_base = group.get("BaseInfo") or {}
-
-        for detail in detail_list:
-            url = detail.get("ContentUrl", "") or ""
-            show_desc = detail.get("ShowDesc", "") or ""
-            counts = show_desc_to_sta(show_desc)
-
-            rows.append(
-                {
-                    "gh_id": gh_id,
-                    "app_msg_id": base_info.get("AppMsgId", 0),
-                    "item_index": detail.get("ItemIndex", 0),
-                    "title": detail.get("Title", "") or "",
-                    "content_url": url,
-                    "wx_sn": extract_wx_sn(url) or "",
-                    "digest": detail.get("Digest", "") or "",
-                    "is_original": 1 if detail.get("IsOriginal") else 0,
-                    "item_show_type": detail.get("ItemShowType", 0) or 0,
-                    "source_url": detail.get("SourceUrl", "") or "",
-                    "cover_img_url": detail.get("CoverImgUrl", "") or "",
-                    "show_desc": show_desc,
-                    "show_view_count": counts["show_view_count"],
-                    "show_like_count": counts["show_like_count"],
-                    "show_pay_count": counts["show_pay_count"],
-                    "show_zs_count": counts["show_zs_count"],
-                    "type": base_info.get("Type", 0) or 0,
-                    "send_time": detail.get("send_time", 0) or 0,
-                    "create_time": base_info.get("CreateTime", 0) or 0,
-                    "update_time": base_info.get("UpdateTime", 0) or 0,
-                    "date_time": top_base.get("DateTime", 0) or 0,
-                    "base_info": top_base,
-                }
-            )
-    return rows
-
-
-__all__ = ["parse_account_response", "parse_article_list_response", "show_desc_to_sta"]

src/domains/demand_search_article/__init__.py

@@ -1,29 +0,0 @@
-"""文章搜索领域 —— 基于需求驱动文章搜索与内容结构化
-
-Phase 1 — 搜索:
-  1. 从 demand_search_queue 取搜索词
-  2. 调微信搜索 API
-  3. 写入 demand_search_article_relation (status=0)
-
-Phase 2 — 拉详情:
-  1. 从 demand_search_article_relation(status=0) 取待拉项
-  2. 调微信详情 API
-  3. 写入 demand_search_article_detail (channel_content_id 去重)
-  4. 回填 relation 状态
-"""
-
-from src.domains.demand_search_article._const import DemandSearchArticleConst
-from src.domains.demand_search_article._mapper import (
-    ArticleSearchRelationMapper,
-    ArticleDetailMapper,
-)
-from src.domains.demand_search_article.article_search import DemandSearchArticle
-from src.domains.demand_search_article.article_fetch_detail import ArticleFetchDetail
-
-__all__ = [
-    "DemandSearchArticleConst",
-    "ArticleSearchRelationMapper",
-    "ArticleDetailMapper",
-    "DemandSearchArticle",
-    "ArticleFetchDetail",
-]

src/domains/demand_search_article/_utils.py

@@ -1,111 +0,0 @@
-"""文章域工具函数 —— API 响应解析、数据转换"""
-
-from datetime import datetime
-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。
-    """
-    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")
-
-
-# 无效搜索词集合 — 仅包含无意义的占位符
-_INVALID_KEYWORDS = frozenset({"-", "--", "---", "/", ".", "。。。"})
-
-
-def is_valid_keyword(keyword: str) -> bool:
-    """搜索词是否有效：非空、非纯占位符"""
-    kw = keyword.strip()
-    if not kw:
-        return False
-    if kw in _INVALID_KEYWORDS:
-        return False
-    return True
-
-
-def parse_search_result(
-    article: Dict,
-    keyword: str,
-    key_type: str,
-    experiment_id: str,
-    page: int,
-) -> Dict:
-    """微信搜索 API 返回的单篇文章 → relation 表 INSERT dict"""
-    ts = article.get("time")
-    publish_time = datetime.fromtimestamp(ts) if ts and ts > 0 else None
-    article_url = article.get("url", "")
-    return {
-        "search_key": keyword,
-        "key_type": key_type,
-        "experiment_id": experiment_id,
-        "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,
-    }
-
-
-def parse_detail(detail: Dict) -> Dict:
-    """微信详情 API 返回的 data.data → detail 表 INSERT dict"""
-    ts = detail.get("publish_timestamp")
-    publish_at = datetime.fromtimestamp(ts / 1000) if ts and ts > 0 else None
-    content_url = detail.get("content_link", "")
-    # API 返回的 wx_sn 通常为空，从 URL query 参数提取
-    wx_sn = detail.get("wx_sn", "") or extract_wx_sn(content_url) or ""
-    return {
-        "channel_content_id": detail.get("channel_content_id", ""),
-        "wx_sn": wx_sn,
-        "url": content_url,
-        "title": detail.get("title", ""),
-        "body_text": detail.get("body_text", ""),
-        "content_type": detail.get("content_type", ""),
-        "channel_account_id": detail.get("channel_account_id", ""),
-        "publish_timestamp": ts or 0,
-        "publish_at": publish_at,
-        "is_original": 1 if detail.get("is_original") else 0,
-        "view_count": detail.get("view_count") or 0,
-        "like_count": detail.get("like_count") or 0,
-        "looking_count": detail.get("looking_count") or 0,
-        "comment_count": detail.get("comment_count") or 0,
-        "share_count": detail.get("share_count") or 0,
-        "image_url_list": detail.get("image_url_list") or [],
-        "video_url_list": detail.get("video_url_list") or [],
-        "is_cache": 1 if detail.get("is_cache") else 0,
-        "extra": {
-            "item_index": detail.get("item_index"),
-            "channel_account_name": detail.get("channel_account_name"),
-        },
-    }
-
-
-__all__ = [
-    "extract_wx_sn",
-    "extract_biz",
-    "is_valid_keyword",
-    "parse_search_result",
-    "parse_detail",
-]

src/handlers/__init__.py

@@ -8,6 +8,6 @@ Handler 通过 discover_and_register() 在应用启动时自动发现。
     discover_and_register()
 """
 
-from src.handlers._discovery import discover_and_register
+from src.handlers._utils import discover_and_register
 
 __all__ = ["discover_and_register"]

src/handlers/_discovery.py

@@ -1,46 +0,0 @@
-"""XXL-JOB handler 自动发现 —— pkgutil + importlib 扫描导入
-
-扫描 src.handlers 包下的所有非私有模块，逐一 import，
-触发 @xxl_job 装饰器将 handler 注册到 _XXL_REGISTRY。
-"""
-
-import importlib
-import logging
-import pkgutil
-
-logger = logging.getLogger(__name__)
-
-
-def discover_and_register(package: str = "src.handlers") -> None:
-    """自动发现并导入 package 下的所有 handler 模块
-
-    跳过 _ 前缀的私有模块和子包。每个模块被 import 时，
-    @xxl_job 装饰器自动执行，将 handler 写入共享的 _XXL_REGISTRY。
-
-    幂等调用：importlib 使用 sys.modules 缓存，重复调用是 no-op。
-
-    Args:
-        package: 点分隔的 Python 包路径，默认 "src.handlers"
-    """
-    try:
-        pkg = importlib.import_module(package)
-    except ImportError:
-        logger.warning("XXL-JOB handler 包 %r 不存在，无 handler 注册", package)
-        return
-
-    registered = 0
-    for finder, name, is_pkg in pkgutil.iter_modules(pkg.__path__, pkg.__name__ + "."):
-        if name.split(".")[-1].startswith("_"):
-            continue
-
-        try:
-            importlib.import_module(name)
-            registered += 1
-            logger.debug("XXL-JOB 发现 handler 模块: %s", name)
-        except Exception:
-            logger.exception("XXL-JOB handler 模块 %r 导入失败，跳过", name)
-
-    if registered:
-        logger.info("XXL-JOB 自动发现完成: 已导入 %d 个 handler 模块", registered)
-    else:
-        logger.info("XXL-JOB 自动发现: 在 %r 中未找到 handler 模块", package)

src/handlers/_utils.py

@@ -0,0 +1,105 @@
+"""handler 共享工具 —— DB 注入 + 参数解析 + 自动发现"""
+
+import importlib
+import logging
+import pkgutil
+from typing import TYPE_CHECKING, Dict
+
+if TYPE_CHECKING:
+    from src.infra.database import DatabaseManager
+
+logger = logging.getLogger(__name__)
+
+# ═══════════════════════════════════════════════════════════════
+# DB 注入槽 — app.py 启动时注入，handler 通过 get_db() 获取
+# ═══════════════════════════════════════════════════════════════
+
+_db_ref: "DatabaseManager | None" = None
+
+
+def set_db(db: "DatabaseManager") -> None:
+    global _db_ref
+    _db_ref = db
+
+
+def get_db() -> "DatabaseManager":
+    """运行时获取已注入的 DatabaseManager，未注入则抛出明确错误"""
+    if _db_ref is None:
+        raise RuntimeError("DatabaseManager 尚未注入，请确认 app.py 中已调用 set_db()")
+    return _db_ref
+
+
+# ═══════════════════════════════════════════════════════════════
+# XXL-JOB executorParams 参数解析
+# ═══════════════════════════════════════════════════════════════
+
+
+def parse_params(param: str) -> Dict[str, str]:
+    """解析 XXL-JOB executorParams 为 key=value dict
+
+    支持格式: "key1=val1 key2=val2"
+    """
+    result: Dict[str, str] = {}
+    if not param or not param.strip():
+        return result
+    for part in param.strip().split():
+        if "=" not in part:
+            continue
+        k, v = part.split("=", 1)
+        result[k] = v
+    return result
+
+
+def parse_limit(param: str, default: int) -> int:
+    """从 param 中提取 limit=N，失败或不存在返回 default"""
+    p = parse_params(param)
+    raw = p.get("limit", "")
+    if not raw:
+        return default
+    try:
+        return int(raw)
+    except ValueError:
+        return default
+
+
+def parse_dt(param: str, default: str) -> str:
+    """从 param 中提取 dt=yyyyMMdd，不存在返回 default"""
+    p = parse_params(param)
+    return p.get("dt", default)
+
+
+# ═══════════════════════════════════════════════════════════════
+# XXL-JOB handler 自动发现
+# ═══════════════════════════════════════════════════════════════
+
+
+def discover_and_register(package: str = "src.handlers") -> None:
+    """自动发现并导入 package 下的所有 handler 模块
+
+    跳过 _ 前缀的私有模块。每个模块被 import 时，
+    @xxl_job 装饰器自动执行，将 handler 写入共享的 _XXL_REGISTRY。
+
+    幂等调用：importlib 使用 sys.modules 缓存，重复调用是 no-op。
+    """
+    try:
+        pkg = importlib.import_module(package)
+    except ImportError:
+        logger.warning("XXL-JOB handler 包 %r 不存在，无 handler 注册", package)
+        return
+
+    registered = 0
+    for finder, name, is_pkg in pkgutil.iter_modules(pkg.__path__, pkg.__name__ + "."):
+        if name.split(".")[-1].startswith("_"):
+            continue
+
+        try:
+            importlib.import_module(name)
+            registered += 1
+            logger.debug("XXL-JOB 发现 handler 模块: %s", name)
+        except Exception:
+            logger.exception("XXL-JOB handler 模块 %r 导入失败，跳过", name)
+
+    if registered:
+        logger.info("XXL-JOB 自动发现完成: 已导入 %d 个 handler 模块", registered)
+    else:
+        logger.info("XXL-JOB 自动发现: 在 %r 中未找到 handler 模块", package)

src/handlers/account_article_fetch.py

@@ -7,34 +7,20 @@ XXL-JOB Admin 配置:
 """
 
 import logging
-from typing import TYPE_CHECKING
 
 from src.infra.xxl_jobs import xxl_job
-from src.domains.demand_search_account import (
-    AccountArticleFetch,
-    DemandSearchAccountConst,
-)
-
-if TYPE_CHECKING:
-    from src.infra.database import DatabaseManager
+from src.domains.content_supply import AccountArticleFetch, ContentSupplyConst
+from src.handlers._utils import get_db, parse_limit
 
 logger = logging.getLogger(__name__)
 
-_db_ref: "DatabaseManager | None" = None
-
-
-def set_db(db: "DatabaseManager") -> None:
-    global _db_ref
-    _db_ref = db
-
-
 @xxl_job("accountArticleFetch")
 async def account_article_fetch(param: str) -> dict:
-    limit = DemandSearchAccountConst.ARTICLE_DEAL_LIMIT
+    limit = parse_limit(param, ContentSupplyConst.ARTICLE_DEAL_LIMIT)
     logger.info("accountArticleFetch started, limit=%d", limit)
 
     try:
-        mysql = _db_ref.mysql_manager("long_articles")
+        mysql = get_db().mysql_manager("long_articles")
         fetcher = AccountArticleFetch(mysql)
         result = await fetcher.deal(limit=limit)
     except Exception:

src/handlers/account_article_update.py

@@ -0,0 +1,41 @@
+"""账号文章指标更新 XXL-JOB Handler —— 已完成深翻的账号 → 第一页文章 → UPDATE 指标
+
+XXL-JOB Admin 配置:
+    执行器: long-articles-agentic-src
+    处理器: accountArticleUpdate
+    参数:   单次处理账号数（可选，默认使用 ARTICLE_DEAL_LIMIT）
+    超时:   建议 ≥600s
+"""
+
+import logging
+
+from src.infra.xxl_jobs import xxl_job
+from src.domains.content_supply import AccountArticleUpdate, ContentSupplyConst
+from src.handlers._utils import get_db, parse_limit
+
+logger = logging.getLogger(__name__)
+
+
+@xxl_job("accountArticleUpdate")
+async def account_article_update(param: str) -> dict:
+    limit = parse_limit(param, ContentSupplyConst.ARTICLE_DEAL_LIMIT)
+    logger.info("accountArticleUpdate started, limit=%d", limit)
+
+    try:
+        mysql = get_db().mysql_manager("long_articles")
+        updater = AccountArticleUpdate(mysql)
+        result = await updater.deal(limit=limit)
+    except Exception:
+        logger.exception("文章指标更新失败")
+        return {"code": 500, "msg": "文章指标更新异常"}
+
+    logger.info(
+        "accountArticleUpdate done: accounts=%d, articles_updated=%d, errors=%d",
+        result["accounts"],
+        result["articles_updated"],
+        result["errors"],
+    )
+    return {
+        "code": 200,
+        "msg": f"accounts={result['accounts']}, updated={result['articles_updated']}, errors={result['errors']}",
+    }

src/handlers/account_fetch.py

@@ -6,31 +6,21 @@ XXL-JOB Admin 配置:
 """
 
 import logging
-from typing import TYPE_CHECKING
 
 from src.infra.xxl_jobs import xxl_job
-from src.domains.demand_search_account import AccountFetch, DemandSearchAccountConst
-
-if TYPE_CHECKING:
-    from src.infra.database import DatabaseManager
+from src.domains.content_supply import AccountFetch, ContentSupplyConst
+from src.handlers._utils import get_db, parse_limit
 
 logger = logging.getLogger(__name__)
 
-_db_ref: "DatabaseManager | None" = None
-
-
-def set_db(db: "DatabaseManager") -> None:
-    global _db_ref
-    _db_ref = db
-
 
 @xxl_job("accountFetch")
 async def account_fetch(param: str) -> dict:
-    limit = DemandSearchAccountConst.ACCOUNT_DEAL_LIMIT
+    limit = parse_limit(param, ContentSupplyConst.ACCOUNT_DEAL_LIMIT)
     logger.info("accountFetch started, limit=%d", limit)
 
     try:
-        mysql = _db_ref.mysql_manager("long_articles")
+        mysql = get_db().mysql_manager("long_articles")
         fetcher = AccountFetch(mysql)
         result = await fetcher.deal(limit=limit)
     except Exception:

src/handlers/article_fetch_detail.py

@@ -7,34 +7,21 @@ XXL-JOB Admin 配置:
 """
 
 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
+from src.domains.content_supply import ArticleFetchDetail, ContentSupplyConst
+from src.handlers._utils import get_db, parse_limit
 
 logger = logging.getLogger(__name__)
 
-_db_ref: "DatabaseManager | None" = None
-
-
-def set_db(db: "DatabaseManager") -> None:
-    global _db_ref
-    _db_ref = db
-
 
 @xxl_job("articleFetchDetail")
 async def article_fetch_detail(param: str) -> dict:
-    limit = DemandSearchArticleConst.DETAIL_DEAL_LIMIT
+    limit = parse_limit(param, ContentSupplyConst.DETAIL_DEAL_LIMIT)
     logger.info("articleFetchDetail started, limit=%d", limit)
 
     try:
-        mysql = _db_ref.mysql_manager("long_articles")
+        mysql = get_db().mysql_manager("long_articles")
         fetcher = ArticleFetchDetail(mysql)
         result = await fetcher.deal(limit=limit)
     except Exception:

src/handlers/article_search.py

@@ -8,44 +8,23 @@ XXL-JOB Admin 配置:
 """
 
 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
+from src.domains.content_supply import DemandSearchArticle, ContentSupplyConst
+from src.handlers._utils import get_db, parse_params
 
 logger = logging.getLogger(__name__)
 
-_db_ref: "DatabaseManager | None" = None
-
-
-def set_db(db: "DatabaseManager") -> None:
-    global _db_ref
-    _db_ref = db
-
-
-def _parse_dt(param: str) -> str | None:
-    if param and param.strip():
-        for part in param.strip().split():
-            if part.startswith("dt="):
-                return part[3:]
-        return param.strip()
-    return None
-
 
 @xxl_job("articleSearch")
 async def article_search(param: str) -> dict:
-    dt = _parse_dt(param)
-    limit = DemandSearchArticleConst.SEARCH_DEAL_LIMIT
+    p = parse_params(param)
+    dt = p.get("dt") or None
+    limit = ContentSupplyConst.SEARCH_DEAL_LIMIT
     logger.info("articleSearch started, dt=%s, limit=%d", dt, limit)
 
     try:
-        mysql = _db_ref.mysql_manager("long_articles")
+        mysql = get_db().mysql_manager("long_articles")
         searcher = DemandSearchArticle(mysql)
         result = await searcher.deal(dt, limit=limit)
     except Exception:

src/handlers/demand_enqueue.py

@@ -9,42 +9,27 @@ XXL-JOB Admin 配置:
 
 import logging
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING
 
 from src.config.odps import OdpsConfig
 from src.infra.xxl_jobs import xxl_job
-from src.domains.fetch_demand import FetchDemands, EnqueueDemands
-
-if TYPE_CHECKING:
-    from src.infra.database import DatabaseManager
+from src.domains.demand_enqueue_task import FetchDemands, EnqueueDemands
+from src.handlers._utils import get_db, parse_dt
 
 logger = logging.getLogger(__name__)
 
-_db_ref: "DatabaseManager | None" = None
-
-
-def set_db(db: "DatabaseManager") -> None:
-    global _db_ref
-    _db_ref = db
-
 
-def _parse_dt(param: str) -> str:
-    if param and param.strip():
-        for part in param.strip().split():
-            if part.startswith("dt="):
-                return part[3:]
-        return param.strip()
+def _default_dt() -> str:
     return (datetime.now() - timedelta(days=2)).strftime("%Y%m%d")
 
 
 @xxl_job("demandEnqueue")
 async def demand_enqueue(param: str) -> dict:
-    dt = _parse_dt(param)
+    dt = parse_dt(param, _default_dt())
     logger.info("demandEnqueue started, dt=%s", dt)
 
     # 1. ODPS 拉取 → PG 查搜索词
     try:
-        pg = _db_ref.pg_manager("pg")
+        pg = get_db().pg_manager("pg")
         fetcher = FetchDemands(OdpsConfig(), pg)
         items = await fetcher.deal(dt)
     except Exception:
@@ -57,7 +42,7 @@ async def demand_enqueue(param: str) -> dict:
 
     # 2. MySQL 入队
     try:
-        mysql = _db_ref.mysql_manager("long_articles")
+        mysql = get_db().mysql_manager("long_articles")
         enqueuer = EnqueueDemands(mysql)
         n = await enqueuer.deal(items)
     except Exception:

src/infra/xxl_jobs/xxl_executor.py

@@ -129,9 +129,7 @@ class XxlJobExecutor:
             except Exception:
                 pass
 
-    async def _callback(
-        self, log_id: int, log_date_time: int, result: dict
-    ) -> None:
+    async def _callback(self, log_id: int, log_date_time: int, result: dict) -> None:
         """向 Admin 回调上报任务执行结果 —— POST /api/callback
 
         对标 Java 版的 TriggerCallbackThread.callback()。
@@ -246,9 +244,7 @@ class XxlJobExecutor:
 
             # 回调 Admin 上报执行结果
             await self._callback(log_id, log_date_time, result)
-            logger.info(
-                "Job %s(id=%s) callback done: %s", handler_name, job_id, result
-            )
+            logger.info("Job %s(id=%s) callback done: %s", handler_name, job_id, result)
 
         task = asyncio.create_task(_execute_and_callback())
         if job_id is not None:

src/server/app.py

@@ -11,12 +11,8 @@ from src.infra.database import DatabaseManager, create_backend
 from src.infra.observability import LogService
 from src.infra.xxl_jobs import XxlJobExecutor
 from src.handlers import discover_and_register
+from src.handlers._utils import set_db
 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
-from src.handlers.account_fetch import set_db as set_db_account_fetch
-from src.handlers.account_article_fetch import set_db as set_db_account_article_fetch
 
 logger = logging.getLogger(__name__)
 
@@ -47,11 +43,7 @@ def create_app(global_config: GlobalConfig = None) -> Quart:
     xxl_executor = XxlJobExecutor(global_config.xxl_job)
 
     set_xxl_executor(xxl_executor)
-    set_db_demand_enqueue(db)
-    set_db_article_search(db)
-    set_db_article_fetch_detail(db)
-    set_db_account_fetch(db)
-    set_db_account_article_fetch(db)
+    set_db(db)
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")