from __future__ import annotations import time import random from copy import deepcopy from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Callable from content_agent.business_modules import content_discovery, platform_access, rule_judgment from content_agent.business_modules.content_discovery import pattern_recall from content_agent.business_modules.content_discovery.fifty_plus import fifty_plus_score from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION from content_agent.errors import ContentAgentError, ErrorCode from content_agent.interfaces import GeminiVideoClient, PlatformSearchClient, RuntimeFileStore from content_agent.record_payload import with_raw_payload INITIAL_BATCH_SIZE = 3 MAX_PAGES = 3 SEARCH_MIN_INTERVAL_SECONDS = 30.0 DOUYIN_SEARCH_MIN_INTERVAL_SECONDS = 10.0 DOUYIN_SEARCH_MAX_INTERVAL_SECONDS = 12.0 SHORT_VIDEO_SEARCH_MIN_INTERVAL_SECONDS = 10.0 SHORT_VIDEO_SEARCH_MAX_INTERVAL_SECONDS = 12.0 PASS_ACTION = "ADD_TO_CONTENT_POOL" # 抖音拉作者画像(适老性)的门:只看 query 相关性,不看平台分,保证适老性始终进三项加权。 # 值=抖音入池 query 下限(score_thresholds.douyin.pool_query=65);低于此靠 query 就被否,不拉省成本。 PORTRAIT_QUERY_GATE = 65 # 画像拉取重试次数(总尝试次数):救热点宝瞬时失败/限速。 PORTRAIT_FETCH_ATTEMPTS = 2 class SearchCallLimiter: def __init__( self, min_interval_seconds: float = SEARCH_MIN_INTERVAL_SECONDS, max_interval_seconds: float | None = None, *, now_fn: Callable[[], float] = time.monotonic, sleep_fn: Callable[[float], None] = time.sleep, random_fn: Callable[[float, float], float] = random.uniform, ) -> None: self.min_interval_seconds = min_interval_seconds self.max_interval_seconds = max_interval_seconds self.now_fn = now_fn self.sleep_fn = sleep_fn self.random_fn = random_fn self._last_call_at: float | None = None def wait(self) -> float: waited_seconds = 0.0 if self._last_call_at is not None: remaining = self._next_interval_seconds() - (self.now_fn() - self._last_call_at) if remaining > 0: self.sleep_fn(remaining) waited_seconds = remaining self._last_call_at = self.now_fn() return waited_seconds def _next_interval_seconds(self) -> float: if self.max_interval_seconds is None or self.max_interval_seconds <= self.min_interval_seconds: return self.min_interval_seconds return self.random_fn(self.min_interval_seconds, self.max_interval_seconds) def make_search_limiter(platform: str = "") -> SearchCallLimiter: if platform == "douyin": return SearchCallLimiter( min_interval_seconds=DOUYIN_SEARCH_MIN_INTERVAL_SECONDS, max_interval_seconds=DOUYIN_SEARCH_MAX_INTERVAL_SECONDS, ) if platform in {"kuaishou", "shipinhao"}: return SearchCallLimiter( min_interval_seconds=SHORT_VIDEO_SEARCH_MIN_INTERVAL_SECONDS, max_interval_seconds=SHORT_VIDEO_SEARCH_MAX_INTERVAL_SECONDS, ) return SearchCallLimiter() @dataclass(frozen=True) class _FetchedPage: results: list[dict[str, Any]] search_duration_ms: int search_wait_ms: int search_request_duration_ms: int def run( *, run_id: str, policy_run_id: str, search_queries: list[dict[str, Any]], source_context: dict[str, Any], policy_bundle: dict[str, Any], platform_client: PlatformSearchClient, runtime: RuntimeFileStore, gemini_video_client: GeminiVideoClient, limiter: SearchCallLimiter | None = None, portrait_limiter: SearchCallLimiter | None = None, archive_dispatcher: Any | None = None, platform: str = "", ) -> dict[str, list[dict[str, Any]]]: if limiter is None and getattr(platform_client, "requires_progressive_search_rate_limit", False): limiter = make_search_limiter(platform) context = _ProgressiveContext( run_id=run_id, policy_run_id=policy_run_id, source_context=source_context, policy_bundle=policy_bundle, platform_client=platform_client, runtime=runtime, gemini_video_client=gemini_video_client, limiter=limiter, portrait_limiter=portrait_limiter, archive_dispatcher=archive_dispatcher, platform=platform, ) try: for query in search_queries: context.process_query(query) if search_queries and context.query_failures and not context.platform_results: raise ContentAgentError( ErrorCode.PLATFORM_REQUEST_FAILED, "all platform queries failed", {"query_failures": context.query_failures}, ) context.write_runtime() return { "platform_results": context.platform_results, "query_failures": context.query_failures, "discovered_content_items": context.discovered_content_items, "content_media_records": context.content_media_records, "pattern_recall_evidence": context.pattern_recall_evidence, "evidence_bundles": context.evidence_bundles, "rule_decisions": context.rule_decisions, } finally: context.close_archive_dispatcher() class _ProgressiveContext: def __init__( self, *, run_id: str, policy_run_id: str, source_context: dict[str, Any], policy_bundle: dict[str, Any], platform_client: PlatformSearchClient, runtime: RuntimeFileStore, gemini_video_client: GeminiVideoClient, archive_dispatcher: Any | None, limiter: SearchCallLimiter | None, portrait_limiter: SearchCallLimiter | None = None, external_seen_content_ids: set[str] | None = None, recall_index_base: int = 0, decision_index_base: int = 0, platform: str = "", portrait_client: Any | None = None, ) -> None: self.run_id = run_id self.policy_run_id = policy_run_id self.source_context = source_context self.policy_bundle = policy_bundle self.platform_client = platform_client self.runtime = runtime self.gemini_video_client = gemini_video_client self.limiter = limiter self.portrait_limiter = portrait_limiter self.archive_dispatcher = archive_dispatcher self._archive_dispatcher_closed = False self.platform_results: list[dict[str, Any]] = [] self.query_failures: list[dict[str, Any]] = [] self.discovered_content_items: list[dict[str, Any]] = [] self.content_media_records: list[dict[str, Any]] = [] self.pattern_recall_evidence: list[dict[str, Any]] = [] self.evidence_bundles: list[dict[str, Any]] = [] self.rule_decisions: list[dict[str, Any]] = [] self._platform_result_by_key: dict[tuple[str, str], dict[str, Any]] = {} self._record_indexes_by_key: dict[tuple[str, str], dict[str, int]] = {} self._queries_with_consumed_next_pages: set[str] = set() # M8C:游走复用统一搜索单元时注入的全 run 级状态。 # external_seen_content_ids:与 _FrontierContext 共享的同一 set,承载首轮 + 跨层 # 已发现 content_id;为 None 时(首轮)行为完全等价。 # *_index_base:游走批次的 id 起算基线(=首轮计数),避免 recall/decision id 撞首轮。 self._external_seen_content_ids = external_seen_content_ids self._recall_index_base = recall_index_base self._decision_index_base = decision_index_base # M9A:抖音 50+ 受众子分。platform!="douyin" 时不注入 content_audience_50plus 块, # evaluator 即走旧 50/50(非抖音零改动)。portrait_client 默认复用 platform_client。 self.platform = platform self.portrait_client = portrait_client or platform_client # M9 修复:作者画像按作者缓存(画像是作者级,同作者结果一致)。避免对首轮大量视频 # 逐条串行拉画像 → 同作者只拉一次,把调用量从"视频数"降到"作者数"。None=该作者拉取失败。 self._portrait_by_author: dict[str, dict[str, Any] | None] = {} self._portrait_error_by_author: dict[str, str | None] = {} self._batch_event_index = 0 def process_query(self, query: dict[str, Any]) -> dict[str, list[dict[str, Any]]]: """统一搜索单元(前3命中→剩余→翻页≤3页)。 M8C:返回本次搜索新增的 discovered_content_items / rule_decisions 切片, 供游走 frontier 取回再判 allow_walk。首轮 run() 忽略返回值,行为等价。 """ start_discovered = len(self.discovered_content_items) start_decisions = len(self.rule_decisions) self._process_query_pages(query) return { "discovered_content_items": self.discovered_content_items[start_discovered:], "rule_decisions": self.rule_decisions[start_decisions:], } def process_prefetched_batch( self, query: dict[str, Any], results: list[dict[str, Any]], *, search_duration_ms: int = 0, search_wait_ms: int = 0, search_request_duration_ms: int = 0, search_timing_source: str = "prefetched_author_works", ) -> dict[str, list[dict[str, Any]]]: """处理已抓取的一批结果(作者作品:列表抓取+截断,不分页、不限速)。 复用 _process_batch 的 discovery→recall→decision+去重+index 路径, 返回本批新增切片。去重/index 与统一搜索单元共享同一 context 状态。 """ start_discovered = len(self.discovered_content_items) start_decisions = len(self.rule_decisions) if results: self._process_batch( query, results, page_number=1, page_item_count=len(results), batch_kind="author_works", rank_offset=0, search_duration_ms=search_duration_ms, search_wait_ms=search_wait_ms, search_request_duration_ms=search_request_duration_ms, search_timing_source=search_timing_source, ) return { "discovered_content_items": self.discovered_content_items[start_discovered:], "rule_decisions": self.rule_decisions[start_decisions:], } def _process_query_pages(self, query: dict[str, Any]) -> None: first_fetch = self._fetch_page(query, page_number=1, cursor=query.get("page_cursor")) if first_fetch is None: return first_page = first_fetch.results page_count = len(first_page) if page_count == 0: return first_batch = first_page[:INITIAL_BATCH_SIZE] remainder_batch = first_page[INITIAL_BATCH_SIZE:] passed_first = self._process_batch( query, first_batch, page_number=1, page_item_count=page_count, batch_kind="initial_top3", rank_offset=0, search_duration_ms=first_fetch.search_duration_ms, search_wait_ms=first_fetch.search_wait_ms, search_request_duration_ms=first_fetch.search_request_duration_ms, search_timing_source="page_fetch", ) if not passed_first or not remainder_batch: return passed_remainder = self._process_batch( query, remainder_batch, page_number=1, page_item_count=page_count, batch_kind="page_remainder", rank_offset=INITIAL_BATCH_SIZE, search_duration_ms=0, search_wait_ms=0, search_request_duration_ms=0, search_timing_source="same_page_remainder", ) if not passed_remainder: return cursor = _page_next_cursor(first_page) has_more = _page_has_more(first_page) page_number = 2 while page_number <= MAX_PAGES and has_more and cursor: fetched_page = self._fetch_page(query, page_number=page_number, cursor=cursor) if fetched_page is None or not fetched_page.results: return page = fetched_page.results self._queries_with_consumed_next_pages.add(str(query["search_query_id"])) page_count = len(page) passed_page = self._process_batch( query, page, page_number=page_number, page_item_count=page_count, batch_kind=f"page_{page_number:02d}", rank_offset=0, search_duration_ms=fetched_page.search_duration_ms, search_wait_ms=fetched_page.search_wait_ms, search_request_duration_ms=fetched_page.search_request_duration_ms, search_timing_source="page_fetch", ) if page_number >= MAX_PAGES or not passed_page: return cursor = _page_next_cursor(page) has_more = _page_has_more(page) page_number += 1 def _fetch_page( self, query: dict[str, Any], *, page_number: int, cursor: Any, ) -> _FetchedPage | None: page_query = dict(query) if page_number > 1: page_query["page_cursor"] = str(cursor or "") try: search_started_at = time.monotonic() waited_seconds = self.limiter.wait() if self.limiter is not None else 0.0 request_started_at = time.monotonic() search = getattr(self.platform_client, "search_full_page_metadata", None) if not callable(search): search = getattr(self.platform_client, "search_full_page", None) if not callable(search): search = self.platform_client.search results = list(search(page_query)) request_ended_at = time.monotonic() return _FetchedPage( results=results, search_duration_ms=_elapsed_ms(search_started_at, request_ended_at), search_wait_ms=int(waited_seconds * 1000), search_request_duration_ms=_elapsed_ms(request_started_at, request_ended_at), ) except Exception as exc: failure_query = page_query if page_number == 1 else { **page_query, "search_query_id": f"{query['search_query_id']}_progressive_page_{page_number:03d}", } failure = platform_access._query_failure(failure_query, exc) # noqa: SLF001 failure["progressive_page_number"] = page_number self.query_failures.append(failure) return None def _process_batch( self, query: dict[str, Any], page_results: list[dict[str, Any]], *, page_number: int, page_item_count: int, batch_kind: str, rank_offset: int, search_duration_ms: int = 0, search_wait_ms: int = 0, search_request_duration_ms: int = 0, search_timing_source: str = "unknown", ) -> bool: batch_started_at = time.monotonic() batch = [] for rank, result in enumerate(page_results, start=1): key = _content_key(result) if key and key in self._platform_result_by_key: existing = self._platform_result_by_key[key] platform_access._append_query_source(existing, query) # noqa: SLF001 self._propagate_query_source_merge(key, existing) continue # M8C:游走跨层 / 首轮去重——已发现的 content_id 静默跳过(不二次判定、不撞 DB 唯一索引)。 if ( key and self._external_seen_content_ids is not None and key[1] in self._external_seen_content_ids ): continue result = self._hydrate_result_media(result, query) prepared = self._prepare_result( query, result, page_number=page_number, page_item_count=page_item_count, batch_kind=batch_kind, rank=rank_offset + rank, ) key = _content_key(prepared) if not key: batch.append(prepared) continue self._platform_result_by_key[key] = prepared if self._external_seen_content_ids is not None: self._external_seen_content_ids.add(key[1]) batch.append(prepared) if not batch: return False discovery_started_at = time.monotonic() discovered = content_discovery.run( self.run_id, self.policy_run_id, batch, self.source_context, self.runtime, write_runtime=False, ) discovery_duration_ms = _elapsed_ms(discovery_started_at, time.monotonic()) recalled = pattern_recall.run( self.run_id, self.policy_run_id, discovered["discovered_content_items"], discovered["content_media_records"], discovered["evidence_bundles"], self.source_context, self.runtime, self.gemini_video_client, start_index=self._recall_index_base + len(self.discovered_content_items) + 1, write_runtime=False, archive_dispatcher=self.archive_dispatcher, ) recall_timing = recalled.get("batch_timing") or {} portrait_started_at = time.monotonic() self._inject_fifty_plus(recalled["evidence_bundles"]) portrait_duration_ms = _elapsed_ms(portrait_started_at, time.monotonic()) rule_started_at = time.monotonic() decisions = rule_judgment.run( self.run_id, self.policy_run_id, recalled["evidence_bundles"], self.policy_bundle, self.runtime, start_index=self._decision_index_base + len(self.rule_decisions) + 1, write_runtime=False, ) rule_duration_ms = _elapsed_ms(rule_started_at, time.monotonic()) self._accumulate(batch, recalled, decisions) has_pass = any(decision.get("decision_action") == PASS_ACTION for decision in decisions) self._record_batch_timing_event( query, batch, decisions, page_number=page_number, page_item_count=page_item_count, batch_kind=batch_kind, search_duration_ms=search_duration_ms, search_wait_ms=search_wait_ms, search_request_duration_ms=search_request_duration_ms, search_timing_source=search_timing_source, oss_batch_duration_ms=int(recall_timing.get("oss_batch_duration_ms") or 0), qwen_batch_duration_ms=int(recall_timing.get("qwen_batch_duration_ms") or 0), portrait_duration_ms=portrait_duration_ms, rule_duration_ms=rule_duration_ms, discovery_duration_ms=discovery_duration_ms, batch_duration_ms=_elapsed_ms(batch_started_at, time.monotonic()), has_pass=has_pass, ) return has_pass def _inject_fifty_plus(self, bundles: list[dict[str, Any]]) -> None: """M9A:仅抖音,对相关性达标(query≥PORTRAIT_QUERY_GATE)的视频拉作者画像补 50+。 门只看 query 相关性、不看平台分:适老性要始终进三项加权(35/35/30), 否则"平台互动一般但极度适老"的视频会被错杀(拍板 2026-06-22)。阈值=抖音入池 query 下限 65, 低于此的视频靠 query 这关就被否、适老性也救不回来,不拉(省成本)。 单点覆盖首轮 + 游走。注入 bundle["content_audience_50plus"]: not_attempted(相关性不够,没拉)/ ok(画像算出)/ unavailable(拉取失败)/ incomplete(字段缺)。 非抖音不注入该块 → evaluator 走旧 50/50。 """ if self.platform != "douyin": return fetch_portrait = getattr(self.portrait_client, "fetch_account_fans_portrait", None) if not callable(fetch_portrait): # 平台 client 不提供画像(mock/replay)→ 不注入,evaluator 走旧 50/50。 return for bundle in bundles: query = _get_path(bundle, "pattern_match_result.query_relevance_score") if not (_is_number(query) and query >= PORTRAIT_QUERY_GATE): bundle["content_audience_50plus"] = {"status": "not_attempted"} continue author_id = _get_path(bundle, "content.author.platform_author_id") if not author_id: bundle["content_audience_50plus"] = {"status": "incomplete"} continue author_id = str(author_id) if author_id in self._portrait_by_author: portrait = self._portrait_by_author[author_id] # 命中缓存,不再 30s 限速 fetch_error = self._portrait_error_by_author.get(author_id) else: portrait, fetch_error = None, None for attempt in range(PORTRAIT_FETCH_ATTEMPTS): # 重试,救瞬时失败/限速 try: portrait = fetch_portrait(author_id) fetch_error = None break except Exception as exc: # 不再吞掉原因:记下来供技术详情展示 fetch_error = f"{type(exc).__name__}: {exc}".strip()[:200] portrait = None self._portrait_by_author[author_id] = portrait self._portrait_error_by_author[author_id] = fetch_error if portrait is None: bundle["content_audience_50plus"] = { "status": "unavailable", "failure_reason": fetch_error or "热点宝画像接口未返回数据", } continue fans_age = _get_path(portrait, "fans.age.data") result = fifty_plus_score(fans_age) block: dict[str, Any] = {"status": result["status"]} if result["status"] == "ok": block["score"] = result["score"] block["components"] = { key: result[key] for key in ("band", "band_tgi", "band_percent", "tgi_score", "percent_score") } bundle["content_audience_50plus"] = block def _record_batch_timing_event( self, query: dict[str, Any], batch: list[dict[str, Any]], decisions: list[dict[str, Any]], *, page_number: int, page_item_count: int, batch_kind: str, search_duration_ms: int, search_wait_ms: int, search_request_duration_ms: int, search_timing_source: str, oss_batch_duration_ms: int, qwen_batch_duration_ms: int, portrait_duration_ms: int, rule_duration_ms: int, discovery_duration_ms: int, batch_duration_ms: int, has_pass: bool, ) -> None: self._batch_event_index += 1 action_counts: dict[str, int] = {} for decision in decisions: action = str(decision.get("decision_action") or "unknown") action_counts[action] = action_counts.get(action, 0) + 1 raw_payload = { "batch_timing_schema_version": "progressive_batch_timing.v1", "progressive_batch_id": ( f"{query['search_query_id']}_p{page_number:02d}_{batch_kind}" ), "search_query_id": query.get("search_query_id"), "search_query": query.get("search_query"), "search_query_generation_method": query.get("search_query_generation_method"), "page_number": page_number, "page_item_count": page_item_count, "batch_kind": batch_kind, "batch_video_count": len(batch), "decision_action_counts": action_counts, "has_add_to_content_pool": has_pass, "search_duration_ms": search_duration_ms, "search_wait_ms": search_wait_ms, "search_request_duration_ms": search_request_duration_ms, "search_timing_source": search_timing_source, "oss_batch_duration_ms": oss_batch_duration_ms, "qwen_batch_duration_ms": qwen_batch_duration_ms, "portrait_duration_ms": portrait_duration_ms, "rule_duration_ms": rule_duration_ms, "discovery_duration_ms": discovery_duration_ms, "batch_duration_ms": batch_duration_ms, "content_ids": [str(item.get("platform_content_id") or "") for item in batch], } row = with_raw_payload( { "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION, "run_id": self.run_id, "policy_run_id": self.policy_run_id, "event_id": ( f"evt_progressive_batch_{self._batch_event_index:04d}_" f"{_safe_token(query.get('search_query_id'))}_p{page_number:02d}_{batch_kind}" ), "event_type": "progressive_batch_timing", "status": "success", "input_ref": "platform_results", "output_ref": "rule_decisions", "error_code": None, "message": ( f"{batch_kind} videos={len(batch)} add={action_counts.get(PASS_ACTION, 0)}" ), "created_at": datetime.now(timezone.utc).isoformat(), } ) row["raw_payload"].update(raw_payload) self.runtime.append_jsonl(self.run_id, "run_events.jsonl", [row]) def _prepare_result( self, query: dict[str, Any], result: dict[str, Any], *, page_number: int, page_item_count: int, batch_kind: str, rank: int, ) -> dict[str, Any]: runtime_result = _strip_transient_fields(result) prepared = platform_access._with_query_source(runtime_result, query) # noqa: SLF001 original_has_more = bool(prepared.get("has_more")) original_next_cursor = str(prepared.get("next_cursor") or "") prepared.update( { "search_query_id": query["search_query_id"], "content_discovery_id": _content_discovery_id(query["search_query_id"], page_number, rank), "progressive_batch_id": ( f"{query['search_query_id']}_p{page_number:02d}_{batch_kind}" ), "progressive_page_number": page_number, "progressive_batch_kind": batch_kind, "progressive_page_item_count": page_item_count, "progressive_item_rank_in_page": rank, "progressive_original_has_more": original_has_more, "progressive_original_next_cursor": original_next_cursor, } ) return prepared def _hydrate_result_media( self, result: dict[str, Any], query: dict[str, Any], ) -> dict[str, Any]: hydrate = getattr(self.platform_client, "hydrate_search_result_media", None) if not callable(hydrate): return result return dict(hydrate(result, query)) def _accumulate( self, batch: list[dict[str, Any]], recalled: dict[str, Any], decisions: list[dict[str, Any]], ) -> None: for offset, result in enumerate(batch): key = _content_key(result) if key: self._record_indexes_by_key[key] = { "platform_result": len(self.platform_results) + offset, "discovered": len(self.discovered_content_items) + offset, "evidence_bundle": len(self.evidence_bundles) + offset, "decision": len(self.rule_decisions) + offset, } self.platform_results.extend(batch) self.discovered_content_items.extend(recalled["discovered_content_items"]) self.content_media_records.extend(recalled["content_media_records"]) self.pattern_recall_evidence.extend(recalled["pattern_recall_evidence"]) self.evidence_bundles.extend(recalled["evidence_bundles"]) self.rule_decisions.extend(decisions) def _propagate_query_source_merge( self, key: tuple[str, str], merged_result: dict[str, Any], ) -> None: indexes = self._record_indexes_by_key.get(key) if not indexes: return fields = [ "query_sources", "matched_search_query_ids", "matched_search_queries", "matched_search_query_generation_methods", ] for collection_name in ["discovered_content_items"]: row = getattr(self, collection_name)[indexes["discovered"]] _copy_fields(row, merged_result, fields) bundle = self.evidence_bundles[indexes["evidence_bundle"]] _copy_fields(bundle.setdefault("source_evidence", {}), merged_result, fields) decision = self.rule_decisions[indexes["decision"]] _copy_fields(decision.setdefault("source_evidence", {}), merged_result, fields) raw_payload = decision.setdefault("raw_payload", {}) if isinstance(raw_payload.get("source_evidence"), dict): _copy_fields(raw_payload["source_evidence"], merged_result, fields) def write_runtime(self) -> None: if self.archive_dispatcher is not None: self._merge_archive_updates(self.archive_dispatcher.drain_completed()) self.runtime.append_jsonl(self.run_id, "content_media_records.jsonl", self.content_media_records) self.runtime.append_jsonl(self.run_id, "pattern_recall_evidence.jsonl", self.pattern_recall_evidence) self._hide_consumed_query_cursors() self.runtime.append_jsonl(self.run_id, "discovered_content_items.jsonl", self.discovered_content_items) self.runtime.append_jsonl(self.run_id, "rule_decisions.jsonl", self.rule_decisions) if self.archive_dispatcher is not None: self.archive_dispatcher.enable_runtime_writes() def close_archive_dispatcher(self) -> None: if self.archive_dispatcher is None or self._archive_dispatcher_closed: return self._archive_dispatcher_closed = True self.archive_dispatcher.shutdown(wait=False) def _merge_archive_updates(self, updates: list[dict[str, Any]]) -> None: if not updates: return by_key = { (row.get("platform"), row.get("platform_content_id")): row for row in updates if row.get("platform_content_id") } if not by_key: return self.content_media_records = [ by_key.get((row.get("platform"), row.get("platform_content_id")), row) for row in self.content_media_records ] def _hide_consumed_query_cursors(self) -> None: if not self._queries_with_consumed_next_pages: return for row in self.platform_results: self._hide_cursor_if_consumed(row) for row in self.discovered_content_items: self._hide_cursor_if_consumed(row) for bundle in self.evidence_bundles: source_evidence = bundle.get("source_evidence") if isinstance(source_evidence, dict): self._hide_cursor_if_consumed(source_evidence) for decision in self.rule_decisions: source_evidence = decision.get("source_evidence") if isinstance(source_evidence, dict): self._hide_cursor_if_consumed(source_evidence) raw_payload = decision.get("raw_payload") if isinstance(raw_payload, dict) and isinstance(raw_payload.get("source_evidence"), dict): self._hide_cursor_if_consumed(raw_payload["source_evidence"]) def _hide_cursor_if_consumed(self, row: dict[str, Any]) -> None: if str(row.get("search_query_id") or "") not in self._queries_with_consumed_next_pages: return row["has_more"] = False row["next_cursor"] = "" raw_payload = row.get("raw_payload") if isinstance(raw_payload, dict): raw_payload["has_more"] = False raw_payload["next_cursor"] = "" def _get_path(data: Any, path: str) -> Any: cur = data for part in path.split("."): if not isinstance(cur, dict): return None cur = cur.get(part) return cur def _elapsed_ms(start: float, end: float) -> int: return max(0, int((end - start) * 1000)) def _safe_token(value: Any) -> str: token = "".join(ch if ch.isalnum() or ch in {"_", "-"} else "_" for ch in str(value or "query")) return token[:80] or "query" def _is_number(value: Any) -> bool: return isinstance(value, (int, float)) and not isinstance(value, bool) def _content_key(result: dict[str, Any]) -> tuple[str, str] | None: content_id = str(result.get("platform_content_id") or "") if not content_id: return None return str(result.get("platform") or ""), content_id def _strip_transient_fields(result: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in result.items() if not str(key).startswith("_platform_")} def _content_discovery_id(search_query_id: str, page_number: int, rank: int) -> str: if page_number == 1: return f"{search_query_id}_content_{rank:03d}" return f"{search_query_id}_page_{page_number:03d}_content_{rank:03d}" def _page_has_more(results: list[dict[str, Any]]) -> bool: return any(bool(result.get("has_more")) for result in results) def _page_next_cursor(results: list[dict[str, Any]]) -> str: for result in results: cursor = str(result.get("next_cursor") or "") if cursor: return cursor return "" def _copy_fields(target: dict[str, Any], source: dict[str, Any], fields: list[str]) -> None: for field in fields: if field in source: target[field] = deepcopy(source[field]) raw_payload = target.get("raw_payload") if isinstance(raw_payload, dict): for field in fields: if field in source: raw_payload[field] = deepcopy(source[field])