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 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_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() 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_partial_when_some_details_fail(): 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 == 1 assert result.done == 0 assert result.failed == 0 assert repo.updated_jobs[-1].status == "partial" assert repo.updated_jobs[-1].metadata["display_count"] == 1 assert repo.updated_jobs[-1].metadata["errors"] == ["detail boom"] 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