Agent/examples/auto_put_ad_mini/test_creative_performance_cleanup.py

import json
import unittest
import os
from contextlib import contextmanager
from datetime import date, datetime
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import Mock, patch
from zoneinfo import ZoneInfo


class CreativePerformanceRuleTests(unittest.TestCase):
    def setUp(self):
        from tools.creative_rejection_cleanup import performance_cleanup_config

        self.config = {
            "new_min_age_days": 5,
            "new_max_age_days": 7,
            "new_impressions_threshold": 100,
            "old_daily_impressions_threshold": 100.0,
            "window_days": 7,
        }
        # Keep an import-time reference so failures in the public config helper
        # are caught by this focused test module too.
        self.assertTrue(callable(performance_cleanup_config))

    @staticmethod
    def _creative(**overrides):
        creative = {
            "configured_status": "AD_STATUS_NORMAL",
            "creative_set_approval_status": (
                "CREATIVE_SET_APPROVAL_STATUS_NORMAL"
            ),
        }
        creative.update(overrides)
        return creative

    @staticmethod
    def _ad(**overrides):
        ad = {"configured_status": "AD_STATUS_NORMAL"}
        ad.update(overrides)
        return ad

    def _action(self, *, age_days, impressions, cost_fen=0, **overrides):
        from tools.creative_rejection_cleanup import (
            determine_performance_cleanup_action,
        )

        as_of = date(2026, 8, 19)
        created = datetime(
            2026,
            8,
            19 - age_days,
            10,
            tzinfo=ZoneInfo("Asia/Shanghai"),
        )
        return determine_performance_cleanup_action(
            overrides.pop("creative", self._creative()),
            overrides.pop("ad", self._ad()),
            creative_created_at=created,
            as_of_date=as_of,
            impressions=impressions,
            cost_fen=cost_fen,
            metric_start_date=date(2026, 8, 12),
            metric_end_date=date(2026, 8, 18),
            config=self.config,
            **overrides,
        )

    def test_new_creative_requires_strict_age_and_impression_boundaries(self):
        from tools.creative_rejection_cleanup import PERFORMANCE_NEW_RULE

        self.assertIsNone(self._action(age_days=5, impressions=0))
        self.assertEqual(
            self._action(age_days=6, impressions=99)["cleanup_rule_type"],
            PERFORMANCE_NEW_RULE,
        )
        self.assertEqual(
            self._action(age_days=7, impressions=0)["cleanup_rule_type"],
            PERFORMANCE_NEW_RULE,
        )
        self.assertIsNone(self._action(age_days=6, impressions=100))
        self.assertIsNone(self._action(age_days=6, impressions=0, cost_fen=1))
        self.assertIsNone(
            self._action(
                age_days=6,
                impressions=0,
                current_day_cost_fen=1,
            )
        )

    def test_old_creative_matches_both_requested_strict_exposure_branches(self):
        from tools.creative_rejection_cleanup import (
            PERFORMANCE_OLD_HIGH_RULE,
            PERFORMANCE_OLD_LOW_RULE,
        )

        self.assertEqual(
            self._action(age_days=8, impressions=699)["cleanup_rule_type"],
            PERFORMANCE_OLD_LOW_RULE,
        )
        self.assertEqual(
            self._action(age_days=8, impressions=701)["cleanup_rule_type"],
            PERFORMANCE_OLD_HIGH_RULE,
        )
        self.assertIsNone(self._action(age_days=8, impressions=700))
        self.assertIsNone(self._action(age_days=8, impressions=701, cost_fen=1))
        self.assertIsNone(
            self._action(
                age_days=8,
                impressions=701,
                current_day_cost_fen=1,
            )
        )

    def test_stopped_or_unapproved_assets_are_never_performance_deleted(self):
        self.assertIsNone(
            self._action(
                age_days=8,
                impressions=0,
                creative=self._creative(configured_status="AD_STATUS_SUSPEND"),
            )
        )
        self.assertIsNone(
            self._action(
                age_days=8,
                impressions=0,
                ad=self._ad(configured_status="AD_STATUS_SUSPEND"),
            )
        )
        self.assertIsNone(
            self._action(
                age_days=8,
                impressions=0,
                ad=self._ad(begin_date="2026-08-20"),
            )
        )
        self.assertIsNone(
            self._action(
                age_days=8,
                impressions=0,
                creative=self._creative(
                    creative_set_approval_status=(
                        "CREATIVE_SET_APPROVAL_STATUS_PENDING"
                    )
                ),
            )
        )


class CreativePerformanceMetricApiTests(unittest.TestCase):
    def test_tencent_metric_query_sums_impressions_and_cost(self):
        from tencent_client import TencentClient

        response = Mock()
        response.raise_for_status.return_value = None
        response.json.return_value = {
            "code": 0,
            "data": {
                "list": [
                    {
                        "dynamic_creative_id": 3,
                        "view_count": 40,
                        "cost": 0,
                    },
                    {
                        "dynamic_creative_id": 3,
                        "view_count": 59,
                        "cost": 0,
                    },
                ],
                "page_info": {"total_page": 1},
            },
        }
        client = TencentClient()
        client.session.get = Mock(return_value=response)
        client._common_params = Mock(return_value={"access_token": "token"})

        metrics = client.get_dynamic_creative_metrics(
            1,
            [3, 4],
            date(2026, 8, 12),
            date(2026, 8, 18),
        )

        self.assertEqual(
            metrics,
            {
                3: {"impressions": 99, "cost_fen": 0},
                4: {"impressions": 0, "cost_fen": 0},
            },
        )

    def test_tencent_ad_metric_query_sums_three_days_and_keeps_zero_rows(self):
        from tencent_client import TencentClient

        response = Mock()
        response.raise_for_status.return_value = None
        response.json.return_value = {
            "code": 0,
            "data": {
                "list": [
                    {"adgroup_id": 2, "cost": 0, "view_count": 10},
                    {"adgroup_id": 2, "cost": 0, "view_count": 20},
                ],
                "page_info": {"total_page": 1},
            },
        }
        client = TencentClient()
        client.session.get = Mock(return_value=response)
        client._common_params = Mock(return_value={"access_token": "token"})

        metrics = client.get_ad_metrics(
            1,
            [2, 4],
            date(2026, 8, 16),
            date(2026, 8, 18),
        )

        self.assertEqual(metrics[2]["cost_fen"], 0)
        self.assertEqual(metrics[2]["impressions"], 30)
        self.assertEqual(metrics[4]["cost_fen"], 0)


class CreativePerformanceScopeTests(unittest.TestCase):
    def test_account_metadata_uses_latest_active_account_row(self):
        import pandas as pd

        from tools.creative_rejection_cleanup import (
            fetch_tencent_account_metadata,
        )

        client = Mock()
        client.execute_sql.return_value = pd.DataFrame([
            {
                "account_id": "1",
                "account_name": " 账户一 ",
                "agent_name": " 代理商甲 ",
            },
            {
                "account_id": "2",
                "account_name": float("nan"),
                "agent_name": float("nan"),
            },
        ])

        metadata = fetch_tencent_account_metadata(client)

        sql = " ".join(client.execute_sql.call_args.args[0].split())
        self.assertIn("FROM loghubods.ad_put_tencent_account", sql)
        self.assertIn("PARTITION BY account_id", sql)
        self.assertLess(sql.index("ROW_NUMBER() OVER"), sql.index("WHERE rn=1"))
        self.assertIn("WHERE rn=1 AND is_delete=0", sql)
        self.assertEqual(
            metadata[1],
            {"account_name": "账户一", "agent_name": "代理商甲"},
        )
        self.assertEqual(
            metadata[2],
            {"account_name": "", "agent_name": ""},
        )

    def test_inventory_filters_is_delete_after_latest_row_selection(self):
        import pandas as pd

        from tools.creative_rejection_cleanup import (
            fetch_active_creative_inventory,
        )

        client = Mock()
        client.execute_sql.return_value = pd.DataFrame([{
            "account_id": 1,
            "ad_id": 2,
            "creative_id": 3,
            "creative_name": "创意三",
            "creative_status": "NORMAL",
            "create_time": "2026-08-13 10:00:00",
            "update_time": "2026-08-18 10:00:00",
        }])

        inventory = fetch_active_creative_inventory(client)

        sql = " ".join(client.execute_sql.call_args.args[0].split())
        self.assertEqual(sql.count("is_delete=0"), 1)
        self.assertIn(
            "ROW_NUMBER() OVER ( PARTITION BY account_id, creative_id",
            sql,
        )
        self.assertLess(sql.index("ROW_NUMBER() OVER"), sql.index("WHERE rn=1"))
        self.assertEqual(inventory[0]["account_id"], 1)
        self.assertEqual(inventory[0]["adgroup_id"], 2)
        self.assertEqual(inventory[0]["creative_id"], 3)
        self.assertEqual(inventory[0]["creative_status"], "NORMAL")
        self.assertIsNotNone(inventory[0]["create_time"])

    def test_scan_uses_only_odps_creatives_and_queries_today_separately(self):
        from tools import creative_rejection_cleanup as cleanup

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [
            {
                "dynamic_creative_id": 3,
                "adgroup_id": 2,
                "configured_status": "AD_STATUS_NORMAL",
                "creative_set_approval_status": (
                    cleanup.CREATIVE_NORMAL_STATUS
                ),
                "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
            },
            {
                "dynamic_creative_id": 9,
                "adgroup_id": 9,
                "configured_status": "AD_STATUS_NORMAL",
                "creative_set_approval_status": (
                    cleanup.CREATIVE_NORMAL_STATUS
                ),
                "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
            },
        ]
        tencent.get_ads.return_value = [
            {
                "adgroup_id": 2,
                "configured_status": "AD_STATUS_NORMAL",
                "created_time": "2026-08-10 00:00:00",
            },
            {
                "adgroup_id": 9,
                "configured_status": "AD_STATUS_NORMAL",
                "created_time": "2026-08-10 00:00:00",
            },
        ]
        tencent.get_dynamic_creative_metrics.side_effect = [
            {3: {"impressions": 99, "cost_fen": 0}},
            {3: {"impressions": 1, "cost_fen": 5}},
        ]
        tencent.get_ad_metrics.side_effect = [
            {2: {"impressions": 0, "cost_fen": 0}},
            {2: {"impressions": 1, "cost_fen": 7}},
        ]
        sources = {
            3: {
                "account_id": 1,
                "adgroup_id": 2,
                "creative_id": 3,
                "create_time": "2026-08-13 10:00:00",
            },
            4: {
                "account_id": 1,
                "adgroup_id": 4,
                "creative_id": 4,
                "create_time": "2026-08-13 10:00:00",
            },
        }

        result = cleanup._scan_one_account(
            {"account_id": 1},
            tencent=tencent,
            review_fetcher=Mock(),
            spend_start_date=date(2026, 8, 16),
            spend_end_date=date(2026, 8, 18),
            review_enabled=False,
            performance_enabled=True,
            performance_source_creatives=sources,
            performance_start_date=date(2026, 8, 12),
            performance_end_date=date(2026, 8, 18),
            current_day_date=date(2026, 8, 19),
            ad_cleanup_enabled=True,
            ad_metric_start_date=date(2026, 8, 16),
            ad_metric_end_date=date(2026, 8, 18),
        )

        self.assertEqual(result[7][3]["current_day_cost_fen"], 5)
        self.assertEqual(result[7][3]["impressions"], 100)
        self.assertEqual(result[9][2]["current_day_cost_fen"], 7)
        self.assertEqual(result[12]["missing_tencent_creatives"], 1)
        creative_calls = tencent.get_dynamic_creative_metrics.call_args_list
        self.assertEqual(creative_calls[0].args[1], [3])
        self.assertEqual(
            creative_calls[0].args[2:],
            (date(2026, 8, 13), date(2026, 8, 18)),
        )
        self.assertEqual(
            creative_calls[1].args[2:],
            (date(2026, 8, 19), date(2026, 8, 19)),
        )
        ad_calls = tencent.get_ad_metrics.call_args_list
        self.assertEqual(ad_calls[0].args[1], [2])
        self.assertEqual(
            ad_calls[1].args[2:],
            (date(2026, 8, 19), date(2026, 8, 19)),
        )

    def test_review_failure_does_not_block_independent_ad_metric_scan(self):
        from tools import creative_rejection_cleanup as cleanup

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [{
            "dynamic_creative_id": 3,
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "creative_set_approval_status": cleanup.CREATIVE_NORMAL_STATUS,
            "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
        }]
        tencent.get_ads.return_value = [{
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }]
        tencent.get_ad_metrics.return_value = {
            2: {"impressions": 0, "cost_fen": 0}
        }

        result = cleanup._scan_one_account(
            {"account_id": 1},
            tencent=tencent,
            review_fetcher=Mock(side_effect=RuntimeError("review unavailable")),
            spend_start_date=date(2026, 8, 16),
            spend_end_date=date(2026, 8, 18),
            review_enabled=True,
            performance_enabled=True,
            performance_source_creatives={
                3: {
                    "account_id": 1,
                    "adgroup_id": 2,
                    "creative_id": 3,
                    "create_time": "2026-08-13 10:00:00",
                }
            },
            performance_start_date=date(2026, 8, 12),
            performance_end_date=date(2026, 8, 18),
            performance_as_of=datetime(
                2026,
                8,
                19,
                11,
                tzinfo=ZoneInfo("Asia/Shanghai"),
            ),
            current_day_date=date(2026, 8, 19),
            ad_cleanup_enabled=True,
            ad_metric_start_date=date(2026, 8, 16),
            ad_metric_end_date=date(2026, 8, 18),
        )

        self.assertEqual(result[11], "review unavailable")
        self.assertIsNone(result[10])
        self.assertIn(2, result[9])
        tencent.get_dynamic_creative_metrics.assert_not_called()

    def test_retry_item_that_left_active_inventory_is_skipped_not_deferred(self):
        from tools import creative_rejection_cleanup as cleanup

        creative_item = {
            "account_id": 1,
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": cleanup.DELETE_CREATIVE,
            "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
        }
        ad_item = {
            "account_id": 1,
            "adgroup_id": 2,
            "dynamic_creative_id": -2,
            "cleanup_action": cleanup.DELETE_AD,
            "cleanup_rule_type": cleanup.PERFORMANCE_AD_ZERO_SPEND_RULE,
        }

        creative_result = cleanup.cleanup_precondition_failure(
            creative_item,
            set(),
            {},
            performance_scanned_accounts={1},
            performance_scope_loaded=True,
            performance_source_keys=set(),
        )
        ad_result = cleanup.cleanup_precondition_failure(
            ad_item,
            set(),
            {},
            ad_scanned_accounts={1},
            performance_scope_loaded=True,
            performance_source_ad_keys=set(),
        )
        scope_error_result = cleanup.cleanup_precondition_failure(
            creative_item,
            set(),
            {},
            performance_scope_loaded=False,
        )

        self.assertEqual(
            creative_result[0], "SKIPPED_REVIEW_NOT_RECONFIRMED"
        )
        self.assertEqual(ad_result[0], "SKIPPED_REVIEW_NOT_RECONFIRMED")
        self.assertEqual(scope_error_result[0], "DEFERRED")

    def test_ad_uses_odps_ad_id_even_when_tencent_creative_mapping_differs(self):
        from tools import creative_rejection_cleanup as cleanup

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [{
            "dynamic_creative_id": 3,
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "creative_set_approval_status": cleanup.CREATIVE_NORMAL_STATUS,
            "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
        }]
        tencent.get_ads.return_value = [{
            "adgroup_id": 5,
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }]
        tencent.get_ad_metrics.return_value = {
            5: {"impressions": 0, "cost_fen": 0}
        }

        result = cleanup._scan_one_account(
            {"account_id": 1},
            tencent=tencent,
            review_fetcher=Mock(),
            spend_start_date=date(2026, 8, 16),
            spend_end_date=date(2026, 8, 18),
            review_enabled=False,
            performance_source_creatives={
                3: {
                    "account_id": 1,
                    "adgroup_id": 5,
                    "creative_id": 3,
                    "create_time": "2026-08-13 10:00:00",
                }
            },
            current_day_date=date(2026, 8, 19),
            ad_cleanup_enabled=True,
            ad_metric_start_date=date(2026, 8, 16),
            ad_metric_end_date=date(2026, 8, 18),
        )

        self.assertIn(5, result[9])
        self.assertEqual(result[12]["ad_mismatches"], 1)
        self.assertEqual(tencent.get_ad_metrics.call_count, 2)
        self.assertEqual(tencent.get_ad_metrics.call_args_list[0].args[1], [5])

    def test_creative_list_failure_does_not_block_ad_scan(self):
        from tools import creative_rejection_cleanup as cleanup

        tencent = Mock()
        tencent.get_dynamic_creatives.side_effect = RuntimeError(
            "creative endpoint unavailable"
        )
        tencent.get_ads.return_value = [{
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }]
        tencent.get_ad_metrics.return_value = {
            2: {"impressions": 0, "cost_fen": 0}
        }

        result = cleanup._scan_one_account(
            {"account_id": 1},
            tencent=tencent,
            review_fetcher=Mock(),
            spend_start_date=date(2026, 8, 16),
            spend_end_date=date(2026, 8, 18),
            review_enabled=True,
            performance_enabled=True,
            performance_source_creatives={
                3: {
                    "account_id": 1,
                    "adgroup_id": 2,
                    "creative_id": 3,
                    "create_time": "2026-08-13 10:00:00",
                }
            },
            performance_start_date=date(2026, 8, 12),
            performance_end_date=date(2026, 8, 18),
            current_day_date=date(2026, 8, 19),
            ad_cleanup_enabled=True,
            ad_metric_start_date=date(2026, 8, 16),
            ad_metric_end_date=date(2026, 8, 18),
        )

        self.assertIn("creative endpoint unavailable", result[11])
        self.assertIn("creative endpoint unavailable", result[8])
        self.assertIsNone(result[10])
        self.assertIn(2, result[9])
        tencent.get_ads.assert_called_once_with(1)


class AdPerformanceRuleTests(unittest.TestCase):
    def test_active_ad_with_zero_three_day_spend_is_deleted(self):
        from tools.creative_rejection_cleanup import (
            DELETE_AD,
            PERFORMANCE_AD_ZERO_SPEND_RULE,
            determine_ad_performance_cleanup_action,
        )

        action = determine_ad_performance_cleanup_action(
            {
                "configured_status": "AD_STATUS_NORMAL",
                "begin_date": "2026-08-01",
                "end_date": "2026-08-31",
                "created_time": "2026-08-10 00:00:00",
            },
            as_of_date=date(2026, 8, 19),
            cost_fen=0,
            metric_start_date=date(2026, 8, 16),
            metric_end_date=date(2026, 8, 18),
            window_days=3,
        )

        self.assertEqual(action["cleanup_action"], DELETE_AD)
        self.assertEqual(
            action["cleanup_rule_type"],
            PERFORMANCE_AD_ZERO_SPEND_RULE,
        )

    def test_ad_with_any_spend_or_inactive_status_is_not_deleted(self):
        from tools.creative_rejection_cleanup import (
            determine_ad_performance_cleanup_action,
        )

        common = {
            "as_of_date": date(2026, 8, 19),
            "metric_start_date": date(2026, 8, 16),
            "metric_end_date": date(2026, 8, 18),
            "window_days": 3,
        }
        self.assertIsNone(
            determine_ad_performance_cleanup_action(
                {
                    "configured_status": "AD_STATUS_NORMAL",
                    "created_time": "2026-08-10 00:00:00",
                },
                cost_fen=1,
                **common,
            )
        )
        self.assertIsNone(
            determine_ad_performance_cleanup_action(
                {
                    "configured_status": "AD_STATUS_SUSPEND",
                    "created_time": "2026-08-10 00:00:00",
                },
                cost_fen=0,
                **common,
            )
        )
        self.assertIsNone(
            determine_ad_performance_cleanup_action(
                {
                    "configured_status": "AD_STATUS_NORMAL",
                    "created_time": "2026-08-10 00:00:00",
                },
                cost_fen=0,
                current_day_cost_fen=1,
                **common,
            )
        )

    def test_ad_requires_more_than_five_complete_days_and_created_time(self):
        from tools.creative_rejection_cleanup import (
            determine_ad_performance_cleanup_action,
        )

        as_of = datetime(
            2026,
            8,
            19,
            11,
            tzinfo=ZoneInfo("Asia/Shanghai"),
        )
        common = {
            "configured_status": "AD_STATUS_NORMAL",
            "begin_date": "2026-08-01",
            "end_date": "2026-08-31",
        }
        kwargs = {
            "as_of_date": as_of,
            "cost_fen": 0,
            "metric_start_date": date(2026, 8, 16),
            "metric_end_date": date(2026, 8, 18),
            "window_days": 3,
            "min_age_days": 5,
        }

        self.assertIsNone(
            determine_ad_performance_cleanup_action(
                {**common, "created_time": "2026-08-14 11:00:00"},
                **kwargs,
            )
        )
        self.assertIsNone(
            determine_ad_performance_cleanup_action(
                {**common, "created_time": "2026-08-13 11:01:00"},
                **kwargs,
            )
        )
        self.assertIsNotNone(
            determine_ad_performance_cleanup_action(
                {**common, "created_time": "2026-08-13 11:00:00"},
                **kwargs,
            )
        )
        self.assertIsNone(
            determine_ad_performance_cleanup_action(common, **kwargs)
        )

    def test_tencent_created_time_epoch_seconds_are_parsed_in_shanghai(self):
        from tools.creative_rejection_cleanup import (
            _as_shanghai_datetime,
            determine_ad_performance_cleanup_action,
        )

        created = datetime(
            2026,
            8,
            10,
            9,
            30,
            tzinfo=ZoneInfo("Asia/Shanghai"),
        )
        epoch_seconds = int(created.timestamp())
        self.assertEqual(_as_shanghai_datetime(epoch_seconds), created)
        self.assertEqual(_as_shanghai_datetime(epoch_seconds * 1000), created)
        self.assertEqual(
            _as_shanghai_datetime(20260810093000),
            created,
        )
        self.assertIsNone(_as_shanghai_datetime(2))

        action = determine_ad_performance_cleanup_action(
            {
                "configured_status": "AD_STATUS_NORMAL",
                "created_time": epoch_seconds,
            },
            as_of_date=datetime(
                2026,
                8,
                19,
                11,
                tzinfo=ZoneInfo("Asia/Shanghai"),
            ),
            cost_fen=0,
            metric_start_date=date(2026, 8, 16),
            metric_end_date=date(2026, 8, 18),
        )
        self.assertEqual(action["creative_created_at"], created)
        self.assertNotEqual(action["creative_created_at"].year, 1970)

    def test_tencent_ad_fields_include_creation_time_and_deleted_flag(self):
        from tencent_client import AD_FIELDS

        self.assertIn("created_time", AD_FIELDS)
        self.assertIn("is_deleted", AD_FIELDS)

    def test_explicit_deleted_ad_states_are_not_cleanup_candidates(self):
        from tools.creative_rejection_cleanup import (
            determine_ad_performance_cleanup_action,
        )

        common = {
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }
        kwargs = {
            "as_of_date": date(2026, 8, 19),
            "cost_fen": 0,
            "metric_start_date": date(2026, 8, 16),
            "metric_end_date": date(2026, 8, 18),
            "window_days": 3,
        }
        for deleted_state in (
            {"is_deleted": True},
            {"is_delete": 1},
            {"system_status": "ADGROUP_STATUS_DELETED"},
            {"system_status": "SMART_ADGROUP_STATUS_DELETED"},
        ):
            with self.subTest(deleted_state=deleted_state):
                self.assertIsNone(
                    determine_ad_performance_cleanup_action(
                        {**common, **deleted_state},
                        **kwargs,
                    )
                )

    def test_get_ad_retries_exact_id_in_deleted_scope(self):
        from tencent_client import TencentClient

        active_response = Mock()
        active_response.raise_for_status.return_value = None
        active_response.json.return_value = {
            "code": 0,
            "data": {"list": []},
        }
        deleted_response = Mock()
        deleted_response.raise_for_status.return_value = None
        deleted_response.json.return_value = {
            "code": 0,
            "data": {
                "list": [{
                    "adgroup_id": 2,
                    "is_deleted": True,
                    "system_status": "ADGROUP_STATUS_DELETED",
                }]
            },
        }
        client = TencentClient()
        client.session.get = Mock(
            side_effect=[active_response, deleted_response]
        )
        client._common_params = Mock(return_value={"access_token": "token"})

        result = client.get_ad(1, 2)

        self.assertTrue(result["is_deleted"])
        self.assertEqual(client.session.get.call_count, 2)
        first_params = client.session.get.call_args_list[0].kwargs["params"]
        second_params = client.session.get.call_args_list[1].kwargs["params"]
        self.assertEqual(
            json.loads(first_params["filtering"])[0]["values"],
            ["2"],
        )
        self.assertNotIn("is_deleted", first_params)
        self.assertIs(second_params["is_deleted"], True)

    def test_cleanup_day_guard_rejects_a_batch_that_crossed_midnight(self):
        from tools.creative_rejection_cleanup import _is_current_cleanup_day

        self.assertTrue(
            _is_current_cleanup_day(
                date(2026, 8, 19),
                datetime(
                    2026,
                    8,
                    19,
                    23,
                    59,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
            )
        )
        self.assertFalse(
            _is_current_cleanup_day(
                date(2026, 8, 19),
                datetime(
                    2026,
                    8,
                    20,
                    0,
                    1,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
            )
        )

    def test_tencent_ad_delete_uses_delete_endpoint_and_readback(self):
        from tencent_client import TencentClient

        response = Mock(status_code=200, text="ok", headers={})
        response.raise_for_status.return_value = None
        response.json.return_value = {"code": 0, "data": {}}
        client = TencentClient()
        client.session.post = Mock(return_value=response)
        client._common_params = Mock(return_value={"access_token": "token"})
        client._user_token = Mock(return_value="user-token")
        client.get_ad = Mock(
            return_value={
                "configured_status": "AD_STATUS_NORMAL",
                "system_status": "ADGROUP_STATUS_DELETED",
                "is_deleted": False,
            }
        )

        result = client.delete_ad(1, 2)

        self.assertEqual(result["system_status"], "ADGROUP_STATUS_DELETED")
        request = client.session.post.call_args
        self.assertTrue(request.args[0].endswith("/adgroups/delete"))
        self.assertEqual(request.kwargs["json"]["adgroup_id"], 2)

    def test_tencent_ad_delete_accepts_is_deleted_flag(self):
        from tencent_client import TencentClient

        response = Mock(status_code=200, text="ok", headers={})
        response.raise_for_status.return_value = None
        response.json.return_value = {"code": 0, "data": {}}
        client = TencentClient()
        client.session.post = Mock(return_value=response)
        client._common_params = Mock(return_value={"access_token": "token"})
        client._user_token = Mock(return_value="user-token")
        client.get_ad = Mock(
            return_value={
                "configured_status": "AD_STATUS_NORMAL",
                "system_status": "ADGROUP_STATUS_ACTIVE",
                "is_deleted": True,
            }
        )

        result = client.delete_ad(1, 2)

        self.assertTrue(result["is_deleted"])

    def test_tencent_ad_delete_does_not_infer_deletion_from_missing_row(self):
        from tencent_client import PostWriteVerificationError, TencentClient

        response = Mock(status_code=200, text="ok", headers={})
        response.raise_for_status.return_value = None
        response.json.return_value = {"code": 0, "data": {}}
        client = TencentClient()
        client.verify_attempts = 1
        client.session.post = Mock(return_value=response)
        client._common_params = Mock(return_value={"access_token": "token"})
        client._user_token = Mock(return_value="user-token")
        client.get_ad = Mock(
            side_effect=RuntimeError(
                "Ad not found after update: account=1 adgroup=2"
            )
        )

        with self.assertRaises(PostWriteVerificationError):
            client.delete_ad(1, 2)


class CreativePerformanceNotificationTests(unittest.TestCase):
    def test_internal_only_agency_uses_internal_route_without_webhook(self):
        from tools import creative_rejection_cleanup as cleanup

        with patch.dict(
            os.environ,
            {
                "CREATIVE_CLEANUP_INTERNAL_ONLY_AGENCIES": "小程序-自动化",
            },
            clear=False,
        ):
            self.assertTrue(
                cleanup._has_cleanup_notification_route("小程序-自动化", {})
            )
            self.assertFalse(
                cleanup._has_cleanup_notification_route("其他代理", {})
            )

    def test_preview_notification_excludes_candidates_not_confirmed_this_run(self):
        from tools import creative_rejection_cleanup as cleanup

        current_row = {
            "id": 1,
            "check_date": date(2026, 8, 20),
            "account_id": 1,
            "dynamic_creative_id": 3,
            "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
            "operator_notified_at": datetime(2026, 8, 20, 11),
        }
        current_ad_row = {
            "id": 2,
            "check_date": date(2026, 8, 20),
            "account_id": 1,
            "dynamic_creative_id": -2,
            "cleanup_rule_type": cleanup.PERFORMANCE_AD_ZERO_SPEND_RULE,
            "creative_created_at": datetime(2026, 8, 10, 10),
            "operator_notified_at": None,
        }
        stale_bad_timestamp_row = {
            **current_ad_row,
            "id": 3,
            "dynamic_creative_id": -4,
            "creative_created_at": datetime(1970, 1, 1, 0, 0, 2),
        }

        selected = cleanup.filter_preview_notification_rows(
            [current_row, current_ad_row, stale_bad_timestamp_row],
            check_date=date(2026, 8, 20),
            confirmed_actions={
                (1, 3): {
                    "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
                },
                (1, -2): {
                    "cleanup_rule_type": (
                        cleanup.PERFORMANCE_AD_ZERO_SPEND_RULE
                    ),
                }
            },
            force_notification=True,
        )

        self.assertEqual([row["id"] for row in selected], [1, 2])
        self.assertIsNone(selected[0]["operator_notified_at"])

    def test_performance_rows_only_enter_the_separate_internal_delivery(self):
        from tools.creative_rejection_cleanup import (
            PERFORMANCE_AD_ZERO_SPEND_RULE,
            PERFORMANCE_NEW_RULE,
            REVIEW_DENIED_RULE,
            split_notification_rows,
        )

        review_row = {
            "id": 1,
            "agency_name": "代理一",
            "cleanup_rule_type": REVIEW_DENIED_RULE,
            "agency_notified_at": None,
            "operator_notified_at": None,
        }
        performance_row = {
            "id": 2,
            "agency_name": "",
            "cleanup_rule_type": PERFORMANCE_NEW_RULE,
            "agency_notified_at": datetime(2026, 8, 19, 10),
            "operator_notified_at": None,
        }
        ad_row = {
            "id": 3,
            "agency_name": "",
            "cleanup_rule_type": PERFORMANCE_AD_ZERO_SPEND_RULE,
            "agency_notified_at": datetime(2026, 8, 19, 10),
            "operator_notified_at": None,
        }
        internal_only_review_row = {
            "id": 4,
            "agency_name": "小程序-自动化",
            "cleanup_rule_type": REVIEW_DENIED_RULE,
            "agency_notified_at": None,
            "operator_notified_at": None,
        }

        with patch.dict(
            os.environ,
            {
                "CREATIVE_CLEANUP_INTERNAL_ONLY_AGENCIES": "小程序-自动化",
            },
            clear=False,
        ):
            agency, review_internal, performance_internal = split_notification_rows(
                [review_row, performance_row, ad_row, internal_only_review_row]
            )

        self.assertEqual(agency, [review_row])
        self.assertEqual(review_internal, [review_row, internal_only_review_row])
        self.assertEqual(performance_internal, [performance_row, ad_row])
        legacy_agency_rows = [
            row
            for row in [review_row, performance_row, ad_row]
            if row.get("agency_notified_at") is None
            and str(row.get("agency_name") or "").strip()
        ]
        self.assertEqual(legacy_agency_rows, [review_row])

    def test_performance_upsert_atomically_suppresses_agency_delivery(self):
        from tools import creative_rejection_cleanup as cleanup

        class Cursor:
            def __init__(self):
                self.calls = []

            def __enter__(self):
                return self

            def __exit__(self, *_args):
                return None

            def execute(self, sql, params=None):
                self.calls.append((sql, params))

            def fetchone(self):
                return {"id": 7}

        cursor = Cursor()
        connection = Mock()
        connection.cursor.return_value = cursor
        with patch.object(cleanup, "get_connection", return_value=connection):
            cleanup.upsert_cleanup_candidate({
                "account_id": 1,
                "account_name": "账户一",
                "agency_name": "代理一",
                "adgroup_id": 2,
                "adgroup_name": "广告二",
                "dynamic_creative_id": 3,
                "dynamic_creative_name": "创意三",
                "check_date": date(2026, 8, 19),
                "cleanup_action": cleanup.DELETE_CREATIVE,
                "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
                "component_ids": [],
                "element_ids": [],
                "recent_cost_fen": 0,
                "cost_start_date": date(2026, 8, 12),
                "cost_end_date": date(2026, 8, 18),
                "action_reason": "未起量",
                "reject_reason": "",
                "review_result": {},
                "pre_state": {},
            })

        insert_sql, insert_params = cursor.calls[0]
        self.assertEqual(insert_params[3], "")
        self.assertEqual(insert_params[10], cleanup.PERFORMANCE_NEW_RULE)
        self.assertIsInstance(insert_params[-1], datetime)
        self.assertIn("agency_notified_at", insert_sql)
        self.assertIn("VALUES(cleanup_rule_type)", insert_sql)
        normalized_insert_sql = " ".join(insert_sql.split())
        terminal_guard = (
            "WHEN cleanup_status IN ( "
            "'DELETING','CREATIVE_DELETED','AD_DELETED' )"
        )
        self.assertEqual(normalized_insert_sql.count(terminal_guard), 2)
        self.assertIn(f"{terminal_guard} THEN agency_name", normalized_insert_sql)
        self.assertIn(
            f"{terminal_guard} THEN agency_notified_at",
            normalized_insert_sql,
        )
        self.assertIn("performance_suppression_at", cursor.calls[1][0])
        self.assertIn("performance_suppression_at", cursor.calls[2][0])
        extended_sql = cursor.calls[2][0]
        self.assertIn(
            "item.creative_created_at <=> incoming.creative_created_at",
            extended_sql,
        )
        self.assertIn(
            "item.agent_name <=> COALESCE(NULLIF(incoming.agent_name,''), item.agent_name)",
            extended_sql,
        )

    def test_schema_backfill_suppresses_existing_performance_rows(self):
        from storage import _suppress_performance_agency_delivery

        cursor = Mock()
        _suppress_performance_agency_delivery(cursor)

        sql = " ".join(cursor.execute.call_args.args[0].split())
        self.assertIn("SET agency_name=''", sql)
        self.assertIn(
            "agency_notified_at=COALESCE(agency_notified_at, NOW())",
            sql,
        )
        self.assertIn(
            "WHERE LEFT(cleanup_rule_type, 12)='PERFORMANCE_'",
            sql,
        )

    def test_schema_backfill_preserves_legacy_partial_review_rule(self):
        from storage import _backfill_legacy_review_rule_types

        cursor = Mock()
        _backfill_legacy_review_rule_types(cursor)

        sql = " ".join(cursor.execute.call_args.args[0].split())
        self.assertIn("SET cleanup_rule_type='REVIEW_PARTIAL'", sql)
        self.assertIn("cleanup_action='ALERT_ONLY'", sql)
        self.assertIn(
            "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL",
            sql,
        )

    def test_performance_report_has_its_own_internal_title_and_sheet(self):
        from openpyxl import load_workbook

        from tools.creative_rejection_cleanup import (
            DELETE_AD,
            PERFORMANCE_AD_ZERO_SPEND_RULE,
            PERFORMANCE_NEW_RULE,
            write_performance_operator_summary,
        )

        row = {
            "id": 2,
            "agency_name": "代理一",
            "account_id": 1,
            "account_name": "账户一",
            "agent_name": "代理商甲",
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": "DELETE_CREATIVE",
            "cleanup_rule_type": PERFORMANCE_NEW_RULE,
            "cleanup_status": "DISCOVERED",
            "reject_reason": "",
            "review_result_json": "{}",
            "pre_state_json": "{}",
            "creative_age_days": 6,
            "metric_impressions": 99,
            "metric_daily_avg_impressions": 99 / 7,
            "recent_cost_fen": 0,
            "cost_start_date": "2026-08-12",
            "cost_end_date": "2026-08-18",
        }
        ad_row = {
            "id": 3,
            "agency_name": "",
            "account_id": 1,
            "account_name": "账户一",
            "agent_name": "代理商甲",
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "dynamic_creative_id": -2,
            "dynamic_creative_name": "",
            "cleanup_action": DELETE_AD,
            "cleanup_rule_type": PERFORMANCE_AD_ZERO_SPEND_RULE,
            "cleanup_status": "DISCOVERED",
            "reject_reason": "",
            "review_result_json": "{}",
            "pre_state_json": '{"configured_status":"AD_STATUS_NORMAL"}',
            "recent_cost_fen": 0,
            "metric_window_days": 3,
            "cost_start_date": "2026-08-16",
            "cost_end_date": "2026-08-18",
        }
        with TemporaryDirectory() as directory:
            report = write_performance_operator_summary(
                [row, ad_row],
                Path(directory),
                "20260819",
            )
            workbook = load_workbook(report["report"])

        self.assertEqual(
            report["title"],
            "长期未起量创意及广告清理汇总通知",
        )
        self.assertEqual(report["notification_kind"], "performance_internal")
        self.assertLessEqual(len(str(report["report_version"])), 64)
        self.assertEqual(report["creative_rows"], 1)
        self.assertEqual(report["ad_rows"], 1)
        self.assertIn("长期未起量清理", workbook.sheetnames)
        sheet = workbook["长期未起量清理"]
        self.assertEqual(sheet["A1"].value, "清理对象")
        self.assertEqual(sheet["D1"].value, "代理商昵称")
        self.assertEqual([sheet["D2"].value, sheet["D3"].value], ["代理商甲"] * 2)
        self.assertEqual([sheet["A2"].value, sheet["A3"].value], ["创意", "广告"])


class CreativePerformanceIntegrationTests(unittest.TestCase):
    def setUp(self):
        flags = patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED": "0",
                "DAILY_UNDERPERFORMING_AD_APPLY_ENABLED": "0",
            },
            clear=False,
        )
        flags.start()
        self.addCleanup(flags.stop)
        account_metadata = patch(
            "tools.creative_rejection_cleanup.fetch_tencent_account_metadata",
            return_value={},
        )
        account_metadata.start()
        self.addCleanup(account_metadata.stop)

    def test_ad_candidate_is_created_after_creative_judgment(self):
        import pandas as pd

        from tools import creative_rejection_cleanup as cleanup

        current = datetime(2026, 8, 19, 11, tzinfo=ZoneInfo("Asia/Shanghai"))
        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [
            {
                "dynamic_creative_id": 3,
                "dynamic_creative_name": "创意三",
                "adgroup_id": 2,
                "configured_status": "AD_STATUS_NORMAL",
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_NORMAL"
                ),
                "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
            },
            {
                "dynamic_creative_id": 4,
                "dynamic_creative_name": "拒审创意四",
                "adgroup_id": 2,
                "configured_status": "AD_STATUS_NORMAL",
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_DENIED"
                ),
                "system_status": "DYNAMIC_CREATIVE_STATUS_DENIED",
            },
            {
                "dynamic_creative_id": 5,
                "dynamic_creative_name": "已删除创意五",
                "adgroup_id": 2,
                "configured_status": "AD_STATUS_DELETED",
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_DENIED"
                ),
                "system_status": "DYNAMIC_CREATIVE_STATUS_DELETED",
            },
        ]
        tencent.get_ads.return_value = [{
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }]
        tencent.get_dynamic_creative_metrics.side_effect = [
            {3: {"impressions": 99, "cost_fen": 0}},
            {3: {"impressions": 0, "cost_fen": 0}},
        ]
        tencent.get_ad_metrics.return_value = {
            2: {"impressions": 0, "cost_fen": 0}
        }
        captured = []
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_AD_APPLY_ENABLED": "0",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260818"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup,
            "fetch_recent_spend_accounts",
            return_value=[{"account_id": 1, "account_name": "账户一"}],
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[{
                "account_id": 1,
                "adgroup_id": 2,
                "creative_id": 3,
                "create_time": datetime(
                    2026,
                    8,
                    13,
                    10,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
            }],
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token"}
        ), patch.object(
            cleanup,
            "upsert_cleanup_candidate",
            side_effect=lambda record: captured.append(record) or {"id": 1},
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[]
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=tencent,
                odps=Mock(),
                review_fetcher=Mock(return_value=[]),
            )

        self.assertEqual(
            [record["cleanup_rule_type"] for record in captured],
            [
                cleanup.REVIEW_DENIED_RULE,
                cleanup.PERFORMANCE_NEW_RULE,
                cleanup.PERFORMANCE_AD_ZERO_SPEND_RULE,
            ],
        )
        self.assertEqual(captured[2]["dynamic_creative_id"], -2)
        self.assertNotIn(5, [record["dynamic_creative_id"] for record in captured])
        self.assertEqual(summary["rejected_discovered"], 1)
        self.assertEqual(summary["ad_discovered"], 1)

    def test_combined_task_discovers_performance_candidate_without_applying(self):
        import pandas as pd

        from tools import creative_rejection_cleanup as cleanup

        current = datetime(2026, 8, 19, 11, tzinfo=ZoneInfo("Asia/Shanghai"))
        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [
            {
                "dynamic_creative_id": 3,
                "dynamic_creative_name": "创意三",
                "adgroup_id": 2,
                "configured_status": "AD_STATUS_NORMAL",
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_NORMAL"
                ),
                "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
            }
        ]
        tencent.get_ads.return_value = [
            {
                "adgroup_id": 2,
                "adgroup_name": "广告二",
                "configured_status": "AD_STATUS_NORMAL",
            }
        ]
        tencent.get_dynamic_creative_metrics.side_effect = [
            {3: {"impressions": 99, "cost_fen": 0}},
            {3: {"impressions": 0, "cost_fen": 0}},
        ]
        review_fetcher = Mock(return_value=[])
        captured = []
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "0",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260818"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup,
            "fetch_recent_spend_accounts",
            return_value=[],
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[{
                "account_id": 1,
                "adgroup_id": 2,
                "creative_id": 3,
                "create_time": datetime(
                    2026,
                    8,
                    13,
                    10,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
            }],
        ), patch.object(
            cleanup,
            "fetch_tencent_account_metadata",
            return_value={
                1: {
                    "account_name": "大数据账户一",
                    "agent_name": "代理商甲",
                }
            },
        ), patch.object(
            cleanup,
            "prefetch_account_access_tokens",
            return_value={1: "access-token"},
        ), patch.object(
            cleanup,
            "upsert_cleanup_candidate",
            side_effect=lambda record: captured.append(record) or {"id": 1},
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[]
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=tencent,
                odps=Mock(),
                review_fetcher=review_fetcher,
            )

        self.assertEqual(len(captured), 1)
        self.assertEqual(
            captured[0]["cleanup_rule_type"],
            cleanup.PERFORMANCE_NEW_RULE,
        )
        self.assertEqual(captured[0]["account_name"], "大数据账户一")
        self.assertEqual(captured[0]["agent_name"], "代理商甲")
        self.assertEqual(summary["performance_discovered"], 1)
        self.assertEqual(summary["rejected_discovered"], 0)
        self.assertEqual(summary["review_account_ids"], [])
        self.assertEqual(summary["performance_account_ids"], [1])
        review_fetcher.assert_not_called()
        tencent.delete_dynamic_creative.assert_not_called()

    def test_current_day_spend_blocks_creative_and_ad_candidates(self):
        import pandas as pd

        from tools import creative_rejection_cleanup as cleanup

        current = datetime(2026, 8, 19, 11, tzinfo=ZoneInfo("Asia/Shanghai"))
        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [{
            "dynamic_creative_id": 3,
            "dynamic_creative_name": "创意三",
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "creative_set_approval_status": cleanup.CREATIVE_NORMAL_STATUS,
            "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
        }]
        tencent.get_ads.return_value = [{
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }]
        tencent.get_dynamic_creative_metrics.side_effect = [
            {3: {"impressions": 99, "cost_fen": 0}},
            {3: {"impressions": 1, "cost_fen": 1}},
        ]
        tencent.get_ad_metrics.side_effect = [
            {2: {"impressions": 0, "cost_fen": 0}},
            {2: {"impressions": 1, "cost_fen": 1}},
        ]
        captured = []
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_AD_APPLY_ENABLED": "0",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260818"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[]
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[{
                "account_id": 1,
                "adgroup_id": 2,
                "creative_id": 3,
                "create_time": datetime(
                    2026,
                    8,
                    13,
                    10,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
            }],
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token"}
        ), patch.object(
            cleanup,
            "upsert_cleanup_candidate",
            side_effect=lambda record: captured.append(record) or {"id": 1},
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[]
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=tencent,
                odps=Mock(),
                review_fetcher=Mock(return_value=[]),
            )

        self.assertEqual(captured, [])
        self.assertEqual(summary["performance_discovered"], 0)
        self.assertEqual(summary["ad_discovered"], 0)
        self.assertEqual(summary["current_day_metric_date"], date(2026, 8, 19))

    def test_review_sources_and_creative_endpoint_can_fail_while_ad_continues(self):
        from tools import creative_rejection_cleanup as cleanup

        current = datetime(2026, 8, 19, 11, tzinfo=ZoneInfo("Asia/Shanghai"))
        tencent = Mock()
        tencent.get_dynamic_creatives.side_effect = RuntimeError(
            "creative endpoint unavailable"
        )
        tencent.get_ads.return_value = [{
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }]
        tencent.get_ad_metrics.return_value = {
            2: {"impressions": 0, "cost_fen": 0}
        }
        captured = []
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup,
            "resolve_end_date",
            side_effect=RuntimeError("review source unavailable"),
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[{
                "account_id": 1,
                "adgroup_id": 2,
                "creative_id": 3,
                "create_time": datetime(
                    2026,
                    8,
                    13,
                    10,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
            }],
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token"}
        ), patch.object(
            cleanup,
            "upsert_cleanup_candidate",
            side_effect=lambda record: captured.append(record) or {"id": 1},
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[]
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=tencent,
                odps=Mock(),
            )

        self.assertEqual(
            [row["cleanup_rule_type"] for row in captured],
            [cleanup.PERFORMANCE_AD_ZERO_SPEND_RULE],
        )
        self.assertEqual(summary["ad_discovered"], 1)
        self.assertEqual(summary["performance_discovered"], 1)
        self.assertIn("review source unavailable", summary["review_scope_error"])
        self.assertTrue(summary["scan_errors"])
        tencent.get_dynamic_creative_metrics.assert_not_called()

    def test_performance_apply_does_not_require_agency_route(self):
        import pandas as pd

        from tools import creative_rejection_cleanup as cleanup

        @contextmanager
        def acquired_lock(_name):
            yield True

        current = datetime(2026, 8, 19, 11, tzinfo=ZoneInfo("Asia/Shanghai"))
        created_at = datetime(
            2026,
            8,
            13,
            10,
            tzinfo=ZoneInfo("Asia/Shanghai"),
        )
        creative = {
            "dynamic_creative_id": 3,
            "dynamic_creative_name": "创意三",
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "creative_set_approval_status": (
                "CREATIVE_SET_APPROVAL_STATUS_NORMAL"
            ),
            "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
        }
        ad = {
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "configured_status": "AD_STATUS_NORMAL",
        }
        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [creative]
        tencent.get_ads.return_value = [ad]
        tencent.get_dynamic_creative.return_value = creative
        tencent.get_ad.return_value = ad
        tencent.get_dynamic_creative_metrics.side_effect = [
            {3: {"impressions": 99, "cost_fen": 0}},
            {3: {"impressions": 0, "cost_fen": 0}},
            {3: {"impressions": 99, "cost_fen": 0}},
            {3: {"impressions": 0, "cost_fen": 0}},
        ]
        tencent.delete_dynamic_creative.return_value = {
            "dynamic_creative_id": 3,
            "configured_status": "AD_STATUS_DELETED",
        }
        retry_item = {
            "id": 7,
            "account_id": 1,
            "account_name": "账户一",
            "agency_name": "",
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "dynamic_creative_id": 3,
            "dynamic_creative_name": "创意三",
            "cleanup_action": cleanup.DELETE_CREATIVE,
            "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
            "cleanup_status": "DISCOVERED",
            "creative_created_at": created_at,
        }
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260818"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[]
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[{
                "account_id": 1,
                "adgroup_id": 2,
                "creative_id": 3,
                "create_time": created_at,
            }],
        ), patch.object(
            cleanup,
            "prefetch_account_access_tokens",
            return_value={1: "access-token"},
        ), patch.object(
            cleanup, "upsert_cleanup_candidate"
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
        ), patch.object(
            cleanup, "claim_cleanup_item", return_value=True
        ), patch.object(
            cleanup, "update_cleanup_item", return_value=True
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ), patch.object(
            cleanup, "advisory_lock", side_effect=acquired_lock
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=tencent,
                odps=Mock(),
                review_fetcher=Mock(return_value=[]),
            )

        tencent.delete_dynamic_creative.assert_called_once_with(1, 3)
        self.assertEqual(summary["deleted"], 1)
        self.assertEqual(summary["deferred"], 0)

    def test_zero_spend_ad_apply_runs_after_reconfirmation(self):
        import pandas as pd

        from tools import creative_rejection_cleanup as cleanup

        @contextmanager
        def acquired_lock(_name):
            yield True

        current = datetime(2026, 8, 19, 11, tzinfo=ZoneInfo("Asia/Shanghai"))
        ad = {
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }
        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [{
            "dynamic_creative_id": 3,
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "creative_set_approval_status": (
                "CREATIVE_SET_APPROVAL_STATUS_PENDING"
            ),
            "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
        }]
        tencent.get_ads.return_value = [ad]
        tencent.get_ad_metrics.return_value = {
            2: {"impressions": 0, "cost_fen": 0}
        }
        tencent.get_ad.return_value = ad
        tencent.delete_ad.return_value = {
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_DELETED",
        }
        retry_item = {
            "id": 9,
            "account_id": 1,
            "account_name": "账户一",
            "agency_name": "",
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "dynamic_creative_id": -2,
            "dynamic_creative_name": "",
            "cleanup_action": cleanup.DELETE_AD,
            "cleanup_rule_type": cleanup.PERFORMANCE_AD_ZERO_SPEND_RULE,
            "cleanup_status": "DISCOVERED",
        }
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_AD_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260818"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[]
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[{
                "account_id": 1,
                "adgroup_id": 2,
                "creative_id": 3,
                "create_time": datetime(
                    2026,
                    8,
                    10,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
            }],
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token"}
        ), patch.object(
            cleanup, "upsert_cleanup_candidate"
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
        ), patch.object(
            cleanup, "claim_cleanup_item", return_value=True
        ), patch.object(
            cleanup, "update_cleanup_item", return_value=True
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ), patch.object(
            cleanup, "advisory_lock", side_effect=acquired_lock
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=tencent,
                odps=Mock(),
                review_fetcher=Mock(return_value=[]),
            )

        tencent.delete_ad.assert_called_once_with(1, 2)
        self.assertEqual(summary["deleted"], 1)
        self.assertEqual(summary["ad_discovered"], 1)

    def test_write_lock_recheck_blocks_deletes_when_today_spend_appears(self):
        import pandas as pd

        from tools import creative_rejection_cleanup as cleanup

        @contextmanager
        def acquired_lock(_name):
            yield True

        current = datetime(2026, 8, 19, 11, tzinfo=ZoneInfo("Asia/Shanghai"))
        created_at = datetime(
            2026,
            8,
            13,
            10,
            tzinfo=ZoneInfo("Asia/Shanghai"),
        )
        creative = {
            "dynamic_creative_id": 3,
            "dynamic_creative_name": "创意三",
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "creative_set_approval_status": cleanup.CREATIVE_NORMAL_STATUS,
            "system_status": "DYNAMIC_CREATIVE_STATUS_NORMAL",
        }
        ad = {
            "adgroup_id": 2,
            "adgroup_name": "广告二",
            "configured_status": "AD_STATUS_NORMAL",
            "created_time": "2026-08-10 00:00:00",
        }
        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [creative]
        tencent.get_ads.return_value = [ad]
        tencent.get_dynamic_creative.return_value = creative
        tencent.get_ad.return_value = ad
        tencent.get_dynamic_creative_metrics.side_effect = [
            {3: {"impressions": 99, "cost_fen": 0}},
            {3: {"impressions": 0, "cost_fen": 0}},
            {3: {"impressions": 99, "cost_fen": 0}},
            {3: {"impressions": 1, "cost_fen": 0}},
        ]
        tencent.get_ad_metrics.side_effect = [
            {2: {"impressions": 0, "cost_fen": 0}},
            {2: {"impressions": 0, "cost_fen": 0}},
            {2: {"impressions": 0, "cost_fen": 0}},
            {2: {"impressions": 0, "cost_fen": 1}},
        ]
        retry_items = [
            {
                "id": 7,
                "account_id": 1,
                "agency_name": "",
                "adgroup_id": 2,
                "dynamic_creative_id": 3,
                "cleanup_action": cleanup.DELETE_CREATIVE,
                "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
                "cleanup_status": "DISCOVERED",
                "creative_created_at": created_at,
            },
            {
                "id": 8,
                "account_id": 1,
                "agency_name": "",
                "adgroup_id": 2,
                "dynamic_creative_id": -2,
                "cleanup_action": cleanup.DELETE_AD,
                "cleanup_rule_type": cleanup.PERFORMANCE_AD_ZERO_SPEND_RULE,
                "cleanup_status": "DISCOVERED",
            },
        ]
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "1",
                "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_AD_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260818"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[]
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[{
                "account_id": 1,
                "adgroup_id": 2,
                "creative_id": 3,
                "create_time": created_at,
            }],
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token"}
        ), patch.object(
            cleanup, "upsert_cleanup_candidate"
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=retry_items
        ), patch.object(
            cleanup, "claim_cleanup_item", return_value=True
        ), patch.object(
            cleanup, "update_cleanup_item", return_value=True
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ), patch.object(
            cleanup, "advisory_lock", side_effect=acquired_lock
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=tencent,
                odps=Mock(),
                review_fetcher=Mock(return_value=[]),
            )

        tencent.delete_dynamic_creative.assert_not_called()
        tencent.delete_ad.assert_not_called()
        self.assertEqual(summary["deleted"], 0)
        self.assertEqual(tencent.get_dynamic_creative_metrics.call_count, 4)
        self.assertEqual(tencent.get_ad_metrics.call_count, 4)

    def test_performance_notification_sends_internal_message_without_webhook(self):
        import pandas as pd

        from tools import creative_rejection_cleanup as cleanup

        @contextmanager
        def acquired_lock(_name):
            yield True

        pending = {
            "id": 7,
            "check_date": date(2026, 8, 19),
            "account_id": 1,
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "agency_name": "代理一",
            "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
            "cleanup_status": "DISCOVERED",
            "operator_notified_at": None,
            "agency_notified_at": None,
        }
        performance_report = {
            "run_id": "performance_20260819_test",
            "report": "unused.xlsx",
            "title": "长期未起量创意清理汇总通知",
            "notification_kind": "performance_internal",
            "creative_rows": 1,
        }
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED": "1",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "0",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260818"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[]
        ), patch.object(
            cleanup, "fetch_active_creative_inventory", return_value=[]
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={}
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[]
        ), patch.object(
            cleanup,
            "load_unnotified_deleted_items",
            side_effect=[[pending], [pending]],
        ), patch.object(
            cleanup, "advisory_lock", side_effect=acquired_lock
        ), patch.object(
            cleanup,
            "write_performance_operator_summary",
            return_value=performance_report,
        ), patch.object(
            cleanup,
            "publish_cleanup_operator_summary",
            return_value={"route": "长期未起量清理汇总", "status": "SENT"},
        ) as publish_internal, patch.object(
            cleanup, "publish_agency_reports"
        ) as publish_agency, patch.object(
            cleanup, "mark_cleanup_items_operator_notified"
        ) as mark_internal:
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=datetime(
                    2026,
                    8,
                    19,
                    11,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
                tencent=Mock(),
                odps=Mock(),
                publisher=Mock(),
            )

        publish_agency.assert_not_called()
        publish_internal.assert_called_once()
        self.assertEqual(
            publish_internal.call_args.kwargs["chat_id"],
            "chat-internal",
        )
        mark_internal.assert_called_once_with(
            [7],
            datetime(2026, 8, 19, 11, tzinfo=ZoneInfo("Asia/Shanghai")),
        )

    def test_force_preview_resends_already_notified_performance_rows(self):
        from tools import creative_rejection_cleanup as cleanup

        @contextmanager
        def acquired_lock(_name):
            yield True

        current = datetime(
            2026, 8, 20, 16, 30, tzinfo=ZoneInfo("Asia/Shanghai")
        )
        pending = {
            "id": 9,
            "check_date": current.date(),
            "account_id": 1,
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
            "cleanup_status": "DISCOVERED",
            "operator_notified_at": datetime(2026, 8, 20, 11),
            "agency_notified_at": datetime(2026, 8, 20, 11),
        }
        report = {
            "run_id": "performance_2026-08-20_digest",
            "report": "unused.xlsx",
            "title": "长期未起量创意及广告清理汇总通知",
            "notification_kind": "performance_internal",
            "creative_rows": 1,
            "ad_rows": 0,
        }
        with TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {"FEISHU_AD_PROJECT_CHAT_ID": "chat-internal"},
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "fetch_active_creative_inventory", return_value=[]
        ), patch.object(
            cleanup, "fetch_tencent_account_metadata", return_value={}
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={}
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[]
        ), patch.object(
            cleanup,
            "load_unnotified_deleted_items",
            side_effect=[[pending], [pending]],
        ) as load_notifications, patch.object(
            cleanup,
            "filter_preview_notification_rows",
            side_effect=lambda rows, **_kwargs: [
                {**row, "operator_notified_at": None} for row in rows
            ],
        ), patch.object(
            cleanup, "advisory_lock", side_effect=acquired_lock
        ), patch.object(
            cleanup,
            "write_performance_operator_summary",
            return_value=report,
        ), patch.object(
            cleanup,
            "publish_cleanup_operator_summary",
            return_value={"route": "长期未起量清理汇总", "status": "SENT"},
        ) as publish, patch.object(
            cleanup, "mark_cleanup_items_operator_notified"
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=Mock(),
                odps=Mock(),
                publisher=Mock(),
                underperformance_preview_only=True,
                force_notification=True,
            )

        self.assertTrue(summary["force_notification"])
        self.assertTrue(
            publish.call_args.kwargs["run_id"].startswith(
                "performance_2026-08-20_digest_force_"
            )
        )
        self.assertIn("手动重发", publish.call_args.kwargs["report"]["title"])
        for call in load_notifications.call_args_list:
            self.assertTrue(call.kwargs["include_notified"])
        self.assertEqual(summary["notification_errors"], [])


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