feat(acquisition): add provider aware pagination strategy

acquisition/crawler.py

@@ -2,7 +2,7 @@
 
 crawler.aiddit.com 各平台 detail 返回字段已归一化（channel_content_id/title/content_type/
 body_text/image_url_list/video_url_list/channel_account_*），所以 parse 一套复用；差异只在
-detail path、URL→content_id 解析、平台名/id 前缀。详见 数据接口与来源/视频音频取数实测.md。
+detail path、URL→content_id 解析、平台名/id 前缀；正式环境 host/超时从 .env 读取。
 
 拆成两层：parse_detail_response 纯函数（可离线用 fixture 测）+ fetch_post_detail 负责 HTTP。
 """
@@ -20,6 +20,8 @@ from core.config import Settings
 from core.models import Card, Post
 
 RATE_LIMIT_SECONDS = 15.0
+CRAWLER_MIN_INTERVAL_SECONDS = 10.0
+CRAWLER_MAX_INTERVAL_SECONDS = 12.0
 
 # 平台路由：detail path + Post.id 前缀
 PLATFORMS = {
@@ -29,6 +31,8 @@ PLATFORMS = {
     "bilibili": {"path": "/crawler/bilibili/detail", "prefix": "bili"},
 }
 
+DOUYIN_PROVIDERS = {"piaoquantv", "aiddit"}
+
 
 def _last_seg(s: str) -> str:
     return s.split("?", 1)[0].rstrip("/").split("/")[-1]
@@ -123,7 +127,26 @@ def _video_urls(inner: dict) -> list[str]:
     return out
 
 
+def _douyin_detail_base_url(settings: Settings, provider: str) -> str:
+    if provider == "piaoquantv":
+        return settings.piaoquantv_douyin_base_url
+    if provider == "aiddit":
+        return settings.aiddit_crawler_base_url
+    raise CrawlerError(f"unsupported douyin provider: {provider}")
+
+
+def _douyin_detail_body(content_id: str, *, settings: Settings, provider: str) -> dict[str, Any]:
+    body: dict[str, Any] = {"content_id": content_id}
+    if provider == "piaoquantv":
+        body["account_id"] = settings.piaoquantv_douyin_account_id
+        body["cookie_batch"] = settings.piaoquantv_douyin_cookie_batch
+    elif provider != "aiddit":
+        raise CrawlerError(f"unsupported douyin provider: {provider}")
+    return body
+
+
 def parse_detail_response(response: dict, *, platform: str = "xiaohongshu",
+                          provider: str = "",
                           fallback_content_id: str = "") -> Post:
     """把 detail 接口返回（信封 {code,msg,data:{data:{...}}}）解析成 Post。字段各平台归一。"""
     if not isinstance(response, dict):
@@ -141,9 +164,13 @@ def parse_detail_response(response: dict, *, platform: str = "xiaohongshu",
     images = _image_urls(inner)
     # 图文帖：每张图一张卡片（1-based）；视频帖的段卡由 extract_video 在提取时写入，覆盖封面卡
     cards = [Card(index=i, kind="image", url=u) for i, u in enumerate(images, start=1)]
+    raw = dict(response)
+    if provider:
+        raw.setdefault("detail_provider", provider)
     return Post(
         id=f"{prefix}_{content_id}",
         platform=platform,
+        provider=provider,
         url=link,
         content_id=content_id,
         title=inner.get("title") or "",
@@ -155,13 +182,15 @@ def parse_detail_response(response: dict, *, platform: str = "xiaohongshu",
         cards=cards,
         author_id=inner.get("channel_account_id"),
         author_name=inner.get("channel_account_name"),
-        raw=response,
+        raw=raw,
     )
 
 
 def fetch_post_detail(
     content_id_or_url: str,
     *,
+    platform: str | None = None,
+    provider: str | None = None,
     settings: Optional[Settings] = None,
     http_client: Any = None,
     rate_limiter: Optional[RateLimiter] = None,
@@ -169,23 +198,43 @@ def fetch_post_detail(
 ) -> Post:
     """真实拉取一条帖子详情，自动识别平台（小红书/抖音/快手/B站），返回 Post。"""
     settings = settings or Settings.from_env(env_file)
-    platform, content_id = detect_platform_and_id(content_id_or_url)
+    detected_platform, content_id = detect_platform_and_id(content_id_or_url)
+    platform = platform or detected_platform
     if not content_id:
         raise CrawlerError(f"cannot parse content_id from: {content_id_or_url}")
     cfg = PLATFORMS[platform]
 
-    rate_limiter = rate_limiter or RateLimiter()
-    rate_limiter.wait(f"{platform}_detail")
+    is_douyin = platform == "douyin"
+    if is_douyin:
+        provider = provider or "piaoquantv"
+        if provider not in DOUYIN_PROVIDERS:
+            raise CrawlerError(f"unsupported douyin provider: {provider}")
+        rate_limiter = rate_limiter or RateLimiter(
+            min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+            max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+        )
+        rate_limiter.wait("douyin")
+    else:
+        provider = provider or ""
+        rate_limiter = rate_limiter or RateLimiter(
+            min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+            max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+        )
+        rate_limiter.wait(f"{platform}_detail")
 
     owns_client = http_client is None
     client = http_client or httpx.Client()
     try:
-        is_douyin = platform == "douyin"
-        base_url = settings.piaoquantv_douyin_base_url if is_douyin else settings.aiddit_crawler_base_url
-        body = {"content_id": content_id}
-        if is_douyin:                                  # 抖音详情走 piaoquantv，需带 account_id + cookie_batch
-            body["account_id"] = settings.piaoquantv_douyin_account_id
-            body["cookie_batch"] = settings.piaoquantv_douyin_cookie_batch
+        base_url = (
+            _douyin_detail_base_url(settings, provider)
+            if is_douyin
+            else settings.aiddit_crawler_base_url
+        )
+        body = (
+            _douyin_detail_body(content_id, settings=settings, provider=provider)
+            if is_douyin
+            else {"content_id": content_id}
+        )
         url = urljoin(base_url, cfg["path"])
         resp = client.post(
             url,
@@ -203,4 +252,9 @@ def fetch_post_detail(
         if owns_client:
             client.close()
 
-    return parse_detail_response(data, platform=platform, fallback_content_id=content_id)
+    return parse_detail_response(
+        data,
+        platform=platform,
+        provider=provider or "",
+        fallback_content_id=content_id,
+    )

acquisition/platforms/base.py

@@ -1,7 +1,7 @@
 """Formal platform adapter contracts for acquisition."""
 from __future__ import annotations
 
-from typing import Any, Protocol
+from typing import Any, Iterable, Protocol
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -13,6 +13,7 @@ class PlatformCandidate(BaseModel):
 
     rank: int
     platform: str
+    provider: str = ""
     source_id: str = ""
     url: str = ""
     title: str = ""
@@ -25,9 +26,11 @@ class PlatformItem(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     platform: str
+    provider: str = ""
     source_id: str = ""
     url: str = ""
     content_type: str = ""
+    content_mode: str = ""
     title: str = ""
     author: str = ""
     body_text: str = ""
@@ -36,9 +39,34 @@ class PlatformItem(BaseModel):
     raw: dict[str, Any] = Field(default_factory=dict)
 
 
+class PlatformSearchPage(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    page_index: int
+    candidates: list[PlatformCandidate] = Field(default_factory=list)
+    raw_count: int = 0
+    cursor: str = ""
+    next_cursor: str = ""
+    has_more: bool = False
+    provider: str = ""
+    request_payload: dict[str, Any] = Field(default_factory=dict)
+    raw_metadata: dict[str, Any] = Field(default_factory=dict)
+
+
 class PlatformAdapter(Protocol):
     platform: str
 
+    def search_pages(
+        self,
+        query: str,
+        *,
+        settings: Settings,
+        limit: int,
+        rate_limiter: Any,
+        max_pages: int = 2,
+    ) -> Iterable[PlatformSearchPage]:
+        ...
+
     def search(
         self,
         query: str,

acquisition/platforms/douyin.py

@@ -1,11 +1,12 @@
 """Douyin formal acquisition adapter."""
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Iterable
 
+from acquisition.content_mode import infer_content_mode
 from acquisition.crawler import fetch_post_detail
-from acquisition.platforms.base import PlatformCandidate, PlatformItem
-from acquisition.search import search_keyword
+from acquisition.platforms.base import PlatformCandidate, PlatformItem, PlatformSearchPage
+from acquisition.search import SearchHit, search_douyin_hits, search_douyin_page
 from core.config import Settings
 
 
@@ -20,9 +21,8 @@ class DouyinAdapter:
         limit: int,
         rate_limiter: Any,
     ) -> list[PlatformCandidate]:
-        ids = search_keyword(
+        hits = search_douyin_hits(
             query,
-            platform=self.platform,
             content_type="视频",
             limit=limit,
             settings=settings,
@@ -32,12 +32,83 @@ class DouyinAdapter:
             PlatformCandidate(
                 rank=i,
                 platform=self.platform,
-                source_id=content_id,
-                raw={"id": content_id},
+                provider=hit.provider,
+                source_id=hit.content_id,
+                raw={
+                    "id": hit.content_id,
+                    "search_provider": hit.provider,
+                    "original": hit.raw,
+                    "request_payload": hit.request_payload,
+                },
             )
-            for i, content_id in enumerate(ids, start=1)
+            for i, hit in enumerate(hits, start=1)
         ]
 
+    def search_pages(
+        self,
+        query: str,
+        *,
+        settings: Settings,
+        limit: int,
+        rate_limiter: Any,
+        max_pages: int = 2,
+    ) -> Iterable[PlatformSearchPage]:
+        provider = "piaoquantv"
+        cursors: dict[str, str] = {}
+        rank = 1
+        seen: set[str] = set()
+        for page_index in range(1, max_pages + 1):
+            page = search_douyin_page(
+                query,
+                provider=provider,
+                cursors=cursors,
+                content_type="视频",
+                limit=limit,
+                settings=settings,
+                rate_limiter=rate_limiter,
+            )
+            candidates: list[PlatformCandidate] = []
+            for page_rank, hit in enumerate(page.rows, start=1):
+                if not isinstance(hit, SearchHit):
+                    continue
+                if not hit.content_id or hit.content_id in seen:
+                    continue
+                seen.add(hit.content_id)
+                candidates.append(
+                    PlatformCandidate(
+                        rank=rank,
+                        platform=self.platform,
+                        provider=hit.provider,
+                        source_id=hit.content_id,
+                        raw={
+                            "id": hit.content_id,
+                            "search_provider": hit.provider,
+                            "original": hit.raw,
+                            "request_payload": hit.request_payload,
+                            "page_index": page_index,
+                            "page_rank": page_rank,
+                            "source_cursor": page.cursor,
+                        },
+                    )
+                )
+                rank += 1
+            yield PlatformSearchPage(
+                page_index=page_index,
+                candidates=candidates,
+                raw_count=page.raw_count,
+                cursor=page.cursor,
+                next_cursor=page.next_cursor,
+                has_more=page.has_more,
+                provider=page.provider,
+                request_payload=page.request_payload,
+                raw_metadata=page.raw_metadata,
+            )
+            provider = page.provider
+            cursors = dict(page.raw_metadata.get("cursors") or cursors)
+            current_cursor = page.cursor
+            if not page.has_more or not page.next_cursor or page.next_cursor == current_cursor:
+                break
+
     def fetch_detail(
         self,
         candidate: PlatformCandidate,
@@ -45,16 +116,30 @@ class DouyinAdapter:
         settings: Settings,
         rate_limiter: Any,
     ) -> PlatformItem:
+        if not candidate.provider:
+            raise RuntimeError("douyin candidate missing search provider")
         post = fetch_post_detail(
             candidate.source_id,
+            platform=self.platform,
+            provider=candidate.provider,
             settings=settings,
             rate_limiter=rate_limiter,
         )
+        content_mode = infer_content_mode(
+            platform=self.platform,
+            content_type=post.content_type,
+            body_text=post.body_text,
+            image_urls=post.image_urls,
+            video_urls=post.video_urls,
+            raw=post.raw if isinstance(post.raw, dict) else {},
+        )
         return PlatformItem(
             platform=self.platform,
+            provider=post.provider or candidate.provider,
             source_id=post.content_id or candidate.source_id,
             url=post.url or candidate.url,
-            content_type=post.content_type or "video",
+            content_type=post.content_type or "",
+            content_mode=content_mode,
             title=post.title or candidate.title,
             author=post.author_name or candidate.author,
             body_text=post.body_text or "",

acquisition/platforms/weixin.py

@@ -2,10 +2,11 @@
 from __future__ import annotations
 
 import hashlib
-from typing import Any
+from typing import Any, Iterable
 
-from acquisition.platforms.base import PlatformCandidate, PlatformItem
-from acquisition.search import fetch_weixin_detail, search_weixin
+from acquisition.content_mode import ARTICLE
+from acquisition.platforms.base import PlatformCandidate, PlatformItem, PlatformSearchPage
+from acquisition.search import fetch_weixin_detail, search_weixin_page
 from core.config import Settings
 
 
@@ -24,25 +25,75 @@ class WeixinAdapter:
         limit: int,
         rate_limiter: Any,
     ) -> list[PlatformCandidate]:
-        rows = search_weixin(
+        pages = self.search_pages(
             query,
-            limit=limit,
             settings=settings,
+            limit=limit,
             rate_limiter=rate_limiter,
+            max_pages=1,
         )
-        return [
-            PlatformCandidate(
-                rank=i,
-                platform=self.platform,
-                source_id=_hash(row.get("url") or ""),
-                url=row.get("url") or "",
-                title=row.get("title") or "",
-                author=row.get("nick_name") or "",
-                cover_url=row.get("cover_url") or "",
-                raw=row,
+        return [candidate for page in pages for candidate in page.candidates]
+
+    def search_pages(
+        self,
+        query: str,
+        *,
+        settings: Settings,
+        limit: int,
+        rate_limiter: Any,
+        max_pages: int = 2,
+    ) -> Iterable[PlatformSearchPage]:
+        cursor = "0"
+        rank = 1
+        seen: set[str] = set()
+        for page_index in range(1, max_pages + 1):
+            page = search_weixin_page(
+                query,
+                cursor=cursor,
+                limit=limit,
+                settings=settings,
+                rate_limiter=rate_limiter,
+            )
+            candidates: list[PlatformCandidate] = []
+            for page_rank, row in enumerate(page.rows, start=1):
+                url = row.get("url") or ""
+                source_id = _hash(url)
+                if not url or source_id in seen:
+                    continue
+                seen.add(source_id)
+                raw = {
+                    **row,
+                    "page_index": page_index,
+                    "page_rank": page_rank,
+                    "source_cursor": page.cursor,
+                    "request_payload": page.request_payload,
+                }
+                candidates.append(
+                    PlatformCandidate(
+                        rank=rank,
+                        platform=self.platform,
+                        source_id=source_id,
+                        url=url,
+                        title=row.get("title") or "",
+                        author=row.get("nick_name") or "",
+                        cover_url=row.get("cover_url") or "",
+                        raw=raw,
+                    )
+                )
+                rank += 1
+            yield PlatformSearchPage(
+                page_index=page_index,
+                candidates=candidates,
+                raw_count=page.raw_count,
+                cursor=page.cursor,
+                next_cursor=page.next_cursor,
+                has_more=page.has_more,
+                provider=page.provider,
+                request_payload=page.request_payload,
             )
-            for i, row in enumerate(rows, start=1)
-        ]
+            if not page.has_more or not page.next_cursor or page.next_cursor == cursor:
+                break
+            cursor = page.next_cursor
 
     def fetch_detail(
         self,
@@ -61,6 +112,7 @@ class WeixinAdapter:
             source_id=candidate.source_id or _hash(candidate.url),
             url=candidate.url,
             content_type="图文",
+            content_mode=ARTICLE,
             title=candidate.title,
             author=candidate.author,
             body_text=body_text,

acquisition/platforms/xiaohongshu.py

@@ -1,11 +1,12 @@
 """Xiaohongshu formal acquisition adapter."""
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Iterable
 
+from acquisition.content_mode import infer_content_mode
 from acquisition.crawler import fetch_post_detail
-from acquisition.platforms.base import PlatformCandidate, PlatformItem
-from acquisition.search import search_xiaohongshu
+from acquisition.platforms.base import PlatformCandidate, PlatformItem, PlatformSearchPage
+from acquisition.search import search_xiaohongshu_page
 from core.config import Settings
 
 
@@ -20,26 +21,75 @@ class XiaohongshuAdapter:
         limit: int,
         rate_limiter: Any,
     ) -> list[PlatformCandidate]:
-        rows = search_xiaohongshu(
+        pages = self.search_pages(
             query,
-            content_type=settings.search_content_type,
-            limit=limit,
             settings=settings,
+            limit=limit,
             rate_limiter=rate_limiter,
+            max_pages=1,
         )
-        return [
-            PlatformCandidate(
-                rank=i,
-                platform=self.platform,
-                source_id=row.get("id") or "",
-                url=row.get("url") or "",
-                title=row.get("title") or "",
-                author=row.get("nick_name") or "",
-                cover_url=row.get("cover_url") or "",
-                raw=row,
+        return [candidate for page in pages for candidate in page.candidates]
+
+    def search_pages(
+        self,
+        query: str,
+        *,
+        settings: Settings,
+        limit: int,
+        rate_limiter: Any,
+        max_pages: int = 2,
+    ) -> Iterable[PlatformSearchPage]:
+        cursor = ""
+        rank = 1
+        seen: set[str] = set()
+        for page_index in range(1, max_pages + 1):
+            page = search_xiaohongshu_page(
+                query,
+                cursor=cursor,
+                content_type=None,
+                limit=limit,
+                settings=settings,
+                rate_limiter=rate_limiter,
+            )
+            candidates: list[PlatformCandidate] = []
+            for page_rank, row in enumerate(page.rows, start=1):
+                source_id = row.get("id") or ""
+                if not source_id or source_id in seen:
+                    continue
+                seen.add(source_id)
+                raw = {
+                    **row,
+                    "page_index": page_index,
+                    "page_rank": page_rank,
+                    "source_cursor": page.cursor,
+                    "request_payload": page.request_payload,
+                }
+                candidates.append(
+                    PlatformCandidate(
+                        rank=rank,
+                        platform=self.platform,
+                        source_id=source_id,
+                        url=row.get("url") or "",
+                        title=row.get("title") or "",
+                        author=row.get("nick_name") or "",
+                        cover_url=row.get("cover_url") or "",
+                        raw=raw,
+                    )
+                )
+                rank += 1
+            yield PlatformSearchPage(
+                page_index=page_index,
+                candidates=candidates,
+                raw_count=page.raw_count,
+                cursor=page.cursor,
+                next_cursor=page.next_cursor,
+                has_more=page.has_more,
+                provider=page.provider,
+                request_payload=page.request_payload,
             )
-            for i, row in enumerate(rows, start=1)
-        ]
+            if not page.has_more or not page.next_cursor or page.next_cursor == cursor:
+                break
+            cursor = page.next_cursor
 
     def fetch_detail(
         self,
@@ -53,15 +103,25 @@ class XiaohongshuAdapter:
             settings=settings,
             rate_limiter=rate_limiter,
         )
+        image_urls = post.image_urls or ([candidate.cover_url] if candidate.cover_url else [])
+        content_mode = infer_content_mode(
+            platform=self.platform,
+            content_type=post.content_type,
+            body_text=post.body_text,
+            image_urls=image_urls,
+            video_urls=post.video_urls,
+            raw={**candidate.raw, "detail_content_type": post.content_type},
+        )
         return PlatformItem(
             platform=self.platform,
             source_id=post.content_id or candidate.source_id,
             url=post.url or candidate.url,
-            content_type=post.content_type or "图文",
+            content_type=post.content_type or "",
+            content_mode=content_mode,
             title=post.title or candidate.title,
             author=post.author_name or candidate.author,
             body_text=post.body_text or "",
-            image_urls=post.image_urls or [candidate.cover_url],
+            image_urls=image_urls,
             video_urls=post.video_urls,
             raw=post.raw if isinstance(post.raw, dict) else {},
         )

acquisition/runner.py

@@ -2,18 +2,30 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, Callable
+from typing import Any, Callable, Iterable
 from uuid import UUID
 
 from acquisition.classification.coarse import ClassificationResult, coarse_classify_item
+from acquisition.content_mode import guard_content_for_processing, infer_content_mode
 from acquisition.crawler import RateLimiter
 from acquisition.domain import AcquisitionJob, Query
 from acquisition.media.service import StabilizedMedia, stabilize_media_urls
 from acquisition.platforms import PlatformAdapter, get_platform_adapter
+from acquisition.platforms.base import PlatformCandidate, PlatformSearchPage
 from acquisition.repositories.base import AcquisitionRepository
+from acquisition.unique_key import build_unique_key
 from core.config import Settings
+from core.text_limits import (
+    BODY_TEXT_MAX_CHARS,
+    ERROR_MESSAGE_MAX_CHARS,
+    RAW_SUMMARY_MAX_CHARS,
+    clip_text,
+)
 
 DEFAULT_PLATFORMS = ("xiaohongshu", "weixin", "douyin")
+PAGINATION_STRATEGY = "creation_ratio_two_page_v1"
+PAGINATION_CREATION_THRESHOLD = 0.5
+PAGINATION_MAX_PAGES = 2
 
 AdapterFactory = Callable[[str], PlatformAdapter]
 Classifier = Callable[..., ClassificationResult]
@@ -31,6 +43,14 @@ class RunBatchResult:
     skipped: int
 
 
+@dataclass(frozen=True)
+class RecordCandidateResult:
+    displayed: bool
+    skipped_processing: bool = False
+    classified: bool = False
+    is_creation_knowledge: bool | None = None
+
+
 class PlatformRateLimiter:
     """Share one rate bucket per platform across search and detail calls."""
 
@@ -75,6 +95,57 @@ def _source_payload(candidate: Any, detail: Any) -> dict[str, Any]:
     }
 
 
+def _candidate_provider(candidate: Any) -> str:
+    provider = getattr(candidate, "provider", "") or ""
+    raw = getattr(candidate, "raw", None)
+    if not provider and isinstance(raw, dict):
+        provider = raw.get("search_provider") or ""
+    return provider
+
+
+def _detail_provider(detail: Any, candidate: Any) -> str:
+    provider = getattr(detail, "provider", "") or ""
+    if not provider:
+        raw = getattr(detail, "raw", None)
+        if isinstance(raw, dict):
+            provider = raw.get("detail_provider") or ""
+    return provider or _candidate_provider(candidate)
+
+
+def _candidate_page_metadata(candidate: Any) -> dict[str, Any]:
+    raw = getattr(candidate, "raw", None)
+    if not isinstance(raw, dict):
+        return {}
+    out: dict[str, Any] = {}
+    for key in ("page_index", "page_rank", "source_cursor"):
+        value = raw.get(key)
+        if value not in (None, ""):
+            out[key] = value
+    return out
+
+
+def _content_mode(platform: str, detail: Any) -> str:
+    explicit = getattr(detail, "content_mode", "") or ""
+    if explicit:
+        return explicit
+    return infer_content_mode(
+        platform=platform,
+        content_type=getattr(detail, "content_type", "") or "",
+        body_text=getattr(detail, "body_text", "") or "",
+        image_urls=getattr(detail, "image_urls", []) or [],
+        video_urls=getattr(detail, "video_urls", []) or [],
+        raw=getattr(detail, "raw", {}) if isinstance(getattr(detail, "raw", None), dict) else {},
+    )
+
+
+def _skip_reason_message(reason: str) -> str:
+    if reason == "video_url_missing":
+        return "Video post has no usable video URL; skipped before classification."
+    if reason == "unsupported_content_mode":
+        return "Unsupported content mode; skipped before classification."
+    return reason or "Skipped before classification."
+
+
 def _record_candidate(
     repo: AcquisitionRepository,
     *,
@@ -87,29 +158,81 @@ def _record_candidate(
     media_stabilizer: MediaStabilizer,
     classifier: Classifier,
     classify: bool,
-) -> bool:
-    media_rows = media_stabilizer(
-        image_urls=detail.image_urls,
+) -> RecordCandidateResult:
+    source_payload = _source_payload(candidate, detail)
+    platform_item_id = detail.source_id or candidate.source_id or None
+    canonical_url = detail.url or candidate.url or None
+    content_mode = _content_mode(platform, detail)
+    guard = guard_content_for_processing(
+        content_mode=content_mode,
         video_urls=detail.video_urls,
-        settings=settings,
     )
+    metadata = {
+        "candidate_rank": candidate.rank,
+        "acquisition_match_status": "created",
+        "content_mode": content_mode,
+        **_candidate_page_metadata(candidate),
+    }
+    search_provider = _candidate_provider(candidate)
+    detail_provider = _detail_provider(detail, candidate)
+    if search_provider:
+        metadata["search_provider"] = search_provider
+    if detail_provider:
+        metadata["detail_provider"] = detail_provider
+    if guard.metadata:
+        metadata.update(guard.metadata)
+    body_text = clip_text(detail.body_text or "", BODY_TEXT_MAX_CHARS)
     item = repo.upsert_candidate_item(
         platform=platform,
         job_id=_job_id(job),
         query_id=_query_id(query),
-        platform_item_id=detail.source_id or candidate.source_id or None,
-        canonical_url=detail.url or candidate.url or None,
+        platform_item_id=platform_item_id,
+        unique_key=build_unique_key(
+            platform,
+            platform_item_id=platform_item_id,
+            url=canonical_url,
+            raw_payload=source_payload,
+        ),
+        canonical_url=canonical_url,
         content_type=detail.content_type or None,
+        content_mode=content_mode,
         title=detail.title or None,
         author_name=detail.author or None,
-        raw_summary=(detail.body_text or "")[:1000],
-        status="candidate",
-        source_payload=_source_payload(candidate, detail),
-        metadata={"candidate_rank": candidate.rank},
+        body_text=body_text or None,
+        raw_summary=clip_text(body_text, RAW_SUMMARY_MAX_CHARS) or None,
+        status="candidate" if guard.can_process else "skipped",
+        source_payload=source_payload,
+        metadata=metadata,
+        error_message=None if guard.can_process else guard.reason,
     )
     if item.id is None:
         raise RuntimeError("repository returned candidate without id")
 
+    if not guard.can_process:
+        repo.add_item_classification(
+            item_id=item.id,
+            is_creation_knowledge=None,
+            label=guard.label,
+            confidence=None,
+            reason=_skip_reason_message(guard.reason),
+            model_name=None,
+            prompt_version="content_mode_guard",
+            result_payload={
+                "content_mode": content_mode,
+                "skip_reason": guard.reason,
+                **(guard.metadata or {}),
+            },
+            status="skipped",
+            error_message=guard.reason,
+        )
+        return RecordCandidateResult(displayed=True, skipped_processing=True)
+
+    media_rows = media_stabilizer(
+        image_urls=detail.image_urls,
+        video_urls=detail.video_urls,
+        settings=settings,
+    )
+
     for row in media_rows:
         repo.add_media_asset(
             item_id=item.id,
@@ -126,6 +249,7 @@ def _record_candidate(
     if classify:
         result = classifier(
             platform=platform,
+            content_mode=content_mode,
             title=detail.title,
             body_text=detail.body_text,
             image_urls=_image_urls(media_rows),
@@ -144,9 +268,102 @@ def _record_candidate(
             status=result.status,
             error_message=result.error_message,
         )
+        return RecordCandidateResult(
+            displayed=True,
+            classified=result.status == "classified" and result.is_creation_knowledge is not None,
+            is_creation_knowledge=result.is_creation_knowledge,
+        )
+    return RecordCandidateResult(displayed=True)
+
+
+def _candidate_unique_key(platform: str, candidate: PlatformCandidate) -> str | None:
+    return build_unique_key(
+        platform,
+        platform_item_id=candidate.source_id or None,
+        url=candidate.url or None,
+        raw_payload={"candidate": candidate.model_dump()},
+    )
+
+
+def _attach_existing_candidate(
+    repo: AcquisitionRepository,
+    *,
+    job: AcquisitionJob,
+    query: Query,
+    candidate: PlatformCandidate,
+    unique_key: str,
+) -> bool:
+    get_by_key = getattr(repo, "get_candidate_item_by_unique_key", None)
+    attach = getattr(repo, "attach_existing_candidate_item", None)
+    if get_by_key is None or attach is None:
+        return False
+    existing = get_by_key(unique_key)
+    if existing is None or existing.id is None:
+        return False
+    attach(
+        existing.id,
+        job_id=_job_id(job),
+        query_id=_query_id(query),
+        metadata={
+            "acquisition_match_status": "existing",
+            "matched_unique_key": unique_key,
+            "matched_candidate": candidate.model_dump(),
+            "candidate_rank": candidate.rank,
+            **_candidate_page_metadata(candidate),
+            **({"search_provider": _candidate_provider(candidate)} if _candidate_provider(candidate) else {}),
+        },
+    )
     return True
 
 
+def _pagination_config(threshold: float, max_pages: int) -> dict[str, Any]:
+    return {
+        "strategy": PAGINATION_STRATEGY,
+        "creation_threshold": threshold,
+        "max_pages": max_pages,
+    }
+
+
+def _ratio(numerator: int, denominator: int) -> float | None:
+    if denominator <= 0:
+        return None
+    return numerator / denominator
+
+
+def _search_pages(
+    adapter: PlatformAdapter,
+    query_text: str,
+    *,
+    settings: Settings,
+    limit: int,
+    rate_limiter: Any,
+    max_pages: int,
+) -> Iterable[PlatformSearchPage]:
+    search_pages = getattr(adapter, "search_pages", None)
+    if callable(search_pages):
+        return search_pages(
+            query_text,
+            settings=settings,
+            limit=limit,
+            rate_limiter=rate_limiter,
+            max_pages=max_pages,
+        )
+    candidates = adapter.search(
+        query_text,
+        settings=settings,
+        limit=limit,
+        rate_limiter=rate_limiter,
+    )
+    return [
+        PlatformSearchPage(
+            page_index=1,
+            candidates=candidates,
+            raw_count=len(candidates),
+            has_more=False,
+        )
+    ]
+
+
 def run_batch(
     repo: AcquisitionRepository,
     *,
@@ -163,6 +380,8 @@ def run_batch(
     media_stabilizer: MediaStabilizer = stabilize_media_urls,
     classifier: Classifier = coarse_classify_item,
     rate_limiter_factory: RateLimiterFactory | None = None,
+    pagination_creation_threshold: float = PAGINATION_CREATION_THRESHOLD,
+    pagination_max_pages: int = PAGINATION_MAX_PAGES,
 ) -> RunBatchResult:
     """Run query x platform acquisition and write formal cloud-state rows."""
     queries = repo.list_queries_for_batch(batch_id, keep=True)
@@ -176,6 +395,10 @@ def run_batch(
             "display_limit": display_limit,
             "classify": classify,
             "resume": resume,
+            "pagination": _pagination_config(
+                pagination_creation_threshold,
+                pagination_max_pages,
+            ),
         },
     )
     if run.id is None:
@@ -211,6 +434,15 @@ def run_batch(
             errors: list[str] = []
             display_count = 0
             searched_count = 0
+            raw_searched_count = 0
+            skipped_item_count = 0
+            search_page_count = 0
+            page_summaries: list[dict[str, Any]] = []
+            first_page_classified_count = 0
+            first_page_creation_count = 0
+            first_page_creation_ratio: float | None = None
+            requested_second_page = False
+            pagination_stop_reason = ""
             try:
                 adapter = adapter_factory(platform)
                 gate = gates.get(platform)
@@ -221,39 +453,102 @@ def run_batch(
                         else PlatformRateLimiter(platform)
                     )
                     gates[platform] = gate
-                candidates = adapter.search(
+                pages = _search_pages(
+                    adapter,
                     query.query_text,
                     settings=settings,
                     limit=search_limit,
                     rate_limiter=gate,
+                    max_pages=pagination_max_pages,
                 )
-                searched_count = len(candidates)
-                for candidate in candidates:
-                    if display_count >= display_limit:
-                        break
-                    try:
-                        detail = adapter.fetch_detail(
-                            candidate,
-                            settings=settings,
-                            rate_limiter=gate,
-                        )
-                        if _record_candidate(
-                            repo,
-                            job=job,
-                            query=query,
-                            platform=platform,
-                            candidate=candidate,
-                            detail=detail,
-                            settings=settings,
-                            media_stabilizer=media_stabilizer,
-                            classifier=classifier,
-                            classify=classify,
-                        ):
-                            display_count += 1
-                    except Exception as exc:
-                        errors.append(str(exc)[:160])
+                for page in pages:
+                    search_page_count += 1
+                    raw_searched_count += page.raw_count
+                    searched_count += len(page.candidates)
+                    page_classified_count = 0
+                    page_creation_count = 0
+
+                    for candidate in page.candidates[:search_limit]:
+                        try:
+                            unique_key = _candidate_unique_key(platform, candidate)
+                            if unique_key and _attach_existing_candidate(
+                                repo,
+                                job=job,
+                                query=query,
+                                candidate=candidate,
+                                unique_key=unique_key,
+                            ):
+                                display_count += 1
+                                continue
+                            detail = adapter.fetch_detail(
+                                candidate,
+                                settings=settings,
+                                rate_limiter=gate,
+                            )
+                            recorded = _record_candidate(
+                                repo,
+                                job=job,
+                                query=query,
+                                platform=platform,
+                                candidate=candidate,
+                                detail=detail,
+                                settings=settings,
+                                media_stabilizer=media_stabilizer,
+                                classifier=classifier,
+                                classify=classify,
+                            )
+                            if recorded.displayed:
+                                display_count += 1
+                                if recorded.skipped_processing:
+                                    skipped_item_count += 1
+                            if recorded.classified:
+                                page_classified_count += 1
+                                if recorded.is_creation_knowledge is True:
+                                    page_creation_count += 1
+                        except Exception as exc:
+                            errors.append(clip_text(exc, ERROR_MESSAGE_MAX_CHARS))
+                            continue
+
+                    page_ratio = _ratio(page_creation_count, page_classified_count)
+                    page_summaries.append(
+                        {
+                            "page_index": page.page_index,
+                            "raw_count": page.raw_count,
+                            "candidate_count": len(page.candidates),
+                            "classified_count": page_classified_count,
+                            "creation_count": page_creation_count,
+                            "creation_ratio": page_ratio,
+                            "has_more": page.has_more,
+                            "next_cursor": page.next_cursor,
+                            "provider": page.provider,
+                        }
+                    )
+
+                    if page.page_index == 1:
+                        first_page_classified_count = page_classified_count
+                        first_page_creation_count = page_creation_count
+                        first_page_creation_ratio = page_ratio
+                        if pagination_max_pages <= 1:
+                            pagination_stop_reason = "max_pages_reached"
+                            break
+                        if not page.has_more or not page.next_cursor:
+                            pagination_stop_reason = "no_more_pages"
+                            break
+                        if page_classified_count <= 0:
+                            pagination_stop_reason = "no_classified_candidates"
+                            break
+                        if page_ratio is None or page_ratio < pagination_creation_threshold:
+                            pagination_stop_reason = "below_threshold"
+                            break
+                        requested_second_page = True
                         continue
 
+                    pagination_stop_reason = "max_pages_reached"
+                    break
+
+                if not pagination_stop_reason:
+                    pagination_stop_reason = "pages_exhausted"
+
                 status = "done" if display_count >= display_limit else (
                     "partial" if display_count else "failed"
                 )
@@ -271,7 +566,22 @@ def run_batch(
                     metadata={
                         "query_text": query.query_text,
                         "searched_count": searched_count,
+                        "raw_searched_count": raw_searched_count,
                         "display_count": display_count,
+                        "skipped_item_count": skipped_item_count,
+                        "search_page_count": search_page_count,
+                        "pagination": {
+                            **_pagination_config(
+                                pagination_creation_threshold,
+                                pagination_max_pages,
+                            ),
+                            "first_page_classified_count": first_page_classified_count,
+                            "first_page_creation_count": first_page_creation_count,
+                            "first_page_creation_ratio": first_page_creation_ratio,
+                            "requested_second_page": requested_second_page,
+                            "stop_reason": pagination_stop_reason,
+                            "pages": page_summaries,
+                        },
                         "errors": errors[-3:],
                     },
                 )
@@ -281,11 +591,26 @@ def run_batch(
                     _job_id(job),
                     status="failed",
                     attempt_count=attempts,
-                    error_message=str(exc)[:300],
+                    error_message=clip_text(exc, ERROR_MESSAGE_MAX_CHARS),
                     metadata={
                         "query_text": query.query_text,
                         "searched_count": searched_count,
+                        "raw_searched_count": raw_searched_count,
                         "display_count": display_count,
+                        "skipped_item_count": skipped_item_count,
+                        "search_page_count": search_page_count,
+                        "pagination": {
+                            **_pagination_config(
+                                pagination_creation_threshold,
+                                pagination_max_pages,
+                            ),
+                            "first_page_classified_count": first_page_classified_count,
+                            "first_page_creation_count": first_page_creation_count,
+                            "first_page_creation_ratio": first_page_creation_ratio,
+                            "requested_second_page": requested_second_page,
+                            "stop_reason": pagination_stop_reason or "failed",
+                            "pages": page_summaries,
+                        },
                         "errors": errors[-3:],
                     },
                 )
@@ -308,6 +633,10 @@ def run_batch(
                 "partial": partial,
                 "failed": failed,
                 "skipped": skipped,
+                "pagination": _pagination_config(
+                    pagination_creation_threshold,
+                    pagination_max_pages,
+                ),
             },
         )
 

acquisition/search.py

@@ -2,7 +2,7 @@
 
 接口（实测 2026-06-26，host = settings.aiddit_crawler_base_url）：
     小红书 POST /crawler/xiao_hong_shu/keyword
-      body{keyword, content_type:"图文"|"视频", sort_type:"综合", publish_time:"", cursor:""}
+      body{keyword, sort_type:"综合", publish_time:"", cursor:""}
       → {code:0, data:{has_more, next_cursor, data:[{id:<content_id>, note_card:{...}}]}}
     抖音   POST /crawler/dou_yin/keyword
       body{keyword, content_type:"视频", sort_type:"综合排序", publish_time:"不限",
@@ -15,6 +15,7 @@ parse_search_response 纯函数可离线测；search_keyword 负责 HTTP + 翻
 """
 from __future__ import annotations
 
+from dataclasses import dataclass, field
 import re
 from typing import Any, Optional
 from urllib.parse import urljoin
@@ -22,7 +23,12 @@ from urllib.parse import urljoin
 import httpx
 
 from core.config import Settings
-from acquisition.crawler import RateLimiter
+from core.text_limits import TITLE_FALLBACK_MAX_CHARS, clip_text
+from acquisition.crawler import (
+    CRAWLER_MAX_INTERVAL_SECONDS,
+    CRAWLER_MIN_INTERVAL_SECONDS,
+    RateLimiter,
+)
 
 # 每平台的搜索参数差异（path / 条目 id 字段 / 排序 / 发布时间 / 起始 cursor / account_id）
 PLATFORM_SEARCH = {
@@ -43,9 +49,77 @@ class SearchError(RuntimeError):
     pass
 
 
-def parse_search_response(response: Any, *, platform: str = "xiaohongshu") -> tuple[list[str], bool, str]:
-    """信封 {code,msg,data:{has_more,next_cursor,data:[...]}} → (content_ids, has_more, next_cursor)。
-    条目 id 字段按平台取（小红书 id / 抖音 aweme_id）。code!=0 抛 SearchError。"""
+@dataclass(frozen=True)
+class SearchHit:
+    content_id: str
+    provider: str = ""
+    raw: dict[str, Any] = field(default_factory=dict)
+    request_payload: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class SearchPage:
+    rows: list[Any]
+    raw_count: int
+    cursor: str = ""
+    next_cursor: str = ""
+    has_more: bool = False
+    provider: str = ""
+    request_payload: dict[str, Any] = field(default_factory=dict)
+    raw_metadata: dict[str, Any] = field(default_factory=dict)
+
+
+DOUYIN_PROVIDERS = ("piaoquantv", "aiddit")
+
+
+def _douyin_cursor0(provider: str) -> str:
+    return "" if provider == "aiddit" else "0"
+
+
+def _douyin_base_url(settings: Settings, provider: str) -> str:
+    if provider == "piaoquantv":
+        return settings.piaoquantv_douyin_base_url
+    if provider == "aiddit":
+        return settings.aiddit_crawler_base_url
+    raise SearchError(f"unsupported douyin provider: {provider}")
+
+
+def _douyin_search_body(
+    keyword: str,
+    *,
+    provider: str,
+    cursor: str,
+    settings: Settings,
+    content_type: str | None,
+    sort_type: str | None,
+    publish_time: str | None,
+    account_id: str | None,
+) -> dict[str, Any]:
+    if provider == "piaoquantv":
+        body: dict[str, Any] = {
+            "keyword": keyword,
+            "sort_type": sort_type or "综合排序",
+            "publish_time": publish_time or "不限",
+            "cursor": cursor,
+        }
+        if content_type:
+            body["content_type"] = content_type
+        body["account_id"] = account_id or settings.piaoquantv_douyin_account_id
+        body["cookie_batch"] = settings.piaoquantv_douyin_cookie_batch
+        return body
+    if provider == "aiddit":
+        body = {
+            "keyword": keyword,
+            "cursor": cursor,
+            "sort_type": sort_type or "最多点赞",
+        }
+        if content_type:
+            body["content_type"] = content_type
+        return body
+    raise SearchError(f"unsupported douyin provider: {provider}")
+
+
+def _parse_search_items(response: Any, *, platform: str = "xiaohongshu") -> tuple[list[dict[str, Any]], int, bool, str]:
     if not isinstance(response, dict):
         raise SearchError("bad_response: not a dict")
     code = response.get("code")
@@ -53,16 +127,180 @@ def parse_search_response(response: Any, *, platform: str = "xiaohongshu") -> tu
         raise SearchError(f"business_error: code={code} msg={response.get('msg')}")
     data = response.get("data") or {}
     items = data.get("data") or []
+    if not isinstance(items, list):
+        raise SearchError("bad_response: data.data is not a list")
+    id_key = (PLATFORM_SEARCH.get(platform) or {}).get("id_key", "id")
+    valid = [it for it in items if isinstance(it, dict) and it.get(id_key)]
+    return valid, len(items), bool(data.get("has_more")), str(data.get("next_cursor") or "")
+
+
+def parse_search_response(response: Any, *, platform: str = "xiaohongshu") -> tuple[list[str], bool, str]:
+    """信封 {code,msg,data:{has_more,next_cursor,data:[...]}} → (content_ids, has_more, next_cursor)。
+    条目 id 字段按平台取（小红书 id / 抖音 aweme_id）。code!=0 抛 SearchError。"""
     id_key = (PLATFORM_SEARCH.get(platform) or {}).get("id_key", "id")
-    ids = [it.get(id_key) for it in items if isinstance(it, dict) and it.get(id_key)]
-    return ids, bool(data.get("has_more")), str(data.get("next_cursor") or "")
+    items, _, has_more, next_cursor = _parse_search_items(response, platform=platform)
+    ids = [it[id_key] for it in items]
+    return ids, has_more, next_cursor
+
+
+def search_douyin_page(
+    keyword: str,
+    *,
+    provider: str = "piaoquantv",
+    cursors: dict[str, str] | None = None,
+    content_type: str | None = "视频",
+    sort_type: Optional[str] = None,
+    publish_time: Optional[str] = None,
+    account_id: Optional[str] = None,
+    limit: int = 10,
+    settings: Optional[Settings] = None,
+    http_client: Any = None,
+    rate_limiter: Optional[RateLimiter] = None,
+    env_file: str = ".env",
+) -> SearchPage:
+    """Fetch one Douyin page, using piaoquantv first and AIDDIT as fallback."""
+    settings = settings or Settings.from_env(env_file)
+    rate_limiter = rate_limiter or RateLimiter(
+        min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+        max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+    )
+    owns_client = http_client is None
+    client = http_client or httpx.Client()
+    cursors = dict(cursors or {p: _douyin_cursor0(p) for p in DOUYIN_PROVIDERS})
+    provider_order = [provider] + [p for p in DOUYIN_PROVIDERS if p != provider]
+    errors: list[str] = []
+    try:
+        for attempt_provider in provider_order:
+            cursor = cursors.get(attempt_provider, _douyin_cursor0(attempt_provider))
+            body = _douyin_search_body(
+                keyword,
+                provider=attempt_provider,
+                cursor=cursor,
+                settings=settings,
+                content_type=content_type,
+                sort_type=sort_type,
+                publish_time=publish_time,
+                account_id=account_id,
+            )
+            rate_limiter.wait("douyin")
+            url = urljoin(_douyin_base_url(settings, attempt_provider), PLATFORM_SEARCH["douyin"]["path"])
+            try:
+                resp = client.post(
+                    url,
+                    json=body,
+                    headers={"Content-Type": "application/json"},
+                    timeout=settings.crawler_timeout,
+                )
+                resp.raise_for_status()
+                data = resp.json()
+                items, raw_count, has_more, next_cursor = _parse_search_items(data, platform="douyin")
+            except httpx.HTTPError as exc:
+                errors.append(f"{attempt_provider}: http_error: {exc}")
+                continue
+            except ValueError:
+                errors.append(f"{attempt_provider}: bad_json")
+                continue
+            except SearchError as exc:
+                errors.append(f"{attempt_provider}: {exc}")
+                continue
+            if not items:
+                errors.append(f"{attempt_provider}: empty_aweme_id")
+                continue
+
+            id_key = PLATFORM_SEARCH["douyin"]["id_key"]
+            hits = [
+                SearchHit(
+                    content_id=str(item[id_key]),
+                    provider=attempt_provider,
+                    raw=item,
+                    request_payload=dict(body),
+                )
+                for item in items[:limit]
+            ]
+            next_cursors = dict(cursors)
+            if has_more and next_cursor and next_cursor != cursor:
+                next_cursors[attempt_provider] = next_cursor
+            return SearchPage(
+                rows=hits,
+                raw_count=raw_count,
+                cursor=cursor,
+                next_cursor=next_cursor,
+                has_more=has_more,
+                provider=attempt_provider,
+                request_payload=dict(body),
+                raw_metadata={"cursors": next_cursors, "errors": errors},
+            )
+    finally:
+        if owns_client:
+            client.close()
+    raise SearchError("; ".join(errors) or "douyin_search_failed")
+
+
+def search_douyin_hits(
+    keyword: str,
+    *,
+    content_type: str | None = "视频",
+    sort_type: Optional[str] = None,
+    publish_time: Optional[str] = None,
+    account_id: Optional[str] = None,
+    limit: int = 5,
+    settings: Optional[Settings] = None,
+    http_client: Any = None,
+    rate_limiter: Optional[RateLimiter] = None,
+    env_file: str = ".env",
+    max_pages: int = 10,
+) -> list[SearchHit]:
+    """Douyin search with piaoquantv primary and AIDDIT fallback per cursor."""
+    settings = settings or Settings.from_env(env_file)
+    rate_limiter = rate_limiter or RateLimiter(
+        min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+        max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+    )
+    owns_client = http_client is None
+    client = http_client or httpx.Client()
+    out: list[SearchHit] = []
+    seen: set[str] = set()
+    provider = "piaoquantv"
+    cursors = {p: _douyin_cursor0(p) for p in DOUYIN_PROVIDERS}
+    try:
+        for _ in range(max_pages):
+            page = search_douyin_page(
+                keyword,
+                provider=provider,
+                cursors=cursors,
+                content_type=content_type,
+                sort_type=sort_type,
+                publish_time=publish_time,
+                account_id=account_id,
+                limit=limit,
+                settings=settings,
+                http_client=client,
+                rate_limiter=rate_limiter,
+                env_file=env_file,
+            )
+            page_hits = list(page.rows)
+            provider = page.provider
+            current_cursor = cursors.get(provider, _douyin_cursor0(provider))
+            for hit in page_hits:
+                if hit.content_id not in seen:
+                    seen.add(hit.content_id)
+                    out.append(hit)
+                    if len(out) >= limit:
+                        return out
+            if not page.has_more or not page.next_cursor or page.next_cursor == current_cursor:
+                break
+            cursors = dict(page.raw_metadata.get("cursors") or cursors)
+    finally:
+        if owns_client:
+            client.close()
+    return out[:limit]
 
 
 def search_keyword(
     keyword: str,
     *,
     platform: str = "xiaohongshu",
-    content_type: str = "图文",
+    content_type: str | None = "图文",
     sort_type: Optional[str] = None,       # None → 平台默认（小红书:综合 / 抖音:综合排序）
     publish_time: Optional[str] = None,     # None → 平台默认（抖音:不限）
     account_id: Optional[str] = None,       # None → 平台默认（抖音:771431222；小红书无）
@@ -80,13 +318,33 @@ def search_keyword(
     if not cfg:
         raise SearchError(f"unsupported platform: {platform}")
     settings = settings or Settings.from_env(env_file)
-    sort_type = cfg["sort_type"] if sort_type is None else sort_type
-    publish_time = cfg["publish_time"] if publish_time is None else publish_time
     is_douyin = platform == "douyin"
     if account_id is None:
         account_id = settings.piaoquantv_douyin_account_id if is_douyin else cfg["account_id"]
+    if is_douyin:
+        return [
+            hit.content_id
+            for hit in search_douyin_hits(
+                keyword,
+                content_type=content_type,
+                sort_type=sort_type,
+                publish_time=publish_time,
+                account_id=account_id,
+                limit=limit,
+                settings=settings,
+                http_client=http_client,
+                rate_limiter=rate_limiter,
+                env_file=env_file,
+                max_pages=max_pages,
+            )
+        ]
+    sort_type = cfg["sort_type"] if sort_type is None else sort_type
+    publish_time = cfg["publish_time"] if publish_time is None else publish_time
     base_url = settings.piaoquantv_douyin_base_url if is_douyin else settings.aiddit_crawler_base_url
-    rate_limiter = rate_limiter or RateLimiter()
+    rate_limiter = rate_limiter or RateLimiter(
+        min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+        max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+    )
 
     owns_client = http_client is None
     client = http_client or httpx.Client()
@@ -97,8 +355,10 @@ def search_keyword(
         for _ in range(max_pages):
             rate_limiter.wait(f"{platform}_keyword")
             url = urljoin(base_url, cfg["path"])
-            body = {"keyword": keyword, "content_type": content_type,
+            body = {"keyword": keyword,
                     "sort_type": sort_type, "publish_time": publish_time, "cursor": cursor}
+            if content_type:
+                body["content_type"] = content_type
             if account_id:
                 body["account_id"] = account_id   # 抖音必带；小红书不带
             if is_douyin:
@@ -163,7 +423,10 @@ def search_weixin(
 ) -> list[dict]:
     """微信公众号搜索：query → 文章列表（带 url/封面，无需 detail）。失败抛 SearchError。"""
     settings = settings or Settings.from_env(env_file)
-    rate_limiter = rate_limiter or RateLimiter()
+    rate_limiter = rate_limiter or RateLimiter(
+        min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+        max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+    )
     owns_client = http_client is None
     client = http_client or httpx.Client()
     try:
@@ -185,6 +448,60 @@ def search_weixin(
             client.close()
 
 
+def search_weixin_page(
+    keyword: str,
+    *,
+    cursor: str = "0",
+    limit: int = 10,
+    settings: Optional[Settings] = None,
+    http_client: Any = None,
+    rate_limiter: Optional[RateLimiter] = None,
+    env_file: str = ".env",
+) -> SearchPage:
+    """微信公众号搜索单页：query + cursor → SearchPage。"""
+    settings = settings or Settings.from_env(env_file)
+    rate_limiter = rate_limiter or RateLimiter(
+        min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+        max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+    )
+    owns_client = http_client is None
+    client = http_client or httpx.Client()
+    body = {"keyword": keyword, "cursor": cursor}
+    try:
+        rate_limiter.wait("weixin_keyword")
+        url = urljoin(settings.aiddit_crawler_base_url, "/crawler/wei_xin/keyword")
+        try:
+            resp = client.post(url, json=body,
+                               headers={"Content-Type": "application/json"},
+                               timeout=settings.crawler_timeout)
+            resp.raise_for_status()
+            data = resp.json()
+        except httpx.HTTPError as exc:
+            raise SearchError(f"http_error: {exc}") from exc
+        except ValueError as exc:
+            raise SearchError("bad_json") from exc
+        if not isinstance(data, dict):
+            raise SearchError("bad_response: not a dict")
+        if data.get("code") not in (0, "0"):
+            raise SearchError(f"business_error: code={data.get('code')} msg={data.get('msg')}")
+        envelope = data.get("data") or {}
+        raw_rows = envelope.get("data") or []
+        if not isinstance(raw_rows, list):
+            raise SearchError("bad_response: data.data is not a list")
+        return SearchPage(
+            rows=parse_weixin(data, limit=limit),
+            raw_count=len(raw_rows),
+            cursor=cursor,
+            next_cursor=str(envelope.get("next_cursor") or ""),
+            has_more=bool(envelope.get("has_more")),
+            provider="aiddit",
+            request_payload=body,
+        )
+    finally:
+        if owns_client:
+            client.close()
+
+
 def fetch_weixin_detail(
     url: str,
     *,
@@ -195,7 +512,10 @@ def fetch_weixin_detail(
 ) -> tuple[str, list[str]]:
     """微信公众号文章详情：content_link → (body_text, image_urls)。实测不需 token。失败抛 SearchError。"""
     settings = settings or Settings.from_env(env_file)
-    rate_limiter = rate_limiter or RateLimiter()
+    rate_limiter = rate_limiter or RateLimiter(
+        min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+        max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+    )
     owns_client = http_client is None
     client = http_client or httpx.Client()
     try:
@@ -246,7 +566,7 @@ def parse_xiaohongshu(response: Any, *, limit: int = 5) -> list[dict]:
             continue
         title = (nc.get("display_title") or "").strip()
         if not title:
-            title = (nc.get("desc") or "").strip().split("\n")[0][:50]
+            title = clip_text((nc.get("desc") or "").strip().split("\n")[0], TITLE_FALLBACK_MAX_CHARS)
         out.append({"id": cid, "title": title, "cover_url": cover,
                     "nick_name": (nc.get("user") or {}).get("nickname") or "",
                     "type": nc.get("type") or "",
@@ -259,7 +579,7 @@ def parse_xiaohongshu(response: Any, *, limit: int = 5) -> list[dict]:
 def search_xiaohongshu(
     keyword: str,
     *,
-    content_type: str = "图文",
+    content_type: str | None = None,
     limit: int = 5,
     settings: Optional[Settings] = None,
     http_client: Any = None,
@@ -268,15 +588,20 @@ def search_xiaohongshu(
 ) -> list[dict]:
     """小红书搜索：query → 帖子列表（带 url/封面/标题，无需 detail；视频帖无直链）。失败抛 SearchError。"""
     settings = settings or Settings.from_env(env_file)
-    rate_limiter = rate_limiter or RateLimiter()
+    rate_limiter = rate_limiter or RateLimiter(
+        min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+        max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+    )
     owns_client = http_client is None
     client = http_client or httpx.Client()
     try:
         rate_limiter.wait("xiaohongshu_keyword")
         url = urljoin(settings.aiddit_crawler_base_url, "/crawler/xiao_hong_shu/keyword")
         try:
-            resp = client.post(url, json={"keyword": keyword, "content_type": content_type,
-                                          "sort_type": "综合", "publish_time": "", "cursor": ""},
+            body = {"keyword": keyword, "sort_type": "综合", "publish_time": "", "cursor": ""}
+            if content_type:
+                body["content_type"] = content_type
+            resp = client.post(url, json=body,
                                headers={"Content-Type": "application/json"},
                                timeout=settings.crawler_timeout)
             resp.raise_for_status()
@@ -289,3 +614,60 @@ def search_xiaohongshu(
     finally:
         if owns_client:
             client.close()
+
+
+def search_xiaohongshu_page(
+    keyword: str,
+    *,
+    cursor: str = "",
+    content_type: str | None = None,
+    limit: int = 10,
+    settings: Optional[Settings] = None,
+    http_client: Any = None,
+    rate_limiter: Optional[RateLimiter] = None,
+    env_file: str = ".env",
+) -> SearchPage:
+    """小红书搜索单页：query + cursor → SearchPage。"""
+    settings = settings or Settings.from_env(env_file)
+    rate_limiter = rate_limiter or RateLimiter(
+        min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS,
+        max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS,
+    )
+    owns_client = http_client is None
+    client = http_client or httpx.Client()
+    body = {"keyword": keyword, "sort_type": "综合", "publish_time": "", "cursor": cursor}
+    if content_type:
+        body["content_type"] = content_type
+    try:
+        rate_limiter.wait("xiaohongshu_keyword")
+        url = urljoin(settings.aiddit_crawler_base_url, "/crawler/xiao_hong_shu/keyword")
+        try:
+            resp = client.post(url, json=body,
+                               headers={"Content-Type": "application/json"},
+                               timeout=settings.crawler_timeout)
+            resp.raise_for_status()
+            data = resp.json()
+        except httpx.HTTPError as exc:
+            raise SearchError(f"http_error: {exc}") from exc
+        except ValueError as exc:
+            raise SearchError("bad_json") from exc
+        if not isinstance(data, dict):
+            raise SearchError("bad_response: not a dict")
+        if data.get("code") not in (0, "0"):
+            raise SearchError(f"business_error: code={data.get('code')} msg={data.get('msg')}")
+        envelope = data.get("data") or {}
+        raw_rows = envelope.get("data") or []
+        if not isinstance(raw_rows, list):
+            raise SearchError("bad_response: data.data is not a list")
+        return SearchPage(
+            rows=parse_xiaohongshu(data, limit=limit),
+            raw_count=len(raw_rows),
+            cursor=cursor,
+            next_cursor=str(envelope.get("next_cursor") or ""),
+            has_more=bool(envelope.get("has_more")),
+            provider="aiddit",
+            request_payload=body,
+        )
+    finally:
+        if owns_client:
+            client.close()

tests/test_acquisition_runner.py

@@ -13,7 +13,7 @@ from acquisition.domain import (
     QueryBatch,
 )
 from acquisition.media.service import StabilizedMedia
-from acquisition.platforms.base import PlatformCandidate, PlatformItem
+from acquisition.platforms.base import PlatformCandidate, PlatformItem, PlatformSearchPage
 from acquisition.runner import run_batch
 from core.config import PgConfig, Settings
 
@@ -52,6 +52,8 @@ class FakeRepo:
         self.items: list[CandidateItem] = []
         self.media: list[MediaAsset] = []
         self.classifications: list[ItemClassification] = []
+        self.existing_by_unique_key: dict[str, CandidateItem] = {}
+        self.attached_items: list[CandidateItem] = []
 
     def create_query_batch(self, **kwargs):
         return QueryBatch(id=self.batch_id, **kwargs)
@@ -86,6 +88,15 @@ class FakeRepo:
         self.items.append(item)
         return item
 
+    def get_candidate_item_by_unique_key(self, unique_key: str):
+        return self.existing_by_unique_key.get(unique_key)
+
+    def attach_existing_candidate_item(self, item_id: UUID, *, job_id: UUID, query_id: UUID, metadata=None):
+        item = next(row for row in self.existing_by_unique_key.values() if row.id == item_id)
+        updated = item.model_copy(update={"job_id": job_id, "query_id": query_id, "metadata": metadata or {}})
+        self.attached_items.append(updated)
+        return updated
+
     def add_media_asset(self, **kwargs):
         row = MediaAsset(id=uuid4(), **kwargs)
         self.media.append(row)
@@ -105,6 +116,10 @@ class FakeRepo:
 
 class FakeAdapter:
     platform = "weixin"
+    url = (
+        "https://mp.weixin.qq.com/s?__biz=MzUxMDg4NDY1Ng=="
+        "&mid=2247484898&idx=2&sn=5069e65919f414b96b7aec53671dc1b5"
+    )
 
     def __init__(self) -> None:
         self.search_calls = 0
@@ -119,7 +134,7 @@ class FakeAdapter:
                 rank=1,
                 platform=self.platform,
                 source_id="wx1",
-                url="https://mp.weixin.qq.com/s/a",
+                url=self.url,
                 title="公众号脚本",
                 author="作者",
                 cover_url="https://img.test/a.jpg",
@@ -172,6 +187,184 @@ class FailingSearchAdapter(FakeAdapter):
         raise RuntimeError("search boom")
 
 
+class UnsupportedModeAdapter(FakeAdapter):
+    def fetch_detail(self, candidate, *, settings, rate_limiter):
+        self.detail_calls += 1
+        return PlatformItem(
+            platform=self.platform,
+            source_id=candidate.source_id,
+            url=candidate.url,
+            content_type="mystery",
+            content_mode="unsupported",
+            title="未知类型",
+            body_text="",
+            image_urls=[],
+            video_urls=[],
+            raw={},
+        )
+
+
+class MissingVideoAdapter(FakeAdapter):
+    def fetch_detail(self, candidate, *, settings, rate_limiter):
+        self.detail_calls += 1
+        return PlatformItem(
+            platform="douyin",
+            source_id="dy-video",
+            url="https://douyin.test/v",
+            content_type="video",
+            content_mode="video_post",
+            title="缺视频地址",
+            body_text="",
+            image_urls=[],
+            video_urls=[],
+            raw={},
+        )
+
+
+class DouyinProviderAdapter(FakeAdapter):
+    platform = "douyin"
+
+    def search(self, query, *, settings, limit, rate_limiter):
+        self.search_calls += 1
+        return [
+            PlatformCandidate(
+                rank=1,
+                platform=self.platform,
+                provider="aiddit",
+                source_id="761",
+                raw={
+                    "id": "761",
+                    "search_provider": "aiddit",
+                    "original": {"aweme_id": "761"},
+                    "request_payload": {"keyword": query},
+                },
+            )
+        ]
+
+    def fetch_detail(self, candidate, *, settings, rate_limiter):
+        self.detail_calls += 1
+        return PlatformItem(
+            platform=self.platform,
+            provider=candidate.provider,
+            source_id=candidate.source_id,
+            url="https://douyin.test/video/761",
+            content_type="video",
+            content_mode="video_post",
+            title="视频标题",
+            author="作者",
+            body_text="视频正文",
+            video_urls=["https://video.test/761.mp4"],
+            raw={
+                "detail_provider": candidate.provider,
+                "code": 0,
+                "data": {"data": {"channel_content_id": candidate.source_id}},
+            },
+        )
+
+
+def _paged_candidate(source_id: str, *, rank: int, page_index: int, platform: str = "douyin") -> PlatformCandidate:
+    return PlatformCandidate(
+        rank=rank,
+        platform=platform,
+        provider="aiddit" if platform == "douyin" else "",
+        source_id=source_id,
+        url=f"https://{platform}.test/{source_id}",
+        title=source_id,
+        raw={
+            "page_index": page_index,
+            "page_rank": rank,
+            "source_cursor": "" if page_index == 1 else str(page_index),
+        },
+    )
+
+
+def _paged_search_page(
+    page_index: int,
+    source_ids: list[str],
+    *,
+    has_more: bool = False,
+    next_cursor: str = "",
+    platform: str = "douyin",
+) -> PlatformSearchPage:
+    return PlatformSearchPage(
+        page_index=page_index,
+        candidates=[
+            _paged_candidate(source_id, rank=i, page_index=page_index, platform=platform)
+            for i, source_id in enumerate(source_ids, start=1)
+        ],
+        raw_count=len(source_ids),
+        cursor="" if page_index == 1 else str(page_index),
+        next_cursor=next_cursor,
+        has_more=has_more,
+        provider="aiddit" if platform == "douyin" else "",
+    )
+
+
+class PagedAdapter:
+    platform = "douyin"
+
+    def __init__(
+        self,
+        pages: list[PlatformSearchPage],
+        details: dict[str, PlatformItem | Exception],
+    ) -> None:
+        self.pages = pages
+        self.details = details
+        self.pages_requested = 0
+        self.detail_calls = 0
+
+    def search_pages(self, query, *, settings, limit, rate_limiter, max_pages=2):
+        for page in self.pages[:max_pages]:
+            self.pages_requested += 1
+            yield page
+
+    def search(self, query, *, settings, limit, rate_limiter):
+        self.pages_requested += 1
+        return self.pages[0].candidates if self.pages else []
+
+    def fetch_detail(self, candidate, *, settings, rate_limiter):
+        self.detail_calls += 1
+        detail = self.details[candidate.source_id]
+        if isinstance(detail, Exception):
+            raise detail
+        return detail
+
+
+def _detail(
+    source_id: str,
+    *,
+    content_mode: str = "image_post",
+    image_urls: list[str] | None = None,
+    video_urls: list[str] | None = None,
+) -> PlatformItem:
+    return PlatformItem(
+        platform="douyin",
+        provider="aiddit",
+        source_id=source_id,
+        url=f"https://douyin.test/{source_id}",
+        content_type="image" if content_mode == "image_post" else "video",
+        content_mode=content_mode,
+        title=source_id,
+        body_text=f"{source_id} 正文",
+        image_urls=image_urls if image_urls is not None else ["https://img.test/a.jpg"],
+        video_urls=video_urls or [],
+        raw={},
+    )
+
+
+def _classification_by_title(creation_titles: set[str]):
+    def classifier(**kwargs):
+        is_creation = kwargs["title"] in creation_titles
+        return ClassificationResult(
+            is_creation_knowledge=is_creation,
+            label="creation" if is_creation else "non_creation",
+            confidence=0.9,
+            reason="ok",
+        )
+
+    return classifier
+
+
 def test_run_batch_writes_jobs_items_media_and_classification():
     repo = FakeRepo()
     adapter = FakeAdapter()
@@ -188,6 +381,7 @@ def test_run_batch_writes_jobs_items_media_and_classification():
         ]
 
     def classifier(**kwargs):
+        assert kwargs["content_mode"] == "article"
         assert kwargs["image_urls"] == ["https://cdn.test/a.jpg"]
         return ClassificationResult(
             is_creation_knowledge=True,
@@ -219,11 +413,349 @@ def test_run_batch_writes_jobs_items_media_and_classification():
     assert adapter.detail_calls == 1
     assert repo.jobs[0].query_id == repo.query.id
     assert repo.items[0].platform_item_id == "wx1"
+    assert repo.items[0].content_mode == "article"
+    assert repo.items[0].body_text == "先定受众再写开头"
+    assert repo.items[0].raw_summary == "先定受众再写开头"
+    assert (
+        repo.items[0].unique_key
+        == "wx:MzUxMDg4NDY1Ng==:2247484898:2:5069e65919f414b96b7aec53671dc1b5"
+    )
     assert repo.media[0].cdn_url == "https://cdn.test/a.jpg"
     assert repo.classifications[0].is_creation_knowledge is True
     assert repo.updated_jobs[-1].status == "done"
 
 
+def test_run_batch_records_unsupported_item_as_skipped_without_processing():
+    repo = FakeRepo()
+    adapter = UnsupportedModeAdapter()
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("weixin",),
+        search_limit=2,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: (_ for _ in ()).throw(AssertionError("media should be skipped")),
+        classifier=lambda **kwargs: (_ for _ in ()).throw(AssertionError("classifier should be skipped")),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.done == 1
+    assert repo.items[0].status == "skipped"
+    assert repo.items[0].content_mode == "unsupported"
+    assert repo.items[0].metadata["skip_reason"] == "unsupported_content_mode"
+    assert repo.media == []
+    assert repo.classifications[0].status == "skipped"
+    assert repo.classifications[0].label == "unsupported_content_mode"
+    assert repo.updated_jobs[-1].metadata["skipped_item_count"] == 1
+
+
+def test_run_batch_records_video_missing_as_skipped_without_processing():
+    repo = FakeRepo()
+    adapter = MissingVideoAdapter()
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=2,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: (_ for _ in ()).throw(AssertionError("media should be skipped")),
+        classifier=lambda **kwargs: (_ for _ in ()).throw(AssertionError("classifier should be skipped")),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.done == 1
+    assert repo.items[0].content_mode == "video_post"
+    assert repo.items[0].metadata["video_url_missing"] is True
+    assert repo.classifications[0].label == "video_missing"
+
+
+def test_run_batch_attaches_existing_unique_key_without_fetching_detail_or_classifying():
+    repo = FakeRepo()
+    adapter = FakeAdapter()
+    unique_key = "wx:MzUxMDg4NDY1Ng==:2247484898:2:5069e65919f414b96b7aec53671dc1b5"
+    repo.existing_by_unique_key[unique_key] = CandidateItem(
+        id=uuid4(),
+        platform="weixin",
+        platform_item_id="wx1",
+        unique_key=unique_key,
+        title="历史帖子",
+        status="candidate",
+    )
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("weixin",),
+        search_limit=2,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: (_ for _ in ()).throw(AssertionError("media should be skipped")),
+        classifier=lambda **kwargs: (_ for _ in ()).throw(AssertionError("classifier should be skipped")),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.done == 1
+    assert adapter.search_calls == 1
+    assert adapter.detail_calls == 0
+    assert repo.items == []
+    assert repo.media == []
+    assert repo.classifications == []
+    assert repo.attached_items[0].job_id == repo.jobs[0].id
+    assert repo.attached_items[0].query_id == repo.query.id
+    assert repo.attached_items[0].metadata["acquisition_match_status"] == "existing"
+    assert repo.attached_items[0].metadata["matched_unique_key"] == unique_key
+
+
+def test_run_batch_records_douyin_provider_in_metadata_and_source_payload():
+    repo = FakeRepo()
+    adapter = DouyinProviderAdapter()
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=1,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: [
+            StabilizedMedia(
+                media_type="video",
+                source_url="https://video.test/761.mp4",
+                cdn_url="https://cdn.test/761.mp4",
+                position=1,
+            )
+        ],
+        classifier=lambda **kwargs: ClassificationResult(
+            is_creation_knowledge=True,
+            label="creation",
+            confidence=0.9,
+            reason="ok",
+        ),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.done == 1
+    item = repo.items[0]
+    assert item.metadata["search_provider"] == "aiddit"
+    assert item.metadata["detail_provider"] == "aiddit"
+    assert item.source_payload["candidate"]["provider"] == "aiddit"
+    assert item.source_payload["candidate"]["raw"]["original"] == {"aweme_id": "761"}
+    assert item.source_payload["detail"]["detail_provider"] == "aiddit"
+
+
+def test_run_batch_records_duplicate_search_provider_without_fetching_detail():
+    repo = FakeRepo()
+    adapter = DouyinProviderAdapter()
+    unique_key = "dy:761"
+    repo.existing_by_unique_key[unique_key] = CandidateItem(
+        id=uuid4(),
+        platform="douyin",
+        platform_item_id="761",
+        unique_key=unique_key,
+        title="历史视频",
+        status="candidate",
+    )
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=1,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: (_ for _ in ()).throw(AssertionError("media should be skipped")),
+        classifier=lambda **kwargs: (_ for _ in ()).throw(AssertionError("classifier should be skipped")),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.done == 1
+    assert adapter.detail_calls == 0
+    assert repo.attached_items[0].metadata["search_provider"] == "aiddit"
+    assert repo.attached_items[0].metadata["matched_candidate"]["provider"] == "aiddit"
+
+
+def test_run_batch_requests_second_page_when_first_page_creation_ratio_reaches_threshold():
+    repo = FakeRepo()
+    adapter = PagedAdapter(
+        pages=[
+            _paged_search_page(1, ["a", "b"], has_more=True, next_cursor="2"),
+            _paged_search_page(2, ["c"], has_more=True, next_cursor="3"),
+            _paged_search_page(3, ["d"]),
+        ],
+        details={source_id: _detail(source_id) for source_id in ["a", "b", "c", "d"]},
+    )
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=10,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: [],
+        classifier=_classification_by_title({"a", "c"}),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    pagination = repo.updated_jobs[-1].metadata["pagination"]
+    assert result.done == 1
+    assert adapter.pages_requested == 2
+    assert pagination["first_page_classified_count"] == 2
+    assert pagination["first_page_creation_count"] == 1
+    assert pagination["first_page_creation_ratio"] == 0.5
+    assert pagination["requested_second_page"] is True
+    assert pagination["stop_reason"] == "max_pages_reached"
+    assert len(pagination["pages"]) == 2
+
+
+def test_run_batch_does_not_request_second_page_below_threshold():
+    repo = FakeRepo()
+    adapter = PagedAdapter(
+        pages=[
+            _paged_search_page(1, ["a", "b"], has_more=True, next_cursor="2"),
+            _paged_search_page(2, ["c"]),
+        ],
+        details={source_id: _detail(source_id) for source_id in ["a", "b", "c"]},
+    )
+
+    run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=10,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: [],
+        classifier=_classification_by_title(set()),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    pagination = repo.updated_jobs[-1].metadata["pagination"]
+    assert adapter.pages_requested == 1
+    assert pagination["first_page_creation_ratio"] == 0.0
+    assert pagination["requested_second_page"] is False
+    assert pagination["stop_reason"] == "below_threshold"
+
+
+def test_run_batch_does_not_request_second_page_when_no_classified_candidates():
+    repo = FakeRepo()
+    adapter = PagedAdapter(
+        pages=[
+            _paged_search_page(1, ["bad"], has_more=True, next_cursor="2"),
+            _paged_search_page(2, ["c"]),
+        ],
+        details={"bad": RuntimeError("detail failed"), "c": _detail("c")},
+    )
+
+    run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=10,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: [],
+        classifier=_classification_by_title({"c"}),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    pagination = repo.updated_jobs[-1].metadata["pagination"]
+    assert adapter.pages_requested == 1
+    assert pagination["first_page_classified_count"] == 0
+    assert pagination["requested_second_page"] is False
+    assert pagination["stop_reason"] == "no_classified_candidates"
+
+
+def test_run_batch_excludes_duplicate_and_skipped_items_from_pagination_denominator():
+    repo = FakeRepo()
+    repo.existing_by_unique_key["dy:dup"] = CandidateItem(
+        id=uuid4(),
+        platform="douyin",
+        platform_item_id="dup",
+        unique_key="dy:dup",
+        status="candidate",
+    )
+    adapter = PagedAdapter(
+        pages=[
+            _paged_search_page(
+                1,
+                ["dup", "unsupported", "missing_video", "good"],
+                has_more=True,
+                next_cursor="2",
+            ),
+            _paged_search_page(2, ["next"]),
+        ],
+        details={
+            "unsupported": _detail("unsupported", content_mode="unsupported", image_urls=[]),
+            "missing_video": _detail("missing_video", content_mode="video_post", image_urls=[]),
+            "good": _detail("good"),
+            "next": _detail("next"),
+        },
+    )
+
+    run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=10,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: [],
+        classifier=_classification_by_title({"good", "next"}),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    pagination = repo.updated_jobs[-1].metadata["pagination"]
+    assert adapter.pages_requested == 2
+    assert pagination["first_page_classified_count"] == 1
+    assert pagination["first_page_creation_count"] == 1
+    assert pagination["first_page_creation_ratio"] == 1.0
+    assert pagination["requested_second_page"] is True
+
+
+def test_run_batch_threshold_is_configurable():
+    repo = FakeRepo()
+    adapter = PagedAdapter(
+        pages=[
+            _paged_search_page(1, ["a", "b"], has_more=True, next_cursor="2"),
+            _paged_search_page(2, ["c"]),
+        ],
+        details={source_id: _detail(source_id) for source_id in ["a", "b", "c"]},
+    )
+
+    run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("douyin",),
+        search_limit=10,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: [],
+        classifier=_classification_by_title({"a"}),
+        rate_limiter_factory=lambda platform: object(),
+        pagination_creation_threshold=0.75,
+    )
+
+    pagination = repo.updated_jobs[-1].metadata["pagination"]
+    assert adapter.pages_requested == 1
+    assert pagination["creation_threshold"] == 0.75
+    assert pagination["stop_reason"] == "below_threshold"
+
+
 def test_run_batch_skip_done_does_not_call_platform():
     repo = FakeRepo(job_status="done")
     adapter = FakeAdapter()

tests/test_crawler.py

@@ -8,10 +8,13 @@ import pytest
 
 from acquisition.crawler import (
     PLATFORMS,
+    RateLimiter,
     detect_platform_and_id,
+    fetch_post_detail,
     parse_content_id,
     parse_detail_response,
 )
+from core.config import PgConfig, Settings
 
 FIXTURES = Path(__file__).parent / "fixtures"
 
@@ -29,6 +32,56 @@ def _load(content_id: str) -> dict:
     return json.loads((FIXTURES / f"xhs_case_{content_id}.json").read_text("utf-8"))
 
 
+def _settings() -> Settings:
+    return Settings(
+        pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
+        aiddit_crawler_base_url="http://crawler.test",
+        crawler_timeout=30,
+        openrouter_timeout_seconds=90,
+        openrouter_model="m",
+        openrouter_base_url="b",
+        openrouter_api_key="k",
+        llm_model="m",
+        max_cards=12,
+        frames_dir="f",
+        douyin_ratio="540p",
+        data_dir="",
+    )
+
+
+class _Resp:
+    def __init__(self, payload):
+        self._p = payload
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._p
+
+
+class _FakeClient:
+    def __init__(self, payload):
+        self.payload = payload
+        self.calls = []
+        self.closed = False
+
+    def post(self, url, json=None, headers=None, timeout=None):
+        self.calls.append({"url": url, "body": json})
+        return _Resp(self.payload)
+
+    def close(self):
+        self.closed = True
+
+
+class _RecordingLimiter:
+    def __init__(self):
+        self.buckets = []
+
+    def wait(self, bucket: str):
+        self.buckets.append(bucket)
+
+
 @pytest.mark.parametrize("content_id", list(EXPECTED))
 def test_parse_detail_fields(content_id: str):
     # 重抓的 xhs_case_*.json 即原始响应 {code,msg,data}，无 response 包裹
@@ -97,3 +150,68 @@ def test_parse_detail_platform_prefix():
                                      fallback_content_id="67e4bdf50000000006028a59")
         assert post.platform == platform
         assert post.id == f"{expected_prefix}_67e4bdf50000000006028a59"
+
+
+def test_fetch_douyin_detail_piaoquantv_body_and_provider():
+    payload = {"code": 0, "data": {"data": {
+        "channel_content_id": "761",
+        "content_link": "https://douyin.test/video/761",
+        "content_type": "video",
+        "title": "视频",
+        "video_url_list": [{"video_url": "https://video.test/761.mp4"}],
+    }}}
+    client = _FakeClient(payload)
+    limiter = _RecordingLimiter()
+
+    post = fetch_post_detail(
+        "761",
+        platform="douyin",
+        provider="piaoquantv",
+        settings=_settings(),
+        http_client=client,
+        rate_limiter=limiter,
+    )
+
+    assert post.provider == "piaoquantv"
+    assert post.raw["detail_provider"] == "piaoquantv"
+    assert client.calls[0]["url"].startswith("http://crawapi.piaoquantv.com/")
+    assert client.calls[0]["body"]["content_id"] == "761"
+    assert client.calls[0]["body"]["account_id"] == "7450041106378522636"
+    assert client.calls[0]["body"]["cookie_batch"] == "default"
+    assert limiter.buckets == ["douyin"]
+
+
+def test_fetch_douyin_detail_aiddit_body_and_provider():
+    payload = {"code": 0, "data": {"data": {
+        "channel_content_id": "762",
+        "content_link": "https://douyin.test/video/762",
+        "content_type": "video",
+        "title": "视频",
+        "video_url_list": [{"video_url": "https://video.test/762.mp4"}],
+    }}}
+    client = _FakeClient(payload)
+
+    post = fetch_post_detail(
+        "762",
+        platform="douyin",
+        provider="aiddit",
+        settings=_settings(),
+        http_client=client,
+        rate_limiter=_RecordingLimiter(),
+    )
+
+    assert post.provider == "aiddit"
+    assert client.calls[0]["url"].startswith("http://crawler.test/")
+    assert client.calls[0]["body"] == {"content_id": "762"}
+
+
+def test_fetch_douyin_detail_rejects_unknown_provider():
+    with pytest.raises(RuntimeError):
+        fetch_post_detail(
+            "762",
+            platform="douyin",
+            provider="other",
+            settings=_settings(),
+            http_client=_FakeClient({}),
+            rate_limiter=RateLimiter(min_interval_seconds=0.0),
+        )

tests/test_platform_adapters.py

@@ -2,31 +2,38 @@ from __future__ import annotations
 
 from types import SimpleNamespace
 
-from acquisition.platforms.douyin import DouyinAdapter
 from acquisition.platforms import douyin as douyin_module
 from acquisition.platforms import weixin as weixin_module
 from acquisition.platforms import xiaohongshu as xhs_module
+from acquisition.platforms.base import PlatformCandidate
+from acquisition.platforms.douyin import DouyinAdapter
 from acquisition.platforms.weixin import WeixinAdapter
 from acquisition.platforms.xiaohongshu import XiaohongshuAdapter
+from acquisition.search import SearchHit, SearchPage
 from core.models import Post
 
 
 def test_xiaohongshu_adapter_maps_search_and_detail(monkeypatch):
     settings = SimpleNamespace(search_content_type="图文")
 
-    def fake_search(query, *, content_type, limit, settings, rate_limiter):
+    def fake_search(query, *, cursor, content_type, limit, settings, rate_limiter):
         assert query == "脚本 开头"
-        assert content_type == "图文"
+        assert cursor == ""
+        assert content_type is None
         assert limit == 3
-        return [
-            {
-                "id": "xhs-1",
-                "url": "https://xhs.test/1",
-                "title": "脚本开头",
-                "nick_name": "作者 A",
-                "cover_url": "https://img.test/cover.jpg",
-            }
-        ]
+        return SearchPage(
+            rows=[
+                {
+                    "id": "xhs-1",
+                    "url": "https://xhs.test/1",
+                    "title": "脚本开头",
+                    "nick_name": "作者 A",
+                    "cover_url": "https://img.test/cover.jpg",
+                }
+            ],
+            raw_count=1,
+            cursor=cursor,
+        )
 
     def fake_fetch(source_id, *, settings, rate_limiter):
         assert source_id == "xhs-1"
@@ -43,7 +50,7 @@ def test_xiaohongshu_adapter_maps_search_and_detail(monkeypatch):
             raw={"source": "mock"},
         )
 
-    monkeypatch.setattr(xhs_module, "search_xiaohongshu", fake_search)
+    monkeypatch.setattr(xhs_module, "search_xiaohongshu_page", fake_search)
     monkeypatch.setattr(xhs_module, "fetch_post_detail", fake_fetch)
 
     adapter = XiaohongshuAdapter()
@@ -52,31 +59,77 @@ def test_xiaohongshu_adapter_maps_search_and_detail(monkeypatch):
 
     assert candidate.source_id == "xhs-1"
     assert candidate.author == "作者 A"
+    assert candidate.raw["page_index"] == 1
+    assert candidate.raw["page_rank"] == 1
     assert item.title == "详情标题"
     assert item.image_urls == ["https://img.test/1.jpg"]
+    assert item.content_mode == "image_post"
     assert item.raw["source"] == "mock"
 
 
+def test_xiaohongshu_adapter_search_pages_uses_next_cursor(monkeypatch):
+    settings = SimpleNamespace(search_content_type="图文")
+    cursors: list[str] = []
+
+    def fake_search(query, *, cursor, content_type, limit, settings, rate_limiter):
+        cursors.append(cursor)
+        if cursor == "":
+            return SearchPage(
+                rows=[{"id": "xhs-1", "url": "https://xhs.test/1", "title": "第一页"}],
+                raw_count=1,
+                cursor=cursor,
+                next_cursor="2",
+                has_more=True,
+            )
+        return SearchPage(
+            rows=[{"id": "xhs-2", "url": "https://xhs.test/2", "title": "第二页"}],
+            raw_count=1,
+            cursor=cursor,
+            has_more=False,
+        )
+
+    monkeypatch.setattr(xhs_module, "search_xiaohongshu_page", fake_search)
+
+    pages = list(
+        XiaohongshuAdapter().search_pages(
+            "脚本",
+            settings=settings,
+            limit=2,
+            rate_limiter=None,
+            max_pages=2,
+        )
+    )
+
+    assert cursors == ["", "2"]
+    assert [page.candidates[0].source_id for page in pages] == ["xhs-1", "xhs-2"]
+    assert pages[1].candidates[0].raw["page_index"] == 2
+
+
 def test_weixin_adapter_hashes_url_and_reads_detail(monkeypatch):
     settings = SimpleNamespace()
 
-    def fake_search(query, *, limit, settings, rate_limiter):
+    def fake_search(query, *, cursor, limit, settings, rate_limiter):
         assert query == "公众号 选题"
+        assert cursor == "0"
         assert limit == 2
-        return [
-            {
-                "url": "https://mp.weixin.qq.com/s/abc",
-                "title": "公众号选题",
-                "nick_name": "公众号",
-                "cover_url": "https://img.test/wx.jpg",
-            }
-        ]
+        return SearchPage(
+            rows=[
+                {
+                    "url": "https://mp.weixin.qq.com/s/abc",
+                    "title": "公众号选题",
+                    "nick_name": "公众号",
+                    "cover_url": "https://img.test/wx.jpg",
+                }
+            ],
+            raw_count=1,
+            cursor=cursor,
+        )
 
     def fake_detail(url, *, settings, rate_limiter):
         assert url == "https://mp.weixin.qq.com/s/abc"
         return "公众号正文", ["https://img.test/wx-1.jpg"]
 
-    monkeypatch.setattr(weixin_module, "search_weixin", fake_search)
+    monkeypatch.setattr(weixin_module, "search_weixin_page", fake_search)
     monkeypatch.setattr(weixin_module, "fetch_weixin_detail", fake_detail)
 
     adapter = WeixinAdapter()
@@ -85,25 +138,74 @@ def test_weixin_adapter_hashes_url_and_reads_detail(monkeypatch):
 
     assert candidate.platform == "weixin"
     assert len(candidate.source_id) == 16
+    assert candidate.raw["page_index"] == 1
     assert item.body_text == "公众号正文"
     assert item.image_urls == ["https://img.test/wx-1.jpg"]
+    assert item.content_mode == "article"
+
+
+def test_weixin_adapter_search_pages_uses_next_cursor(monkeypatch):
+    settings = SimpleNamespace()
+    cursors: list[str] = []
+
+    def fake_search(query, *, cursor, limit, settings, rate_limiter):
+        cursors.append(cursor)
+        if cursor == "0":
+            return SearchPage(
+                rows=[{"url": "https://mp.weixin.qq.com/s/a", "title": "第一页"}],
+                raw_count=1,
+                cursor=cursor,
+                next_cursor="1",
+                has_more=True,
+            )
+        return SearchPage(
+            rows=[{"url": "https://mp.weixin.qq.com/s/b", "title": "第二页"}],
+            raw_count=1,
+            cursor=cursor,
+            has_more=False,
+        )
+
+    monkeypatch.setattr(weixin_module, "search_weixin_page", fake_search)
+
+    pages = list(
+        WeixinAdapter().search_pages(
+            "公众号",
+            settings=settings,
+            limit=2,
+            rate_limiter=None,
+            max_pages=2,
+        )
+    )
+
+    assert cursors == ["0", "1"]
+    assert len(pages) == 2
+    assert pages[1].candidates[0].raw["page_index"] == 2
 
 
 def test_douyin_adapter_maps_video_search_and_detail(monkeypatch):
     settings = SimpleNamespace()
 
-    def fake_search(query, *, platform, content_type, limit, settings, rate_limiter):
+    def fake_search(query, *, content_type, limit, settings, rate_limiter):
         assert query == "短视频 讲法"
-        assert platform == "douyin"
         assert content_type == "视频"
         assert limit == 1
-        return ["dy-1"]
+        return [
+            SearchHit(
+                content_id="dy-1",
+                provider="piaoquantv",
+                raw={"aweme_id": "dy-1"},
+                request_payload={"keyword": query},
+            )
+        ]
 
-    def fake_fetch(source_id, *, settings, rate_limiter):
+    def fake_fetch(source_id, *, platform, provider, settings, rate_limiter):
         assert source_id == "dy-1"
+        assert platform == "douyin"
+        assert provider == "piaoquantv"
         return Post(
             id="dy_dy-1",
             platform="douyin",
+            provider=provider,
             url="https://douyin.test/1",
             content_id="dy-1",
             title="短视频讲法",
@@ -113,7 +215,7 @@ def test_douyin_adapter_maps_video_search_and_detail(monkeypatch):
             raw={"source": "mock"},
         )
 
-    monkeypatch.setattr(douyin_module, "search_keyword", fake_search)
+    monkeypatch.setattr(douyin_module, "search_douyin_hits", fake_search)
     monkeypatch.setattr(douyin_module, "fetch_post_detail", fake_fetch)
 
     adapter = DouyinAdapter()
@@ -121,5 +223,102 @@ def test_douyin_adapter_maps_video_search_and_detail(monkeypatch):
     item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)
 
     assert candidate.source_id == "dy-1"
+    assert candidate.provider == "piaoquantv"
+    assert candidate.raw["search_provider"] == "piaoquantv"
+    assert candidate.raw["original"] == {"aweme_id": "dy-1"}
     assert item.content_type == "video"
+    assert item.provider == "piaoquantv"
+    assert item.content_mode == "video_post"
     assert item.video_urls == ["https://video.test/1.mp4"]
+
+
+def test_douyin_adapter_search_pages_preserves_provider_and_page_metadata(monkeypatch):
+    settings = SimpleNamespace()
+    cursors_seen: list[dict[str, str]] = []
+
+    def fake_search_page(query, *, provider, cursors, content_type, limit, settings, rate_limiter):
+        cursors_seen.append(dict(cursors))
+        return SearchPage(
+            rows=[
+                SearchHit(
+                    content_id="dy-1",
+                    provider="aiddit",
+                    raw={"aweme_id": "dy-1"},
+                    request_payload={"keyword": query},
+                )
+            ],
+            raw_count=1,
+            cursor="",
+            provider="aiddit",
+            request_payload={"keyword": query},
+            raw_metadata={"cursors": {"aiddit": "20"}},
+        )
+
+    monkeypatch.setattr(douyin_module, "search_douyin_page", fake_search_page)
+
+    page = next(
+        iter(
+            DouyinAdapter().search_pages(
+                "短视频",
+                settings=settings,
+                limit=1,
+                rate_limiter=None,
+                max_pages=1,
+            )
+        )
+    )
+
+    assert cursors_seen == [{}]
+    assert page.provider == "aiddit"
+    assert page.candidates[0].provider == "aiddit"
+    assert page.candidates[0].raw["page_index"] == 1
+    assert page.candidates[0].raw["search_provider"] == "aiddit"
+
+
+def test_douyin_adapter_falls_back_to_image_post(monkeypatch):
+    settings = SimpleNamespace()
+
+    def fake_search(query, *, content_type, limit, settings, rate_limiter):
+        assert content_type == "视频"
+        return [SearchHit(content_id="dy-img-1", provider="aiddit")]
+
+    def fake_fetch(source_id, *, platform, provider, settings, rate_limiter):
+        assert provider == "aiddit"
+        return Post(
+            id="dy_dy-img-1",
+            platform="douyin",
+            provider=provider,
+            url="https://douyin.test/img/1",
+            content_id="dy-img-1",
+            title="图片内容",
+            content_type="image",
+            body_text="图片正文",
+            image_urls=["https://img.test/dy.jpg"],
+            raw={"source": "mock"},
+        )
+
+    monkeypatch.setattr(douyin_module, "search_douyin_hits", fake_search)
+    monkeypatch.setattr(douyin_module, "fetch_post_detail", fake_fetch)
+
+    item = DouyinAdapter().fetch_detail(
+        DouyinAdapter().search("图片", settings=settings, limit=1, rate_limiter=None)[0],
+        settings=settings,
+        rate_limiter=None,
+    )
+
+    assert item.content_mode == "image_post"
+    assert item.provider == "aiddit"
+    assert item.image_urls == ["https://img.test/dy.jpg"]
+
+
+def test_douyin_adapter_requires_provider_for_detail():
+    try:
+        DouyinAdapter().fetch_detail(
+            PlatformCandidate(rank=1, platform="douyin", source_id="dy-1"),
+            settings=SimpleNamespace(),
+            rate_limiter=None,
+        )
+    except RuntimeError as exc:
+        assert "missing search provider" in str(exc)
+    else:
+        raise AssertionError("missing provider should fail")

tests/test_search.py

@@ -9,7 +9,13 @@ import pytest
 from core.config import PgConfig, Settings
 from acquisition.crawler import RateLimiter
 from acquisition.search import (
-    SearchError, parse_search_response, parse_weixin, parse_xiaohongshu, search_keyword,
+    SearchError,
+    parse_search_response,
+    parse_weixin,
+    parse_xiaohongshu,
+    search_douyin_hits,
+    search_keyword,
+    search_xiaohongshu,
 )
 
 FIX = Path(__file__).parent / "fixtures"
@@ -44,6 +50,23 @@ class _FakeClient:
     def close(self): self.closed = True
 
 
+class _FakeClientWithUrls:
+    """按调用次序返回预置 payload；记录每次 post 的 url/body。"""
+    def __init__(self, pages): self.pages = list(pages); self.calls = []; self.closed = False
+    def post(self, url, json=None, headers=None, timeout=None):
+        self.calls.append({"url": url, "body": json})
+        return _Resp(self.pages.pop(0))
+    def close(self): self.closed = True
+
+
+class _RecordingLimiter:
+    def __init__(self):
+        self.buckets = []
+
+    def wait(self, bucket: str):
+        self.buckets.append(bucket)
+
+
 def test_parse_from_fixture():
     resp = json.loads((FIX / "xhs_search_分镜脚本.json").read_text("utf-8"))
     ids, has_more, cursor = parse_search_response(resp)
@@ -114,6 +137,124 @@ def test_douyin_body_has_account_id_and_params():
     assert body["cursor"] == "0"                       # 抖音起始 cursor
 
 
+def test_douyin_search_hits_success_does_not_request_aiddit():
+    page = {"code": 0, "data": {"has_more": False, "next_cursor": "",
+            "data": [{"aweme_id": "a1", "title": "ok"}]}}
+    client = _FakeClientWithUrls([page])
+    limiter = _RecordingLimiter()
+
+    out = search_douyin_hits(
+        "科普",
+        settings=_settings(),
+        http_client=client,
+        rate_limiter=limiter,
+        limit=1,
+    )
+
+    assert [hit.content_id for hit in out] == ["a1"]
+    assert out[0].provider == "piaoquantv"
+    assert out[0].raw["title"] == "ok"
+    assert len(client.calls) == 1
+    assert client.calls[0]["url"].startswith("http://crawapi.piaoquantv.com/")
+    assert limiter.buckets == ["douyin"]
+
+
+def test_douyin_search_hits_falls_back_to_aiddit_with_separate_body():
+    piao_error = {"code": 10000, "msg": "captcha", "data": None}
+    aiddit_page = {"code": 0, "data": {"has_more": False, "next_cursor": "",
+            "data": [{"aweme_id": "aid-1", "desc": "aiddit"}]}}
+    client = _FakeClientWithUrls([piao_error, aiddit_page])
+    limiter = _RecordingLimiter()
+
+    out = search_douyin_hits(
+        "厦门",
+        settings=_settings(),
+        http_client=client,
+        rate_limiter=limiter,
+        limit=1,
+    )
+
+    assert out[0].content_id == "aid-1"
+    assert out[0].provider == "aiddit"
+    assert len(client.calls) == 2
+    piao_body = client.calls[0]["body"]
+    aiddit_body = client.calls[1]["body"]
+    assert piao_body["account_id"] == "7450041106378522636"
+    assert piao_body["cookie_batch"] == "default"
+    assert piao_body["publish_time"] == "不限"
+    assert piao_body["cursor"] == "0"
+    assert aiddit_body["cursor"] == ""
+    assert aiddit_body["sort_type"] == "最多点赞"
+    assert "account_id" not in aiddit_body
+    assert "cookie_batch" not in aiddit_body
+    assert "publish_time" not in aiddit_body
+    assert client.calls[1]["url"].startswith("http://crawler.test/")
+    assert limiter.buckets == ["douyin", "douyin"]
+
+
+def test_douyin_search_hits_falls_back_when_no_usable_aweme_id():
+    piao_empty = {"code": 0, "data": {"has_more": False, "next_cursor": "",
+            "data": [{"id": "not-aweme"}]}}
+    aiddit_page = {"code": 0, "data": {"has_more": False, "next_cursor": "",
+            "data": [{"aweme_id": "aid-2"}]}}
+    client = _FakeClientWithUrls([piao_empty, aiddit_page])
+
+    out = search_douyin_hits(
+        "厦门",
+        settings=_settings(),
+        http_client=client,
+        rate_limiter=_RecordingLimiter(),
+        limit=1,
+    )
+
+    assert out[0].provider == "aiddit"
+    assert out[0].content_id == "aid-2"
+
+
+def test_douyin_search_hits_paginates_with_successful_provider_cursor():
+    page1 = {"code": 0, "data": {"has_more": True, "next_cursor": "10",
+             "data": [{"aweme_id": "a1"}]}}
+    page2 = {"code": 0, "data": {"has_more": False, "next_cursor": "",
+             "data": [{"aweme_id": "a2"}]}}
+    client = _FakeClientWithUrls([page1, page2])
+
+    out = search_douyin_hits(
+        "厦门",
+        settings=_settings(),
+        http_client=client,
+        rate_limiter=_RecordingLimiter(),
+        limit=2,
+    )
+
+    assert [hit.content_id for hit in out] == ["a1", "a2"]
+    assert client.calls[0]["body"]["cursor"] == "0"
+    assert client.calls[1]["body"]["cursor"] == "10"
+
+
+def test_douyin_search_hits_switches_provider_with_provider_own_cursor():
+    piao_page1 = {"code": 0, "data": {"has_more": True, "next_cursor": "10",
+                  "data": [{"aweme_id": "p1"}]}}
+    piao_page2_error = {"code": 10000, "msg": "captcha", "data": None}
+    aiddit_page1 = {"code": 0, "data": {"has_more": True, "next_cursor": "12",
+                   "data": [{"aweme_id": "a1"}]}}
+    client = _FakeClientWithUrls([piao_page1, piao_page2_error, aiddit_page1])
+
+    out = search_douyin_hits(
+        "厦门",
+        settings=_settings(),
+        http_client=client,
+        rate_limiter=_RecordingLimiter(),
+        limit=2,
+    )
+
+    assert [(hit.content_id, hit.provider) for hit in out] == [
+        ("p1", "piaoquantv"),
+        ("a1", "aiddit"),
+    ]
+    assert client.calls[1]["body"]["cursor"] == "10"
+    assert client.calls[2]["body"]["cursor"] == ""
+
+
 def test_parse_weixin_cleans_title_and_keeps_url_cover():
     resp = {"code": 0, "data": {"data": [
         {"title": "复旦教授<em class='highlight'>历史科普</em>", "url": "http://mp.weixin.qq.com/s?x=1",
@@ -147,6 +288,28 @@ def test_parse_xiaohongshu_takes_cover_title_from_note_card():
     assert out[0]["url"] == "https://www.xiaohongshu.com/explore/abc123"
 
 
+def test_xiaohongshu_search_omits_content_type_by_default():
+    page = {"code": 0, "data": {"data": [
+        {"id": "abc", "note_card": {
+            "type": "video",
+            "display_title": "混合结果",
+            "image_list": [{"image_url": "https://ci.test/c.jpg"}],
+        }}
+    ]}}
+    client = _FakeClient([page])
+
+    out = search_xiaohongshu(
+        "符号 视频 灵感 怎么做",
+        settings=_settings(),
+        http_client=client,
+        rate_limiter=_no_wait(),
+        limit=1,
+    )
+
+    assert out[0]["id"] == "abc"
+    assert "content_type" not in client.calls[0]
+
+
 def test_parse_xiaohongshu_title_falls_back_to_desc():
     resp = {"code": 0, "data": {"data": [
         {"id": "x1", "note_card": {"display_title": "", "desc": "那些赤裸裸的社会真相！\n第二行",