content-find-agent-new/tests/test_pattern_recall_archive.py

from __future__ import annotations

from content_agent.business_modules.content_discovery.pattern_recall import recall_decision
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from tests.gemini_helpers import FakeGeminiVideoClient, fake_gemini_fail, fake_gemini_pool, fake_gemini_review


def test_pattern_recall_marks_all_judged_videos_for_oss_archive(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    items = [
        _item(run_id, policy_run_id, "content_pool", "d_001"),
        _item(run_id, policy_run_id, "content_review", "d_002"),
        _item(run_id, policy_run_id, "content_failed", "d_003"),
    ]
    media = [_media(run_id, policy_run_id, item["platform_content_id"]) for item in items]
    bundles = [_bundle(item) for item in items]
    pool_result = fake_gemini_pool()
    pool_result["timing_metrics"] = {
        "video_fetch": {"download_duration_ms": 100, "ffmpeg_duration_ms": 200},
        "gemini_request": {"total_duration_ms": 300},
    }
    gemini = FakeGeminiVideoClient(
        result_by_content_id={
            "content_pool": pool_result,
            "content_review": fake_gemini_review(),
            "content_failed": fake_gemini_fail("gemini_client_timeout"),
        }
    )

    result = recall_decision.run(
        run_id,
        policy_run_id,
        items,
        media,
        bundles,
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
    )

    assert {
        row["platform_content_id"]: row["content_media_status"]
        for row in result["content_media_records"]
    } == {
        "content_pool": "oss_upload_pending",
        "content_review": "oss_upload_pending",
        "content_failed": "oss_upload_pending",
    }
    for row in result["content_media_records"]:
        assert row["raw_payload"]["oss_archive_status"] == "pending"
    media_by_id = {row["platform_content_id"]: row for row in result["content_media_records"]}
    assert media_by_id["content_pool"]["raw_payload"]["gemini_timing_metrics"]["video_fetch"] == {
        "download_duration_ms": 100,
        "ffmpeg_duration_ms": 200,
    }
    evidence_by_id = {row["platform_content_id"]: row for row in result["pattern_recall_evidence"]}
    assert evidence_by_id["content_pool"]["evidence_summary"]["timing_metrics"]["gemini_request"] == {
        "total_duration_ms": 300,
    }


def test_pattern_recall_submits_oss_archive_before_gemini_judgment(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    item = _item(run_id, policy_run_id, "content_pool", "d_001")
    media = _media(run_id, policy_run_id, item["platform_content_id"])
    bundle = _bundle(item)
    dispatcher = RecordingArchiveDispatcher()
    gemini = ArchiveAwareGeminiVideoClient(dispatcher)

    recall_decision.run(
        run_id,
        policy_run_id,
        [item],
        [media],
        [bundle],
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
        archive_dispatcher=dispatcher,
    )

    assert dispatcher.submitted_content_ids == ["content_pool"]
    assert gemini.calls


def test_pattern_recall_waits_for_oss_archive_before_gemini_judgment(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    item = _item(run_id, policy_run_id, "content_pool", "d_001")
    media = _media(run_id, policy_run_id, item["platform_content_id"])
    bundle = _bundle(item)
    dispatcher = SynchronousArchiveDispatcher()
    gemini = OssAwareGeminiVideoClient()

    result = recall_decision.run(
        run_id,
        policy_run_id,
        [item],
        [media],
        [bundle],
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
        archive_dispatcher=dispatcher,
    )

    assert dispatcher.archived_content_ids == ["content_pool"]
    assert gemini.calls[0]["media"]["oss_url"] == "https://res.example/content_pool.mp4"
    [row] = result["content_media_records"]
    assert row["content_media_status"] == "oss_uploaded"
    assert row["oss_url"] == "https://res.example/content_pool.mp4"
    assert row["raw_payload"]["gemini_timing_metrics"]["video_fetch"]["gemini_video_source"] == "oss_url"


def test_pattern_recall_falls_back_to_play_url_when_oss_archive_is_pending(tmp_path):
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    item = _item(run_id, policy_run_id, "content_pool", "d_001")
    media = _media(run_id, policy_run_id, item["platform_content_id"])
    bundle = _bundle(item)
    dispatcher = PendingArchiveDispatcher()
    gemini = PlayUrlAwareGeminiVideoClient()

    result = recall_decision.run(
        run_id,
        policy_run_id,
        [item],
        [media],
        [bundle],
        {"ext_data": {"evidence_pack": {"seed_terms": ["养生"]}}},
        runtime,
        gemini,
        archive_dispatcher=dispatcher,
    )

    assert gemini.calls[0]["media"]["oss_url"] is None
    assert gemini.calls[0]["media"]["play_url"] == "https://source.example/content_pool.mp4"
    [row] = result["content_media_records"]
    assert row["content_media_status"] == "oss_upload_pending"
    assert row["raw_payload"]["oss_archive_last_error"] == "oss_upload_http_error"
    assert row["raw_payload"]["gemini_timing_metrics"]["video_fetch"]["gemini_video_source"] == "play_url"


class RecordingArchiveDispatcher:
    def __init__(self) -> None:
        self.submitted_content_ids: list[str] = []

    def submit_records(self, records: list[dict]) -> None:
        self.submitted_content_ids.extend(row["platform_content_id"] for row in records)


class SynchronousArchiveDispatcher:
    def __init__(self) -> None:
        self.archived_content_ids: list[str] = []

    def archive_records(self, records: list[dict]) -> list[dict]:
        archived = []
        for row in records:
            content_id = row["platform_content_id"]
            self.archived_content_ids.append(content_id)
            raw_payload = {
                **(row.get("raw_payload") or {}),
                "content_media_status": "oss_uploaded",
                "oss_url": f"https://res.example/{content_id}.mp4",
                "oss_archive_status": "uploaded",
                "oss_object_key": f"crawler/video/{content_id}.mp4",
            }
            archived.append(
                {
                    **row,
                    "content_media_status": "oss_uploaded",
                    "oss_url": f"https://res.example/{content_id}.mp4",
                    "raw_payload": raw_payload,
                }
            )
        return archived


class PendingArchiveDispatcher:
    def archive_records(self, records: list[dict]) -> list[dict]:
        archived = []
        for row in records:
            raw_payload = {
                **(row.get("raw_payload") or {}),
                "content_media_status": "oss_upload_pending",
                "oss_archive_status": "pending",
                "oss_archive_last_error": "oss_upload_http_error",
                "oss_archive_last_exception_type": "HTTPStatusError",
            }
            archived.append(
                {
                    **row,
                    "content_media_status": "oss_upload_pending",
                    "oss_url": None,
                    "raw_payload": raw_payload,
                }
            )
        return archived


class ArchiveAwareGeminiVideoClient(FakeGeminiVideoClient):
    def __init__(self, dispatcher: RecordingArchiveDispatcher) -> None:
        super().__init__(default_result=fake_gemini_pool())
        self.dispatcher = dispatcher

    def analyze(self, content: dict, media: dict, source_context: dict) -> dict:
        assert self.dispatcher.submitted_content_ids == [content["platform_content_id"]]
        return super().analyze(content, media, source_context)


class PlayUrlAwareGeminiVideoClient(FakeGeminiVideoClient):
    def __init__(self) -> None:
        result = fake_gemini_pool()
        result["timing_metrics"] = {
            "video_fetch": {"gemini_video_source": "play_url"},
        }
        super().__init__(default_result=result)

    def analyze(self, content: dict, media: dict, source_context: dict) -> dict:
        assert media.get("oss_url") is None
        assert media["play_url"].startswith("https://source.example/")
        return super().analyze(content, media, source_context)


class OssAwareGeminiVideoClient(FakeGeminiVideoClient):
    def __init__(self) -> None:
        result = fake_gemini_pool()
        result["timing_metrics"] = {
            "video_fetch": {"gemini_video_source": "oss_url"},
        }
        super().__init__(default_result=result)

    def analyze(self, content: dict, media: dict, source_context: dict) -> dict:
        assert media["oss_url"] == f"https://res.example/{content['platform_content_id']}.mp4"
        assert media["content_media_status"] == "oss_uploaded"
        return super().analyze(content, media, source_context)


def _item(run_id: str, policy_run_id: str, content_id: str, discovery_id: str) -> dict:
    return {
        "record_schema_version": "runtime_record.v1",
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "content_discovery_id": discovery_id,
        "search_query_id": "q_001",
        "platform": "douyin",
        "platform_content_id": content_id,
        "platform_content_format": "video",
        "description": "stub",
        "platform_author_id": "author_001",
        "author_display_name": "author",
        "statistics": {"digg_count": 100, "comment_count": 10, "share_count": 5},
        "tags": [],
        "discovery_start_source": "pattern_itemset",
        "previous_discovery_step": "query_direct",
        "raw_payload": {"platform_content_id": content_id},
    }


def _media(run_id: str, policy_run_id: str, content_id: str) -> dict:
    return {
        "record_schema_version": "runtime_record.v1",
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "platform": "douyin",
        "platform_content_id": content_id,
        "content_media_status": "metadata_only",
        "content_metadata_source": "douyin_keyword_search",
        "play_url": f"https://source.example/{content_id}.mp4",
        "local_path": None,
        "oss_url": None,
        "raw_payload": {"platform_content_id": content_id},
    }


def _bundle(item: dict) -> dict:
    return {
        "source_evidence": {"discovered_platform_content_id": item["platform_content_id"]},
        "content": {
            "decision_target_type": "content",
            "decision_target_id": item["platform_content_id"],
            "content_discovery_id": item["content_discovery_id"],
            "search_query_id": item["search_query_id"],
            "platform": item["platform"],
            "platform_content_id": item["platform_content_id"],
            "platform_content_format": "video",
            "description": item["description"],
        },
        "pattern_match_result": {},
        "content_engagement_metrics": {},
        "run_context": {
            "decision_input_snapshot_id": f"evidence_bundle:{item['platform_content_id']}",
        },
    }