Create-knowledge-find-decode/tests/test_acquisition_runner.py

from __future__ import annotations

from uuid import UUID, uuid4

from acquisition.classification.coarse import ClassificationResult
from acquisition.domain import (
    AcquisitionJob,
    AcquisitionRun,
    CandidateItem,
    ItemClassification,
    MediaAsset,
    Query,
    QueryBatch,
)
from acquisition.media.service import StabilizedMedia
from acquisition.platforms.base import PlatformCandidate, PlatformItem, PlatformSearchPage
from acquisition.runner import run_batch
from core.config import PgConfig, Settings


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="http://openrouter.test",
        openrouter_api_key="k",
        llm_model="m",
        max_cards=12,
        frames_dir="f",
        douyin_ratio="540p",
        data_dir="data",
    )


class FakeRepo:
    def __init__(self, *, job_status: str = "pending") -> None:
        self.batch_id = uuid4()
        self.query = Query(
            id=uuid4(),
            batch_id=self.batch_id,
            query_text="脚本 开头 怎么做",
            keep=True,
            status="ready",
        )
        self.job_status = job_status
        self.runs: list[AcquisitionRun] = []
        self.jobs: list[AcquisitionJob] = []
        self.updated_jobs: list[AcquisitionJob] = []
        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)

    def add_query(self, **kwargs):
        return Query(id=uuid4(), **kwargs)

    def list_queries_for_batch(self, batch_id: UUID, *, keep: bool | None = None):
        assert batch_id == self.batch_id
        assert keep is True
        return [self.query]

    def create_acquisition_run(self, **kwargs):
        run = AcquisitionRun(id=uuid4(), **kwargs)
        self.runs.append(run)
        return run

    def ensure_acquisition_job(self, **kwargs):
        kwargs["status"] = self.job_status
        job = AcquisitionJob(id=uuid4(), **kwargs)
        self.jobs.append(job)
        return job

    def update_acquisition_job(self, job_id: UUID, **kwargs):
        original = next(job for job in self.jobs if job.id == job_id)
        updated = original.model_copy(update=kwargs)
        self.updated_jobs.append(updated)
        return updated

    def upsert_candidate_item(self, **kwargs):
        item = CandidateItem(id=uuid4(), **kwargs)
        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)
        return row

    def add_item_classification(self, **kwargs):
        row = ItemClassification(id=uuid4(), **kwargs)
        self.classifications.append(row)
        return row

    def get_run_summary(self, run_id: UUID):
        return {}

    def get_query_detail(self, *, run_id: UUID, query_id: UUID):
        return {}


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
        self.detail_calls = 0

    def search(self, query, *, settings, limit, rate_limiter):
        self.search_calls += 1
        assert query == "脚本 开头 怎么做"
        assert limit == 2
        return [
            PlatformCandidate(
                rank=1,
                platform=self.platform,
                source_id="wx1",
                url=self.url,
                title="公众号脚本",
                author="作者",
                cover_url="https://img.test/a.jpg",
            )
        ]

    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="图文",
            title="公众号脚本",
            author="作者",
            body_text="先定受众再写开头",
            image_urls=["https://img.test/a.jpg"],
            raw={"ok": True},
        )


class PartialAdapter(FakeAdapter):
    def search(self, query, *, settings, limit, rate_limiter):
        self.search_calls += 1
        return [
            PlatformCandidate(rank=1, platform=self.platform, source_id="bad", url="https://bad"),
            PlatformCandidate(rank=2, platform=self.platform, source_id="good", url="https://good"),
        ]

    def fetch_detail(self, candidate, *, settings, rate_limiter):
        self.detail_calls += 1
        if candidate.source_id == "bad":
            raise RuntimeError("detail boom")
        return PlatformItem(
            platform=self.platform,
            source_id=candidate.source_id,
            url=candidate.url,
            content_type="图文",
            title="可用详情",
            author="作者",
            body_text="先定受众",
            image_urls=[],
            raw={},
        )


class FailingSearchAdapter(FakeAdapter):
    def search(self, query, *, settings, limit, rate_limiter):
        self.search_calls += 1
        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


class RetryPagedAdapter:
    platform = "douyin"

    def __init__(
        self,
        page_batches: list[list[PlatformSearchPage]],
        details: dict[str, PlatformItem | Exception],
    ) -> None:
        self.page_batches = page_batches
        self.details = details
        self.search_calls = 0
        self.pages_requested = 0
        self.detail_calls = 0
        self.limiter_ids: list[int] = []

    def search_pages(self, query, *, settings, limit, rate_limiter, max_pages=2):
        self.limiter_ids.append(id(rate_limiter))
        batch_index = min(self.search_calls, len(self.page_batches) - 1)
        self.search_calls += 1
        for page in self.page_batches[batch_index][:max_pages]:
            self.pages_requested += 1
            yield page

    def fetch_detail(self, candidate, *, settings, rate_limiter):
        self.limiter_ids.append(id(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()

    def media_stabilizer(**kwargs):
        assert kwargs["image_urls"] == ["https://img.test/a.jpg"]
        return [
            StabilizedMedia(
                media_type="image",
                source_url="https://img.test/a.jpg",
                cdn_url="https://cdn.test/a.jpg",
                position=1,
            )
        ]

    def classifier(**kwargs):
        assert kwargs["content_mode"] == "article"
        assert kwargs["image_urls"] == ["https://cdn.test/a.jpg"]
        return ClassificationResult(
            is_creation_knowledge=True,
            label="creation",
            confidence=1.0,
            reason="是创作知识",
            knowledge="先定受众",
            prompt_version="pv1",
            result_payload={"knowledge": "先定受众"},
        )

    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=media_stabilizer,
        classifier=classifier,
        rate_limiter_factory=lambda platform: object(),
    )

    assert result.done == 1
    assert result.failed == 0
    assert result.total_jobs == 1
    assert adapter.search_calls == 1
    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_skips_douyin_video_when_oss_transfer_fails():
    repo = FakeRepo()
    adapter = DouyinProviderAdapter()

    def media_stabilizer(**kwargs):
        assert kwargs["video_fallback_to_source"] is False
        assert kwargs["video_retry_delays_seconds"] == (5.0, 10.0)
        return [
            StabilizedMedia(
                media_type="video",
                source_url="https://video.test/761.mp4",
                cdn_url="",
                position=1,
                status="failed",
                error_message="oss timeout",
            )
        ]

    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=media_stabilizer,
        classifier=lambda **kwargs: (_ for _ in ()).throw(AssertionError("classifier should be skipped")),
        rate_limiter_factory=lambda platform: object(),
    )

    assert result.done == 1
    assert repo.media[0].status == "failed"
    assert repo.media[0].metadata["error_message"] == "oss timeout"
    assert repo.classifications[0].status == "skipped"
    assert repo.classifications[0].label == "video_oss_failed"
    assert repo.classifications[0].result_payload["media_errors"][0]["error_message"] == "oss timeout"


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(),
        low_result_max_retries=0,
    )

    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["creation_threshold"] == 0.4
    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(),
        low_result_max_retries=0,
    )

    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(),
        low_result_max_retries=0,
    )

    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(),
        low_result_max_retries=0,
    )

    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,
        low_result_max_retries=0,
    )

    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()

    result = run_batch(
        repo,
        batch_id=repo.batch_id,
        settings=_settings(),
        platforms=("weixin",),
        adapter_factory=lambda platform: adapter,
    )

    assert result.skipped == 1
    assert adapter.search_calls == 0
    assert repo.updated_jobs == []


def test_run_batch_marks_done_when_some_details_fail_but_an_item_displays():
    repo = FakeRepo()
    adapter = PartialAdapter()

    result = run_batch(
        repo,
        batch_id=repo.batch_id,
        settings=_settings(),
        platforms=("weixin",),
        search_limit=2,
        display_limit=2,
        adapter_factory=lambda platform: adapter,
        media_stabilizer=lambda **kwargs: [],
        classifier=lambda **kwargs: ClassificationResult(
            is_creation_knowledge=True,
            label="creation",
            confidence=1.0,
            reason="ok",
        ),
        rate_limiter_factory=lambda platform: object(),
    )

    assert result.partial == 0
    assert result.done == 1
    assert result.failed == 0
    assert repo.updated_jobs[-1].status == "done"
    assert repo.updated_jobs[-1].metadata["display_count"] == 1
    assert repo.updated_jobs[-1].metadata["errors"] == ["detail boom"]


def test_run_batch_retries_low_result_platform_once_and_merges_unique_results():
    repo = FakeRepo()
    adapter = RetryPagedAdapter(
        page_batches=[
            [_paged_search_page(1, ["a", "b", "dup"])],
            [_paged_search_page(1, ["dup", "c", "d", "e", "f", "g"])],
        ],
        details={source_id: _detail(source_id) for source_id in ["a", "b", "dup", "c", "d", "e", "f", "g"]},
    )
    limiter = object()

    result = run_batch(
        repo,
        batch_id=repo.batch_id,
        settings=_settings(),
        platforms=("douyin",),
        search_limit=10,
        display_limit=5,
        adapter_factory=lambda platform: adapter,
        media_stabilizer=lambda **kwargs: [],
        classifier=_classification_by_title(set()),
        rate_limiter_factory=lambda platform: limiter,
    )

    metadata = repo.updated_jobs[-1].metadata
    assert result.done == 1
    assert result.partial == 0
    assert repo.updated_jobs[-1].status == "done"
    assert adapter.search_calls == 2
    assert adapter.detail_calls == 8
    assert len(repo.items) == 8
    assert metadata["display_count"] == 8
    assert metadata["searched_count"] == 8
    assert metadata["low_result_retry"]["triggered"] is True
    assert metadata["low_result_retry"]["attempt_count"] == 2
    assert metadata["low_result_retry"]["attempts"][0]["display_count"] == 3
    assert metadata["low_result_retry"]["attempts"][1]["display_count"] == 5
    assert {item.platform_item_id for item in repo.items} == {"a", "b", "dup", "c", "d", "e", "f", "g"}
    assert set(adapter.limiter_ids) == {id(limiter)}


def test_run_batch_does_not_retry_when_first_attempt_is_above_low_result_threshold():
    repo = FakeRepo()
    adapter = RetryPagedAdapter(
        page_batches=[
            [_paged_search_page(1, ["a", "b", "c", "d", "e", "f"])],
            [_paged_search_page(1, ["g"])],
        ],
        details={source_id: _detail(source_id) for source_id in ["a", "b", "c", "d", "e", "f", "g"]},
    )

    run_batch(
        repo,
        batch_id=repo.batch_id,
        settings=_settings(),
        platforms=("douyin",),
        search_limit=10,
        display_limit=5,
        adapter_factory=lambda platform: adapter,
        media_stabilizer=lambda **kwargs: [],
        classifier=_classification_by_title(set()),
        rate_limiter_factory=lambda platform: object(),
    )

    metadata = repo.updated_jobs[-1].metadata
    assert adapter.search_calls == 1
    assert metadata["display_count"] == 6
    assert metadata["low_result_retry"]["triggered"] is False


def test_run_batch_marks_failed_when_search_fails_before_any_item():
    repo = FakeRepo()
    adapter = FailingSearchAdapter()

    result = run_batch(
        repo,
        batch_id=repo.batch_id,
        settings=_settings(),
        platforms=("weixin",),
        adapter_factory=lambda platform: adapter,
        rate_limiter_factory=lambda platform: object(),
    )

    assert result.failed == 1
    assert result.done == 0
    assert repo.updated_jobs[-1].status == "failed"
    assert "search boom" in (repo.updated_jobs[-1].error_message or "")


def test_formal_runner_does_not_import_legacy_store():
    source = __import__("pathlib").Path("acquisition/runner.py").read_text(encoding="utf-8")
    assert "acquisition.store" not in source
    assert "creation_demo.json" not in source