content-find-agent-new/content_agent/business_modules/content_discovery/content_discovery_builder.py

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

from content_agent.business_modules.content_discovery.platform_observable_performance import (
    performance_score,
)
from content_agent.business_modules.content_discovery.source_evidence import (
    build_source_evidence,
)
from content_agent.constants import EVIDENCE_BUNDLE_SCHEMA_VERSION, RUNTIME_RECORD_SCHEMA_VERSION
from content_agent.interfaces import RuntimeFileStore
from content_agent.record_payload import with_raw_payload


def run(
    run_id: str,
    policy_run_id: str,
    platform_results: list[dict[str, Any]],
    source_context: dict[str, Any],
    runtime: RuntimeFileStore,
    write_runtime: bool = True,
) -> dict[str, list[dict[str, Any]]]:
    created_at = datetime.now(timezone.utc).isoformat()
    discovered_content_items: list[dict[str, Any]] = []
    content_media_records: list[dict[str, Any]] = []
    evidence_bundles: list[dict[str, Any]] = []

    for result in platform_results:
        discovered_content_item = _build_discovered_content_item(
            run_id, policy_run_id, result, created_at
        )
        content_media_record = _build_content_media_record(
            run_id, policy_run_id, result, created_at
        )
        evidence_bundles.append(
            _build_evidence_bundle(
                run_id, policy_run_id, discovered_content_item, result, source_context
            )
        )
        discovered_content_items.append(discovered_content_item)
        content_media_records.append(content_media_record)

    if write_runtime:
        runtime.append_jsonl(run_id, "content_media_records.jsonl", content_media_records)
    return {
        "discovered_content_items": discovered_content_items,
        "content_media_records": content_media_records,
        "evidence_bundles": evidence_bundles,
    }


def _build_discovered_content_item(
    run_id: str,
    policy_run_id: str,
    result: dict[str, Any],
    created_at: str,
) -> dict[str, Any]:
    item = {
        "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "content_discovery_id": result["content_discovery_id"],
        "search_query_id": result["search_query_id"],
        "platform": result.get("platform", "douyin"),
        "platform_content_id": result["platform_content_id"],
        "platform_content_format": result.get("platform_content_format", "video"),
        "platform_content_url": result.get("platform_content_url"),
        "description": result["description"],
        "platform_author_id": result.get("platform_author_id", ""),
        "author_display_name": result.get("author_display_name", ""),
        "statistics": result["statistics"],
        "tags": result.get("tags", []),
        "discovery_start_source": result["discovery_start_source"],
        "previous_discovery_step": result["previous_discovery_step"],
        "created_at": created_at,
    }
    for field in [
        "text_extra",
        "create_time",
        "has_more",
        "next_cursor",
        "platform_auth_mode",
        "platform_raw_payload",
        "query_sources",
        "matched_search_query_ids",
        "matched_search_queries",
        "matched_search_query_generation_methods",
        "progressive_batch_id",
        "progressive_page_number",
        "progressive_batch_kind",
        "progressive_page_item_count",
        "progressive_item_rank_in_page",
        "progressive_original_has_more",
        "progressive_original_next_cursor",
    ]:
        if field in result:
            item[field] = result[field]
    return with_raw_payload(item)


def _build_content_media_record(
    run_id: str,
    policy_run_id: str,
    result: dict[str, Any],
    created_at: str,
) -> dict[str, Any]:
    record = {
        "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "platform": result.get("platform", "douyin"),
        "platform_content_id": result["platform_content_id"],
        "content_media_status": "metadata_only",
        "content_metadata_source": result.get("content_metadata_source", "mock_platform_search"),
        "play_url": result.get("play_url"),
        "local_path": None,
        "oss_url": None,
        "created_at": created_at,
    }
    if not record["play_url"]:
        record["failure_reason"] = result.get("media_failure_reason", "no_play_url_returned")
    record = with_raw_payload(record)
    if result.get("video_url_candidates"):
        record["raw_payload"]["video_url_candidates"] = result["video_url_candidates"]
    platform_raw_payload = result.get("platform_raw_payload") or {}
    if isinstance(platform_raw_payload, dict):
        for field in [
            "no_valid_play_url",
            "selected_video_url_path",
            "selected_video_url_host",
            "selected_video_url_probe_status",
            "selected_video_url_content_type",
            "selected_video_url_content_range",
            "video_url_candidate_counts",
            "video_url_reject_reasons",
            "kuaishou_detail_fallback_attempted",
            "kuaishou_detail_fallback_status",
            "kuaishou_detail_fallback_exception_type",
            "kuaishou_detail_fallback_error",
            "douyin_detail_fallback_attempted",
            "douyin_detail_fallback_status",
            "douyin_detail_fallback_exception_type",
            "douyin_detail_fallback_error",
            "shipinhao_research_source",
            "shipinhao_research_status",
            "shipinhao_research_exception_type",
            "shipinhao_research_error",
            "shipinhao_research_item_count",
        ]:
            if field in platform_raw_payload:
                record["raw_payload"][field] = platform_raw_payload[field]
    return record


def _build_evidence_bundle(
    run_id: str,
    policy_run_id: str,
    discovered_content_item: dict[str, Any],
    result: dict[str, Any],
    source_context: dict[str, Any],
) -> dict[str, Any]:
    source_evidence = build_source_evidence(
        run_id, policy_run_id, discovered_content_item, result, source_context
    )
    return {
        "schema_version": EVIDENCE_BUNDLE_SCHEMA_VERSION,
        "source_evidence": source_evidence,
        "content": {
            "decision_target_type": "content",
            "decision_target_id": discovered_content_item["platform_content_id"],
            "content_discovery_id": discovered_content_item["content_discovery_id"],
            "search_query_id": discovered_content_item["search_query_id"],
            "platform": discovered_content_item["platform"],
            "platform_content_id": discovered_content_item["platform_content_id"],
            "platform_content_format": discovered_content_item["platform_content_format"],
            "description": discovered_content_item["description"],
            "create_time": result.get("create_time"),
            "author": {
                "platform_author_id": discovered_content_item.get("platform_author_id", ""),
                "author_display_name": discovered_content_item.get("author_display_name", ""),
            },
            "platform_author_id": discovered_content_item.get("platform_author_id", ""),
        },
        "pattern_match_result": {
            "level": result.get("relevance_level", "related"),
            "score": result.get("score"),
        },
        "content_engagement_metrics": {
            "statistics": result["statistics"],
            **result["statistics"],
            "platform_performance": performance_score(
                result["statistics"],
                discovered_content_item["platform"],
            ),
        },
        "author_audience_profile": {
            "platform_author_id": discovered_content_item.get("platform_author_id", ""),
            "author": {
                "platform_author_id": discovered_content_item.get("platform_author_id", "")
            },
        },
        "content_risk_check": {
            "risk_level": result.get("risk_level", "unknown"),
            "age_50_plus_safety": result.get("age_50_plus_safety", "safe"),
            "availability": result.get("availability", "metadata_only"),
        },
        "discovery_path_context": {
            "discovery_start_source": discovered_content_item["discovery_start_source"],
            "previous_discovery_step": discovered_content_item["previous_discovery_step"],
        },
        "run_context": {
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "runtime_record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
            "evidence_bundle_schema_version": EVIDENCE_BUNDLE_SCHEMA_VERSION,
            "decision_input_snapshot_id": (
                f"evidence_bundle:{run_id}:{discovered_content_item['content_discovery_id']}"
            ),
        },
    }