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

from __future__ import annotations

import json

from content_agent.business_modules import platform_access
from content_agent.business_modules.content_discovery import content_discovery_builder
from content_agent.integrations.database_runtime import DatabaseRuntimeStore
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from tests.test_database_runtime import FakeConnection, _config, _insert_values


RUN_ID = "run_m2_replay"
POLICY_RUN_ID = "policy_m2_replay"


class SearchOnlyPlatformClient:
    def __init__(self, platform: str, content_id: str):
        self.platform = platform
        self.content_id = content_id

    def search(self, search_query):
        return [
            {
                "content_discovery_id": f"{search_query['search_query_id']}_content_001",
                "search_query_id": search_query["search_query_id"],
                "platform": self.platform,
                "platform_content_id": self.content_id,
                "platform_content_format": "video",
                "description": f"{self.platform} 内容",
                "platform_author_id": f"{self.platform}_author",
                "author_display_name": f"{self.platform} 作者",
                "statistics": {
                    "digg_count": 10,
                    "comment_count": 2,
                    "share_count": 3,
                    "collect_count": 4,
                    "play_count": 100,
                },
                "tags": ["#祝福"],
                "play_url": f"https://video.test/{self.content_id}.mp4",
                "has_more": False,
                "next_cursor": "",
                "platform_raw_payload": {
                    "channel_content_id": self.content_id,
                    "channel_account_id": f"{self.platform}_author",
                },
                "discovery_start_source": search_query["discovery_start_source"],
                "previous_discovery_step": "search_query_direct",
            }
        ]


def test_v4_m2_platform_sources_replay_preserves_observable_containers(tmp_path):
    platform_results = []
    for platform in ["douyin", "kuaishou", "shipinhao"]:
        query = {
            "search_query_id": f"q_{platform}",
            "search_query": f"{platform} 祝福",
            "search_query_generation_method": "item_single",
            "discovery_start_source": "pattern_itemset",
        }
        result = platform_access.run(
            [query],
            SearchOnlyPlatformClient(platform, f"{platform}_content_001"),
        )
        assert result["query_failures"] == []
        platform_results.extend(result["platform_results"])

    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(RUN_ID)
    discovery_result = content_discovery_builder.run(
        RUN_ID,
        POLICY_RUN_ID,
        platform_results,
        _source_context(),
        runtime,
    )

    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)
    store.append_jsonl(
        RUN_ID,
        "discovered_content_items.jsonl",
        discovery_result["discovered_content_items"],
    )

    inserted = [_insert_values(sql, params) for sql, params in connection.statements]
    assert len(inserted) == 3
    for values in inserted:
        statistics = json.loads(values["statistics"])
        platform_raw_payload = json.loads(values["platform_raw_payload"])
        raw_payload = json.loads(values["raw_payload"])

        assert statistics["digg_count"] == 10
        assert statistics["play_count"] == 100
        assert platform_raw_payload["channel_content_id"].endswith("_content_001")
        assert raw_payload["statistics"] == statistics
        assert raw_payload["platform_raw_payload"] == platform_raw_payload
        assert raw_payload["matched_search_query_ids"] == [values["search_query_id"]]


def _source_context():
    return {
        "schema_version": "runtime_record.v1",
        "run_id": RUN_ID,
        "demand_content_id": "demand_001",
        "ext_data": {
            "evidence_pack": {
                "pattern_source_system": "pg_pattern_v2",
                "source_kind": "pattern_itemset",
                "source_post_id": "post_001",
                "pattern_execution_id": 581,
                "mining_config_id": 2082,
                "itemset_ids": [1608352],
                "seed_terms": ["祝福"],
            }
        },
    }