Agent/examples/auto_put_ad_mini/test_creative_review_scan.py

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

import pandas as pd
from openpyxl import load_workbook


class CreativeReviewScanTests(unittest.TestCase):
    def test_tencent_get_retries_transient_tls_connection_failure(self):
        from tools import ad_api

        response = Mock()
        response.raise_for_status.return_value = None
        response.json.return_value = {"code": 0, "data": {"list": []}}
        connection_error = ad_api.httpx.ConnectError("temporary TLS EOF")

        with patch.dict(
            os.environ,
            {
                "TENCENT_AD_GET_RETRY_ATTEMPTS": "3",
                "TENCENT_AD_GET_RETRY_BACKOFF_SECONDS": "0",
            },
            clear=False,
        ), patch.object(
            ad_api, "_common_params", side_effect=[{"nonce": "1"}, {"nonce": "2"}]
        ) as common_params, patch.object(
            ad_api.httpx,
            "get",
            side_effect=[connection_error, response],
            create=True,
        ) as get:
            payload = ad_api._get("/test/get", {"account_id": 123})

        self.assertEqual(payload, {"code": 0, "data": {"list": []}})
        self.assertEqual(get.call_count, 2)
        self.assertEqual(common_params.call_count, 2)

    def test_access_tokens_are_prefetched_concurrently_and_deduplicated(self):
        import threading

        from tools import ad_api

        barrier = threading.Barrier(3, timeout=2)

        def fetch(account_id):
            barrier.wait()
            return f"access-token-{account_id}"

        with patch.dict(ad_api._token_cache, {}, clear=True), patch.object(
            ad_api,
            "_get_access_token",
            side_effect=fetch,
        ) as get_token:
            tokens = ad_api.prefetch_access_tokens([3, 1, 2, 2], max_workers=3)
            cached_accounts = set(ad_api._token_cache)

        self.assertEqual(tokens, {
            1: "access-token-1",
            2: "access-token-2",
            3: "access-token-3",
        })
        self.assertEqual(get_token.call_count, 3)
        self.assertEqual(cached_accounts, {1, 2, 3})

    def test_parse_rejected_result_collects_reasons_and_locations(self):
        from tools.creative_review import parse_review_result

        parsed = parse_review_result({
            "dynamic_creative_id": 123,
            "reject_message_list": ["整体拒绝"],
            "site_set_result_list": [
                {
                    "site_set": "SITE_SET_MOMENTS",
                    "system_status": "DYNAMIC_CREATIVE_STATUS_DENIED",
                    "reject_message": "朋友圈拒绝",
                }
            ],
            "element_result_list": [
                {
                    "element_name": "image",
                    "element_type": "ELEMENT_TYPE_IMAGE",
                    "review_status": "REVIEW_STATUS_REJECTED",
                    "element_reject_detail_info": [
                        {
                            "reason": "图片含违规信息",
                            "reject_info_location": [
                                {"x": 1, "y": 2, "width": 3, "height": 4}
                            ],
                        }
                    ],
                }
            ],
        })

        self.assertEqual(parsed.review_status, "rejected")
        self.assertIn("整体拒绝", parsed.reject_messages)
        self.assertIn("朋友圈拒绝", parsed.reject_messages)
        self.assertIn("图片含违规信息", parsed.reject_messages)
        self.assertEqual(len(parsed.rejection_facts), 2)

    def test_parse_pending_result(self):
        from tools.creative_review import parse_review_result

        parsed = parse_review_result({
            "dynamic_creative_id": 456,
            "is_all_component_compose_pending": True,
            "delay_message_list": ["审核延迟"],
        })

        self.assertEqual(parsed.review_status, "pending")
        self.assertIn("审核延迟", parsed.delay_messages)

    def test_review_granularity_fields_lists_element_and_site_results(self):
        from tools.creative_review import review_granularity_fields

        fields = review_granularity_fields(
            {
                "element_result_list": [
                    {
                        "element_name": "主图",
                        "image_id": "10001",
                        "review_status": "REVIEW_STATUS_REJECTED",
                        "element_reject_detail_info": [
                            {"reason": "图片违规"},
                            {"reason": "文字违规"},
                        ],
                    },
                    {
                        "element_type": "ELEMENT_TYPE_VIDEO",
                        "video_id": "20002",
                        "review_status": "REVIEW_STATUS_APPROVED",
                    },
                    {
                        "element_name": "正常元素",
                        "element_id": "30003",
                        "system_status": "AD_STATUS_NORMAL",
                    },
                ],
                "site_set_result_list": [
                    {
                        "site_set": "SITE_SET_MOMENTS",
                        "site_set_id": "40004",
                        "system_status": "DYNAMIC_CREATIVE_STATUS_DENIED",
                        "reject_message": "朋友圈版位拒绝",
                        "element_reject_detail_info": [
                            {"reason": "版位素材不适配"}
                        ],
                    },
                    {
                        "site_set": "SITE_SET_WECHAT_CHANNELS",
                        "review_status": "REVIEW_STATUS_APPROVED",
                    },
                    {
                        "site_set": "SITE_SET_NORMAL",
                        "review_status": "AD_STATUS_NORMAL",
                    },
                ],
            }
        )

        self.assertEqual(
            fields["element_review_status"],
            "主图(id=10001): 审核拒绝\nELEMENT_TYPE_VIDEO: 审核通过",
        )
        self.assertEqual(
            fields["element_reject_reason"],
            "主图(id=10001): 图片违规\n主图(id=10001): 文字违规",
        )
        self.assertEqual(
            fields["site_review_status"],
            "SITE_SET_MOMENTS(id=40004): 审核拒绝\n"
            "SITE_SET_WECHAT_CHANNELS: 审核通过",
        )
        self.assertEqual(
            fields["site_reject_reason"],
            "SITE_SET_MOMENTS(id=40004): 朋友圈版位拒绝\n"
            "SITE_SET_MOMENTS(id=40004): 版位素材不适配",
        )

    def test_status_desc_preserves_unknown_enum(self):
        from tools.creative_review import status_desc

        self.assertEqual(status_desc("AD_STATUS_NORMAL"), "投放中")
        self.assertEqual(
            status_desc("CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"),
            "部分投放中",
        )
        self.assertEqual(status_desc("FUTURE_STATUS_NEW"), "FUTURE_STATUS_NEW")

    def test_chunk_ids_by_account_limits_to_100(self):
        from tools.creative_review import group_review_tasks

        tasks = [
            {"account_id": 1, "dynamic_creative_id": i}
            for i in range(250)
        ] + [{"account_id": 2, "dynamic_creative_id": 999}]

        groups = list(group_review_tasks(tasks, batch_size=100))

        self.assertEqual(len(groups), 4)
        self.assertEqual(groups[0][0], 1)
        self.assertEqual(len(groups[0][1]), 100)
        self.assertEqual(len(groups[1][1]), 100)
        self.assertEqual(len(groups[2][1]), 50)
        self.assertEqual(groups[3], (2, [999]))


class CreativeRejectionCleanupTests(unittest.TestCase):
    def test_dynamic_creative_api_requests_creative_approval_status(self):
        from tencent_client import DYNAMIC_CREATIVE_FIELDS, TencentClient

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

        client.get_dynamic_creatives(1)

        self.assertIn("creative_set_approval_status", DYNAMIC_CREATIVE_FIELDS)
        requested_fields = json.loads(client.session.get.call_args.kwargs["params"]["fields"])
        self.assertIn("creative_set_approval_status", requested_fields)

    def test_dynamic_creative_costs_query_sums_three_day_api_rows(self):
        from datetime import date

        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, "date": "2026-08-10", "cost": 1200},
                    {"dynamic_creative_id": 3, "date": "2026-08-11", "cost": 1799},
                    {"dynamic_creative_id": 4, "date": "2026-08-12", "cost": 3000},
                ],
                "page_info": {"total_page": 1},
            },
        }
        client = TencentClient()
        client.session.get = Mock(return_value=response)
        client._common_params = Mock(return_value={"access_token": "token"})

        costs = client.get_dynamic_creative_costs(
            1,
            [3, 4],
            date(2026, 8, 10),
            date(2026, 8, 12),
        )

        self.assertEqual(costs, {3: 2999, 4: 3000})
        request = client.session.get.call_args
        self.assertTrue(
            request.args[0].endswith("/daily_reports/get")
        )
        self.assertEqual(
            request.kwargs["params"]["level"],
            "REPORT_LEVEL_DYNAMIC_CREATIVE",
        )
        self.assertEqual(
            json.loads(request.kwargs["params"]["date_range"]),
            {"start_date": "2026-08-10", "end_date": "2026-08-12"},
        )
        self.assertEqual(
            json.loads(request.kwargs["params"]["filtering"]),
            [
                {
                    "field": "dynamic_creative_id",
                    "operator": "IN",
                    "values": ["3", "4"],
                }
            ],
        )
        self.assertEqual(request.kwargs["params"]["time_line"], "REQUEST_TIME")

    def test_account_scope_uses_recent_three_day_opengid_spend(self):
        from roi_control.data_source import (
            build_recent_spend_accounts_sql,
            fetch_recent_spend_accounts,
        )

        sql = build_recent_spend_accounts_sql("20260808", "20260810")
        self.assertIn("FROM loghubods.opengid_base_data", sql)
        self.assertIn("dt BETWEEN '20260808' AND '20260810'", sql)
        self.assertNotIn("usersharedepth", sql)
        self.assertNotIn("videoid", sql)
        self.assertNotIn("hotsencetype", sql)
        self.assertIn("GROUP BY 账号id", sql)
        self.assertIn("HAVING SUM(NVL(成本, 0)) > 0", sql)
        self.assertNotIn("account_whitelist", sql)

        client = Mock()
        client.execute_sql.return_value = pd.DataFrame(
            [
                {"account_id": "1", "account_name": "账户一", "cost_yuan": 10},
                {"account_id": "", "account_name": "无效", "cost_yuan": 20},
                {"account_id": "2", "account_name": "账户二", "cost_yuan": 30},
            ]
        )
        self.assertEqual(
            fetch_recent_spend_accounts(client, "20260808", "20260810"),
            [
                {"account_id": 1, "account_name": "账户一", "cost_yuan": 10.0},
                {"account_id": 2, "account_name": "账户二", "cost_yuan": 30.0},
            ],
        )

    def test_account_agency_fallback_uses_latest_active_account_record(self):
        from roi_control.data_source import (
            build_account_agency_fallback_sql,
            fetch_account_agency_fallbacks,
        )

        sql = build_account_agency_fallback_sql([2, 1, 2])
        self.assertIn("FROM loghubods.ad_put_tencent_account", sql)
        self.assertIn("account_id IN ('1', '2')", sql)
        self.assertIn("NVL(is_delete, 0) = 0", sql)
        self.assertIn("status = 1", sql)
        self.assertIn("TRIM(agent_name) <> ''", sql)
        self.assertIn("ORDER BY id DESC", sql)
        self.assertIn("WHERE row_number = 1", sql)
        inner_projection = sql.split("FROM (", 1)[1].split(
            "FROM loghubods.ad_put_tencent_account", 1
        )[0]
        self.assertIn("is_delete", inner_projection)
        self.assertIn("status", inner_projection)

        client = Mock()
        client.execute_sql.return_value = pd.DataFrame(
            [
                {"account_id": "1", "agent_name": " 代理 A "},
                {"account_id": "2", "agent_name": float("nan")},
                {"account_id": "invalid", "agent_name": "代理 B"},
            ]
        )
        self.assertEqual(fetch_account_agency_fallbacks(client, [1, 2]), {1: "代理A"})

    def test_agency_resolution_only_uses_account_table_as_last_fallback(self):
        from tools.creative_rejection_cleanup import _resolve_agency

        context = {
            "creative_agencies": {(1, 101): "创意代理"},
            "account_agencies": {1: "账户代理"},
            "fallback_account_agencies": {1: "表中代理", 2: "兜底代理"},
            "account_names": {},
        }

        self.assertEqual(_resolve_agency(context, 1, 101), "创意代理")
        self.assertEqual(_resolve_agency(context, 1, 102), "账户代理")
        self.assertEqual(_resolve_agency(context, 2, 201), "兜底代理")

    def test_cleanup_requires_current_rejection_confirmation(self):
        from tools.creative_rejection_cleanup import (
            DELETE_CREATIVE,
            cleanup_precondition_failure,
        )

        item = {
            "account_id": 1,
            "dynamic_creative_id": 3,
            "cleanup_status": "DISCOVERED",
            "cleanup_action": DELETE_CREATIVE,
        }
        self.assertEqual(
            cleanup_precondition_failure(item, set(), set())[0],
            "DEFERRED",
        )
        self.assertEqual(
            cleanup_precondition_failure(item, {1}, {})[0],
            "SKIPPED_REVIEW_NOT_RECONFIRMED",
        )
        confirmed = {
            (1, 3): {
                "cleanup_action": DELETE_CREATIVE,
            }
        }
        self.assertIsNone(cleanup_precondition_failure(item, {1}, confirmed))
        item["cleanup_action"] = "DELETE_COMPONENTS"
        self.assertEqual(
            cleanup_precondition_failure(item, {1}, confirmed)[0],
            "SKIPPED_REVIEW_NOT_RECONFIRMED",
        )

    def test_cleanup_action_follows_creative_approval_and_element_statuses(self):
        from tools.creative_rejection_cleanup import (
            ALERT_ONLY,
            DELETE_CREATIVE,
            determine_cleanup_action,
        )

        denied_element = {
            "element_result_list": [
                {
                    "element_id": 101,
                    "review_status": "AD_STATUS_DENIED",
                    "component_info": {"component_id": 201},
                }
            ]
        }
        denied_action = determine_cleanup_action(
                {
                    "configured_status": "AD_STATUS_NORMAL",
                    "creative_set_approval_status": (
                        "CREATIVE_SET_APPROVAL_STATUS_DENIED"
                    ),
                },
                denied_element,
        )
        self.assertEqual(denied_action["cleanup_action"], DELETE_CREATIVE)
        self.assertEqual(denied_action["component_ids"], [])
        normal_action = determine_cleanup_action(
                {
                    "configured_status": "AD_STATUS_SUSPEND",
                    "creative_set_approval_status": (
                        "CREATIVE_SET_APPROVAL_STATUS_NORMAL"
                    ),
                },
                denied_element,
        )
        self.assertIsNone(normal_action)

        partial = {
            "creative_set_approval_status": (
                "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
            )
        }
        low_cost = determine_cleanup_action(
            partial,
            denied_element,
            recent_cost_fen=4999,
            current_day_cost_fen=0,
            creative_created_at="2026-08-10 11:00:00",
            as_of_datetime="2026-08-20 11:00:00",
        )
        self.assertEqual(low_cost["cleanup_action"], DELETE_CREATIVE)
        at_threshold = determine_cleanup_action(
            partial,
            denied_element,
            recent_cost_fen=5000,
            current_day_cost_fen=0,
            creative_created_at="2026-08-10 11:00:00",
            as_of_datetime="2026-08-20 11:00:00",
        )
        self.assertEqual(at_threshold["cleanup_action"], ALERT_ONLY)

    def test_partial_creative_with_rejected_wechat_mini_program_is_deleted(self):
        from tools.creative_rejection_cleanup import (
            DELETE_CREATIVE,
            determine_cleanup_action,
        )

        action = determine_cleanup_action(
            {
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                )
            },
            {
                "element_result_list": [
                    {
                        "element_name": "微信小程序",
                        "element_id": -8101,
                        "review_status": "AD_STATUS_DENIED",
                    }
                ]
            },
            recent_cost_fen=9999,
            current_day_cost_fen=0,
            creative_created_at="2026-08-10 11:00:00",
            as_of_datetime="2026-08-20 11:00:00",
        )

        self.assertEqual(action["cleanup_action"], DELETE_CREATIVE)
        self.assertIn("微信小程序元素审核拒绝", action["action_reason"])

    def test_partial_creative_with_high_cost_rejected_wechat_requires_review(self):
        from tools.creative_rejection_cleanup import ALERT_ONLY, determine_cleanup_action

        action = determine_cleanup_action(
            {
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                )
            },
            {
                "element_result_list": [{
                    "element_name": "微信小程序",
                    "review_status": "AD_STATUS_DENIED",
                }]
            },
            recent_cost_fen=10001,
            current_day_cost_fen=0,
            creative_created_at="2026-08-10 11:00:00",
            as_of_datetime="2026-08-20 11:00:00",
        )

        self.assertEqual(action["cleanup_action"], ALERT_ONLY)
        self.assertIn("需人工判断", action["action_reason"])

    def test_partial_creative_spend_error_only_alerts(self):
        from tools.creative_rejection_cleanup import ALERT_ONLY, determine_cleanup_action

        action = determine_cleanup_action(
            {
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                )
            },
            {},
            creative_created_at="2026-08-10 11:00:00",
            as_of_datetime="2026-08-20 11:00:00",
            spend_error="timeout",
        )

        self.assertEqual(action["cleanup_action"], ALERT_ONLY)
        self.assertIn("消耗读取失败", action["action_reason"])

    def test_partial_creative_not_older_than_three_days_only_alerts(self):
        from tools.creative_rejection_cleanup import ALERT_ONLY, determine_cleanup_action

        action = determine_cleanup_action(
            {
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                )
            },
            {},
            recent_cost_fen=0,
            current_day_cost_fen=0,
            creative_created_at="2026-08-17 11:00:00",
            as_of_datetime="2026-08-20 11:00:00",
        )

        self.assertEqual(action["cleanup_action"], ALERT_ONLY)
        self.assertEqual(action["creative_age_days"], 3)
        self.assertIn("未超过3天", action["action_reason"])

    def test_partial_creative_uses_historical_plus_current_day_cost(self):
        from tools.creative_rejection_cleanup import (
            ALERT_ONLY,
            DELETE_CREATIVE,
            determine_cleanup_action,
        )

        common = {
            "creative_set_approval_status": (
                "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
            )
        }
        protected_by_total = determine_cleanup_action(
            common,
            {},
            recent_cost_fen=4000,
            current_day_cost_fen=1000,
            creative_created_at="2026-08-10 11:00:00",
            as_of_datetime="2026-08-20 11:00:00",
        )
        still_below_total = determine_cleanup_action(
            common,
            {},
            recent_cost_fen=4000,
            current_day_cost_fen=999,
            creative_created_at="2026-08-10 11:00:00",
            as_of_datetime="2026-08-20 11:00:00",
        )

        self.assertEqual(protected_by_total["cleanup_action"], ALERT_ONLY)
        self.assertEqual(protected_by_total["total_cost_fen"], 5000)
        self.assertEqual(still_below_total["cleanup_action"], DELETE_CREATIVE)
        self.assertEqual(still_below_total["total_cost_fen"], 4999)

    def test_partial_creative_missing_creation_time_only_alerts(self):
        from tools.creative_rejection_cleanup import ALERT_ONLY, determine_cleanup_action

        action = determine_cleanup_action(
            {
                "creative_set_approval_status": (
                    "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                )
            },
            {},
            recent_cost_fen=0,
            current_day_cost_fen=0,
            creative_created_at=None,
            as_of_datetime="2026-08-20 11:00:00",
        )

        self.assertEqual(action["cleanup_action"], ALERT_ONLY)
        self.assertIn("搭建时间缺失", action["action_reason"])

    def test_partial_creative_protection_days_config_is_validated(self):
        from tools.creative_rejection_cleanup import partial_creative_protection_days

        with patch.dict(
            os.environ,
            {"DAILY_PARTIAL_CREATIVE_PROTECTION_DAYS": "5"},
            clear=False,
        ):
            self.assertEqual(partial_creative_protection_days(), 5)
        with patch.dict(
            os.environ,
            {"DAILY_PARTIAL_CREATIVE_PROTECTION_DAYS": "-1"},
            clear=False,
        ):
            with self.assertRaises(ValueError):
                partial_creative_protection_days()

    def test_cleanup_candidate_insert_has_one_value_per_column(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):
            result = 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": "2026-08-12",
                    "cleanup_action": cleanup.ALERT_ONLY,
                    "component_ids": [],
                    "element_ids": [],
                    "recent_cost_fen": 3000,
                    "cost_start_date": "2026-08-10",
                    "cost_end_date": "2026-08-12",
                    "reject_reason": "待人工判断",
                    "review_result": {},
                    "pre_state": {},
                }
            )

        insert_sql, insert_params = cursor.calls[0]
        values_sql = insert_sql.split("ON DUPLICATE KEY UPDATE", 1)[0]
        self.assertEqual(values_sql.count("%s"), 22)
        self.assertEqual(len(insert_params), 22)
        self.assertEqual(insert_params[-2], "ALERT_PENDING")
        self.assertIsNone(insert_params[-1])
        update_sql, update_params = cursor.calls[1]
        self.assertIn(
            "NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)",
            update_sql,
        )
        normalized_update_sql = " ".join(update_sql.split())
        self.assertIn(
            "'DELETING','CREATIVE_DELETED','AD_DELETED'",
            normalized_update_sql,
        )
        self.assertEqual(update_params, insert_params)
        self.assertEqual(result, {"id": 7})
        connection.close.assert_called_once()

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

        connections = []

        def connection_factory():
            cursor = Mock()
            cursor.__enter__ = Mock(return_value=cursor)
            cursor.__exit__ = Mock(return_value=None)
            connection = Mock()
            connection.cursor.return_value = cursor
            connections.append((connection, cursor))
            return connection

        records = [
            {
                "account_id": 1,
                "account_name": "账户一",
                "agency_name": "",
                "adgroup_id": 2,
                "adgroup_name": "广告二",
                "dynamic_creative_id": creative_id,
                "dynamic_creative_name": f"创意{creative_id}",
                "check_date": date(2026, 8, 20),
                "cleanup_action": cleanup.DELETE_CREATIVE,
                "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
                "reject_reason": "",
                "review_result": {},
                "pre_state": {},
            }
            for creative_id in range(1, 5)
        ]
        with patch.dict(
            os.environ,
            {
                "DAILY_CLEANUP_CANDIDATE_BATCH_SIZE": "2",
                "DAILY_CLEANUP_CANDIDATE_STORE_WORKERS": "2",
            },
            clear=False,
        ), patch.object(
            cleanup,
            "get_connection",
            side_effect=connection_factory,
        ):
            stored, errors = cleanup.upsert_cleanup_candidates(records)

        self.assertEqual(len(stored), 4)
        self.assertEqual(errors, [])
        self.assertEqual(len(connections), 2)
        self.assertEqual(
            sum(cursor.execute.call_count for _, cursor in connections),
            12,
        )
        for connection, _cursor in connections:
            connection.begin.assert_called_once_with()
            connection.commit.assert_called_once_with()
            connection.close.assert_called_once_with()

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

        connections = []

        def connection_factory():
            cursor = Mock()
            cursor.__enter__ = Mock(return_value=cursor)
            cursor.__exit__ = Mock(return_value=None)
            cursor.fetchone.return_value = {"id": 7}
            connection = Mock()
            connection.cursor.return_value = cursor
            if not connections:
                cursor.execute.side_effect = RuntimeError("batch write failed")
            connections.append((connection, cursor))
            return connection

        records = [
            {
                "account_id": 1,
                "account_name": "账户一",
                "agency_name": "",
                "adgroup_id": 2,
                "adgroup_name": "广告二",
                "dynamic_creative_id": creative_id,
                "dynamic_creative_name": f"创意{creative_id}",
                "check_date": date(2026, 8, 20),
                "cleanup_action": cleanup.DELETE_CREATIVE,
                "cleanup_rule_type": cleanup.PERFORMANCE_NEW_RULE,
                "reject_reason": "",
                "review_result": {},
                "pre_state": {},
            }
            for creative_id in (1, 2)
        ]
        with patch.dict(
            os.environ,
            {
                "DAILY_CLEANUP_CANDIDATE_BATCH_SIZE": "10",
                "DAILY_CLEANUP_CANDIDATE_STORE_WORKERS": "1",
            },
            clear=False,
        ), patch.object(
            cleanup,
            "get_connection",
            side_effect=connection_factory,
        ):
            stored, errors = cleanup.upsert_cleanup_candidates(records)

        self.assertEqual(len(stored), 2)
        self.assertEqual(errors, [])
        self.assertEqual(len(connections), 3)
        connections[0][0].rollback.assert_called_once_with()
        for connection, _cursor in connections:
            connection.close.assert_called_once_with()

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

        cursor = Mock()
        cursor.__enter__ = Mock(return_value=cursor)
        cursor.__exit__ = Mock(return_value=None)
        cursor.fetchall.return_value = []
        connection = Mock()
        connection.cursor.return_value = cursor

        with patch.object(cleanup, "get_connection", return_value=connection):
            self.assertEqual(cleanup.load_retryable_cleanup_items(), [])

        sql = cursor.execute.call_args.args[0]
        self.assertIn(
            "cleanup_action IN ('DELETE_CREATIVE','DELETE_AD')",
            sql,
        )
        self.assertIn("item.cleanup_status='DELETING'", sql)
        self.assertIn("DATE_SUB(NOW(), INTERVAL %s MINUTE)", sql)
        self.assertNotIn("COMPONENTS_PARTIAL", sql)

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

        cursor = Mock()
        cursor.__enter__ = Mock(return_value=cursor)
        cursor.__exit__ = Mock(return_value=None)
        cursor.rowcount = 0
        connection = Mock()
        connection.cursor.return_value = cursor

        with patch.object(cleanup, "get_connection", return_value=connection):
            self.assertFalse(cleanup.claim_cleanup_item(7))

        sql, params = cursor.execute.call_args.args
        self.assertIn("SET cleanup_status='DELETING'", sql)
        self.assertIn(
            "cleanup_action IN ('DELETE_CREATIVE','DELETE_AD')",
            sql,
        )
        self.assertIn("cleanup_status IN", sql)
        self.assertIn("cleanup_status='DELETING'", sql)
        self.assertEqual(params, (7, cleanup.DEFAULT_DELETE_CLAIM_STALE_MINUTES))

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

        cursor = Mock()
        cursor.__enter__ = Mock(return_value=cursor)
        cursor.__exit__ = Mock(return_value=None)
        cursor.rowcount = 1
        connection = Mock()
        connection.cursor.return_value = cursor

        with patch.object(cleanup, "get_connection", return_value=connection):
            self.assertTrue(
                cleanup._update_owned_cleanup_item(
                    7,
                    cleanup_status="CREATIVE_DELETED",
                    error_message=None,
                )
            )

        sql, params = cursor.execute.call_args.args
        self.assertIn("WHERE id=%s AND cleanup_status=%s", sql)
        self.assertEqual(params, ["CREATIVE_DELETED", None, 7, "DELETING"])

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

        cursor = Mock()
        cursor.__enter__ = Mock(return_value=cursor)
        cursor.__exit__ = Mock(return_value=None)
        cursor.fetchall.return_value = []
        connection = Mock()
        connection.cursor.return_value = cursor

        with patch.object(cleanup, "get_connection", return_value=connection):
            cleanup.load_pending_notification_items(include_discovered=True)

        sql = cursor.execute.call_args.args[0]
        self.assertIn("'DISCOVERED'", sql)

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

        cursor = Mock()
        cursor.__enter__ = Mock(return_value=cursor)
        cursor.__exit__ = Mock(return_value=None)
        cursor.fetchall.return_value = []
        connection = Mock()
        connection.cursor.return_value = cursor
        target_date = date(2026, 8, 20)

        with patch.object(cleanup, "get_connection", return_value=connection):
            cleanup.load_pending_notification_items(
                include_discovered=True,
                check_date=target_date,
                performance_only=True,
            )

        sql, params = cursor.execute.call_args.args
        self.assertIn("AND check_date=%s", sql)
        self.assertIn(
            "LEFT(cleanup_rule_type, 12)='PERFORMANCE_'",
            sql,
        )
        self.assertEqual(params, [target_date])

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

        cursor = Mock()
        cursor.__enter__ = Mock(return_value=cursor)
        cursor.__exit__ = Mock(return_value=None)
        cursor.fetchall.return_value = []
        connection = Mock()
        connection.cursor.return_value = cursor

        with patch.object(cleanup, "get_connection", return_value=connection):
            cleanup.load_pending_notification_items(
                include_discovered=True,
                check_date=date(2026, 8, 20),
                performance_only=True,
                include_notified=True,
            )

        sql = cursor.execute.call_args.args[0]
        self.assertNotIn("operator_notified_at IS NULL", sql)

    def test_dry_run_reports_mark_delete_as_not_executed(self):
        from tools.creative_rejection_cleanup import (
            write_cleanup_operator_summary,
            write_cleanup_reports,
        )

        row = {
            "id": 7,
            "agency_name": "棱镜",
            "account_id": 1,
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": "DELETE_CREATIVE",
            "cleanup_status": "DISCOVERED",
            "reject_reason": "审核拒绝",
            "review_result_json": "{}",
            "pre_state_json": "{}",
        }
        with tempfile.TemporaryDirectory() as directory:
            output_dir = Path(directory)
            _, reports, _ = write_cleanup_reports(
                [row], output_dir, "20260813"
            )
            operator = write_cleanup_operator_summary(
                [row], output_dir, "20260813", "unused"
            )
            agency_sheet = load_workbook(reports[0]["report"])[
                "审核不通过创意清理"
            ]
            operator_sheet = load_workbook(operator["report"])[
                "审核不通过创意清理"
            ]

        for sheet in (agency_sheet, operator_sheet):
            columns = {cell.value: cell.column for cell in sheet[1]}
            self.assertEqual(
                sheet.cell(2, columns["执行操作"]).value,
                "建议删除创意（未执行）",
            )
        self.assertEqual(
            reports[0]["notification_type"],
            "creative_rejection_dry_run",
        )
        self.assertTrue(operator["dry_run"])

    def test_cleanup_action_does_not_delete_for_overall_or_site_denial(self):
        from tools.creative_rejection_cleanup import determine_cleanup_action

        creative = {
            "configured_status": "AD_STATUS_NORMAL",
            "system_status": "DYNAMIC_CREATIVE_STATUS_DENIED",
        }
        result = {
            "site_set_result_list": [
                {"system_status": "DYNAMIC_CREATIVE_STATUS_DENIED"}
            ]
        }
        self.assertIsNone(determine_cleanup_action(creative, result))

    def test_denied_element_is_not_a_cleanup_candidate(self):
        from tools.creative_rejection_cleanup import determine_cleanup_action

        action = determine_cleanup_action(
            {"configured_status": "AD_STATUS_NORMAL"},
            {
                "element_result_list": [
                    {"element_id": 101, "system_status": "AD_STATUS_DENIED"}
                ]
            },
        )
        self.assertIsNone(action)

    def test_tencent_client_uses_official_creative_delete_endpoint(self):
        from tencent_client import TencentClient

        response = Mock(status_code=200, text="ok")
        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={"nonce": "1"})
        client._user_token = Mock(return_value="user-token")
        client.get_dynamic_creative = Mock(
            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
        )

        result = client.delete_dynamic_creative(1, 3)

        request = client.session.post.call_args
        self.assertTrue(request.args[0].endswith("/dynamic_creatives/delete"))
        self.assertEqual(
            request.kwargs["json"],
            {"account_id": 1, "dynamic_creative_id": 3},
        )
        self.assertEqual(result["configured_status"], "AD_STATUS_DELETED")

    def test_tencent_creative_delete_accepts_deleted_system_status(self):
        from tencent_client import TencentClient

        response = Mock(status_code=200, text="ok")
        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={"nonce": "1"})
        client._user_token = Mock(return_value="user-token")
        client.get_dynamic_creative = Mock(return_value={
            "dynamic_creative_id": 3,
            "configured_status": "AD_STATUS_NORMAL",
            "system_status": "DYNAMIC_CREATIVE_STATUS_DELETED",
            "creative_set_approval_status": (
                "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
            ),
        })

        result = client.delete_dynamic_creative(1, 3)

        self.assertEqual(result["system_status"], "DYNAMIC_CREATIVE_STATUS_DELETED")
        self.assertFalse(result["deleted_from_listing"])
        self.assertEqual(client.get_dynamic_creative.call_count, 1)

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

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [{
            "dynamic_creative_id": 3,
            "configured_status": "AD_STATUS_NORMAL",
            "system_status": "DYNAMIC_CREATIVE_STATUS_DELETED",
            "creative_set_approval_status": cleanup.CREATIVE_PARTIAL_NORMAL_STATUS,
        }]
        tencent.get_ads.return_value = []
        review_fetcher = Mock(return_value=[])

        result = cleanup._scan_one_account(
            {"account_id": 1},
            tencent=tencent,
            review_fetcher=review_fetcher,
            spend_start_date=date(2026, 8, 14),
            spend_end_date=date(2026, 8, 16),
        )

        self.assertEqual(result[5], 0)
        self.assertEqual(result[0], [])
        review_fetcher.assert_not_called()
        tencent.get_dynamic_creative_costs.assert_not_called()

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

        tencent = Mock()
        tencent.get_dynamic_creatives.side_effect = RuntimeError(
            "get_dynamic_creatives failed: code=11002 "
            "message=您的 access_token 无效，请通过 oauth/authorize 接口获取 token。"
        )
        review_fetcher = Mock()

        result = cleanup._scan_one_account(
            {"account_id": 1},
            tencent=tencent,
            review_fetcher=review_fetcher,
            spend_start_date=date(2026, 8, 14),
            spend_end_date=date(2026, 8, 16),
            performance_enabled=True,
            ad_cleanup_enabled=True,
        )

        self.assertTrue(cleanup._is_token_skipped_scan_error(result[6]))
        tencent.get_ads.assert_not_called()
        review_fetcher.assert_not_called()
        tencent.get_dynamic_creative_costs.assert_not_called()
        tencent.get_dynamic_creative_metrics.assert_not_called()
        tencent.get_ad_metrics.assert_not_called()

    def test_tencent_creative_delete_retries_explicit_rate_limit_only(self):
        import tencent_client

        TencentClient = tencent_client.TencentClient

        limited = Mock(status_code=429, text="too many requests", headers={})
        limited.raise_for_status.side_effect = AssertionError(
            "429 should be handled before raise_for_status"
        )
        success = Mock(status_code=200, text="ok", headers={})
        success.raise_for_status.return_value = None
        success.json.return_value = {"code": 0, "data": {}}
        client = TencentClient()
        client.session.post = Mock(side_effect=[limited, success])
        client._common_params = Mock(
            side_effect=[{"nonce": "1"}, {"nonce": "2"}]
        )
        client._user_token = Mock(return_value="user-token")
        client.get_dynamic_creative = Mock(
            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
        )

        limiter = Mock()
        with patch.dict(
            os.environ,
            {
                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "1",
                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "0",
            },
            clear=False,
        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
            result = client.delete_dynamic_creative(1, 3)

        self.assertEqual(result["configured_status"], "AD_STATUS_DELETED")
        self.assertEqual(client.session.post.call_count, 2)
        self.assertEqual(client._common_params.call_count, 2)
        self.assertEqual(limiter.wait.call_args_list, [call(0.0)] * 2)
        limiter.defer.assert_called_once_with(0.0)

    def test_tencent_creative_delete_honors_shared_retry_after_cooldown(self):
        import tencent_client

        limited = Mock(
            status_code=429,
            text="too many requests",
            headers={"Retry-After": "2"},
        )
        success = Mock(status_code=200, text="ok", headers={})
        success.raise_for_status.return_value = None
        success.json.return_value = {"code": 0, "data": {}}
        client = tencent_client.TencentClient()
        client.session.post = Mock(side_effect=[limited, success])
        client._common_params = Mock(return_value={"nonce": "1"})
        client._user_token = Mock(return_value="user-token")
        client.get_dynamic_creative = Mock(
            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
        )
        limiter = Mock()

        with patch.dict(
            os.environ,
            {
                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0.25",
                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "1",
                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "1",
            },
            clear=False,
        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
            client.delete_dynamic_creative(1, 3)

        self.assertEqual(limiter.wait.call_args_list, [call(0.25)] * 2)
        limiter.defer.assert_called_once_with(2.0)

    def test_tencent_creative_delete_exhausted_rate_limit_still_cools_workers(self):
        import tencent_client

        client = tencent_client.TencentClient()
        limited = Mock(
            status_code=429,
            text="too many requests",
            headers={"Retry-After": "3"},
        )
        client.session.post = Mock(return_value=limited)
        client._common_params = Mock(return_value={"nonce": "1"})
        client._user_token = Mock(return_value="user-token")
        limiter = Mock()

        with patch.dict(
            os.environ,
            {
                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0.25",
                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "0",
                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "1",
            },
            clear=False,
        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
            with self.assertRaises(tencent_client.TencentWriteRateLimitedError):
                client.delete_dynamic_creative(1, 3)

        limiter.wait.assert_called_once_with(0.25)
        limiter.defer.assert_called_once_with(3.0)
        self.assertEqual(client.session.post.call_count, 1)

    def test_tencent_creative_delete_invalid_retry_after_uses_backoff(self):
        import tencent_client

        client = tencent_client.TencentClient()
        limited = Mock(
            status_code=429,
            text="too many requests",
            headers={"Retry-After": "NaN"},
        )
        client.session.post = Mock(return_value=limited)
        client._common_params = Mock(return_value={"nonce": "1"})
        client._user_token = Mock(return_value="user-token")
        limiter = Mock()

        with patch.dict(
            os.environ,
            {
                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "0",
                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "1.5",
            },
            clear=False,
        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
            with self.assertRaises(tencent_client.TencentWriteRateLimitedError):
                client.delete_dynamic_creative(1, 3)

        limiter.defer.assert_called_once_with(1.5)

    def test_tencent_creative_delete_retries_business_rate_limit_payload(self):
        import tencent_client

        client = tencent_client.TencentClient()
        limited = Mock(status_code=200, text="limited", headers={})
        limited.raise_for_status.return_value = None
        limited.json.return_value = {
            "code": 12002,
            "message_cn": "请求频繁，请稍后重试",
        }
        success = Mock(status_code=200, text="ok", headers={})
        success.raise_for_status.return_value = None
        success.json.return_value = {"code": 0, "data": {}}
        client.session.post = Mock(side_effect=[limited, success])
        client._common_params = Mock(
            side_effect=[{"nonce": "1"}, {"nonce": "2"}]
        )
        client._user_token = Mock(return_value="user-token")
        client.get_dynamic_creative = Mock(
            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
        )
        limiter = Mock()

        with patch.dict(
            os.environ,
            {
                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "1",
                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "2",
            },
            clear=False,
        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
            result = client.delete_dynamic_creative(1, 3)

        self.assertEqual(result["configured_status"], "AD_STATUS_DELETED")
        self.assertEqual(client.session.post.call_count, 2)
        self.assertEqual(limiter.defer.call_args_list, [call(2.0)])

    def test_tencent_creative_delete_does_not_retry_success_message_about_qps(self):
        import tencent_client

        client = tencent_client.TencentClient()
        success = Mock(status_code=200, text="ok", headers={})
        success.raise_for_status.return_value = None
        success.json.return_value = {
            "code": 0,
            "message": "success; current QPS quota is available",
            "data": {},
        }
        client.session.post = Mock(return_value=success)
        client._common_params = Mock(return_value={"nonce": "1"})
        client._user_token = Mock(return_value="user-token")
        client.get_dynamic_creative = Mock(
            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
        )
        limiter = Mock()

        with patch.dict(
            os.environ,
            {
                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "2",
                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "1",
            },
            clear=False,
        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
            result = client.delete_dynamic_creative(1, 3)

        self.assertEqual(result["configured_status"], "AD_STATUS_DELETED")
        self.assertEqual(client.session.post.call_count, 1)
        limiter.defer.assert_not_called()

    def test_tencent_creative_delete_does_not_retry_unknown_network_failure(self):
        from tencent_client import TencentClient, TencentWriteOutcomeUnknownError

        client = TencentClient()
        client.session.post = Mock(
            side_effect=__import__("requests").RequestException("connection reset")
        )
        client._common_params = Mock(return_value={"nonce": "1"})
        client._user_token = Mock(return_value="user-token")

        with patch.dict(
            os.environ,
            {
                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "3",
                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "0",
            },
            clear=False,
        ):
            with self.assertRaises(TencentWriteOutcomeUnknownError):
                client.delete_dynamic_creative(1, 3)

        self.assertEqual(client.session.post.call_count, 1)

    def test_tencent_client_uses_component_id_for_component_delete(self):
        from tencent_client import TencentClient

        response = Mock(status_code=200, text="ok")
        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={"nonce": "1"})
        client.get_creative_component = Mock(
            return_value={"component_id": 201, "is_deleted": True}
        )

        result = client.delete_creative_component(1, 201)

        request = client.session.post.call_args
        self.assertTrue(request.args[0].endswith("/components/delete"))
        self.assertEqual(request.kwargs["json"]["component_id"], 201)
        self.assertEqual(
            request.kwargs["json"]["delete_strategy"],
            "DELETE_STRATEGY_FORCE",
        )
        self.assertTrue(result["is_deleted"])

    def test_agency_context_prefers_creative_and_rejects_ambiguous_fallback(self):
        from tools.creative_rejection_cleanup import build_agency_context

        daily = pd.DataFrame(
            [
                {
                    "entity_type": "self",
                    "账号id": "1",
                    "账号名称": "账户一",
                    "创意id": "101",
                    "代理名称": "小程序-代投-棱镜",
                },
                {
                    "entity_type": "self",
                    "账号id": "2",
                    "账号名称": "账户二",
                    "创意id": "201",
                    "代理名称": "代理A",
                },
                {
                    "entity_type": "self",
                    "账号id": "2",
                    "账号名称": "账户二",
                    "创意id": "202",
                    "代理名称": "代理B",
                },
                {
                    "entity_type": "self",
                    "账号id": "3",
                    "账号名称": "账户三",
                    "创意id": "301",
                    "代理名称": float("nan"),
                },
            ]
        )
        context = build_agency_context(daily)

        self.assertEqual(
            context["creative_agencies"][(1, 101)],
            "小程序-代投-棱镜",
        )
        self.assertEqual(context["account_agencies"][1], "小程序-代投-棱镜")
        self.assertNotIn(2, context["account_agencies"])
        self.assertNotIn(3, context["account_agencies"])

    def test_agency_context_uses_latest_date_regardless_of_row_order(self):
        from tools.creative_rejection_cleanup import build_agency_context

        rows = [
            {
                "entity_type": "self",
                "dt": "20260810",
                "账号id": "1",
                "创意id": "101",
                "代理名称": "旧代理",
            },
            {
                "entity_type": "self",
                "dt": "20260812",
                "账号id": "1",
                "创意id": "101",
                "代理名称": "新代理",
            },
        ]
        for ordered_rows in (rows, list(reversed(rows))):
            context = build_agency_context(pd.DataFrame(ordered_rows))
            self.assertEqual(context["creative_agencies"][(1, 101)], "新代理")
            self.assertEqual(context["account_agencies"][1], "新代理")

    def test_agency_context_rejects_conflicting_agencies_on_latest_date(self):
        from tools.creative_rejection_cleanup import build_agency_context

        context = build_agency_context(pd.DataFrame([
            {
                "entity_type": "self",
                "dt": "20260812",
                "账号id": "1",
                "创意id": "101",
                "代理名称": "代理A",
            },
            {
                "entity_type": "self",
                "dt": "20260812",
                "账号id": "1",
                "创意id": "101",
                "代理名称": "代理B",
            },
        ]))

        self.assertNotIn((1, 101), context["creative_agencies"])
        self.assertNotIn(1, context["account_agencies"])

    def test_agency_report_run_id_is_independent_per_destination(self):
        from tools.creative_rejection_cleanup import write_cleanup_reports

        def row(item_id, agency):
            return {
                "id": item_id,
                "agency_name": agency,
                "account_id": item_id,
                "adgroup_id": item_id + 10,
                "dynamic_creative_id": item_id + 20,
                "cleanup_action": "ALERT_ONLY",
                "cleanup_status": "ALERT_PENDING",
                "reject_reason": "需人工判断",
            }

        with tempfile.TemporaryDirectory() as directory:
            _, one_report, _ = write_cleanup_reports(
                [row(1, "代理A")], Path(directory), "20260812"
            )
            _, two_reports, _ = write_cleanup_reports(
                [row(1, "代理A"), row(2, "代理B")],
                Path(directory),
                "20260812",
            )

        run_id_a = next(
            report["run_id"]
            for report in two_reports
            if report["agency_name"] == "代理A"
        )
        self.assertEqual(one_report[0]["run_id"], run_id_a)

    def test_daily_cleanup_schema_uses_check_date_in_unique_key(self):
        schema = (
            Path(__file__).parents[1]
            / "tencent_realtime_control"
            / "schema.sql"
        ).read_text(encoding="utf-8")
        self.assertIn(
            "(account_id, dynamic_creative_id, check_date)",
            schema,
        )

    def test_cleanup_report_has_required_business_columns(self):
        from tools.creative_rejection_cleanup import (
            AGENCY_REPORT_COLUMNS,
            write_cleanup_reports,
        )

        now = datetime(2026, 8, 11, 11, 0)
        with tempfile.TemporaryDirectory() as directory:
            _, reports, item_ids = write_cleanup_reports(
                [
                    {
                        "id": 7,
                        "agency_name": "棱镜",
                        "account_id": 1000000000001,
                        "account_name": "账户一",
                        "adgroup_id": 2000000000002,
                        "adgroup_name": "广告二",
                        "dynamic_creative_id": 3000000000003,
                        "dynamic_creative_name": "创意三",
                        "cleanup_action": "DELETE_CREATIVE",
                        "target_element_ids_json": "[]",
                        "target_component_ids_json": "[]",
                        "cleanup_status": "CREATIVE_DELETED",
                        "recent_cost_fen": 54261,
                        "cost_start_date": "2026-08-08",
                        "cost_end_date": "2026-08-10",
                        "action_reason": "创意审核状态为审核拒绝",
                        "reject_reason": "图片违规",
                        "review_result_json": json.dumps(
                            {
                                "element_result_list": [
                                    {
                                        "element_name": "主图",
                                        "image_id": "10001",
                                        "review_status": "REVIEW_STATUS_REJECTED",
                                        "element_reject_detail_info": [
                                            {"reason": "图片违规"}
                                        ],
                                    },
                                    {
                                        "element_name": "正常元素",
                                        "element_id": "30003",
                                        "system_status": "AD_STATUS_NORMAL",
                                    },
                                ],
                                "site_set_result_list": [
                                    {
                                        "site_set": "SITE_SET_MOMENTS",
                                        "site_set_id": "40004",
                                        "system_status": (
                                            "DYNAMIC_CREATIVE_STATUS_DENIED"
                                        ),
                                        "reject_message": "朋友圈版位拒绝",
                                    },
                                    {
                                        "site_set": "SITE_SET_NORMAL",
                                        "review_status": "AD_STATUS_NORMAL",
                                    },
                                ],
                            },
                            ensure_ascii=False,
                        ),
                        "pre_state_json": json.dumps(
                            {
                                "configured_status": "AD_STATUS_NORMAL",
                                "creative_set_approval_status": (
                                    "CREATIVE_SET_APPROVAL_STATUS_DENIED"
                                ),
                            }
                        ),
                        "deleted_at": now,
                    }
                ],
                Path(directory),
                "20260811",
            )
            workbook = load_workbook(reports[0]["report"])
            sheet = workbook["审核不通过创意清理"]

        headers = [cell.value for cell in sheet[1]]
        self.assertEqual(
            headers,
            [
                "代理名称",
                "账户ID",
                "账户名称",
                "广告ID",
                "广告名称",
                "创意ID",
                "创意名称",
                "近3天累计历史消耗(元)",
                "当日消耗(元)",
                "近3天及当日累计消耗(元)",
                "配置状态",
                "创意审核状态",
                "审核不通过原因",
                "执行操作",
            ],
        )
        self.assertEqual(headers, list(AGENCY_REPORT_COLUMNS))
        for column_index, expected in (
            (2, "1000000000001"),
            (4, "2000000000002"),
            (6, "3000000000003"),
        ):
            cell = sheet.cell(2, column_index)
            self.assertEqual(cell.value, expected)
            self.assertEqual(cell.data_type, "s")
            self.assertEqual(cell.number_format, "@")
        by_header = {cell.value: cell.column for cell in sheet[1]}
        self.assertEqual(sheet.cell(2, by_header["配置状态"]).value, "投放中")
        self.assertEqual(
            sheet.cell(2, by_header["创意审核状态"]).value,
            "审核拒绝",
        )
        self.assertEqual(
            sheet.cell(2, by_header["近3天累计历史消耗(元)"]).value,
            "542.61",
        )
        self.assertEqual(
            sheet.cell(2, by_header["执行操作"]).value,
            "删除创意",
        )
        self.assertEqual(sheet.cell(2, by_header["代理名称"]).value, "棱镜")
        self.assertEqual(
            sheet.cell(2, by_header["审核不通过原因"]).value,
            "图片违规",
        )
        for removed_header in (
            "消耗日期范围",
            "操作判断原因",
            "元素粒度审核状态",
            "元素粒度审核不通过原因",
            "版位粒度审核状态",
            "版位粒度审核不通过原因",
            "检查时间",
            "元素ID",
            "组件ID",
            "处理结果",
            "处理时间",
        ):
            self.assertNotIn(removed_header, by_header)
        self.assertEqual(item_ids, {"棱镜": [7]})
        self.assertEqual(reports[0]["notification_type"], "creative_rejection_cleanup")

    def test_cleanup_report_displays_partial_creative_manual_alert(self):
        from tools.creative_rejection_cleanup import write_cleanup_reports

        with tempfile.TemporaryDirectory() as directory:
            _, reports, _ = write_cleanup_reports(
                [
                    {
                        "id": 8,
                        "agency_name": "棱镜",
                        "account_id": 1,
                        "account_name": "账户一",
                        "adgroup_id": 2,
                        "adgroup_name": "广告二",
                        "dynamic_creative_id": 3,
                        "dynamic_creative_name": "创意三",
                        "cleanup_action": "ALERT_ONLY",
                        "target_element_ids_json": "[]",
                        "target_component_ids_json": "[]",
                        "cleanup_status": "ALERT_PENDING",
                        "recent_cost_fen": 3000,
                        "cost_start_date": "2026-08-10",
                        "cost_end_date": "2026-08-12",
                        "reject_reason": "部分投放中，需人工判断是否删除",
                        "review_result_json": "{}",
                        "pre_state_json": json.dumps(
                            {
                                "creative_set_approval_status": (
                                    "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                                )
                            }
                        ),
                        "updated_at": datetime(2026, 8, 12, 11, 0),
                    }
                ],
                Path(directory),
                "20260812",
            )
            sheet = load_workbook(reports[0]["report"])["审核不通过创意清理"]

        columns = {cell.value: cell.column for cell in sheet[1]}
        self.assertEqual(sheet.cell(2, columns["执行操作"]).value, "需人工判断")
        self.assertEqual(sheet.cell(2, columns["创意审核状态"]).value, "部分投放中")
        self.assertEqual(
            sheet.cell(2, columns["近3天累计历史消耗(元)"]).value,
            "30.00",
        )
        self.assertNotIn("操作判断原因", columns)

    def test_operator_summary_report_contains_all_agencies(self):
        from tools.creative_rejection_cleanup import (
            OPERATOR_REPORT_COLUMNS,
            write_cleanup_operator_summary,
        )

        rows = [
            {
                "id": index,
                "agency_name": agency,
                "account_id": index,
                "account_name": f"账户{index}",
                "adgroup_id": index + 10,
                "adgroup_name": f"广告{index}",
                "dynamic_creative_id": index + 20,
                "dynamic_creative_name": f"创意{index}",
                "cleanup_action": "DELETE_CREATIVE",
                "recent_cost_fen": index * 100,
                "cost_start_date": "2026-08-09",
                "cost_end_date": "2026-08-11",
                "action_reason": f"判断原因{index}",
                "reject_reason": "图片违规",
                "review_result_json": "{}",
                "pre_state_json": "{}",
                "updated_at": datetime(2026, 8, 12, 11, 0),
            }
            for index, agency in ((1, "代理A"), (2, "代理B"), (3, ""))
        ]
        with tempfile.TemporaryDirectory() as directory:
            report = write_cleanup_operator_summary(
                rows,
                Path(directory),
                "20260812",
                "reject_20260812_abc123",
            )
            sheet = load_workbook(report["report"])["审核不通过创意清理"]

        agencies = [sheet.cell(row, 1).value or "" for row in range(2, 5)]
        headers = [cell.value for cell in sheet[1]]
        columns = {cell.value: cell.column for cell in sheet[1]}
        self.assertEqual(agencies, ["代理A", "代理B", ""])
        self.assertEqual(headers, list(OPERATOR_REPORT_COLUMNS))
        self.assertEqual(
            sheet.cell(2, columns["近3天累计历史消耗(元)"]).value,
            "1.00",
        )
        self.assertEqual(
            sheet.cell(2, columns["消耗日期范围"]).value,
            "2026-08-09 ~ 2026-08-11",
        )
        self.assertEqual(sheet.cell(2, columns["操作判断原因"]).value, "判断原因1")
        self.assertEqual(headers[-2:], ["执行操作", "操作判断原因"])
        self.assertEqual(report["creative_rows"], 3)
        self.assertIn("投放调控", Path(report["report"]).name)

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

        publisher = Mock()
        publisher.send_report_card.return_value = "message-1"
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "summary.xlsx"
            path.write_bytes(b"xlsx")
            report = {
                "report_version": "summary-v1",
                "report": str(path),
                "title": "创意审核异常处理汇总",
                "creative_rows": 3,
            }
            with patch.object(
                cleanup,
                "upsert_cleanup_delivery",
                return_value={
                    "id": 9,
                    "status": "FAILED",
                    "sheet_url": "https://example.test/existing-summary",
                    "sheet_token": "existing-token",
                },
            ), patch.object(cleanup, "update_cleanup_delivery") as update:
                outcome = cleanup.publish_cleanup_operator_summary(
                    run_id="reject_20260812_abc123",
                    report=report,
                    chat_id="chat-operator",
                    publisher=publisher,
                    now=datetime(2026, 8, 12, 11, 0),
                )

        publisher.upload_workbook.assert_not_called()
        publisher.send_report_card.assert_called_once_with(
            title="创意审核异常处理汇总",
            content="本批次共 **3** 条创意，包含各代理自动删除及需人工判断的完整汇总。",
            sheet_url="https://example.test/existing-summary",
            chat_id="chat-operator",
            button_text="查看全部处理明细",
        )
        self.assertEqual(outcome["status"], "SENT")
        self.assertEqual(update.call_args.kwargs["status"], "SENT")

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

        with patch.dict(
            os.environ,
            {
                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-project",
            },
            clear=False,
        ):
            self.assertEqual(cleanup._operator_summary_chat_id(), "chat-project")

        with patch.dict(
            os.environ,
            {
                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
                "FEISHU_AD_PROJECT_CHAT_ID": "",
            },
            clear=False,
        ):
            self.assertEqual(cleanup._operator_summary_chat_id(), "")

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

        @contextmanager
        def acquired_lock(_name):
            yield True

        class FakeTencent:
            def __init__(self):
                self.deleted = []
                self.seeded_tokens = {}

            def seed_access_tokens(self, tokens):
                self.seeded_tokens.update(tokens)

            def get_dynamic_creatives(self, account_id):
                return [
                    {
                        "dynamic_creative_id": 3,
                        "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",
                    }
                ]

            def get_ads(self, account_id):
                return [{"adgroup_id": 2, "adgroup_name": "广告二"}]

            def get_dynamic_creative(self, account_id, creative_id):
                return {
                    "dynamic_creative_id": creative_id,
                    "adgroup_id": 2,
                    "configured_status": "AD_STATUS_NORMAL",
                    "creative_set_approval_status": (
                        "CREATIVE_SET_APPROVAL_STATUS_DENIED"
                    ),
                    "system_status": "DYNAMIC_CREATIVE_STATUS_DENIED",
                }

            def delete_dynamic_creative(self, account_id, creative_id):
                self.deleted.append((account_id, creative_id))
                return {
                    "dynamic_creative_id": creative_id,
                    "configured_status": "AD_STATUS_DELETED",
                }

        tencent = FakeTencent()
        current = datetime(2026, 8, 11, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai"))
        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": "DELETE_CREATIVE",
            "target_component_ids_json": "[]",
            "target_element_ids_json": "[]",
            "reject_reason": "图片违规",
            "cleanup_status": "DISCOVERED",
        }
        deleted_item = {
            **retry_item,
            "cleanup_status": "CREATIVE_DELETED",
            "deleted_at": current,
        }
        updates = []
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps(
                    {
                        "棱镜": (
                            "https://open.feishu.cn/open-apis/bot/v2/hook/"
                            "test-cleanup-route"
                        )
                    }
                ),
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260810"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame(
                [
                    {
                        "entity_type": "self",
                        "账号id": "1",
                        "账号名称": "账户一",
                        "创意id": "3",
                        "代理名称": "棱镜",
                    }
                ]
            )
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup,
            "prefetch_account_access_tokens",
            return_value={1: "access-token-account-1"},
        ), 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
        ) as claim_item, patch.object(
            cleanup,
            "update_cleanup_item",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[deleted_item]
        ), patch.object(
            cleanup, "advisory_lock", side_effect=acquired_lock
        ), patch.object(
            cleanup,
            "publish_agency_reports",
            return_value=[{"agency_name": "棱镜", "status": "SENT"}],
        ), patch.object(
            cleanup,
            "publish_cleanup_operator_summary",
            return_value={"route": "投放调控汇总", "status": "SENT"},
        ), patch.object(
            cleanup, "mark_cleanup_items_notified"
        ) as mark_notified, patch.object(
            cleanup, "mark_cleanup_items_operator_notified"
        ) as mark_operator_notified:
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=current,
                tencent=tencent,
                odps=Mock(),
                review_fetcher=lambda _account, _ids: [
                    {
                        "dynamic_creative_id": 3,
                        "reject_message_list": ["图片违规"],
                    }
                ],
                publisher=Mock(),
            )

        self.assertEqual(tencent.deleted, [(1, 3)])
        self.assertEqual(tencent.seeded_tokens, {1: "access-token-account-1"})
        self.assertEqual(summary["account_scope"], "opengid_recent_3d_spend")
        self.assertEqual(summary["account_scope_start_date"], "20260808")
        self.assertEqual(summary["account_scope_end_date"], "20260810")
        self.assertEqual(summary["account_ids"], [1])
        self.assertEqual(summary["tokens_prefetched"], 1)
        self.assertEqual(summary["deleted"], 1)
        claim_item.assert_called_once_with(7)
        self.assertTrue(
            any(
                values.get("cleanup_status") == "CREATIVE_DELETED"
                for _, values in updates
            )
        )
        mark_notified.assert_called_once_with([7], current)
        mark_operator_notified.assert_called_once_with([7], current)

    def test_cleanup_defers_item_when_delete_rate_limit_is_exhausted(self):
        from tencent_client import TencentWriteRateLimitedError
        from tools import creative_rejection_cleanup as cleanup

        @contextmanager
        def acquired_lock(_name):
            yield True

        tencent = Mock()
        tencent.get_dynamic_creative.return_value = {
            "dynamic_creative_id": 3,
            "adgroup_id": 2,
            "creative_set_approval_status": cleanup.CREATIVE_DENIED_STATUS,
        }
        tencent.delete_dynamic_creative.side_effect = TencentWriteRateLimitedError(
            "Tencent HTTP 429"
        )
        retry_item = {
            "id": 7,
            "account_id": 1,
            "agency_name": "代理A",
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": cleanup.DELETE_CREATIVE,
            "cleanup_status": "WRITE_OUTCOME_UNKNOWN",
        }
        updates = []

        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps({
                    "代理A": (
                        "https://open.feishu.cn/open-apis/bot/v2/hook/"
                        "test-cleanup-route"
                    )
                }),
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260812"
        ), 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, "prefetch_account_access_tokens", return_value={}
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
        ), patch.object(
            cleanup, "claim_cleanup_item", return_value=True
        ), patch.object(
            cleanup, "cleanup_precondition_failure", return_value=None
        ), patch.object(
            cleanup,
            "update_cleanup_item",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), 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=datetime(2026, 8, 13, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai")),
                tencent=tencent,
                odps=Mock(),
            )

        self.assertEqual(summary["deleted"], 0)
        self.assertEqual(summary["deferred"], 1)
        self.assertTrue(any(
            values.get("cleanup_status") == "DEFERRED"
            and "429" in values.get("error_message", "")
            for _, values in updates
        ))
        self.assertFalse(any(
            values.get("cleanup_status") == "WRITE_OUTCOME_UNKNOWN"
            for _, values in updates
        ))

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

        @contextmanager
        def acquired_lock(_name):
            yield True

        class FakeTencent:
            def __init__(self):
                self.deleted = []
                self.cost_requests = []

            def seed_access_tokens(self, _tokens):
                return None

            def get_dynamic_creatives(self, _account_id):
                return [
                    {
                        "dynamic_creative_id": creative_id,
                        "dynamic_creative_name": f"创意{creative_id}",
                        "adgroup_id": creative_id + 10,
                        "configured_status": "AD_STATUS_NORMAL",
                        "creative_set_approval_status": (
                            "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                        ),
                    }
                    for creative_id in (3, 4)
                ]

            def get_ads(self, _account_id):
                return [
                    {"adgroup_id": creative_id + 10, "adgroup_name": f"广告{creative_id}"}
                    for creative_id in (3, 4)
                ]

            def get_dynamic_creative(self, _account_id, creative_id):
                return {
                    "dynamic_creative_id": creative_id,
                    "adgroup_id": creative_id + 10,
                    "configured_status": "AD_STATUS_NORMAL",
                    "creative_set_approval_status": (
                        "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                    ),
                }

            def get_dynamic_creative_costs(
                self, account_id, creative_ids, start_date, end_date
            ):
                self.cost_requests.append(
                    (account_id, list(creative_ids), start_date, end_date)
                )
                costs = (
                    {3: 2999, 4: 3000}
                    if start_date == date(2026, 8, 9)
                    else {3: 0, 4: 0}
                )
                return {creative_id: costs[creative_id] for creative_id in creative_ids}

            def delete_dynamic_creative(self, account_id, creative_id):
                self.deleted.append((account_id, creative_id))
                return {"dynamic_creative_id": creative_id, "deleted_from_listing": True}

        current = datetime(2026, 8, 12, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai"))
        retry_item = {
            "id": 7,
            "account_id": 1,
            "agency_name": "棱镜",
            "adgroup_id": 13,
            "dynamic_creative_id": 3,
            "cleanup_action": "DELETE_CREATIVE",
            "target_component_ids_json": "[]",
            "target_element_ids_json": "[]",
            "cleanup_status": "DISCOVERED",
        }
        candidates = []
        updates = []
        tencent = FakeTencent()
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "DAILY_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN": "30",
                "DAILY_PARTIAL_CREATIVE_PROTECTION_DAYS": "3",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps(
                    {
                        "棱镜": (
                            "https://open.feishu.cn/open-apis/bot/v2/hook/"
                            "test-cleanup-route"
                        )
                    }
                ),
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260811"
        ), patch.object(
            cleanup,
            "fetch_daily_data",
            return_value=pd.DataFrame(
                [
                    {
                        "entity_type": "self",
                        "账号id": "1",
                        "账号名称": "账户一",
                        "创意id": str(creative_id),
                        "代理名称": "棱镜",
                    }
                    for creative_id in (3, 4)
                ]
            ),
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[
                {
                    "account_id": 1,
                    "adgroup_id": creative_id + 10,
                    "creative_id": creative_id,
                    "create_time": datetime(
                        2026, 8, 1, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai")
                    ),
                }
                for creative_id in (3, 4)
            ],
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token-account-1"}
        ), patch.object(
            cleanup,
            "upsert_cleanup_candidate",
            side_effect=lambda record: candidates.append(record) or {"id": len(candidates)},
        ), 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",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), 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=lambda _account, _ids: [],
            )

        actions = {
            row["dynamic_creative_id"]: row["cleanup_action"] for row in candidates
        }
        self.assertEqual(actions, {3: "DELETE_CREATIVE", 4: "ALERT_ONLY"})
        self.assertEqual(tencent.deleted, [(1, 3)])
        self.assertEqual(summary["deleted"], 1)
        self.assertEqual(len(tencent.cost_requests), 2)
        self.assertEqual(
            [(start.isoformat(), end.isoformat()) for _, _, start, end in tencent.cost_requests],
            [
                ("2026-08-09", "2026-08-11"),
                ("2026-08-12", "2026-08-12"),
            ],
        )
        self.assertTrue(
            any(
                values.get("cleanup_status") == "CREATIVE_DELETED"
                for _, values in updates
            )
        )

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

        @contextmanager
        def acquired_lock(_name):
            yield True

        class FakeTencent:
            def __init__(self):
                self.cost_call_count = 0
                self.deleted = []

            def seed_access_tokens(self, _tokens):
                return None

            def get_dynamic_creatives(self, _account_id):
                return [{
                    "dynamic_creative_id": 3,
                    "adgroup_id": 13,
                    "configured_status": "AD_STATUS_NORMAL",
                    "creative_set_approval_status": (
                        "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
                    ),
                }]

            def get_ads(self, _account_id):
                return [{"adgroup_id": 13}]

            def get_dynamic_creative(self, _account_id, _creative_id):
                return self.get_dynamic_creatives(1)[0]

            def get_dynamic_creative_costs(self, *_args):
                self.cost_call_count += 1
                if self.cost_call_count <= 2:
                    return {3: 2999 if self.cost_call_count == 1 else 0}
                raise AssertionError("review delete must not query spend inside write lock")

            def delete_dynamic_creative(self, account_id, creative_id):
                self.deleted.append((account_id, creative_id))

        retry_item = {
            "id": 7,
            "account_id": 1,
            "agency_name": "棱镜",
            "adgroup_id": 13,
            "dynamic_creative_id": 3,
            "cleanup_action": "DELETE_CREATIVE",
            "target_component_ids_json": "[]",
            "target_element_ids_json": "[]",
            "cleanup_status": "DISCOVERED",
        }
        updates = []
        tencent = FakeTencent()
        current = datetime(2026, 8, 12, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai"))
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "DAILY_PARTIAL_CREATIVE_PROTECTION_DAYS": "3",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps(
                    {
                        "棱镜": (
                            "https://open.feishu.cn/open-apis/bot/v2/hook/"
                            "test-cleanup-route"
                        )
                    }
                ),
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260811"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(
            cleanup,
            "fetch_active_creative_inventory",
            return_value=[
                {
                    "account_id": 1,
                    "adgroup_id": 13,
                    "creative_id": 3,
                    "create_time": datetime(
                        2026, 8, 1, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai")
                    ),
                }
            ],
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token-account-1"}
        ), 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",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), 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=lambda _account, _ids: [],
            )

        self.assertEqual(tencent.cost_call_count, 2)
        self.assertEqual(tencent.deleted, [(1, 3)])
        self.assertEqual(summary["deleted"], 1)
        self.assertTrue(
            any(
                values.get("cleanup_status") == "CREATIVE_DELETED"
                for _, values in updates
            )
        )

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

        @contextmanager
        def acquired_lock(_name):
            yield True

        class FakeTencent:
            def __init__(self):
                self.deleted_creatives = []
                self.deleted_components = []

            def seed_access_tokens(self, _tokens):
                return None

            def get_dynamic_creatives(self, _account_id):
                return [{
                    "dynamic_creative_id": 3,
                    "adgroup_id": 2,
                    "configured_status": "AD_STATUS_NORMAL",
                }]

            def get_ads(self, _account_id):
                return [{"adgroup_id": 2}]

            def get_dynamic_creative(self, _account_id, _creative_id):
                return {
                    "dynamic_creative_id": 3,
                    "adgroup_id": 2,
                    "configured_status": "AD_STATUS_NORMAL",
                }

            def get_creative_component(
                self, _account_id, component_id, *, include_deleted=True
            ):
                return {"component_id": component_id, "is_deleted": False}

            def delete_creative_component(self, account_id, component_id):
                self.deleted_components.append((account_id, component_id))
                return {"component_id": component_id, "is_deleted": True}

            def delete_dynamic_creative(self, account_id, creative_id):
                self.deleted_creatives.append((account_id, creative_id))

        review = {
            "dynamic_creative_id": 3,
            "element_result_list": [
                {
                    "element_id": 102,
                    "review_status": "AD_STATUS_DENIED",
                    "component_info": {"component_id": 12},
                },
            ],
        }
        retry_item = {
            "id": 7,
            "account_id": 1,
            "agency_name": "",
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": "DELETE_COMPONENTS",
            "target_component_ids_json": "[11, 12]",
            "target_element_ids_json": "[101, 102]",
            "readback_json": json.dumps(
                {"component_results": {"11": {"status": "DELETED"}}}
            ),
            "cleanup_status": "COMPONENTS_PARTIAL",
        }
        tencent = FakeTencent()
        updates = []
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                "ROI_AGENCY_WEBHOOKS_JSON": "{}",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260810"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token-account-1"}
        ), patch.object(
            cleanup, "upsert_cleanup_candidate"
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
        ), patch.object(
            cleanup,
            "update_cleanup_item",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), 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),
                tencent=tencent,
                odps=Mock(),
                review_fetcher=lambda _account, _ids: [review],
            )

        self.assertEqual(tencent.deleted_creatives, [])
        self.assertEqual(tencent.deleted_components, [])
        self.assertEqual(summary["deleted"], 0)
        self.assertTrue(
            any(
                values.get("cleanup_status") == "SKIPPED_REVIEW_NOT_RECONFIRMED"
                for _, values in updates
            )
        )

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

        @contextmanager
        def acquired_lock(_name):
            yield True

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [{
            "dynamic_creative_id": 3,
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
        }]
        tencent.get_ads.return_value = [{"adgroup_id": 2}]
        tencent.get_dynamic_creative.return_value = {
            "dynamic_creative_id": 3,
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
        }
        review = {
            "dynamic_creative_id": 3,
            "element_result_list": [
                {"element_id": 101, "review_status": "AD_STATUS_DENIED"}
            ],
        }
        retry_item = {
            "id": 7,
            "account_id": 1,
            "agency_name": "",
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": "DELETE_COMPONENTS",
            "target_component_ids_json": "[]",
            "target_element_ids_json": "[101]",
            "cleanup_status": "DISCOVERED",
        }
        updates = []
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                "ROI_AGENCY_WEBHOOKS_JSON": "{}",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260810"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={1: "token-account-1"}
        ), patch.object(
            cleanup, "upsert_cleanup_candidate"
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
        ), patch.object(
            cleanup,
            "update_cleanup_item",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ), patch.object(
            cleanup, "advisory_lock", side_effect=acquired_lock
        ):
            cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                tencent=tencent,
                odps=Mock(),
                review_fetcher=lambda _account, _ids: [review],
            )

        tencent.delete_dynamic_creative.assert_not_called()
        tencent.delete_creative_component.assert_not_called()
        self.assertTrue(
            any(
                values.get("cleanup_status") == "SKIPPED_REVIEW_NOT_RECONFIRMED"
                for _, values in updates
            )
        )

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

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [
            {
                "dynamic_creative_id": 3,
                "adgroup_id": 2,
                "system_status": "DYNAMIC_CREATIVE_STATUS_DENIED",
            }
        ]
        tencent.get_ads.return_value = [{"adgroup_id": 2}]
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260810"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(
            cleanup, "fetch_account_agency_fallbacks", return_value={}
        ), patch.object(
            cleanup,
            "prefetch_account_access_tokens",
            return_value={1: "access-token-account-1"},
        ), patch.object(
            cleanup, "upsert_cleanup_candidate"
        ), patch.object(
            cleanup, "load_retryable_cleanup_items", return_value=[{"id": 7}]
        ), patch.object(
            cleanup, "load_unnotified_deleted_items", return_value=[]
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                tencent=tencent,
                odps=Mock(),
                review_fetcher=lambda _account, _ids: [],
            )

        tencent.delete_dynamic_creative.assert_not_called()
        self.assertFalse(summary["apply_enabled"])
        self.assertEqual(summary["pending_cleanup"], 1)

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

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = []
        tencent.get_ads.return_value = []
        pending_alert = {
            "id": 7,
            "agency_name": "棱镜",
            "cleanup_status": "ALERT_PENDING",
        }
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
                "FEISHU_APP_ID": "",
                "FEISHU_APP_SECRET": "",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260810"
        ), patch.object(
            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
        ), patch.object(
            cleanup, "fetch_recent_spend_accounts", 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", return_value=[pending_alert]
        ), patch.object(
            cleanup, "RoiFeishuPublisher"
        ) as publisher, patch.object(
            cleanup, "publish_agency_reports"
        ) as publish:
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                tencent=tencent,
                odps=Mock(),
            )

        publisher.assert_not_called()
        publish.assert_not_called()
        self.assertEqual(summary["deliveries"], [])

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

        with patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps(
                    {
                        "棱镜": (
                            "https://open.feishu.cn/open-apis/bot/v2/hook/"
                            "test-cleanup-route"
                        )
                    }
                ),
                "FEISHU_OPERATOR_CHAT_ID": "",
                "FEISHU_AD_PROJECT_CHAT_ID": "",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema") as initialize:
            with self.assertRaisesRegex(RuntimeError, "FEISHU_AD_PROJECT_CHAT_ID"):
                cleanup.run_rejected_creative_cleanup(output_dir=Path("unused"))

        initialize.assert_not_called()

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

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = [{
            "dynamic_creative_id": 3,
            "adgroup_id": 2,
            "creative_set_approval_status": cleanup.CREATIVE_DENIED_STATUS,
        }]
        tencent.get_ads.return_value = [{"adgroup_id": 2}]
        retry_item = {
            "id": 7,
            "account_id": 1,
            "agency_name": "",
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": cleanup.DELETE_CREATIVE,
            "cleanup_status": "DISCOVERED",
        }
        updates = []
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps({
                    "代理A": (
                        "https://open.feishu.cn/open-apis/bot/v2/hook/"
                        "test-cleanup-route"
                    )
                }),
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260812"
        ), patch.object(cleanup, "fetch_daily_data", return_value=pd.DataFrame()), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(cleanup, "fetch_account_agency_fallbacks", return_value={}), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={}
        ), 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",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), patch.object(cleanup, "load_unnotified_deleted_items", return_value=[]), patch.object(
            cleanup, "advisory_lock"
        ) as lock:
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path(directory),
                now=datetime(2026, 8, 13, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai")),
                tencent=tencent,
                odps=Mock(),
                review_fetcher=lambda _account, _ids: [],
            )

        tencent.delete_dynamic_creative.assert_not_called()
        lock.assert_not_called()
        self.assertEqual(summary["deferred"], 1)
        self.assertTrue(any(
            values.get("cleanup_status") == "DEFERRED"
            and "代理商归属为空" in values.get("error_message", "")
            for _, values in updates
        ))

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

        @contextmanager
        def acquired_lock(_name):
            yield True

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = []
        tencent.get_ads.return_value = []
        tencent.get_dynamic_creative.side_effect = RuntimeError(
            "Dynamic creative not found: account=1 creative=3"
        )
        retry_item = {
            "id": 7,
            "account_id": 1,
            "agency_name": "代理A",
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": cleanup.DELETE_CREATIVE,
            "cleanup_status": "WRITE_OUTCOME_UNKNOWN",
        }
        updates = []
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps({
                    "代理A": (
                        "https://open.feishu.cn/open-apis/bot/v2/hook/"
                        "test-cleanup-route"
                    )
                }),
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260812"
        ), patch.object(cleanup, "fetch_daily_data", return_value=pd.DataFrame()), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(cleanup, "fetch_account_agency_fallbacks", return_value={}), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={}
        ), 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",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), 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=datetime(2026, 8, 13, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai")),
                tencent=tencent,
                odps=Mock(),
            )

        tencent.delete_dynamic_creative.assert_not_called()
        self.assertEqual(summary["deleted"], 1)
        self.assertTrue(any(
            values.get("cleanup_status") == "CREATIVE_DELETED"
            for _, values in updates
        ))

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

        @contextmanager
        def acquired_lock(_name):
            yield True

        tencent = Mock()
        tencent.get_dynamic_creatives.return_value = []
        tencent.get_ads.return_value = []
        tencent.get_dynamic_creative.return_value = {
            "dynamic_creative_id": 3,
            "adgroup_id": 2,
            "configured_status": "AD_STATUS_NORMAL",
            "system_status": "DYNAMIC_CREATIVE_STATUS_DELETED",
            "creative_set_approval_status": cleanup.CREATIVE_PARTIAL_NORMAL_STATUS,
        }
        retry_item = {
            "id": 7,
            "account_id": 1,
            "agency_name": "代理A",
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": cleanup.DELETE_CREATIVE,
            "cleanup_status": "WRITE_OUTCOME_UNKNOWN",
        }
        updates = []
        with tempfile.TemporaryDirectory() as directory, patch.dict(
            os.environ,
            {
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps({
                    "代理A": (
                        "https://open.feishu.cn/open-apis/bot/v2/hook/"
                        "test-cleanup-route"
                    )
                }),
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date", return_value="20260812"
        ), patch.object(cleanup, "fetch_daily_data", return_value=pd.DataFrame()), patch.object(
            cleanup, "fetch_recent_spend_accounts", return_value=[{"account_id": 1}]
        ), patch.object(cleanup, "fetch_account_agency_fallbacks", return_value={}), patch.object(
            cleanup, "prefetch_account_access_tokens", return_value={}
        ), 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",
            side_effect=lambda item_id, **values: updates.append((item_id, values)),
        ), 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=datetime(2026, 8, 13, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai")),
                tencent=tencent,
                odps=Mock(),
            )

        tencent.delete_dynamic_creative.assert_not_called()
        self.assertEqual(summary["deleted"], 1)
        self.assertTrue(any(
            values.get("cleanup_status") == "CREATIVE_DELETED"
            and "DYNAMIC_CREATIVE_STATUS_DELETED" in values.get("readback_json", "")
            for _, values in updates
        ))

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

        base_row = {
            "id": 7,
            "check_date": date(2026, 8, 13),
            "agency_name": "代理A",
            "account_id": 1,
            "adgroup_id": 2,
            "dynamic_creative_id": 3,
            "cleanup_action": cleanup.ALERT_ONLY,
            "cleanup_status": "ALERT_PENDING",
            "reject_reason": "需人工判断",
        }

        def run_with(row, *, lock_acquired=True):
            @contextmanager
            def notification_lock(_name):
                yield lock_acquired

            with tempfile.TemporaryDirectory() as directory, patch.dict(
                os.environ,
                {
                    "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "0",
                    "ROI_AGENCY_WEBHOOK_ENABLED": "1",
                    "ROI_AGENCY_WEBHOOKS_JSON": json.dumps({
                        "代理A": (
                            "https://open.feishu.cn/open-apis/bot/v2/hook/"
                            "test-cleanup-route"
                        )
                    }),
                    "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                },
                clear=False,
            ), patch.object(cleanup, "initialize_schema"), patch.object(
                cleanup, "resolve_end_date", return_value="20260812"
            ), patch.object(cleanup, "fetch_daily_data", return_value=pd.DataFrame()), patch.object(
                cleanup, "fetch_recent_spend_accounts", 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", return_value=[row]), patch.object(
                cleanup, "advisory_lock", side_effect=notification_lock
            ), patch.object(
                cleanup, "publish_agency_reports", return_value=[{
                    "agency_name": "代理A", "status": "SENT"
                }]
            ) as agency_publish, patch.object(
                cleanup, "publish_cleanup_operator_summary", return_value={
                    "route": cleanup.OPERATOR_SUMMARY_ROUTE, "status": "SENT"
                }
            ) as operator_publish, patch.object(
                cleanup, "mark_cleanup_items_notified"
            ) as mark_agency, patch.object(
                cleanup, "mark_cleanup_items_operator_notified"
            ) as mark_operator:
                cleanup.run_rejected_creative_cleanup(
                    output_dir=Path(directory),
                    now=datetime(2026, 8, 13, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai")),
                    tencent=Mock(),
                    odps=Mock(),
                    publisher=Mock(),
                )
            return agency_publish, operator_publish, mark_agency, mark_operator

        agency, operator, mark_agency, mark_operator = run_with({
            **base_row,
            "agency_notified_at": datetime(2026, 8, 13, 11, 0),
            "operator_notified_at": None,
        })
        agency.assert_not_called()
        operator.assert_called_once()
        mark_agency.assert_not_called()
        mark_operator.assert_called_once()

        agency, operator, mark_agency, mark_operator = run_with({
            **base_row,
            "agency_notified_at": None,
            "operator_notified_at": datetime(2026, 8, 13, 11, 0),
        })
        agency.assert_called_once()
        operator.assert_not_called()
        mark_agency.assert_called_once()
        mark_operator.assert_not_called()

        agency, operator, mark_agency, mark_operator = run_with(
            {
                **base_row,
                "agency_notified_at": None,
                "operator_notified_at": None,
            },
            lock_acquired=False,
        )
        agency.assert_not_called()
        operator.assert_not_called()
        mark_agency.assert_not_called()
        mark_operator.assert_not_called()

    def test_cleanup_entry_returns_nonzero_for_notification_failure(self):
        from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry

        summary = {
            "deleted": 3,
            "deferred": 1,
            "scan_errors": [],
            "delete_errors": [],
            "notification_errors": ["operator failed"],
        }
        with patch("logging_setup.setup_logging"), patch.object(
            entry, "run_rejected_creative_cleanup", return_value=summary
        ), patch.object(entry.logger, "error") as error_log:
            self.assertEqual(entry.main(), 1)

        error_log.assert_called_once()
        message, *values = error_log.call_args.args
        self.assertIn("result=partial_failure", message)
        self.assertEqual(values[:5], [3, 1, 0, 0, 1])
        self.assertIn("operator failed", values[5])

    def test_cleanup_entry_delete_failure_notifies_but_returns_zero(self):
        from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry

        delete_errors = [
            "account=1 creative=2: delete_dynamic_creative failed: code=30000"
        ]
        summary = {
            "deleted": 3,
            "deferred": 1,
            "scan_errors": [],
            "delete_errors": delete_errors,
            "notification_errors": [],
        }
        with patch("logging_setup.setup_logging"), patch.object(
            entry, "run_rejected_creative_cleanup", return_value=summary
        ), patch.object(
            entry, "_notify_delete_failures"
        ) as notify_delete_failures:
            self.assertEqual(entry.main(), 0)

        notify_delete_failures.assert_called_once_with(delete_errors)

    def test_delete_failure_notification_reuses_cleanup_failure_destination(self):
        from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry

        publisher = Mock()
        with patch.dict(
            os.environ,
            {"CREATIVE_CLEANUP_FAILURE_FEISHU_CHAT_ID": "chat-cleanup-failure"},
            clear=False,
        ), patch.object(
            entry, "RoiFeishuPublisher", return_value=publisher
        ):
            entry._notify_delete_failures(["account=1 creative=2: 腾讯系统繁忙"])

        publisher.send_service_alert.assert_called_once()
        alert = publisher.send_service_alert.call_args.kwargs
        self.assertEqual(alert["chat_id"], "chat-cleanup-failure")
        self.assertIn("删除失败", alert["title"])
        self.assertIn("不影响任务成功状态", alert["content"])
        publisher.close.assert_called_once_with()

    def test_cleanup_entry_exposes_underperformance_preview_only_mode(self):
        from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry

        summary = {
            "scan_errors": [],
            "delete_errors": [],
            "notification_errors": [],
        }
        output_dir = Path("/tmp/underperformance-preview-test")
        with patch("logging_setup.setup_logging"), patch.object(
            entry, "run_rejected_creative_cleanup", return_value=summary
        ) as run:
            self.assertEqual(
                entry.main([
                    "--underperformance-preview-only",
                    "--output-dir",
                    str(output_dir),
                ]),
                0,
            )

        run.assert_called_once_with(
            output_dir=output_dir,
            underperformance_preview_only=True,
            force_notification=False,
        )

    def test_force_notification_requires_underperformance_preview(self):
        from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry

        with self.assertRaises(SystemExit):
            entry.parse_args(["--force-notification"])

    def test_cleanup_entry_exposes_force_notification_mode(self):
        from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry

        summary = {
            "scan_errors": [],
            "delete_errors": [],
            "notification_errors": [],
        }
        with patch("logging_setup.setup_logging"), patch.object(
            entry, "run_rejected_creative_cleanup", return_value=summary
        ) as run:
            self.assertEqual(
                entry.main([
                    "--underperformance-preview-only",
                    "--force-notification",
                ]),
                0,
            )

        run.assert_called_once_with(
            output_dir=entry.HERE / "outputs" / "creative_rejection_cleanup",
            underperformance_preview_only=True,
            force_notification=True,
        )

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

        review_pending_row = {
            "id": 99,
            "check_date": date(2026, 8, 20),
            "cleanup_rule_type": cleanup.REVIEW_DENIED_RULE,
        }
        tencent = Mock()
        with patch.dict(
            os.environ,
            {
                "FEISHU_AD_PROJECT_CHAT_ID": "chat-internal",
                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED": "1",
                "DAILY_UNDERPERFORMING_AD_APPLY_ENABLED": "1",
            },
            clear=False,
        ), patch.object(cleanup, "initialize_schema"), patch.object(
            cleanup, "resolve_end_date"
        ) as resolve_review_date, patch.object(
            cleanup, "fetch_daily_data"
        ) as fetch_review_context, patch.object(
            cleanup, "fetch_recent_spend_accounts"
        ) as fetch_review_accounts, patch.object(
            cleanup, "fetch_account_agency_fallbacks"
        ) as fetch_agency_fallbacks, 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",
            return_value=[review_pending_row],
        ):
            summary = cleanup.run_rejected_creative_cleanup(
                output_dir=Path("/tmp/underperformance-preview-test"),
                now=datetime(
                    2026,
                    8,
                    20,
                    11,
                    tzinfo=ZoneInfo("Asia/Shanghai"),
                ),
                tencent=tencent,
                odps=Mock(),
                publisher=Mock(),
                underperformance_preview_only=True,
            )

        self.assertTrue(summary["underperformance_preview_only"])
        self.assertTrue(summary["performance_cleanup_enabled"])
        self.assertTrue(summary["ad_cleanup_enabled"])
        self.assertFalse(summary["apply_enabled"])
        self.assertFalse(summary["performance_apply_enabled"])
        self.assertFalse(summary["ad_apply_enabled"])
        resolve_review_date.assert_not_called()
        fetch_review_context.assert_not_called()
        fetch_review_accounts.assert_not_called()
        fetch_agency_fallbacks.assert_not_called()
        tencent.delete_dynamic_creative.assert_not_called()
        tencent.delete_ad.assert_not_called()

    def test_logging_accepts_only_safe_inherited_child_trace_id(self):
        from examples.auto_put_ad_mini import logging_setup

        with patch.dict(
            os.environ,
            {"AUTO_PUT_TRACE_ID": "parent.cleanup.1234.abcd1234"},
            clear=False,
        ):
            self.assertEqual(
                logging_setup._inherited_trace_id(),
                "parent.cleanup.1234.abcd1234",
            )
        with patch.dict(
            os.environ,
            {"AUTO_PUT_TRACE_ID": "unsafe\nforged-log"},
            clear=False,
        ):
            self.assertIsNone(logging_setup._inherited_trace_id())

    def test_scheduled_subprocess_receives_searchable_child_trace_id(self):
        import run_daily_service as daily_service

        @contextmanager
        def acquired_lock():
            yield True

        with patch.object(
            daily_service, "advisory_lock", return_value=acquired_lock()
        ), patch.object(
            daily_service, "get_trace_id", return_value="parent-trace"
        ), patch.object(
            daily_service.time, "time_ns", return_value=123456789
        ), patch.object(
            daily_service.secrets, "token_hex", return_value="abcd1234"
        ), patch.object(
            daily_service.subprocess,
            "run",
            return_value=Mock(returncode=7),
        ) as run:
            with self.assertRaises(RuntimeError) as raised:
                daily_service._run_script("failing_task.py", "test-lock")

        expected_trace = "parent-trace.failing_task.123456789.abcd1234"
        self.assertIn(expected_trace, str(raised.exception))
        self.assertEqual(
            run.call_args.kwargs["env"]["AUTO_PUT_TRACE_ID"],
            expected_trace,
        )


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