Agent/examples/auto_put_ad_mini/test_landing_video_dedupe.py

import sys
import unittest
from pathlib import Path
from unittest.mock import patch


_HERE = Path(__file__).parent
sys.path.insert(0, str(_HERE))

import execute_creation_once  # noqa: E402


class _Strategy:
    def __init__(self, material_source):
        self.material_source = material_source
        self.use_ai_generated = material_source == "ai_generated"
        self.ai_fallback_to_history = False


class LandingVideoDedupeTest(unittest.TestCase):
    def test_same_crowd_package_excludes_landing_video_across_accounts_in_one_run(self):
        calls = []

        def fake_prepare(account_id, adgroup_id, *, excluded_material_ids, excluded_landing_ids, **kwargs):
            calls.append((account_id, adgroup_id, set(excluded_landing_ids)))
            return {
                "account_id": account_id,
                "adgroup_id": adgroup_id,
                "audience_tier": "wx*商业",
                "landing_video_id": 1000 + account_id,
                "_material_id": f"m-{account_id}",
                "material_source": "history",
            }

        with patch.object(execute_creation_once, "get_creation_account_ids", return_value=[1, 2]), \
            patch.object(execute_creation_once, "load_excluded_ad_ids_from_adjustment", return_value=set()), \
            patch.object(execute_creation_once, "get_account_crowd_package", return_value="wx*商业"), \
            patch.object(execute_creation_once, "load_recent_landing_usage_counts", return_value={"history": {}, "ai_generated": {}}), \
            patch.object(execute_creation_once, "load_account_material_strategy", return_value=_Strategy("history")), \
            patch.object(execute_creation_once, "find_ads_needing_creatives", side_effect=[
                [{"adgroup_id": 101, "creative_count": 3}],
                [{"adgroup_id": 201, "creative_count": 3}],
            ]), \
            patch.object(execute_creation_once, "build_landing_candidate_pool", return_value=object()), \
            patch.object(execute_creation_once, "prepare_one_creative_for_ad", side_effect=fake_prepare), \
            patch.object(execute_creation_once, "record_prepared_material_usage"):
            records = execute_creation_once.phase1_prepare(target_creatives=4)

        self.assertEqual(2, len(records))
        self.assertEqual(set(), calls[0][2])
        self.assertEqual({1001}, calls[1][2])

    def test_ai_landing_dedupe_is_independent_from_history_and_limits_once_per_ai_pool(self):
        calls = []

        def fake_prepare(account_id, adgroup_id, *, excluded_material_ids, excluded_landing_ids, **kwargs):
            calls.append(set(excluded_landing_ids))
            return {
                "account_id": account_id,
                "adgroup_id": adgroup_id,
                "audience_tier": "wx*商业",
                "landing_video_id": 777,
                "_material_id": f"ai:{account_id}-{adgroup_id}-{len(calls)}",
                "material_source": "ai_generated",
            }

        with patch.object(execute_creation_once, "get_creation_account_ids", return_value=[1, 2]), \
            patch.object(execute_creation_once, "load_excluded_ad_ids_from_adjustment", return_value=set()), \
            patch.object(execute_creation_once, "get_account_crowd_package", return_value="wx*商业"), \
            patch.object(
                execute_creation_once,
                "load_recent_landing_usage_counts",
                return_value={"history": {777: 1}, "ai_generated": {}},
            ), \
            patch.object(execute_creation_once, "load_account_material_strategy", return_value=_Strategy("ai_generated")), \
            patch.object(execute_creation_once, "find_ads_needing_creatives", side_effect=[
                [{"adgroup_id": 101, "creative_count": 11}],
                [{"adgroup_id": 201, "creative_count": 11}],
            ]), \
            patch.object(execute_creation_once, "build_landing_candidate_pool", return_value=object()), \
            patch.object(execute_creation_once, "prepare_one_creative_for_ad", side_effect=fake_prepare), \
            patch.object(execute_creation_once, "record_prepared_material_usage"):
            records = execute_creation_once.phase1_prepare(target_creatives=12)

        self.assertEqual(1, len(records))
        self.assertEqual(set(), calls[0])
        self.assertEqual({777}, calls[1])

    def test_phase1_reuses_landing_candidate_pool_for_multiple_creative_attempts(self):
        pool = object()
        build_calls = []
        prepare_calls = []

        def fake_build_pool(account_id):
            build_calls.append(account_id)
            return pool

        def fake_prepare(
            account_id,
            adgroup_id,
            *,
            excluded_material_ids,
            excluded_landing_ids,
            landing_candidates,
            failed_landing_ids,
        ):
            prepare_calls.append((landing_candidates, failed_landing_ids))
            return {
                "account_id": account_id,
                "adgroup_id": adgroup_id,
                "audience_tier": "R330",
                "landing_video_id": 1000 + len(prepare_calls),
                "_material_id": f"ai:{len(prepare_calls)}",
                "material_source": "ai_generated",
            }

        with patch.object(execute_creation_once, "get_creation_account_ids", return_value=[1]), \
            patch.object(execute_creation_once, "load_excluded_ad_ids_from_adjustment", return_value=set()), \
            patch.object(execute_creation_once, "get_account_crowd_package", return_value="R330"), \
            patch.object(execute_creation_once, "load_recent_landing_usage_counts", return_value={"history": {}, "ai_generated": {}}), \
            patch.object(execute_creation_once, "load_account_material_strategy", return_value=_Strategy("ai_generated")), \
            patch.object(execute_creation_once, "find_ads_needing_creatives", return_value=[
                {"adgroup_id": 101, "creative_count": 0},
            ]), \
            patch.object(execute_creation_once, "build_landing_candidate_pool", side_effect=fake_build_pool), \
            patch.object(execute_creation_once, "prepare_one_creative_for_ad", side_effect=fake_prepare), \
            patch.object(execute_creation_once, "record_prepared_material_usage"):
            records = execute_creation_once.phase1_prepare(target_creatives=2)

        self.assertEqual(2, len(records))
        self.assertEqual([1], build_calls)
        self.assertEqual([(pool, set()), (pool, set())], prepare_calls)

    def test_phase1_carries_failed_landing_ids_between_attempts(self):
        calls = []

        def fake_prepare(
            account_id,
            adgroup_id,
            *,
            excluded_material_ids,
            excluded_landing_ids,
            landing_candidates,
            failed_landing_ids,
        ):
            calls.append(set(failed_landing_ids))
            if len(calls) == 1:
                failed_landing_ids.add(777)
                return None
            return {
                "account_id": account_id,
                "adgroup_id": adgroup_id,
                "audience_tier": "R330",
                "landing_video_id": 888,
                "_material_id": "ai:ok",
                "material_source": "ai_generated",
            }

        with patch.object(execute_creation_once, "get_creation_account_ids", return_value=[1]), \
            patch.object(execute_creation_once, "load_excluded_ad_ids_from_adjustment", return_value=set()), \
            patch.object(execute_creation_once, "get_account_crowd_package", return_value="R330"), \
            patch.object(execute_creation_once, "load_recent_landing_usage_counts", return_value={"history": {}, "ai_generated": {}}), \
            patch.object(execute_creation_once, "load_account_material_strategy", return_value=_Strategy("ai_generated")), \
            patch.object(execute_creation_once, "find_ads_needing_creatives", return_value=[
                {"adgroup_id": 101, "creative_count": 0},
            ]), \
            patch.object(execute_creation_once, "build_landing_candidate_pool", return_value=object()), \
            patch.object(execute_creation_once, "prepare_one_creative_for_ad", side_effect=fake_prepare), \
            patch.object(execute_creation_once, "record_prepared_material_usage"):
            records = execute_creation_once.phase1_prepare(target_creatives=2)

        self.assertEqual(1, len(records))
        self.assertEqual([set(), {777}], calls)


if __name__ == "__main__":
    unittest.main()