init

src/__init__.py

@@ -62,8 +62,7 @@ from src.domains import (
     DemandQueueMapper,
     DemandSearchArticle,
     ArticleFetchDetail,
-    Account,
-    AccountService,
+    AccountFetch,
 )
 
 # 服务
@@ -72,7 +71,6 @@ from src.server import create_app
 __all__ = [
     # Config
     "GlobalConfig",
-
     # Infra
     "DatabaseManager",
     "MysqlManager",
@@ -104,8 +102,7 @@ __all__ = [
     "DemandQueueMapper",
     "DemandSearchArticle",
     "ArticleFetchDetail",
-    "Account",
-    "AccountService",
+    "AccountFetch",
     # Server
     "create_app",
 ]

src/config/_global_config.py

@@ -75,7 +75,6 @@ class GlobalConfig(BaseSettings):
     # ODPS
     odps: OdpsConfig = Field(default_factory=OdpsConfig)
 
-
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

src/config/apollo.py

@@ -6,7 +6,9 @@ from typing import Dict
 class ApolloConfig(BaseSettings):
     """Apollo 配置中心配置"""
 
-    app_id: str = Field(default="long-articles-agentic-supply", description="Apollo 应用 ID")
+    app_id: str = Field(
+        default="long-articles-agentic-supply", description="Apollo 应用 ID"
+    )
     env: str = Field(default="prod", description="Apollo 环境: dev/pre/prod")
 
     # apollo 配置中心地址映射

src/config/volcengine.py

@@ -23,4 +23,3 @@ class VolcengineConfig(BaseSettings):
         case_sensitive=False,
         extra="ignore",
     )
-

src/domains/__init__.py

@@ -12,9 +12,14 @@
 数据流: fetch_demand → demand_search_article → demand_search_account（下游消费上游产出）
 """
 
-from src.domains.fetch_demand import FetchDemands, EnqueueDemands, DemandQueueMapper, VideoDeconstructMapper
+from src.domains.fetch_demand import (
+    FetchDemands,
+    EnqueueDemands,
+    DemandQueueMapper,
+    VideoDeconstructMapper,
+)
 from src.domains.demand_search_article import DemandSearchArticle, ArticleFetchDetail
-from src.domains.demand_search_account import Account, AccountService
+from src.domains.demand_search_account import AccountFetch
 
 __all__ = [
     # fetch_demand
@@ -26,6 +31,5 @@ __all__ = [
     "DemandSearchArticle",
     "ArticleFetchDetail",
     # demand_search_account
-    "Account",
-    "AccountService",
+    "AccountFetch",
 ]

src/domains/demand_search_account/__init__.py

@@ -1,16 +1,18 @@
-"""账号领域 —— 从抓取到的文章中识别和管理内容账号
+"""账号领域 —— 从 demand_search_article_relation 中提取公众号信息
 
 核心流程:
-  1. 接收 demand_search_article 领域产出中的账号信息
-  2. 账号去重、画像构建（领域、活跃度、质量评分）
-  3. 输出采集策略建议（哪些账号值得持续跟踪）
-  4. 反哺 fetch_demand 领域：高质量账号可作为新的需求来源
+  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
 """
 
-from src.domains.demand_search_account.models import Account
-from src.domains.demand_search_account.service import AccountService
+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
 
 __all__ = [
-    "Account",
-    "AccountService",
+    "DemandSearchAccountConst",
+    "AccountMapper",
+    "AccountFetch",
 ]

src/domains/demand_search_account/_const.py

@@ -0,0 +1,20 @@
+"""账号域常量"""
+
+
+class DemandSearchAccountConst:
+    """账号域所有常量的聚合根"""
+
+    ACCOUNT_TABLE = "demand_search_accounts"
+    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 调用间隔（秒）
+
+
+__all__ = ["DemandSearchAccountConst"]

src/domains/demand_search_account/_mapper.py

@@ -0,0 +1,178 @@
+"""账号域 DB 读写 —— demand_search_accounts + demand_search_article_relation.save_account_status"""
+
+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"""
+        query = f"""
+            SELECT id
+            FROM {self.RELATION_TABLE}
+            WHERE biz = %s AND save_account_status = %s
+        """
+        rows = await self.pool.fetch(
+            query=query,
+            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，用于异常恢复"""
+        query = f"""
+            SELECT id
+            FROM {self.RELATION_TABLE}
+            WHERE biz = %s
+              AND save_account_status IN (%s, %s)
+        """
+        rows = await self.pool.fetch(
+            query=query,
+            params=(
+                biz,
+                self.AccountSaveStatus.INIT,
+                self.AccountSaveStatus.PROCESSING,
+            ),
+        )
+        return [r["id"] for r in rows]
+
+    # ═══════════════════════════════════════════════════════════════
+    # relation 表 — save_account_status 状态更新
+    # ═══════════════════════════════════════════════════════════════
+
+    async def relation_mark_processing(self, ids: List[int]) -> int:
+        """INIT → PROCESSING，CAS 防重复"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.RELATION_TABLE}
+            SET save_account_status = %s
+            WHERE id IN ({placeholders}) AND save_account_status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(
+                self.AccountSaveStatus.PROCESSING,
+                *ids,
+                self.AccountSaveStatus.INIT,
+            ),
+        )
+
+    async def relation_mark_success(self, ids: List[int]) -> int:
+        """PROCESSING → SUCCESS"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.RELATION_TABLE}
+            SET save_account_status = %s
+            WHERE id IN ({placeholders}) AND save_account_status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(
+                self.AccountSaveStatus.SUCCESS,
+                *ids,
+                self.AccountSaveStatus.PROCESSING,
+            ),
+        )
+
+    async def relation_mark_success_direct(self, ids: List[int]) -> int:
+        """INIT → SUCCESS，跳过 PROCESSING（查重命中时用）"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.RELATION_TABLE}
+            SET save_account_status = %s
+            WHERE id IN ({placeholders}) AND save_account_status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.AccountSaveStatus.SUCCESS, *ids, self.AccountSaveStatus.INIT),
+        )
+
+    async def relation_mark_failed(self, ids: List[int]) -> int:
+        """INIT 或 PROCESSING → FAIL，异常恢复用，不挑前置状态"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.RELATION_TABLE}
+            SET save_account_status = %s
+            WHERE id IN ({placeholders})
+              AND save_account_status IN (%s, %s)
+        """
+        return await self.pool.save(
+            query=query,
+            params=(
+                self.AccountSaveStatus.FAIL,
+                *ids,
+                self.AccountSaveStatus.INIT,
+                self.AccountSaveStatus.PROCESSING,
+            ),
+        )
+
+    # ═══════════════════════════════════════════════════════════════
+    # demand_search_accounts 表
+    # ═══════════════════════════════════════════════════════════════
+
+    async def find_by_biz(self, biz: str) -> Optional[Dict]:
+        """按 biz 查账号是否已存在"""
+        query = f"""
+            SELECT * FROM {self.ACCOUNT_TABLE}
+            WHERE biz = %s
+            LIMIT 1
+        """
+        return await self.pool.fetch_one(query=query, 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"],
+            ),
+        )
+
+
+__all__ = ["AccountMapper"]

src/domains/demand_search_account/_utils.py

@@ -0,0 +1,21 @@
+"""账号域工具函数 —— API 响应解析"""
+
+from typing import Dict
+
+
+def parse_account_response(api_data: Dict, biz: str = "") -> Dict:
+    """get_account_from_url 返回的 data.data → accounts 表 INSERT dict
+
+    biz 参数来自 relation 表（URL __biz 提取），比 API 返回值更可靠。
+    """
+    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 "",
+    }
+
+
+__all__ = ["parse_account_response"]

src/domains/demand_search_account/account_fetch.py

@@ -0,0 +1,150 @@
+"""Phase 3: demand_search_article_relation(save_account_status=0) → 账号 API → demand_search_accounts
+
+流程:
+  1. 从 relation 取 DISTINCT biz（save_account_status=0）
+  2. biz 在 accounts 表查重，命中 → 直接 0→2
+  3. 未命中 → 加锁 0→1 → 调 get_account_from_url → 写入 accounts → 1→2
+"""
+
+import asyncio
+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 ._mapper import AccountMapper
+from ._utils import parse_account_response
+
+logger = logging.getLogger(__name__)
+
+
+class AccountFetch(DemandSearchAccountConst):
+    """从 relation 取待抓 biz → 查重 → 调账号 API → 写入 accounts 表"""
+
+    def __init__(self, pool: MysqlManager):
+        self.mapper = AccountMapper(pool)
+
+    async def deal(self, *, limit: int = 100) -> dict:
+        """串行处理待抓账号
+
+        Returns:
+            {"pending": int, "success": int, "skipped": int, "failed": int}
+        """
+        rows = await self.mapper.fetch_pending_relations(limit=limit)
+        if not rows:
+            logger.info("无待抓账号")
+            return {"pending": 0, "success": 0, "skipped": 0, "failed": 0}
+
+        success = 0
+        skipped = 0
+        failed = 0
+        first = True
+        for row in rows:
+            biz = row["biz"]
+            url = row["url"]
+
+            if not first:
+                await asyncio.sleep(self.ACCOUNT_INTERVAL_SEC)
+            first = False
+
+            try:
+                outcome = await self._fetch_account(biz, url)
+            except Exception:
+                logger.exception("账号抓取异常: biz=%s, url=%s", biz, url)
+                # 异常恢复：尝试找到所有未完成的行（含 INIT 和 PROCESSING）标失败
+                ids = await self.mapper.fetch_relation_ids_by_biz_any_status(biz)
+                if ids:
+                    await self.mapper.relation_mark_failed(ids)
+                failed += 1
+                continue
+
+            if outcome == "success":
+                success += 1
+            elif outcome == "skipped":
+                skipped += 1
+            else:
+                failed += 1
+
+        logger.info(
+            "账号抓取完成: total=%d, success=%d, skipped=%d, failed=%d",
+            len(rows),
+            success,
+            skipped,
+            failed,
+        )
+        return {
+            "pending": len(rows),
+            "success": success,
+            "skipped": skipped,
+            "failed": failed,
+        }
+
+    # ═══════════════════════════════════════════════════════════════
+    # 单 biz 处理
+    # ═══════════════════════════════════════════════════════════════
+
+    async def _fetch_account(self, biz: str, url: str) -> str:
+        """查重 → 命中直接 0→2 / 未命中 0→1→API→2
+
+        Returns:
+            "success" / "skipped" / "failed"
+        """
+        # 1. 获取该 biz 下所有待处理的 relation 行
+        relation_ids = await self.mapper.fetch_relation_ids_by_biz(biz)
+        if not relation_ids:
+            logger.warning("biz=%s 无待处理 relation 行", biz)
+            return "failed"
+
+        # 2. 先查重（不加锁），命中则直接 0→2，不调 API
+        existing = await self.mapper.find_by_biz(biz)
+        if existing:
+            logger.info(
+                "账号已存在，跳过: biz=%s, channel_account_id=%s",
+                biz,
+                existing.get("channel_account_id"),
+            )
+            await self.mapper.relation_mark_success_direct(relation_ids)
+            return "skipped"
+
+        # 3. 未命中 → 加锁 0→1，检查受影响行数防并发重复调用
+        locked = await self.mapper.relation_mark_processing(relation_ids)
+        if locked == 0:
+            logger.info("biz=%s 已被其他实例认领，跳过", biz)
+            return "skipped"
+
+        # 4. 调 API
+        resp = await get_account_from_url(url)
+        if not resp or resp.get("code") != 0:
+            logger.warning(
+                "get_account_from_url 返回异常: biz=%s, url=%s, code=%s",
+                biz,
+                url,
+                resp.get("code") if resp else "None",
+            )
+            await self.mapper.relation_mark_failed(relation_ids)
+            return "failed"
+
+        api_data = (resp.get("data") or {}).get("data")
+        if not api_data:
+            logger.warning("账号数据为空: biz=%s, url=%s", biz, url)
+            await self.mapper.relation_mark_failed(relation_ids)
+            return "failed"
+
+        # 5. 解析并写入 accounts 表（传入 biz 保证和 relation 表一致）
+        account_row = parse_account_response(api_data, biz=biz)
+        await self.mapper.insert_account(account_row)
+
+        # 6. 1→2
+        await self.mapper.relation_mark_success(relation_ids)
+
+        logger.info(
+            "账号抓取成功: biz=%s, channel_account_id=%s, name=%s",
+            biz,
+            account_row["channel_account_id"],
+            account_row["account_name"],
+        )
+        return "success"
+
+
+__all__ = ["AccountFetch"]

src/domains/demand_search_account/models.py

@@ -1,23 +0,0 @@
-"""账号领域模型"""
-
-from dataclasses import dataclass, field
-from datetime import datetime
-
-
-@dataclass
-class Account:
-    """内容账号模型
-
-    从文章元信息中反推识别，逐步构建画像。
-    """
-    account_id: str                                 # 平台唯一 ID（如公众号 fakeid）
-    account_name: str                               # 账号名称
-    platform: str = "wechat"                        # 所属平台
-    description: str = ""                           # 账号简介
-    domain_tags: list[str] = field(default_factory=list)   # 领域标签
-    article_count_crawled: int = 0                  # 已抓取文章数
-    quality_score: float = 0.0                      # 质量评分 (0-100)
-    is_active: bool = True                          # 是否活跃采集
-    first_seen_at: datetime | None = None
-    last_crawled_at: datetime | None = None
-    extra: dict = field(default_factory=dict)

src/domains/demand_search_account/service.py

@@ -1,49 +0,0 @@
-"""账号管理服务"""
-
-import logging
-from dataclasses import dataclass, field
-from datetime import datetime
-
-from src.domains.demand_search_account.models import Account
-
-
-@dataclass
-class Article:
-    """文章内存对象 —— 从 demand_search_article_detail 结构化后供 account 域消费"""
-    channel_content_id: str
-    url: str
-    title: str = ""
-    body_text: str = ""
-    channel_account_id: str = ""
-    channel_account_name: str = ""
-    publish_at: datetime | None = None
-    source_platform: str = "wechat"
-    extra: dict = field(default_factory=dict)
-
-logger = logging.getLogger(__name__)
-
-
-class AccountService:
-    """账号管理服务
-
-    职责: 从 Article 中提取账号信息，去重合并，构建账号画像，输出采集策略。
-    不负责: 文章的抓取（由 demand_search_article 领域负责）。
-    """
-
-    async def extract_account_from_article(self, article: Article) -> Account | None:
-        """从单篇文章提取账号信息"""
-        raise NotImplementedError
-
-    async def upsert_account(self, account: Account) -> Account:
-        """新建或更新账号信息（去重合并）"""
-        raise NotImplementedError
-
-    async def build_profile(self, account_id: str) -> Account:
-        """构建/更新账号画像（领域标签、质量评分等）"""
-        raise NotImplementedError
-
-    async def list_active_accounts(
-        self, platform: str = "wechat", min_score: float = 0.0
-    ) -> list[Account]:
-        """列出活跃采集的账号列表，按质量评分降序"""
-        raise NotImplementedError

src/domains/demand_search_article/_const.py

@@ -47,26 +47,26 @@ class DemandSearchArticleConst:
     # ═══════════════════════════════════════════════════════════════
 
     class SearchStatus:
-        INIT = 0        # 待拉取详情
+        INIT = 0  # 待拉取详情
         PROCESSING = 1  # 拉取详情中
-        SUCCESS = 2     # 拉取详情成功
-        FAIL = 99       # 详情拉取失败
+        SUCCESS = 2  # 拉取详情成功
+        FAIL = 99  # 详情拉取失败
 
     # ═══════════════════════════════════════════════════════════════
     # 搜索参数
     # ═══════════════════════════════════════════════════════════════
 
-    MAX_SEARCH_PAGES = 3        # 每个搜索词最多翻多少页
-    CACHE_TTL_DAYS = 10         # 搜索缓存有效期（天）
-    SEARCH_INTERVAL_SEC = 5     # 搜索页间延迟（秒）
-    DETAIL_INTERVAL_SEC = 3     # 详情拉取条间延迟（秒）
+    MAX_SEARCH_PAGES = 3  # 每个搜索词最多翻多少页
+    CACHE_TTL_DAYS = 10  # 搜索缓存有效期（天）
+    SEARCH_INTERVAL_SEC = 5  # 搜索页间延迟（秒）
+    DETAIL_INTERVAL_SEC = 3  # 详情拉取条间延迟（秒）
 
     # ═══════════════════════════════════════════════════════════════
     # 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 详情：单次拉取条数上限（串行 3s/条，小批次高频）
 
 
 __all__ = ["DemandSearchArticleConst"]

src/domains/demand_search_article/_mapper.py

@@ -18,7 +18,9 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
     # demand_search_queue 读取 & 状态更新
     # ═══════════════════════════════════════════════════════════════
 
-    async def fetch_queue_pending(self, dt: str | None = None, *, limit: int = 500) -> List[Dict]:
+    async def fetch_queue_pending(
+        self, dt: str | None = None, *, limit: int = 500
+    ) -> List[Dict]:
         """取待搜索的队列项（id 升序，FIFO）
 
         传 dt 则按分区过滤，不传则全表扫描。
@@ -67,7 +69,12 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         """
         return await self.pool.save(
             query=query,
-            params=(self.QueueStatus.SUCCESS, remark, item_id, self.QueueStatus.PROCESSING),
+            params=(
+                self.QueueStatus.SUCCESS,
+                remark,
+                item_id,
+                self.QueueStatus.PROCESSING,
+            ),
         )
 
     async def queue_mark_failed(self, item_id: int, reason: str = "") -> int:
@@ -79,7 +86,12 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         """
         return await self.pool.save(
             query=query,
-            params=(self.QueueStatus.FAIL, reason[:512], item_id, self.QueueStatus.PROCESSING),
+            params=(
+                self.QueueStatus.FAIL,
+                reason[:512],
+                item_id,
+                self.QueueStatus.PROCESSING,
+            ),
         )
 
     # ═══════════════════════════════════════════════════════════════
@@ -183,15 +195,15 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         )
 
     async def mark_success(self, item_id: int, *, channel_content_id: str = "") -> int:
-        """标记为成功，回填 channel_content_id (→2)"""
+        """PROCESSING → SUCCESS，回填 channel_content_id，CAS 防覆盖终态"""
         query = f"""
             UPDATE {self.RELATION_TABLE}
             SET status = %s, channel_content_id = %s
-            WHERE id = %s
+            WHERE id = %s AND status = %s
         """
         return await self.pool.save(
             query=query,
-            params=(self.SearchStatus.SUCCESS, channel_content_id, item_id),
+            params=(self.SearchStatus.SUCCESS, channel_content_id, item_id, self.SearchStatus.PROCESSING),
         )
 
     async def mark_failed(self, ids: List[int]) -> int:
@@ -209,7 +221,6 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
             params=(self.SearchStatus.FAIL, *ids, self.SearchStatus.PROCESSING),
         )
 
-
     # ═══════════════════════════════════════════════════════════════
     # 搜索缓存
     # ═══════════════════════════════════════════════════════════════
@@ -235,7 +246,10 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         return resp
 
     async def cache_set(
-        self, keyword: str, page: int, response: Dict,
+        self,
+        keyword: str,
+        page: int,
+        response: Dict,
     ) -> int:
         """写入缓存（upsert）"""
         query = f"""

src/domains/demand_search_article/_utils.py

@@ -53,9 +53,7 @@ def parse_search_result(
 ) -> Dict:
     """微信搜索 API 返回的单篇文章 → relation 表 INSERT dict"""
     ts = article.get("time")
-    publish_time = (
-        datetime.fromtimestamp(ts) if ts and ts > 0 else None
-    )
+    publish_time = datetime.fromtimestamp(ts) if ts and ts > 0 else None
     article_url = article.get("url", "")
     return {
         "search_key": keyword,
@@ -74,9 +72,7 @@ def parse_search_result(
 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
-    )
+    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 ""
@@ -106,4 +102,10 @@ def parse_detail(detail: Dict) -> Dict:
     }
 
 
-__all__ = ["extract_wx_sn", "extract_biz", "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

@@ -54,7 +54,8 @@ class ArticleFetchDetail(DemandSearchArticleConst):
                 outcome = await self._fetch_detail_and_save(item)
             except Exception:
                 logger.exception(
-                    "详情拉取异常: relation_id=%s", item["id"],
+                    "详情拉取异常: relation_id=%s",
+                    item["id"],
                 )
                 await self.relation_mapper.mark_failed([item["id"]])
                 failed += 1
@@ -67,7 +68,9 @@ class ArticleFetchDetail(DemandSearchArticleConst):
 
         logger.info(
             "详情拉取完成: total=%d, success=%d, failed=%d",
-            len(items), success, failed,
+            len(items),
+            success,
+            failed,
         )
         return {
             "pending": len(items),
@@ -95,7 +98,9 @@ class ArticleFetchDetail(DemandSearchArticleConst):
             resp = await get_article_detail(url)
         except Exception:
             logger.exception(
-                "get_article_detail 调用异常: url=%s, relation_id=%s", url, item["id"],
+                "get_article_detail 调用异常: url=%s, relation_id=%s",
+                url,
+                item["id"],
             )
             await self.relation_mapper.mark_failed([item["id"]])
             return "failed"
@@ -103,7 +108,9 @@ class ArticleFetchDetail(DemandSearchArticleConst):
         if not resp or resp.get("code") != 0:
             logger.warning(
                 "get_article_detail 返回异常: url=%s, relation_id=%s, code=%s",
-                url, item["id"], resp.get("code") if resp else "None",
+                url,
+                item["id"],
+                resp.get("code") if resp else "None",
             )
             await self.relation_mapper.mark_failed([item["id"]])
             return "failed"
@@ -111,7 +118,9 @@ class ArticleFetchDetail(DemandSearchArticleConst):
         detail_data = (resp.get("data") or {}).get("data")
         if not detail_data:
             logger.warning(
-                "详情数据为空: url=%s, relation_id=%s", url, item["id"],
+                "详情数据为空: url=%s, relation_id=%s",
+                url,
+                item["id"],
             )
             await self.relation_mapper.mark_failed([item["id"]])
             return "failed"
@@ -119,7 +128,9 @@ class ArticleFetchDetail(DemandSearchArticleConst):
         channel_content_id = detail_data.get("channel_content_id", "")
         if not channel_content_id:
             logger.warning(
-                "channel_content_id 为空: url=%s, relation_id=%s", url, item["id"],
+                "channel_content_id 为空: url=%s, relation_id=%s",
+                url,
+                item["id"],
             )
             await self.relation_mapper.mark_failed([item["id"]])
             return "failed"
@@ -127,12 +138,15 @@ class ArticleFetchDetail(DemandSearchArticleConst):
         detail_row = parse_detail(detail_data)
         await self.detail_mapper.insert_one(detail_row)
         await self.relation_mapper.mark_success(
-            item["id"], channel_content_id=channel_content_id,
+            item["id"],
+            channel_content_id=channel_content_id,
         )
 
         logger.debug(
             "详情拉取成功: relation_id=%s, channel_content_id=%s, title=%s",
-            item["id"], channel_content_id, detail_row.get("title", "")[:40],
+            item["id"],
+            channel_content_id,
+            detail_row.get("title", "")[:40],
         )
         return "success"
 

src/domains/demand_search_article/article_search.py

@@ -54,28 +54,37 @@ class DemandSearchArticle(DemandSearchArticleConst):
             except Exception as e:
                 logger.exception(
                     "搜索失败: queue_id=%s, experiment_id=%s",
-                    item["id"], item.get("experiment_id"),
+                    item["id"],
+                    item.get("experiment_id"),
                 )
                 await self.relation_mapper.queue_mark_failed(
-                    item["id"], reason=str(e),
+                    item["id"],
+                    reason=str(e),
                 )
                 continue
 
             if n > 0:
                 total_written += n
                 processed += 1
-                await self.relation_mapper.queue_mark_done(item["id"], remark="搜索成功")
+                await self.relation_mapper.queue_mark_done(
+                    item["id"], remark="搜索成功"
+                )
             elif from_cache_only:
                 processed += 1
-                await self.relation_mapper.queue_mark_done(item["id"], remark="命中缓存")
+                await self.relation_mapper.queue_mark_done(
+                    item["id"], remark="命中缓存"
+                )
             else:
                 await self.relation_mapper.queue_mark_failed(
-                    item["id"], reason="搜索词未返回结果",
+                    item["id"],
+                    reason="搜索词未返回结果",
                 )
 
         logger.info(
             "搜索完成: 处理 %d/%d 项, 写入 %d 条结果",
-            processed, len(items), total_written,
+            processed,
+            len(items),
+            total_written,
         )
         return {
             "queued": len(items),
@@ -136,7 +145,10 @@ class DemandSearchArticle(DemandSearchArticleConst):
                 newly_written += n
                 logger.info(
                     "搜索写入: keyword=%s, page=%d, 返回 %d 条, 写入 %d 条",
-                    keyword, page, len(articles), n,
+                    keyword,
+                    page,
+                    len(articles),
+                    n,
                 )
                 await asyncio.sleep(self.SEARCH_INTERVAL_SEC)
 
@@ -150,7 +162,10 @@ class DemandSearchArticle(DemandSearchArticleConst):
         return newly_written, from_cache_only
 
     async def _search_with_cache(
-        self, keyword: str, page: int, cursor: str,
+        self,
+        keyword: str,
+        page: int,
+        cursor: str,
     ) -> tuple[Dict | None, bool]:
         """返回 (result, from_cache)。miss 调 API 并写缓存。"""
         cached = await self.relation_mapper.cache_get(keyword, page)

src/domains/fetch_demand/_const.py

@@ -30,9 +30,9 @@ class DemandConst:
 
     class KeyType:
         INSPIRATION_SUBSTANCE = "INSPIRATION_SUBSTANCE"
-        VIDEO_INSPIRATION =    "VIDEO_INSPIRATION"
-        VIDEO_KEYPOINT =       "VIDEO_KEYPOINT"
-        VIDEO_TITLE =          "VIDEO_TITLE"
+        VIDEO_INSPIRATION = "VIDEO_INSPIRATION"
+        VIDEO_KEYPOINT = "VIDEO_KEYPOINT"
+        VIDEO_TITLE = "VIDEO_TITLE"
 
     # ═══════════════════════════════════════════════════════════════
     # 队列
@@ -42,10 +42,10 @@ class DemandConst:
     DRIVE_TYPE_VIDEO = "VIDEO"
 
     class QueueStatus:
-        INIT = 0        # 初始，待处理
+        INIT = 0  # 初始，待处理
         PROCESSING = 1  # 处理中
-        SUCCESS = 2     # 处理成功
-        FAIL = 99       # 处理失败
+        SUCCESS = 2  # 处理成功
+        FAIL = 99  # 处理失败
 
 
-__all__ = ["DemandConst"]
+__all__ = ["DemandConst"]

src/domains/fetch_demand/_mapper.py

@@ -129,7 +129,8 @@ class DemandQueueMapper(DemandConst):
             WHERE id = %s
         """
         return await self.pool.save(
-            query=query, params=(self.QueueStatus.SUCCESS, item_id),
+            query=query,
+            params=(self.QueueStatus.SUCCESS, item_id),
         )
 
     async def mark_failed(self, item_id: int, reason: str = "") -> int:

src/domains/fetch_demand/enqueue_demands.py

@@ -44,7 +44,9 @@ class EnqueueDemands(DemandConst):
         videos = len(set(it["video_id"] for it in items))
         logger.info(
             "入队完成: %d 条, %d 个账号, %d 个视频 → demand_search_queue",
-            n, accounts, videos,
+            n,
+            accounts,
+            videos,
         )
         return n
 

src/domains/fetch_demand/fetch_demands.py

@@ -164,7 +164,8 @@ class FetchDemands(DemandConst):
         result = list(groups.values())
         logger.info(
             "total_rov 去重: %d 行 → %d 条 (按 账号+视频 分组)",
-            len(rows), len(result),
+            len(rows),
+            len(result),
         )
         return result
 
@@ -199,19 +200,27 @@ class FetchDemands(DemandConst):
             return []
 
         # 分页批量查询（带进度条）
-        total_batches = (len(ordered_ids) + self._VIDEO_BATCH_SIZE - 1) // self._VIDEO_BATCH_SIZE
+        total_batches = (
+            len(ordered_ids) + self._VIDEO_BATCH_SIZE - 1
+        ) // self._VIDEO_BATCH_SIZE
         vectors_map: dict[int, list[Dict]] = {}
         with tqdm(total=len(ordered_ids), desc="video_vectors", unit="vid") as pbar:
             for i in range(0, len(ordered_ids), self._VIDEO_BATCH_SIZE):
-                batch = ordered_ids[i:i + self._VIDEO_BATCH_SIZE]
+                batch = ordered_ids[i : i + self._VIDEO_BATCH_SIZE]
                 chunk = await self._deconstruct_mapper.find_by_video_ids(batch)
                 vectors_map.update(chunk)
                 pbar.update(len(batch))
-                pbar.set_postfix({"hit": len(vectors_map), "batch": f"{i // self._VIDEO_BATCH_SIZE + 1}/{total_batches}"})
+                pbar.set_postfix(
+                    {
+                        "hit": len(vectors_map),
+                        "batch": f"{i // self._VIDEO_BATCH_SIZE + 1}/{total_batches}",
+                    }
+                )
 
         logger.info(
             "video_vectors 批量查询完成: %d 个 video_id → %d 个有结果",
-            len(ordered_ids), len(vectors_map),
+            len(ordered_ids),
+            len(vectors_map),
         )
 
         # 带回分发到每条 ODPS 行
@@ -251,12 +260,14 @@ class FetchDemands(DemandConst):
         }
 
         # 仅保留四种有效 key_type，过滤 video_vectors 中其他 config_code
-        _VALID_KEY_TYPES = frozenset({
-            DemandConst.KeyType.INSPIRATION_SUBSTANCE,
-            DemandConst.KeyType.VIDEO_INSPIRATION,
-            DemandConst.KeyType.VIDEO_KEYPOINT,
-            DemandConst.KeyType.VIDEO_TITLE,
-        })
+        _VALID_KEY_TYPES = frozenset(
+            {
+                DemandConst.KeyType.INSPIRATION_SUBSTANCE,
+                DemandConst.KeyType.VIDEO_INSPIRATION,
+                DemandConst.KeyType.VIDEO_KEYPOINT,
+                DemandConst.KeyType.VIDEO_TITLE,
+            }
+        )
 
         items: List[Dict] = []
         seen: Set[str] = set()
@@ -272,11 +283,13 @@ class FetchDemands(DemandConst):
             if key in seen:
                 continue
             seen.add(key)
-            items.append({
-                **base,
-                "search_key": text_val,
-                "key_type": key_type,
-            })
+            items.append(
+                {
+                    **base,
+                    "search_key": text_val,
+                    "key_type": key_type,
+                }
+            )
 
         return items
 

src/handlers/account_fetch.py

@@ -0,0 +1,50 @@
+"""账号信息抓取 XXL-JOB Handler —— relation(save_account_status=0) → 账号 API → accounts 表
+
+XXL-JOB Admin 配置:
+    执行器: long-articles-agentic-src
+    处理器: accountFetch
+"""
+
+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
+
+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
+    logger.info("accountFetch started, limit=%d", limit)
+
+    try:
+        mysql = _db_ref.mysql_manager("long_articles")
+        fetcher = AccountFetch(mysql)
+        result = await fetcher.deal(limit=limit)
+    except Exception:
+        logger.exception("账号抓取失败")
+        return {"code": 500, "msg": "账号抓取异常"}
+
+    logger.info(
+        "accountFetch done: pending=%d, success=%d, skipped=%d, failed=%d",
+        result["pending"],
+        result["success"],
+        result["skipped"],
+        result["failed"],
+    )
+    return {
+        "code": 200,
+        "msg": f"success={result['success']}, skipped={result['skipped']}, failed={result['failed']}",
+    }

src/handlers/article_fetch_detail.py

@@ -3,14 +3,9 @@
 XXL-JOB Admin 配置:
     执行器: long-articles-agentic-src
     处理器: articleFetchDetail
-
-设计:
-    - Admin 调度的 HTTP 请求秒级返回，实际拉取在后台异步执行
-    - limit 取自 DemandSearchArticleConst.DETAIL_DEAL_LIMIT，全表取 status=0 的行
-    - MySQL 复用 app 级别的连接池（set_db() 注入），不每次自建
+    超时:   建议 ≥600s（100条 × 3s间隔 + API 调用）
 """
 
-import asyncio
 import logging
 from typing import TYPE_CHECKING
 
@@ -22,38 +17,32 @@ if TYPE_CHECKING:
 
 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 表写入"""
+@xxl_job("articleFetchDetail")
+async def article_fetch_detail(param: str) -> dict:
     limit = DemandSearchArticleConst.DETAIL_DEAL_LIMIT
-    logger.info("articleFetchDetail background task started, limit=%d", limit)
+    logger.info("articleFetchDetail 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.exception("详情拉取失败")
+        return {"code": 500, "msg": "详情拉取异常"}
 
     logger.info(
         "articleFetchDetail done: pending=%d, success=%d, failed=%d",
         result["pending"], result["success"], result["failed"],
     )
-
-
-@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"}
+    return {
+        "code": 200,
+        "msg": f"success={result['success']}, failed={result['failed']}",
+    }

src/handlers/article_search.py

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

src/handlers/demand_enqueue.py

@@ -4,13 +4,9 @@ XXL-JOB Admin 配置:
     执行器: long-articles-agentic-src
     处理器: demandEnqueue
     参数:   dt=yyyyMMdd (可选，不传则默认前天)
-
-设计:
-    - Admin 调度的 HTTP 请求秒级返回，实际入队在后台异步执行
-    - MySQL 复用 app 级别的连接池（set_db() 注入），不每次自建
+    超时:   建议 ≥300s（ODPS + PG + MySQL 多段查询）
 """
 
-import asyncio
 import logging
 from datetime import datetime, timedelta
 from typing import TYPE_CHECKING
@@ -24,12 +20,10 @@ if TYPE_CHECKING:
 
 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
 
@@ -43,40 +37,36 @@ def _parse_dt(param: str) -> str:
     return (datetime.now() - timedelta(days=2)).strftime("%Y%m%d")
 
 
-async def _do_enqueue(dt: str) -> None:
-    """后台执行: ODPS 拉取 → 白名单过滤 → rov 去重 → 视频解构 → MySQL 入队"""
-    logger.info("demandEnqueue background task started, dt=%s", dt)
+@xxl_job("demandEnqueue")
+async def demand_enqueue(param: str) -> dict:
+    dt = _parse_dt(param)
+    logger.info("demandEnqueue started, dt=%s", dt)
 
+    # 1. ODPS 拉取 → PG 查搜索词
     try:
         pg = _db_ref.pg_manager("pg")
         fetcher = FetchDemands(OdpsConfig(), pg)
         items = await fetcher.deal(dt)
-        if not items:
-            logger.info("无队列项生成, dt=%s", dt)
-            return
     except Exception:
-        logger.exception("ODPS 查询失败, dt=%s", dt)
-        return
+        logger.exception("ODPS 拉取失败, dt=%s", dt)
+        return {"code": 500, "msg": "ODPS 拉取异常"}
+
+    if not items:
+        logger.info("无队列项生成, dt=%s", dt)
+        return {"code": 200, "msg": "no items"}
 
+    # 2. MySQL 入队
     try:
         mysql = _db_ref.mysql_manager("long_articles")
         enqueuer = EnqueueDemands(mysql)
         n = await enqueuer.deal(items)
     except Exception:
         logger.exception("MySQL 入队失败, dt=%s", dt)
-        return
+        return {"code": 500, "msg": "入队异常"}
 
     accounts = len(set(it["account"] for it in items))
     logger.info(
         "demandEnqueue done: enqueued %d items (%d accounts) for dt=%s",
         n, accounts, dt,
     )
-
-
-@xxl_job("demandEnqueue")
-async def demand_enqueue(param: str) -> dict:
-    """需求入队 handler — 秒级返回，实际工作在后台执行"""
-    dt = _parse_dt(param)
-    asyncio.create_task(_do_enqueue(dt))
-    logger.info("demandEnqueue accepted, dt=%s (background task spawned)", dt)
-    return {"code": 200, "msg": f"accepted, dt={dt}"}
+    return {"code": 200, "msg": f"enqueued={n}, accounts={accounts}"}

src/infra/database/manager.py

@@ -97,17 +97,20 @@ class DatabaseManager:
     def mysql_manager(self, db_name: str = "long_articles"):
         """获取 MySQL 专用门面"""
         from .mysql.manager import MysqlManager
+
         return MysqlManager(self._get_sql(db_name))
 
     def pg_manager(self, db_name: str = "vector"):
         """获取 PostgreSQL 专用门面"""
         from .postgresql.manager import PgManager
+
         return PgManager(self._get_sql(db_name))
 
     def redis_manager(self, db_name: str = "default"):
         """获取 Redis 专用门面"""
         from .redis.backend import RedisBackend
         from .redis.manager import RedisManager
+
         return RedisManager(self._get_typed(db_name, RedisBackend))
 
     # ==================== 生命周期 ====================
@@ -118,7 +121,9 @@ class DatabaseManager:
             for name, backend in self._backends.items():
                 await backend.init()
                 initialized.append((name, backend))
-                logger.info("存储后端 [%s] 初始化完成 → %s", name, type(backend).__name__)
+                logger.info(
+                    "存储后端 [%s] 初始化完成 → %s", name, type(backend).__name__
+                )
         except Exception as exc:
             logger.exception("存储后端初始化失败，开始回滚已初始化连接: %s", exc)
             for name, backend in reversed(initialized):
@@ -138,18 +143,22 @@ class DatabaseManager:
 
     def mysql(self, name: Optional[str] = None):
         from .mysql.backend import MySQLBackend
+
         return self._get_typed(name or self._default_db, MySQLBackend)
 
     def pg(self, name: str):
         from .postgresql.backend import PgBackend
+
         return self._get_typed(name, PgBackend)
 
     def milvus(self, name: str):
         from .milvus.backend import MilvusBackend
+
         return self._get_typed(name, MilvusBackend)
 
     def redis(self, name: str):
         from .redis.backend import RedisBackend
+
         return self._get_typed(name, RedisBackend)
 
     # ==================== 辅助 ====================

src/infra/external/__init__.py

@@ -1,4 +1,5 @@
 """外部服务集成 —— 第三方 API 客户端"""
+
 from .apollo import AsyncApolloApi
 from .feishu import FeishuBotApi, FeishuSheetApi
 from .odps import fetch_from_odps
@@ -11,4 +12,4 @@ __all__ = [
     "FeishuSheetApi",
     "fetch_from_odps",
     "fetch_deepseek_completion",
-]
+]

src/infra/shared/tools.py

@@ -8,6 +8,7 @@ import string
 from datetime import datetime, timezone, date, timedelta
 from typing import Any, Dict, List, Optional
 
+
 def safe_json_parse(text: str) -> Optional[Dict | List]:
     """多层降级解析 JSON：直接解析 → 提取代码块 → 提取 JSON 对象/数组
 
@@ -140,7 +141,9 @@ def request_retry(
 
     return {
         "stop": stop_after_attempt(retry_times),
-        "wait": wait_exponential(multiplier=1, min=min_retry_delay, max=max_retry_delay),
+        "wait": wait_exponential(
+            multiplier=1, min=min_retry_delay, max=max_retry_delay
+        ),
         "retry": retry_if_exception_type(Exception),
         "reraise": True,
     }

src/infra/spider/__init__.py

@@ -1,4 +1,5 @@
 """爬虫组件 —— 多平台内容抓取"""
+
 from src.infra.spider.wechat import (
     get_article_detail,
     get_article_list_from_account,

src/infra/spider/wechat/__init__.py

@@ -9,4 +9,4 @@ __all__ = [
     "get_article_list_from_account",
     "get_source_account_from_article",
     "weixin_search",
-]
+]

src/infra/spider/wechat/gzh.py

@@ -53,9 +53,7 @@ async def get_article_detail(
         async with AsyncHttpClient(timeout=10) as http_client:
             response = await http_client.post(target_url, headers=HEADERS, data=payload)
     except Exception as e:
-        logger.error(
-            "get_article_detail API请求失败: %s, link=%s", e, article_link
-        )
+        logger.error("get_article_detail API请求失败: %s, link=%s", e, article_link)
         return None
     return response
 
@@ -86,7 +84,9 @@ async def get_article_list_from_account(
     except Exception as e:
         logger.error(
             "get_article_list_from_account API请求失败: %s, account_id=%s, index=%s",
-            e, account_id, index,
+            e,
+            account_id,
+            index,
         )
         return None
     return response
@@ -148,4 +148,24 @@ async def weixin_search(keyword: str, page: str = "1") -> dict | None:
             "weixin_search API请求失败: %s, keyword=%s, page=%s", e, keyword, page
         )
         return None
-    return response
+    return response
+
+
+@retry(**retry_desc)
+async def get_account_from_url(article_url) -> dict | None:
+    """微信关键词搜索
+
+    Args:
+        article_url: 文章链接
+    """
+    url = f"{BASE_URL}/account_info"
+    payload = json.dumps({"content_link": article_url})
+    try:
+        async with AsyncHttpClient(timeout=120) as http_client:
+            response = await http_client.post(url=url, headers=HEADERS, data=payload)
+    except Exception as e:
+        logger.error(
+            "get_account_from_url API请求失败: %s, article_url=%s", e, article_url
+        )
+        return None
+    return response

src/server/app.py

@@ -15,6 +15,7 @@ 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
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +49,7 @@ def create_app(global_config: GlobalConfig = None) -> Quart:
     set_db_demand_enqueue(db)
     set_db_article_search(db)
     set_db_article_fetch_detail(db)
+    set_db_account_fetch(db)
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")