"""Formal acquisition runner backed by the repository boundary.""" from __future__ import annotations from dataclasses import dataclass 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.4 PAGINATION_MAX_PAGES = 2 LOW_RESULT_RETRY_THRESHOLD = 5 LOW_RESULT_MAX_RETRIES = 1 AdapterFactory = Callable[[str], PlatformAdapter] Classifier = Callable[..., ClassificationResult] MediaStabilizer = Callable[..., list[StabilizedMedia]] RateLimiterFactory = Callable[[str], Any] @dataclass(frozen=True) class RunBatchResult: run_id: UUID total_jobs: int done: int partial: int failed: int 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.""" def __init__(self, platform: str, delegate: RateLimiter | None = None) -> None: self.platform = platform self.delegate = delegate or RateLimiter( min_interval_seconds=10.0, max_interval_seconds=12.0, ) def wait(self, bucket: str) -> None: self.delegate.wait(self.platform) def _query_id(query: Query) -> UUID: if query.id is None: raise RuntimeError("formal acquisition queries must have id before running") return query.id def _job_id(job: AcquisitionJob) -> UUID: if job.id is None: raise RuntimeError("formal acquisition jobs must have id before running") return job.id def _best_video_url(media: list[StabilizedMedia]) -> str: for row in media: if row.media_type == "video" and row.cdn_url: return row.cdn_url return "" def _image_urls(media: list[StabilizedMedia]) -> list[str]: return [row.cdn_url for row in media if row.media_type == "image" and row.cdn_url] def _source_payload(candidate: Any, detail: Any) -> dict[str, Any]: return { "candidate": candidate.model_dump() if hasattr(candidate, "model_dump") else {}, "detail": detail.raw if isinstance(getattr(detail, "raw", None), dict) else {}, } 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, *, job: AcquisitionJob, query: Query, platform: str, candidate: Any, detail: Any, settings: Settings, media_stabilizer: MediaStabilizer, classifier: Classifier, classify: bool, attempt_index: int = 1, ) -> 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, ) metadata = { "candidate_rank": candidate.rank, "acquisition_match_status": "created", "content_mode": content_mode, "retry_attempt_index": attempt_index, **_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=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, 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, media_type=row.media_type, source_url=row.source_url, oss_url=row.cdn_url, cdn_url=row.cdn_url, position=row.position, status=row.status, source_payload={}, metadata={}, ) if classify: result = classifier( platform=platform, content_mode=content_mode, title=detail.title, body_text=detail.body_text, image_urls=_image_urls(media_rows), video_url=_best_video_url(media_rows), settings=settings, ) repo.add_item_classification( item_id=item.id, is_creation_knowledge=result.is_creation_knowledge, label=result.label, confidence=result.confidence, reason=result.reason, model_name=settings.video_model, prompt_version=result.prompt_version, result_payload=result.result_payload, 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 _candidate_dedupe_key(platform: str, candidate: PlatformCandidate) -> str | None: unique_key = _candidate_unique_key(platform, candidate) if unique_key: return f"unique:{unique_key}" if candidate.source_id: return f"source:{platform}:{candidate.source_id}" if candidate.url: return f"url:{platform}:{candidate.url}" return None def _attach_existing_candidate( repo: AcquisitionRepository, *, job: AcquisitionJob, query: Query, candidate: PlatformCandidate, unique_key: str, attempt_index: int = 1, ) -> 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, "retry_attempt_index": attempt_index, **_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, *, batch_id: UUID, settings: Settings, platforms: tuple[str, ...] | list[str] = DEFAULT_PLATFORMS, search_limit: int = 10, display_limit: int = 5, classify: bool = True, resume: bool = True, skip_done: bool = True, run_key: str | None = None, adapter_factory: AdapterFactory = get_platform_adapter, 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, low_result_retry_threshold: int = LOW_RESULT_RETRY_THRESHOLD, low_result_max_retries: int = LOW_RESULT_MAX_RETRIES, ) -> RunBatchResult: """Run query x platform acquisition and write formal cloud-state rows.""" queries = repo.list_queries_for_batch(batch_id, keep=True) run = repo.create_acquisition_run( batch_id=batch_id, run_key=run_key or f"acquisition:{batch_id}", status="running", metadata={ "platforms": list(platforms), "search_limit": search_limit, "display_limit": display_limit, "classify": classify, "resume": resume, "pagination": _pagination_config( pagination_creation_threshold, pagination_max_pages, ), "low_result_retry": { "threshold": low_result_retry_threshold, "max_retries": low_result_max_retries, }, }, ) if run.id is None: raise RuntimeError("repository returned acquisition run without id") total_jobs = len(queries) * len(platforms) done = partial = failed = skipped = 0 gates: dict[str, Any] = {} for query in queries: query_id = _query_id(query) for platform in platforms: job = repo.ensure_acquisition_job( run_id=run.id, query_id=query_id, platform=platform, search_limit=search_limit, display_limit=display_limit, status="pending", metadata={"query_text": query.query_text}, ) if skip_done and job.status == "done": skipped += 1 continue attempts = job.attempt_count + 1 job = repo.update_acquisition_job( _job_id(job), status="running", attempt_count=attempts, error_message=None, ) 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) if gate is None: gate = ( rate_limiter_factory(platform) if rate_limiter_factory else PlatformRateLimiter(platform) ) gates[platform] = gate retry_enabled = ( low_result_max_retries > 0 and low_result_retry_threshold >= 0 and search_limit > low_result_retry_threshold ) max_attempts = 1 + (low_result_max_retries if retry_enabled else 0) seen_candidate_keys: set[str] = set() retry_attempts: list[dict[str, Any]] = [] retry_triggered = False for attempt_index in range(1, max_attempts + 1): attempt_display_start = display_count attempt_searched_start = searched_count attempt_raw_start = raw_searched_count attempt_pages_start = search_page_count attempt_page_summaries: list[dict[str, Any]] = [] attempt_first_page_classified_count = 0 attempt_first_page_creation_count = 0 attempt_first_page_creation_ratio: float | None = None attempt_requested_second_page = False attempt_stop_reason = "" pages = _search_pages( adapter, query.query_text, settings=settings, limit=search_limit, rate_limiter=gate, max_pages=pagination_max_pages, ) for page in pages: search_page_count += 1 raw_searched_count += page.raw_count page_classified_count = 0 page_creation_count = 0 unique_candidate_count = 0 for candidate in page.candidates[:search_limit]: dedupe_key = _candidate_dedupe_key(platform, candidate) if dedupe_key and dedupe_key in seen_candidate_keys: continue if dedupe_key: seen_candidate_keys.add(dedupe_key) unique_candidate_count += 1 searched_count += 1 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, attempt_index=attempt_index, ): 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, attempt_index=attempt_index, ) 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_summary = { "attempt_index": attempt_index, "page_index": page.page_index, "raw_count": page.raw_count, "candidate_count": len(page.candidates), "unique_candidate_count": unique_candidate_count, "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, } page_summaries.append(page_summary) attempt_page_summaries.append(page_summary) if page.page_index == 1: attempt_first_page_classified_count = page_classified_count attempt_first_page_creation_count = page_creation_count attempt_first_page_creation_ratio = page_ratio if pagination_max_pages <= 1: attempt_stop_reason = "max_pages_reached" break if not page.has_more or not page.next_cursor: attempt_stop_reason = "no_more_pages" break if page_classified_count <= 0: attempt_stop_reason = "no_classified_candidates" break if page_ratio is None or page_ratio < pagination_creation_threshold: attempt_stop_reason = "below_threshold" break attempt_requested_second_page = True continue attempt_stop_reason = "max_pages_reached" break if not attempt_stop_reason: attempt_stop_reason = "pages_exhausted" first_page_classified_count = attempt_first_page_classified_count first_page_creation_count = attempt_first_page_creation_count first_page_creation_ratio = attempt_first_page_creation_ratio requested_second_page = attempt_requested_second_page pagination_stop_reason = attempt_stop_reason attempt_display_count = display_count - attempt_display_start retry_attempts.append( { "attempt_index": attempt_index, "display_count": attempt_display_count, "searched_count": searched_count - attempt_searched_start, "raw_searched_count": raw_searched_count - attempt_raw_start, "search_page_count": search_page_count - attempt_pages_start, "stop_reason": attempt_stop_reason, "requested_second_page": attempt_requested_second_page, "first_page_classified_count": attempt_first_page_classified_count, "first_page_creation_count": attempt_first_page_creation_count, "first_page_creation_ratio": attempt_first_page_creation_ratio, "pages": attempt_page_summaries, } ) if ( attempt_index <= low_result_max_retries and display_count <= low_result_retry_threshold ): retry_triggered = True continue break status = "done" if display_count else "failed" if status == "done": done += 1 else: failed += 1 repo.update_acquisition_job( _job_id(job), status=status, attempt_count=attempts, error_message=None if status != "failed" else "; ".join(errors[-3:]), 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, }, "low_result_retry": { "threshold": low_result_retry_threshold, "max_retries": low_result_max_retries if retry_enabled else 0, "triggered": retry_triggered, "attempt_count": len(retry_attempts), "attempts": retry_attempts, }, "errors": errors[-3:], }, ) except Exception as exc: failed += 1 repo.update_acquisition_job( _job_id(job), status="failed", attempt_count=attempts, 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:], }, ) run_status = "done" if done > 0 else "failed" update_run = getattr(repo, "update_acquisition_run", None) if update_run: update_run( run.id, status=run_status, metadata={ "total_jobs": total_jobs, "done": done, "partial": partial, "failed": failed, "skipped": skipped, "pagination": _pagination_config( pagination_creation_threshold, pagination_max_pages, ), }, ) return RunBatchResult( run_id=run.id, total_jobs=total_jobs, done=done, partial=partial, failed=failed, skipped=skipped, )