init

src/domains/demand_search_article/_const.py

@@ -0,0 +1,62 @@
+"""文章域常量"""
+
+
+class DemandSearchArticleConst:
+    """文章域所有常量的聚合根，服务类继承此类即可通过 self. 访问"""
+
+    # ═══════════════════════════════════════════════════════════════
+    # 表名
+    # ═══════════════════════════════════════════════════════════════
+
+    QUEUE_TABLE = "demand_search_queue"
+    RELATION_TABLE = "demand_search_article_relation"
+    DETAIL_TABLE = "demand_search_article_detail"
+    CACHE_TABLE = "demand_search_cache"
+
+    # ═══════════════════════════════════════════════════════════════
+    # 队列状态
+    # ═══════════════════════════════════════════════════════════════
+
+    class QueueStatus:
+        INIT = 0
+        PROCESSING = 1
+        SUCCESS = 2
+        FAIL = 99
+
+    # ═══════════════════════════════════════════════════════════════
+    # 搜索词类型（来源列）
+    # ═══════════════════════════════════════════════════════════════
+
+    class KeyType:
+        """搜索词来源类型，对应 demand_search_queue 的三列"""
+        STANDARD_ELEMENT = "standard_element"
+        MATCH_TEXT = "match_text"
+        MATCH_GENERALIZED_ELEMENT = "match_generalized_element"
+
+        @classmethod
+        def all_types(cls) -> tuple:
+            return cls.STANDARD_ELEMENT, cls.MATCH_TEXT, cls.MATCH_GENERALIZED_ELEMENT
+
+    # ═══════════════════════════════════════════════════════════════
+    # 搜索状态: 0→1→2 / 0→1→99
+    # ═══════════════════════════════════════════════════════════════
+
+    class SearchStatus:
+        INIT = 0        # 待拉取详情
+        PROCESSING = 1  # 拉取详情中
+        SUCCESS = 2     # 拉取详情成功
+        FAIL = 99       # 详情拉取失败
+
+    # ═══════════════════════════════════════════════════════════════
+    # 搜索参数
+    # ═══════════════════════════════════════════════════════════════
+
+    MAX_SEARCH_PAGES = 3        # 每个搜索词最多翻多少页
+    SEARCH_PAGE_SIZE = 10       # 每页结果数（微信 API 默认）
+    BATCH_INSERT_SIZE = 100     # 批量写入每批行数
+    CACHE_TTL_DAYS = 10         # 搜索缓存有效期（天）
+    SEARCH_INTERVAL_SEC = 5     # 搜索页间延迟（秒）
+    DETAIL_INTERVAL_SEC = 3     # 详情拉取条间延迟（秒）
+
+
+__all__ = ["DemandSearchArticleConst"]

src/domains/demand_search_article/_mapper.py

@@ -0,0 +1,288 @@
+"""文章域 DB 读写 —— demand_search_article_relation + demand_search_article_detail"""
+
+import json
+from typing import Dict, List, Optional
+
+from src.infra.database.mysql.manager import MysqlManager
+
+from ._const import DemandSearchArticleConst
+
+
+class ArticleSearchRelationMapper(DemandSearchArticleConst):
+    """demand_search_article_relation + demand_search_queue 的读写"""
+
+    def __init__(self, pool: MysqlManager):
+        self.pool = pool
+
+    # ═══════════════════════════════════════════════════════════════
+    # demand_search_queue 读取 & 状态更新
+    # ═══════════════════════════════════════════════════════════════
+
+    async def fetch_queue_pending(self, *, limit: int = 500) -> List[Dict]:
+        """取待搜索的队列项，按 match_score 降序"""
+        query = f"""
+            SELECT *
+            FROM {self.QUEUE_TABLE}
+            WHERE status = %s
+            ORDER BY match_score DESC
+            LIMIT %s
+        """
+        return await self.pool.fetch(
+            query=query,
+            params=(self.QueueStatus.INIT, limit),
+        )
+
+    async def queue_mark_processing(self, ids: List[int]) -> int:
+        """批量标记队列项为处理中"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.QUEUE_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders})
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.QueueStatus.PROCESSING, *ids),
+        )
+
+    async def queue_mark_done(self, item_id: int) -> int:
+        """标记队列项为成功"""
+        query = f"""
+            UPDATE {self.QUEUE_TABLE}
+            SET status = %s
+            WHERE id = %s
+        """
+        return await self.pool.save(
+            query=query, params=(self.QueueStatus.SUCCESS, item_id),
+        )
+
+    async def queue_mark_failed(self, item_id: int, reason: str = "") -> int:
+        """标记队列项为失败"""
+        query = f"""
+            UPDATE {self.QUEUE_TABLE}
+            SET status = %s, fail_reason = %s
+            WHERE id = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.QueueStatus.FAIL, reason[:512], item_id),
+        )
+
+    # ═══════════════════════════════════════════════════════════════
+    # relation 表 — 写入
+    # ═══════════════════════════════════════════════════════════════
+
+    async def insert_batch(self, items: List[Dict]) -> int:
+        """批量写入搜索结果，返回受影响行数
+
+        items 每条须包含:
+          search_key, key_type, experiment_id, search_cursor,
+          url, wx_sn, title, cover_url, publish_time
+        """
+        if not items:
+            return 0
+        query = f"""
+            INSERT IGNORE INTO {self.RELATION_TABLE}
+                (search_key, key_type, experiment_id, search_cursor,
+                 url, wx_sn, title, cover_url, publish_time)
+            VALUES
+                (%s, %s, %s, %s, %s, %s, %s, %s, %s)
+        """
+        params = [
+            (
+                it["search_key"],
+                it["key_type"],
+                it["experiment_id"],
+                it["search_cursor"],
+                it["url"],
+                it["wx_sn"],
+                it["title"],
+                it["cover_url"],
+                it["publish_time"],
+            )
+            for it in items
+        ]
+        return await self.pool.save(query=query, params=params, batch=True)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 读取
+    # ═══════════════════════════════════════════════════════════════
+
+    async def fetch_pending(
+        self,
+        experiment_id: str,
+        *,
+        limit: int = 500,
+    ) -> List[Dict]:
+        """查询待拉详情的行 (status=0)"""
+        query = f"""
+            SELECT *
+            FROM {self.RELATION_TABLE}
+            WHERE experiment_id = %s AND status = %s
+            LIMIT %s
+        """
+        return await self.pool.fetch(
+            query=query,
+            params=(experiment_id, self.SearchStatus.INIT, limit),
+        )
+
+    async def count_by_status(
+        self,
+        experiment_id: str,
+        status: int,
+    ) -> int:
+        """统计指定状态的条目数"""
+        query = f"""
+            SELECT COUNT(*) AS cnt
+            FROM {self.RELATION_TABLE}
+            WHERE experiment_id = %s AND status = %s
+        """
+        row = await self.pool.fetch_one(query=query, params=(experiment_id, status))
+        return row["cnt"] if row else 0
+
+    # ═══════════════════════════════════════════════════════════════
+    # 状态更新
+    # ═══════════════════════════════════════════════════════════════
+
+    async def mark_processing(self, ids: List[int]) -> int:
+        """批量标记为拉取中 (0→1)"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.RELATION_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders}) AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.SearchStatus.PROCESSING, *ids, self.SearchStatus.INIT),
+        )
+
+    async def mark_success(self, item_id: int, *, channel_content_id: str = "") -> int:
+        """标记为成功，回填 channel_content_id (→2)"""
+        query = f"""
+            UPDATE {self.RELATION_TABLE}
+            SET status = %s, channel_content_id = %s
+            WHERE id = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.SearchStatus.SUCCESS, channel_content_id, item_id),
+        )
+
+    async def mark_failed(self, ids: List[int]) -> int:
+        """批量标记为失败 (→99)"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.RELATION_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders})
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.SearchStatus.FAIL, *ids),
+        )
+
+
+    # ═══════════════════════════════════════════════════════════════
+    # 搜索缓存
+    # ═══════════════════════════════════════════════════════════════
+
+    async def cache_get(self, keyword: str, page: int) -> Optional[Dict]:
+        """读取缓存 (keyword, page)，TTL 过期返回 None"""
+        query = f"""
+            SELECT response
+            FROM {self.CACHE_TABLE}
+            WHERE keyword = %s
+              AND page = %s
+              AND updated_at > DATE_SUB(NOW(), INTERVAL %s DAY)
+        """
+        row = await self.pool.fetch_one(
+            query=query,
+            params=(keyword, page, self.CACHE_TTL_DAYS),
+        )
+        if not row:
+            return None
+        resp = row["response"]
+        if isinstance(resp, str):
+            return json.loads(resp)
+        return resp
+
+    async def cache_set(
+        self, keyword: str, page: int, response: Dict,
+    ) -> int:
+        """写入缓存（upsert）"""
+        query = f"""
+            INSERT INTO {self.CACHE_TABLE} (keyword, page, response)
+            VALUES (%s, %s, %s)
+            ON DUPLICATE KEY UPDATE
+                response = VALUES(response),
+                updated_at = CURRENT_TIMESTAMP
+        """
+        return await self.pool.save(
+            query=query,
+            params=(keyword, page, json.dumps(response, ensure_ascii=False)),
+        )
+
+
+class ArticleDetailMapper(DemandSearchArticleConst):
+    """demand_search_article_detail 表的数据访问层"""
+
+    def __init__(self, pool: MysqlManager):
+        self.pool = pool
+
+    # ═══════════════════════════════════════════════════════════════
+    # 写入
+    # ═══════════════════════════════════════════════════════════════
+
+    async def insert_one(self, detail: Dict) -> int:
+        """写入单篇详情，channel_content_id 冲突则跳过
+
+        detail 须包含:
+          channel_content_id, wx_sn, url, title, body_text, content_type,
+          channel_account_id, publish_timestamp, publish_at, is_original,
+          view_count, like_count, looking_count, comment_count, share_count,
+          image_url_list, video_url_list, is_cache, extra
+        """
+        query = f"""
+            INSERT IGNORE INTO {self.DETAIL_TABLE}
+                (channel_content_id, wx_sn, url, title, body_text, content_type,
+                 channel_account_id, publish_timestamp, publish_at, is_original,
+                 view_count, like_count, looking_count, comment_count, share_count,
+                 image_url_list, video_url_list, is_cache, extra)
+            VALUES
+                (%s, %s, %s, %s, %s, %s,
+                 %s, %s, %s, %s,
+                 %s, %s, %s, %s, %s,
+                 %s, %s, %s, %s)
+        """
+        params = (
+            detail["channel_content_id"],
+            detail["wx_sn"],
+            detail.get("url", ""),
+            detail.get("title", ""),
+            detail.get("body_text", ""),
+            detail.get("content_type", ""),
+            detail.get("channel_account_id", ""),
+            detail.get("publish_timestamp", 0),
+            detail.get("publish_at"),
+            detail.get("is_original", 0),
+            detail.get("view_count", 0),
+            detail.get("like_count", 0),
+            detail.get("looking_count", 0),
+            detail.get("comment_count", 0),
+            detail.get("share_count", 0),
+            json.dumps(detail.get("image_url_list") or [], ensure_ascii=False),
+            json.dumps(detail.get("video_url_list") or [], ensure_ascii=False),
+            detail.get("is_cache", 0),
+            json.dumps(detail.get("extra") or {}, ensure_ascii=False),
+        )
+        return await self.pool.save(query=query, params=params)
+
+
+__all__ = ["ArticleSearchRelationMapper", "ArticleDetailMapper"]

src/domains/demand_search_article/_utils.py

@@ -0,0 +1,96 @@
+"""文章域工具函数 —— API 响应解析、数据转换"""
+
+from datetime import datetime
+from typing import Dict
+from urllib.parse import parse_qs, urlparse
+
+
+def extract_wx_sn(content_url: str | None) -> str | None:
+    """从文章 URL 中提取 sn 参数作为 wx_sn
+
+    文章链接格式: http://mp.weixin.qq.com/s?__biz=...&sn=xxx...
+    sn 参数值即为微信文章链接 ID。
+    """
+    if not content_url:
+        return None
+    query = urlparse(content_url).query
+    return parse_qs(query).get("sn", [None])[0]
+
+
+# 无效搜索词集合 — 仅包含无意义的占位符
+_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 "",
+        "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": {
+            "biz_info": detail.get("biz_info"),
+            "item_index": detail.get("item_index"),
+            "channel_account_name": detail.get("channel_account_name"),
+        },
+    }
+
+
+__all__ = ["extract_wx_sn", "is_valid_keyword", "parse_search_result", "parse_detail"]

src/domains/demand_search_article/article_fetch_detail.py

@@ -0,0 +1,147 @@
+"""Phase 2: demand_search_article_relation(status=0) → 微信详情 API → demand_search_article_detail
+
+流程:
+  RelationMapper.fetch_pending() → 详情 API → DetailMapper.insert_one() + RelationMapper.mark_success/failed
+"""
+
+import asyncio
+import logging
+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 ._mapper import ArticleSearchRelationMapper, ArticleDetailMapper
+from ._utils import parse_detail
+
+logger = logging.getLogger(__name__)
+
+
+class ArticleFetchDetail(DemandSearchArticleConst):
+    """从 relation 表取待拉项 → 调详情 API → 写入 detail 表"""
+
+    def __init__(self, pool: MysqlManager):
+        self.relation_mapper = ArticleSearchRelationMapper(pool)
+        self.detail_mapper = ArticleDetailMapper(pool)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 入口
+    # ═══════════════════════════════════════════════════════════════
+
+    async def deal(
+        self,
+        experiment_id: str,
+        *,
+        limit: int = 500,
+    ) -> dict:
+        """串行处理待拉详情项
+
+        Returns:
+            {"pending": int, "success": int, "failed": int, "skipped": int}
+        """
+        items = await self.relation_mapper.fetch_pending(experiment_id, limit=limit)
+        if not items:
+            logger.info("无待拉详情项, experiment_id=%s", experiment_id)
+            return {"pending": 0, "success": 0, "failed": 0, "skipped": 0}
+
+        ids = [it["id"] for it in items]
+        await self.relation_mapper.mark_processing(ids)
+
+        success = 0
+        failed = 0
+        skipped = 0
+        for item in items:
+            if success + failed > 0:  # 非首条，等待后执行
+                await asyncio.sleep(self.DETAIL_INTERVAL_SEC)
+            try:
+                outcome, _ = await self._fetch_detail_and_save(item)
+            except Exception:
+                logger.exception(
+                    "详情拉取异常: relation_id=%s", item["id"],
+                )
+                await self.relation_mapper.mark_failed([item["id"]])
+                failed += 1
+                continue
+
+            if outcome == "success":
+                success += 1
+            elif outcome == "failed":
+                failed += 1
+            else:
+                skipped += 1
+
+        logger.info(
+            "详情拉取完成: experiment_id=%s, total=%d, success=%d, failed=%d, skipped=%d",
+            experiment_id, len(items), success, failed, skipped,
+        )
+        return {
+            "pending": len(items),
+            "success": success,
+            "failed": failed,
+            "skipped": skipped,
+        }
+
+    # ═══════════════════════════════════════════════════════════════
+    # 单条处理
+    # ═══════════════════════════════════════════════════════════════
+
+    async def _fetch_detail_and_save(self, item: Dict) -> tuple[str, int | None]:
+        """拉取详情 → 写 detail 表 → 更新 relation 状态
+
+        Returns:
+            ("success"/"failed"/"skipped", relation_id)
+        """
+        url = item.get("url", "")
+        if not url:
+            logger.warning("空 URL, 跳过 relation_id=%s", item["id"])
+            await self.relation_mapper.mark_failed([item["id"]])
+            return ("failed", item["id"])
+
+        try:
+            resp = await get_article_detail(url)
+        except Exception:
+            logger.exception(
+                "get_article_detail 调用异常: url=%s, relation_id=%s", url, item["id"],
+            )
+            await self.relation_mapper.mark_failed([item["id"]])
+            return ("failed", item["id"])
+
+        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",
+            )
+            await self.relation_mapper.mark_failed([item["id"]])
+            return ("failed", item["id"])
+
+        detail_data = (resp.get("data") or {}).get("data")
+        if not detail_data:
+            logger.warning(
+                "详情数据为空: url=%s, relation_id=%s", url, item["id"],
+            )
+            await self.relation_mapper.mark_failed([item["id"]])
+            return ("failed", item["id"])
+
+        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"],
+            )
+            await self.relation_mapper.mark_failed([item["id"]])
+            return ("failed", item["id"])
+
+        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,
+        )
+
+        logger.debug(
+            "详情拉取成功: relation_id=%s, channel_content_id=%s, title=%s",
+            item["id"], channel_content_id, detail_row.get("title", "")[:40],
+        )
+        return ("success", item["id"])
+
+
+__all__ = ["ArticleFetchDetail"]

src/domains/demand_search_article/article_search.py

@@ -0,0 +1,164 @@
+"""Phase 1: 从 demand_search_queue 取搜索词 → 微信搜索 → 写入 demand_search_article_relation
+
+流程:
+  ArticleSearchRelationMapper.fetch_queue_pending() → 微信搜索 → RelationMapper.insert_batch()
+  experiment_id 从上到下透传，不做过滤条件。
+"""
+
+import asyncio
+import logging
+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 ._mapper import ArticleSearchRelationMapper
+from ._utils import parse_search_result, is_valid_keyword
+
+logger = logging.getLogger(__name__)
+
+# demand_search_queue 列名 → key_type 映射
+_SEARCH_KEY_COLUMNS = [
+    ("standard_element", DemandSearchArticleConst.KeyType.STANDARD_ELEMENT),
+    ("match_text", DemandSearchArticleConst.KeyType.MATCH_TEXT),
+    ("match_generalized_element", DemandSearchArticleConst.KeyType.MATCH_GENERALIZED_ELEMENT),
+]
+
+
+class DemandSearchArticle(DemandSearchArticleConst):
+    """从需求队列取搜索词 → 微信搜索 → 写入关系表
+
+    experiment_id 只透传不过滤——取 match_score 最高的待处理项。
+    """
+
+    def __init__(self, pool: MysqlManager):
+        self.relation_mapper = ArticleSearchRelationMapper(pool)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 入口
+    # ═══════════════════════════════════════════════════════════════
+
+    async def deal(self, *, limit: int = 500) -> dict:
+        """按 match_score 降序取待处理项 → 微信搜索 → 写入 relation
+
+        Returns:
+            {"queued": int, "processed": int, "results_written": int}
+        """
+        items = await self.relation_mapper.fetch_queue_pending(limit=limit)
+        if not items:
+            logger.info("无待处理队列项")
+            return {"queued": 0, "processed": 0, "results_written": 0}
+
+        ids = [it["id"] for it in items]
+        await self.relation_mapper.queue_mark_processing(ids)
+
+        total_written = 0
+        processed = 0
+        for item in items:
+            n = 0
+            try:
+                n = await self._search_and_write(item)
+            except Exception as e:
+                logger.exception(
+                    "搜索失败: queue_id=%s, experiment_id=%s",
+                    item["id"], item.get("experiment_id"),
+                )
+                await self.relation_mapper.queue_mark_failed(
+                    item["id"], reason=str(e),
+                )
+                continue
+
+            if n > 0:
+                total_written += n
+                processed += 1
+                await self.relation_mapper.queue_mark_done(item["id"])
+            else:
+                await self.relation_mapper.queue_mark_failed(
+                    item["id"], reason="所有搜索词均未返回结果",
+                )
+
+        logger.info(
+            "搜索完成: 处理 %d/%d 项, 写入 %d 条结果",
+            processed, len(items), total_written,
+        )
+        return {
+            "queued": len(items),
+            "processed": processed,
+            "results_written": total_written,
+        }
+
+    async def _search_and_write(self, item: Dict) -> int:
+        """对单个队列项的每个搜索词列执行搜索，累计写入行数"""
+        total = 0
+        for col, key_type in _SEARCH_KEY_COLUMNS:
+            keyword = (item.get(col) or "").strip()
+            if not is_valid_keyword(keyword):
+                continue
+            n = await self._search_one_keyword(
+                keyword=keyword,
+                key_type=key_type,
+                experiment_id=item["experiment_id"],
+            )
+            total += n
+        return total
+
+    async def _search_one_keyword(
+        self,
+        keyword: str,
+        key_type: str,
+        experiment_id: str,
+    ) -> int:
+        """单个搜索词: 翻页搜索 → 批量写入 relation 表"""
+        total = 0
+        cursor = "0"
+        for page in range(self.MAX_SEARCH_PAGES):
+            result, from_cache = await self._search_with_cache(keyword, page, cursor)
+            if result is None:
+                break
+
+            # 非缓存命中时才写 relation + 等待（缓存命中的数据已经写过了）
+            if not from_cache:
+                data = result.get("data", {})
+                articles = data.get("data", [])
+                if not articles:
+                    break
+
+                items = [
+                    parse_search_result(a, keyword, key_type, experiment_id, page)
+                    for a in articles
+                ]
+                n = await self.relation_mapper.insert_batch(items)
+                total += n
+                logger.info(
+                    "搜索写入: keyword=%s, page=%d, 返回 %d 条, 写入 %d 条",
+                    keyword, page, len(articles), n,
+                )
+                await asyncio.sleep(self.SEARCH_INTERVAL_SEC)
+
+            # 从响应中取翻页游标（缓存和非缓存都要翻页）
+            data = result.get("data", {})
+            has_more = data.get("has_more", False)
+            next_cursor = data.get("next_cursor")
+            if not has_more or not next_cursor:
+                break
+            cursor = str(next_cursor)
+
+        return total
+
+    async def _search_with_cache(
+        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)
+        if cached:
+            logger.info("缓存命中: keyword=%s, page=%d", keyword, page)
+            return cached, True
+
+        result = await weixin_search(keyword, page=cursor)
+        if result and result.get("code") == 0:
+            await self.relation_mapper.cache_set(keyword, page, result)
+        return result, False
+
+
+__all__ = ["DemandSearchArticle"]