init

src/__init__.py

@@ -63,6 +63,7 @@ from src.domains import (
     DemandSearchArticle,
     ArticleFetchDetail,
     AccountFetch,
+    AccountArticleFetch,
 )
 
 # 服务
@@ -103,6 +104,7 @@ __all__ = [
     "DemandSearchArticle",
     "ArticleFetchDetail",
     "AccountFetch",
+    "AccountArticleFetch",
     # Server
     "create_app",
 ]

src/domains/__init__.py

@@ -19,7 +19,7 @@ from src.domains.fetch_demand import (
     VideoDeconstructMapper,
 )
 from src.domains.demand_search_article import DemandSearchArticle, ArticleFetchDetail
-from src.domains.demand_search_account import AccountFetch
+from src.domains.demand_search_account import AccountFetch, AccountArticleFetch
 
 __all__ = [
     # fetch_demand
@@ -32,4 +32,5 @@ __all__ = [
     "ArticleFetchDetail",
     # demand_search_account
     "AccountFetch",
+    "AccountArticleFetch",
 ]

src/domains/demand_search_account/__init__.py

@@ -1,18 +1,26 @@
 """账号领域 —— 从 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

@@ -5,6 +5,7 @@ class DemandSearchAccountConst:
     """账号域所有常量的聚合根"""
 
     ACCOUNT_TABLE = "demand_search_accounts"
+    ARTICLES_TABLE = "demand_account_articles"
     RELATION_TABLE = "demand_search_article_relation"
 
     class AccountSaveStatus:
@@ -13,8 +14,13 @@ class DemandSearchAccountConst:
         SUCCESS = 2
         FAIL = 99
 
-    ACCOUNT_DEAL_LIMIT = 100  # 单次取待抓账号条数
-    ACCOUNT_INTERVAL_SEC = 1  # API 调用间隔（秒）
+    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,5 +1,6 @@
-"""账号域 DB 读写 —— demand_search_accounts + demand_search_article_relation.save_account_status"""
+"""账号域 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
@@ -28,38 +29,22 @@ class AccountMapper(DemandSearchAccountConst):
             LIMIT %s
         """
         return await self.pool.fetch(
-            query=query,
-            params=(self.AccountSaveStatus.INIT, limit),
+            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,
+            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，用于异常恢复"""
-        query = f"""
-            SELECT id
-            FROM {self.RELATION_TABLE}
-            WHERE biz = %s
-              AND save_account_status IN (%s, %s)
-        """
+        """取某个 biz 下未完成（INIT/PROCESSING）的 relation id，异常恢复用"""
         rows = await self.pool.fetch(
-            query=query,
-            params=(
-                biz,
-                self.AccountSaveStatus.INIT,
-                self.AccountSaveStatus.PROCESSING,
-            ),
+            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]
 
@@ -67,78 +52,38 @@ class AccountMapper(DemandSearchAccountConst):
     # relation 表 — save_account_status 状态更新
     # ═══════════════════════════════════════════════════════════════
 
-    async def relation_mark_processing(self, ids: List[int]) -> int:
-        """INIT → PROCESSING，CAS 防重复"""
+    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 = %s
+            WHERE id IN ({placeholders}) AND save_account_status IN ({status_placeholders})
         """
         return await self.pool.save(
             query=query,
-            params=(
-                self.AccountSaveStatus.PROCESSING,
-                *ids,
-                self.AccountSaveStatus.INIT,
-            ),
+            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"""
-        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,
-            ),
-        )
+        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（查重命中时用）"""
-        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),
-        )
+        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，异常恢复用，不挑前置状态"""
-        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,
-            ),
+        """INIT/PROCESSING → FAIL，异常恢复用"""
+        return await self._mark_batch(
+            self.AccountSaveStatus.FAIL, ids, [self.AccountSaveStatus.INIT, self.AccountSaveStatus.PROCESSING],
         )
 
     # ═══════════════════════════════════════════════════════════════
@@ -147,32 +92,86 @@ class AccountMapper(DemandSearchAccountConst):
 
     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,))
+        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)
+            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"],
-            ),
+            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,13 +1,13 @@
 """账号域工具函数 —— API 响应解析"""
 
-from typing import Dict
+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
 
-    biz 参数来自 relation 表（URL __biz 提取），比 API 返回值更可靠。
-    """
+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 "",
@@ -18,4 +18,133 @@ def parse_account_response(api_data: Dict, biz: str = "") -> Dict:
     }
 
 
-__all__ = ["parse_account_response"]
+# ═══════════════════════════════════════════════════════════════
+# 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_account/account_article_fetch.py

@@ -0,0 +1,150 @@
+"""Phase 4: 从 accounts(gh_id) → get_article_list_from_account → demand_account_articles
+
+流程:
+  1. 取 accounts 表中所有有 gh_id 的账号（crawl_deep_done=0 的优先）
+  2. 每个账号:
+     - 0: 首次深翻 → 翻页 → send_time < 一年前停止 → 标记 deep_done=1
+     - 1: 增量更新 → 只取第一页
+  3. 文章 INSERT IGNORE (wx_sn 去重)
+"""
+
+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 ._mapper import AccountMapper
+from ._utils import parse_article_list_response
+
+logger = logging.getLogger(__name__)
+
+
+class AccountArticleFetch(DemandSearchAccountConst):
+    """从 accounts 取 gh_id → 调文章列表 API → 写入 articles 表"""
+
+    def __init__(self, pool: MysqlManager):
+        self.mapper = AccountMapper(pool)
+
+    async def deal(self, *, limit: int = 10) -> dict:
+        """处理 N 个账号的文章抓取
+
+        Returns:
+            {"accounts": int, "articles_written": int, "errors": int}
+        """
+        accounts = await self.mapper.list_accounts_for_article_crawl()
+        if not accounts:
+            logger.info("无待抓取文章的账号")
+            return {"accounts": 0, "articles_written": 0, "errors": 0}
+
+        accounts = accounts[:limit]
+        total_articles = 0
+        errors = 0
+
+        for i, acct in enumerate(accounts):
+            gh_id = acct["gh_id"]
+            deep_done = acct.get("crawl_deep_done", 0)
+            cursor = acct.get("crawl_cursor") or None
+
+            if i > 0:
+                await asyncio.sleep(self.ARTICLE_PAGE_INTERVAL_SEC)
+
+            try:
+                if deep_done:
+                    n = await self._incremental_crawl(gh_id, cursor)
+                else:
+                    n = await self._deep_crawl(gh_id)
+                    await self.mapper.update_crawl_status(gh_id, cursor="", deep_done=1)
+            except Exception:
+                logger.exception("账号文章抓取异常: gh_id=%s", gh_id)
+                errors += 1
+                continue
+
+            total_articles += n
+            prev_count = await self.mapper.count_articles_by_gh_id(gh_id)
+            logger.info(
+                "gh_id=%s 抓取完成: 本批写入 %d 条, 累计 %d 条",
+                gh_id, n, prev_count,
+            )
+
+        logger.info(
+            "文章抓取完成: 账号 %d 个, 写入 %d 条, 异常 %d",
+            len(accounts), total_articles, errors,
+        )
+        return {"accounts": len(accounts), "articles_written": total_articles, "errors": errors}
+
+    # ═══════════════════════════════════════════════════════════════
+    # 抓取模式
+    # ═══════════════════════════════════════════════════════════════
+
+    async def _deep_crawl(self, gh_id: str) -> int:
+        """首次深翻：逐页拉取，直到遇到一年前的文章"""
+        total = 0
+        cursor = None
+        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)
+
+            if articles:
+                # 筛选一年内的文章，过滤一年前的
+                fresh = [a for a in articles if a["send_time"] >= cutoff]
+                old_count = len(articles) - len(fresh)
+                if fresh:
+                    n = await self.mapper.insert_articles_batch(fresh)
+                    total += n
+
+                # 出现一年前的文章 → 停止翻页
+                if old_count > 0 or (oldest_send_time and oldest_send_time < cutoff):
+                    logger.info(
+                        "gh_id=%s 深翻到达一年前边界: 本页 %d 条中 %d 条超一年, 停止",
+                        gh_id, len(articles), old_count,
+                    )
+                    break
+
+            if not has_more or not next_cursor:
+                break
+
+            cursor = next_cursor
+            await asyncio.sleep(self.ARTICLE_PAGE_INTERVAL_SEC)
+
+        return total
+
+    async def _incremental_crawl(self, gh_id: str, _cursor: str | None = None) -> int:
+        """增量更新：只取第一页即停"""
+        articles, _, _, _ = await self._fetch_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/demand_search_article/_const.py

@@ -58,7 +58,7 @@ class DemandSearchArticleConst:
 
     MAX_SEARCH_PAGES = 3  # 每个搜索词最多翻多少页
     CACHE_TTL_DAYS = 10  # 搜索缓存有效期（天）
-    SEARCH_INTERVAL_SEC = 5  # 搜索页间延迟（秒）
+    SEARCH_INTERVAL_SEC = 15  # 搜索页间延迟（秒）
     DETAIL_INTERVAL_SEC = 3  # 详情拉取条间延迟（秒）
 
     # ═══════════════════════════════════════════════════════════════

src/server/app.py

@@ -16,6 +16,7 @@ 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__)
 
@@ -50,6 +51,7 @@ def create_app(global_config: GlobalConfig = None) -> Quart:
     set_db_article_search(db)
     set_db_article_fetch_detail(db)
     set_db_account_fetch(db)
+    set_db_account_article_fetch(db)
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")