Agent/examples/auto_put_ad_mini/tools/creative_rejection_cleanup.py

"""先清理符合规则的创意，再清理零消耗广告，并隔离不同通知渠道。"""

from __future__ import annotations

import hashlib
import json
import logging
import math
import os
import threading
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Callable
from zoneinfo import ZoneInfo

import pandas as pd
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter

from db.connection import get_connection
from roi_control.agency_delivery import publish_agency_reports, resolve_agency_webhook
from roi_control.config import AgencyWebhookConfig
from roi_control.feishu import RoiFeishuPublisher
from storage import advisory_lock, initialize_schema
from tools.creative_review import (
    fetch_dynamic_creative_review_results,
    parse_review_result,
    review_granularity_fields,
    status_desc,
)


logger = logging.getLogger(__name__)
SHANGHAI = ZoneInfo("Asia/Shanghai")
REPORT_VERSION = "creative_rejection_cleanup_v13"
OPERATOR_SUMMARY_ROUTE = "投放调控汇总"
PERFORMANCE_SUMMARY_ROUTE = "长期未起量清理汇总"
PERFORMANCE_REPORT_VERSION = (
    "creative_rejection_v13_performance_internal_acct_meta_v2"
)
DENIED_SYSTEM_STATUS = "DYNAMIC_CREATIVE_STATUS_DENIED"
DELETED_STATUS = "AD_STATUS_DELETED"
DYNAMIC_CREATIVE_DELETED_STATUS = "DYNAMIC_CREATIVE_STATUS_DELETED"
CREATIVE_DENIED_STATUS = "CREATIVE_SET_APPROVAL_STATUS_DENIED"
CREATIVE_NORMAL_STATUS = "CREATIVE_SET_APPROVAL_STATUS_NORMAL"
CREATIVE_PARTIAL_NORMAL_STATUS = "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
DELETE_CREATIVE = "DELETE_CREATIVE"
ALERT_ONLY = "ALERT_ONLY"
REVIEW_DENIED_RULE = "REVIEW_DENIED"
REVIEW_PARTIAL_RULE = "REVIEW_PARTIAL"
PERFORMANCE_NEW_RULE = "PERFORMANCE_NEW_ZERO_DELIVERY"
PERFORMANCE_OLD_LOW_RULE = "PERFORMANCE_OLD_LOW_EXPOSURE_ZERO_COST"
PERFORMANCE_OLD_HIGH_RULE = "PERFORMANCE_OLD_HIGH_EXPOSURE_ZERO_COST"
PERFORMANCE_AD_ZERO_SPEND_RULE = "PERFORMANCE_AD_3D_ZERO_SPEND"
PERFORMANCE_RULE_PREFIX = "PERFORMANCE_"
DELETE_AD = "DELETE_AD"
DEFAULT_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN = 50.0
DEFAULT_WECHAT_MINI_PROGRAM_COST_THRESHOLD_YUAN = 100.0
DEFAULT_DELETE_CLAIM_STALE_MINUTES = 30
DEFAULT_PERFORMANCE_NEW_MIN_AGE_DAYS = 5
DEFAULT_PERFORMANCE_NEW_MAX_AGE_DAYS = 7
DEFAULT_PERFORMANCE_NEW_IMPRESSIONS_THRESHOLD = 100
DEFAULT_PERFORMANCE_OLD_DAILY_IMPRESSIONS_THRESHOLD = 100.0
DEFAULT_PERFORMANCE_WINDOW_DAYS = 7
DEFAULT_AD_PERFORMANCE_WINDOW_DAYS = 3
DEFAULT_AD_PERFORMANCE_MIN_AGE_DAYS = 5
AGENCY_REPORT_COLUMNS = (
    "代理名称",
    "账户ID",
    "账户名称",
    "广告ID",
    "广告名称",
    "创意ID",
    "创意名称",
    "近3天累计历史消耗(元)",
    "配置状态",
    "创意审核状态",
    "审核不通过原因",
    "执行操作",
)
OPERATOR_REPORT_COLUMNS = (
    "代理名称",
    "账户ID",
    "账户名称",
    "广告ID",
    "广告名称",
    "创意ID",
    "创意名称",
    "清理规则",
    "创意搭建时间",
    "创意年龄(天)",
    "规则窗口累计曝光",
    "规则窗口日均曝光",
    "规则窗口累计消耗(元)",
    "规则指标日期范围",
    "近3天累计历史消耗(元)",
    "消耗日期范围",
    "配置状态",
    "创意审核状态",
    "元素粒度审核状态",
    "元素粒度审核不通过原因",
    "版位粒度审核状态",
    "版位粒度审核不通过原因",
    "审核不通过原因",
    "检查时间",
    "执行操作",
    "操作判断原因",
)
PERFORMANCE_REPORT_COLUMNS = (
    "清理对象",
    "账户ID",
    "账户名称",
    "代理商昵称",
    "广告ID",
    "广告名称",
    "创意ID",
    "创意名称",
    "清理规则",
    "对象创建时间",
    "对象年龄(天)",
    "规则窗口累计曝光",
    "规则窗口日均曝光",
    "规则窗口累计消耗(元)",
    "广告近3日日均消耗(元)",
    "规则指标日期范围",
    "配置状态",
    "创意审核状态",
    "检查时间",
    "执行操作",
    "操作判断原因",
)
# 兼容仍将代理商报表视为默认报表的旧调用方。
REPORT_COLUMNS = AGENCY_REPORT_COLUMNS


def _operator_summary_chat_id() -> str:
    """读取投放运营清理汇总使用的内部群 ID。"""

    return os.getenv("FEISHU_AD_PROJECT_CHAT_ID", "").strip()


def _json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, default=str)


def _is_rejected_status(value: Any) -> bool:
    upper = str(value or "").strip().upper()
    return "REJECT" in upper or "DENIED" in upper


def _is_deleted_creative(creative: dict[str, Any]) -> bool:
    return (
        creative.get("configured_status") == DELETED_STATUS
        or creative.get("system_status") == DYNAMIC_CREATIVE_DELETED_STATUS
    )


def _is_deleted_ad(ad: dict[str, Any]) -> bool:
    """所有读取链路统一使用腾讯广告的显式删除字段判断状态。"""

    from tencent_client import is_deleted_ad

    return is_deleted_ad(ad)


def has_rejected_wechat_mini_program_element(raw_result: dict | None) -> bool:
    """判断审核结果中是否存在明确名为“微信小程序”的拒审元素。"""

    raw = raw_result if isinstance(raw_result, dict) else {}
    for element in raw.get("element_result_list") or []:
        if not isinstance(element, dict):
            continue
        element_name = "".join(str(element.get("element_name") or "").split())
        if element_name != "微信小程序":
            continue
        if any(
            _is_rejected_status(element.get(field))
            for field in ("review_status", "system_status")
        ):
            return True
    return False


def partial_creative_cost_threshold_fen() -> int:
    raw = os.getenv(
        "DAILY_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN",
        str(DEFAULT_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN),
    )
    try:
        yuan = float(raw)
    except (TypeError, ValueError) as exc:
        raise ValueError(
            "DAILY_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN must be numeric"
        ) from exc
    if yuan < 0:
        raise ValueError(
            "DAILY_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN must not be negative"
        )
    return int(round(yuan * 100))


def wechat_mini_program_cost_threshold_fen() -> int:
    raw = os.getenv(
        "DAILY_WECHAT_MINI_PROGRAM_CREATIVE_COST_THRESHOLD_YUAN",
        str(DEFAULT_WECHAT_MINI_PROGRAM_COST_THRESHOLD_YUAN),
    )
    try:
        yuan = float(raw)
    except (TypeError, ValueError) as exc:
        raise ValueError(
            "DAILY_WECHAT_MINI_PROGRAM_CREATIVE_COST_THRESHOLD_YUAN must be numeric"
        ) from exc
    if yuan < 0:
        raise ValueError(
            "DAILY_WECHAT_MINI_PROGRAM_CREATIVE_COST_THRESHOLD_YUAN must not be negative"
        )
    return int(round(yuan * 100))


def performance_cleanup_config() -> dict[str, int | float]:
    """读取并校验长期未起量清理阈值。"""

    names_and_defaults: tuple[tuple[str, int | float, type], ...] = (
        (
            "DAILY_UNDERPERFORMING_CREATIVE_NEW_MIN_AGE_DAYS",
            DEFAULT_PERFORMANCE_NEW_MIN_AGE_DAYS,
            int,
        ),
        (
            "DAILY_UNDERPERFORMING_CREATIVE_NEW_MAX_AGE_DAYS",
            DEFAULT_PERFORMANCE_NEW_MAX_AGE_DAYS,
            int,
        ),
        (
            "DAILY_UNDERPERFORMING_CREATIVE_NEW_IMPRESSIONS_THRESHOLD",
            DEFAULT_PERFORMANCE_NEW_IMPRESSIONS_THRESHOLD,
            int,
        ),
        (
            "DAILY_UNDERPERFORMING_CREATIVE_OLD_DAILY_IMPRESSIONS_THRESHOLD",
            DEFAULT_PERFORMANCE_OLD_DAILY_IMPRESSIONS_THRESHOLD,
            float,
        ),
        (
            "DAILY_UNDERPERFORMING_CREATIVE_WINDOW_DAYS",
            DEFAULT_PERFORMANCE_WINDOW_DAYS,
            int,
        ),
    )
    values: dict[str, int | float] = {}
    for name, default, converter in names_and_defaults:
        raw = os.getenv(name, str(default))
        try:
            value = converter(raw)
        except (TypeError, ValueError) as exc:
            raise ValueError(f"{name} must be numeric") from exc
        if isinstance(value, float) and not math.isfinite(value):
            raise ValueError(f"{name} must be finite")
        if value < 0:
            raise ValueError(f"{name} must not be negative")
        values[name] = value
    min_age = int(values["DAILY_UNDERPERFORMING_CREATIVE_NEW_MIN_AGE_DAYS"])
    max_age = int(values["DAILY_UNDERPERFORMING_CREATIVE_NEW_MAX_AGE_DAYS"])
    window_days = int(values["DAILY_UNDERPERFORMING_CREATIVE_WINDOW_DAYS"])
    if max_age <= min_age:
        raise ValueError(
            "DAILY_UNDERPERFORMING_CREATIVE_NEW_MAX_AGE_DAYS must be greater "
            "than DAILY_UNDERPERFORMING_CREATIVE_NEW_MIN_AGE_DAYS"
        )
    if window_days <= 0:
        raise ValueError(
            "DAILY_UNDERPERFORMING_CREATIVE_WINDOW_DAYS must be positive"
        )
    return {
        "new_min_age_days": min_age,
        "new_max_age_days": max_age,
        "new_impressions_threshold": int(
            values[
                "DAILY_UNDERPERFORMING_CREATIVE_NEW_IMPRESSIONS_THRESHOLD"
            ]
        ),
        "old_daily_impressions_threshold": float(
            values[
                "DAILY_UNDERPERFORMING_CREATIVE_OLD_DAILY_IMPRESSIONS_THRESHOLD"
            ]
        ),
        "window_days": window_days,
    }


def ad_performance_window_days() -> int:
    raw = os.getenv(
        "DAILY_UNDERPERFORMING_AD_WINDOW_DAYS",
        str(DEFAULT_AD_PERFORMANCE_WINDOW_DAYS),
    )
    try:
        value = int(raw)
    except (TypeError, ValueError) as exc:
        raise ValueError(
            "DAILY_UNDERPERFORMING_AD_WINDOW_DAYS must be an integer"
        ) from exc
    if value <= 0:
        raise ValueError(
            "DAILY_UNDERPERFORMING_AD_WINDOW_DAYS must be positive"
        )
    return value


def ad_performance_min_age_days() -> int:
    raw = os.getenv(
        "DAILY_UNDERPERFORMING_AD_MIN_AGE_DAYS",
        str(DEFAULT_AD_PERFORMANCE_MIN_AGE_DAYS),
    )
    try:
        value = int(raw)
    except (TypeError, ValueError) as exc:
        raise ValueError(
            "DAILY_UNDERPERFORMING_AD_MIN_AGE_DAYS must be an integer"
        ) from exc
    if value < 0:
        raise ValueError(
            "DAILY_UNDERPERFORMING_AD_MIN_AGE_DAYS must not be negative"
        )
    return value


def _is_performance_rule(rule_type: Any) -> bool:
    return str(rule_type or "").startswith(PERFORMANCE_RULE_PREFIX)


def _is_ad_performance_rule(rule_type: Any) -> bool:
    return str(rule_type or "") == PERFORMANCE_AD_ZERO_SPEND_RULE


def _as_shanghai_datetime(value: Any) -> datetime | None:
    if value is None or (isinstance(value, str) and not value.strip()):
        return None
    if not isinstance(value, (date, datetime)):
        try:
            if bool(pd.isna(value)):
                return None
        except (TypeError, ValueError):
            return None
    try:
        raw_text = str(value).strip()
        unsigned_text = raw_text.lstrip("+-")
        numeric_value: float | None = None
        if unsigned_text.isdigit():
            # ODPS 导出结果可能把紧凑日期值表示为整数。
            if len(unsigned_text) in {8, 14} and not raw_text.startswith("-"):
                calendar_format = (
                    "%Y%m%d" if len(unsigned_text) == 8 else "%Y%m%d%H%M%S"
                )
                parsed_calendar = datetime.strptime(
                    unsigned_text,
                    calendar_format,
                )
                return parsed_calendar.replace(tzinfo=SHANGHAI)
            numeric_value = float(raw_text)
        elif not isinstance(value, (str, date, datetime, pd.Timestamp)):
            try:
                numeric_value = float(value)
            except (TypeError, ValueError, OverflowError):
                numeric_value = None

        if numeric_value is not None:
            if not math.isfinite(numeric_value) or numeric_value < 0:
                return None
            absolute_value = abs(numeric_value)
            # 腾讯 adgroups/get 的 created_time 是整数 Unix 时间戳；若不指定
            # 单位，pandas 会按纳秒解释，产生 1970-01-01 00:00:02 一类错误时间。
            if 100_000_000 <= absolute_value < 100_000_000_000:
                unit = "s"
            elif 100_000_000_000 <= absolute_value < 100_000_000_000_000:
                unit = "ms"
            elif (
                100_000_000_000_000
                <= absolute_value
                < 100_000_000_000_000_000
            ):
                unit = "us"
            elif (
                100_000_000_000_000_000
                <= absolute_value
                < 100_000_000_000_000_000_000
            ):
                unit = "ns"
            else:
                return None
            timestamp = pd.to_datetime(numeric_value, unit=unit, utc=True)
            shanghai_timestamp = timestamp.tz_convert(SHANGHAI)
            try:
                return shanghai_timestamp.to_pydatetime(warn=False)
            except TypeError:
                return shanghai_timestamp.to_pydatetime()

        timestamp = pd.Timestamp(value)
        if pd.isna(timestamp):
            return None
        try:
            parsed = timestamp.to_pydatetime(warn=False)
        except TypeError:
            # pandas 2.0 以前的版本不支持 ``warn`` 参数。
            parsed = timestamp.to_pydatetime()
    except (TypeError, ValueError, OverflowError):
        return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=SHANGHAI)
    return parsed.astimezone(SHANGHAI)


def _as_date(value: Any) -> date | None:
    if value is None or (isinstance(value, str) and not value.strip()):
        return None
    try:
        timestamp = pd.Timestamp(value)
    except (TypeError, ValueError, OverflowError):
        return None
    if pd.isna(timestamp):
        return None
    return timestamp.date()


def _is_current_cleanup_day(run_date: date, current_time: Any) -> bool:
    current = _as_shanghai_datetime(current_time)
    return current is not None and current.date() == run_date


def determine_performance_cleanup_action(
    creative: dict[str, Any],
    ad: dict[str, Any],
    *,
    creative_created_at: Any,
    as_of_date: date | datetime,
    impressions: int,
    cost_fen: int,
    metric_start_date: date,
    metric_end_date: date,
    current_day_cost_fen: int = 0,
    config: dict[str, int | float] | None = None,
) -> dict[str, Any] | None:
    """严格应用新素材和老素材的未起量规则。

    只有审核通过、创意启用且所属广告启用的对象才允许进入候选，避免把父广告
    主动暂停、创意人工暂停或审核未完成造成的零投放误判为应删除对象。
    """

    if creative.get("configured_status") != "AD_STATUS_NORMAL":
        return None
    if ad.get("configured_status") != "AD_STATUS_NORMAL":
        return None
    if creative.get("creative_set_approval_status") not in {
        CREATIVE_NORMAL_STATUS,
        CREATIVE_PARTIAL_NORMAL_STATUS,
    }:
        return None
    created_at = _as_shanghai_datetime(creative_created_at)
    if created_at is None:
        return None
    try:
        impressions_value = int(impressions)
        cost_value = int(cost_fen)
        current_day_cost_value = int(current_day_cost_fen)
    except (TypeError, ValueError):
        return None
    if (
        impressions_value < 0
        or cost_value < 0
        or current_day_cost_value < 0
    ):
        return None
    if current_day_cost_value != 0:
        return None
    settings = config or performance_cleanup_config()
    if isinstance(as_of_date, datetime):
        as_of_datetime = _as_shanghai_datetime(as_of_date)
        if as_of_datetime is None:
            return None
        age_days = (as_of_datetime - created_at).days
    else:
        age_days = (as_of_date - created_at.date()).days
    evaluation_date = (
        as_of_date.astimezone(SHANGHAI).date()
        if isinstance(as_of_date, datetime) and as_of_date.tzinfo is not None
        else (
            as_of_date.date()
            if isinstance(as_of_date, datetime)
            else as_of_date
        )
    )
    begin_date = _as_date(ad.get("begin_date"))
    end_date = _as_date(ad.get("end_date"))
    if begin_date and begin_date > evaluation_date:
        return None
    if end_date and end_date < evaluation_date:
        return None
    if age_days < 0:
        return None
    window_days = int(settings["window_days"])
    daily_average = impressions_value / window_days
    common = {
        "cleanup_action": DELETE_CREATIVE,
        "component_ids": [],
        "element_ids": [],
        "recent_cost_fen": cost_value,
        "cost_start_date": metric_start_date,
        "cost_end_date": metric_end_date,
        "creative_created_at": created_at,
        "creative_age_days": age_days,
        "metric_impressions": impressions_value,
        "metric_daily_avg_impressions": daily_average,
        "metric_window_days": window_days,
    }
    min_age = int(settings["new_min_age_days"])
    max_age = int(settings["new_max_age_days"])
    new_threshold = int(settings["new_impressions_threshold"])
    old_threshold = float(settings["old_daily_impressions_threshold"])
    if min_age < age_days <= max_age:
        if impressions_value < new_threshold and cost_value == 0:
            return {
                **common,
                "cleanup_rule_type": PERFORMANCE_NEW_RULE,
                "action_reason": (
                    f"新素材搭建{age_days}天（>{min_age}且≤{max_age}），"
                    f"累计曝光{impressions_value}<{new_threshold}、"
                    "创建日至当日累计消耗为0元"
                ),
            }
        return None
    if age_days <= max_age or cost_value != 0:
        return None
    if daily_average < old_threshold:
        return {
            **common,
            "cleanup_rule_type": PERFORMANCE_OLD_LOW_RULE,
            "action_reason": (
                f"老素材搭建{age_days}天（>{max_age}），近{window_days}天"
                f"日均曝光{daily_average:.2f}<{old_threshold:g}，"
                "历史窗口及当日消耗均为0元"
            ),
        }
    if daily_average > old_threshold:
        return {
            **common,
            "cleanup_rule_type": PERFORMANCE_OLD_HIGH_RULE,
            "action_reason": (
                f"老素材搭建{age_days}天（>{max_age}），近{window_days}天"
                f"日均曝光{daily_average:.2f}>{old_threshold:g}，"
                "历史窗口及当日消耗均为0元"
            ),
        }
    # 业务规则使用严格大于/小于，因此日均恰好为 100（或配置边界）不命中任何规则。
    return None


def determine_ad_performance_cleanup_action(
    ad: dict[str, Any],
    *,
    as_of_date: date | datetime,
    cost_fen: int,
    metric_start_date: date,
    metric_end_date: date,
    current_day_cost_fen: int = 0,
    window_days: int = DEFAULT_AD_PERFORMANCE_WINDOW_DAYS,
    min_age_days: int = DEFAULT_AD_PERFORMANCE_MIN_AGE_DAYS,
) -> dict[str, Any] | None:
    """仅将当前启用且完整指标窗口内零消耗的广告判定为删除候选。"""

    if _is_deleted_ad(ad):
        return None
    if ad.get("configured_status") != "AD_STATUS_NORMAL":
        return None
    try:
        cost_value = int(cost_fen)
        current_day_cost_value = int(current_day_cost_fen)
    except (TypeError, ValueError):
        return None
    if (
        cost_value != 0
        or current_day_cost_value != 0
        or current_day_cost_value < 0
        or window_days <= 0
        or min_age_days < 0
    ):
        return None
    created_at = _as_shanghai_datetime(ad.get("created_time"))
    if created_at is None:
        return None
    if isinstance(as_of_date, datetime):
        as_of_datetime = _as_shanghai_datetime(as_of_date)
        if as_of_datetime is None:
            return None
        age_days = (as_of_datetime - created_at).days
    else:
        age_days = (as_of_date - created_at.date()).days
    if age_days <= min_age_days:
        return None
    evaluation_date = (
        as_of_date.astimezone(SHANGHAI).date()
        if isinstance(as_of_date, datetime) and as_of_date.tzinfo is not None
        else as_of_date.date()
        if isinstance(as_of_date, datetime)
        else as_of_date
    )
    begin_date = _as_date(ad.get("begin_date"))
    end_date = _as_date(ad.get("end_date"))
    if begin_date and begin_date > evaluation_date:
        return None
    if end_date and end_date < evaluation_date:
        return None
    return {
        "cleanup_action": DELETE_AD,
        "cleanup_rule_type": PERFORMANCE_AD_ZERO_SPEND_RULE,
        "component_ids": [],
        "element_ids": [],
        "recent_cost_fen": 0,
        "cost_start_date": metric_start_date,
        "cost_end_date": metric_end_date,
        "creative_created_at": created_at,
        "creative_age_days": age_days,
        "metric_window_days": window_days,
        "action_reason": (
            f"广告创建{age_days}天（>{min_age_days}），"
            f"近{window_days}个完整日累计消耗为0元，"
            "且当日消耗为0元"
        ),
    }


def determine_cleanup_action(
    creative: dict[str, Any],
    raw_result: dict | None,
    *,
    recent_cost_fen: int | None = None,
    cost_threshold_fen: int = 5000,
    wechat_cost_threshold_fen: int = 10000,
    spend_error: str | None = None,
) -> dict[str, Any] | None:
    """从三种整创意删除规则和人工告警中选择唯一处理动作。"""

    approval_status = str(creative.get("creative_set_approval_status") or "")
    if approval_status == CREATIVE_DENIED_STATUS:
        return {
            "cleanup_action": DELETE_CREATIVE,
            "cleanup_rule_type": REVIEW_DENIED_RULE,
            "component_ids": [],
            "element_ids": [],
            "action_reason": "创意审核状态为审核拒绝",
            "recent_cost_fen": recent_cost_fen,
        }
    if approval_status == CREATIVE_PARTIAL_NORMAL_STATUS:
        if has_rejected_wechat_mini_program_element(raw_result):
            if spend_error or recent_cost_fen is None:
                reason = (
                    "部分投放中且微信小程序元素审核拒绝，"
                    "近3天历史消耗读取失败，需人工判断"
                )
                if spend_error:
                    reason = f"{reason}：{spend_error}"
                return {
                    "cleanup_action": ALERT_ONLY,
                    "cleanup_rule_type": REVIEW_PARTIAL_RULE,
                    "component_ids": [],
                    "element_ids": [],
                    "action_reason": reason,
                    "recent_cost_fen": None,
                }
            if recent_cost_fen >= wechat_cost_threshold_fen:
                return {
                    "cleanup_action": ALERT_ONLY,
                    "cleanup_rule_type": REVIEW_PARTIAL_RULE,
                    "component_ids": [],
                    "element_ids": [],
                    "action_reason": (
                        "部分投放中且微信小程序元素审核拒绝，近3天历史消耗"
                        f"{recent_cost_fen / 100:.2f}元不低于"
                        f"{wechat_cost_threshold_fen / 100:.2f}元，需人工判断是否删除"
                    ),
                    "recent_cost_fen": recent_cost_fen,
                }
            return {
                "cleanup_action": DELETE_CREATIVE,
                "cleanup_rule_type": REVIEW_PARTIAL_RULE,
                "component_ids": [],
                "element_ids": [],
                "action_reason": (
                    "部分投放中且微信小程序元素审核拒绝，近3天历史消耗"
                    f"{recent_cost_fen / 100:.2f}元低于"
                    f"{wechat_cost_threshold_fen / 100:.2f}元"
                ),
                "recent_cost_fen": recent_cost_fen,
            }
        if spend_error or recent_cost_fen is None:
            reason = "部分投放中，近3天历史消耗读取失败，需人工判断"
            if spend_error:
                reason = f"{reason}：{spend_error}"
            return {
                "cleanup_action": ALERT_ONLY,
                "cleanup_rule_type": REVIEW_PARTIAL_RULE,
                "component_ids": [],
                "element_ids": [],
                "action_reason": reason,
                "recent_cost_fen": None,
            }
        if recent_cost_fen < cost_threshold_fen:
            return {
                "cleanup_action": DELETE_CREATIVE,
                "cleanup_rule_type": REVIEW_PARTIAL_RULE,
                "component_ids": [],
                "element_ids": [],
                "action_reason": (
                    f"部分投放中且近3天历史消耗{recent_cost_fen / 100:.2f}元"
                    f"低于{cost_threshold_fen / 100:.2f}元"
                ),
                "recent_cost_fen": recent_cost_fen,
            }
        return {
            "cleanup_action": ALERT_ONLY,
            "cleanup_rule_type": REVIEW_PARTIAL_RULE,
            "component_ids": [],
            "element_ids": [],
            "action_reason": (
                f"部分投放中且近3天历史消耗{recent_cost_fen / 100:.2f}元，"
                "需人工判断是否删除"
            ),
            "recent_cost_fen": recent_cost_fen,
        }
    return None


def _as_int(value: Any) -> int | None:
    try:
        number = int(value)
    except (TypeError, ValueError):
        return None
    return number if number > 0 else None


def _agency_name(value: Any) -> str:
    if value is None or pd.isna(value):
        return ""
    return "".join(str(value or "").split())


def _env_flag(name: str, default: bool = False) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def resolve_end_date(client, requested=None, *, now=None):
    from roi_control.data_source import resolve_end_date as resolve

    return resolve(client, requested, now=now)


def date_window(end_date: str) -> tuple[str, str]:
    end = datetime.strptime(end_date, "%Y%m%d")
    return (end - timedelta(days=2)).strftime("%Y%m%d"), end_date


def fetch_daily_data(client, start_date: str, end_date: str) -> pd.DataFrame:
    from roi_control.data_source import fetch_daily_data as fetch

    return fetch(client, start_date, end_date)


def fetch_recent_spend_accounts(client, start_date: str, end_date: str) -> list[dict]:
    from roi_control.data_source import fetch_recent_spend_accounts as fetch

    return fetch(client, start_date, end_date)


def fetch_account_agency_fallbacks(client, account_ids: list[int]) -> dict[int, str]:
    from roi_control.data_source import fetch_account_agency_fallbacks as fetch

    return fetch(client, account_ids)


def fetch_active_creative_inventory(client) -> list[dict[str, Any]]:
    """从 ODPS 读取每个创意最新一行，并仅保留当前 is_delete=0 的记录。"""

    frame = client.execute_sql(
        """
        SELECT account_id, ad_id, creative_id, creative_name,
               creative_status, create_time, update_time
        FROM (
            SELECT id, account_id, ad_id, creative_id, creative_name,
                   creative_status, is_delete, create_time, update_time,
                   ROW_NUMBER() OVER (
                       PARTITION BY account_id, creative_id
                       ORDER BY COALESCE(update_time, create_time) DESC, id DESC
                   ) AS rn
            FROM loghubods.ad_put_tencent_creative_day
            WHERE account_id IS NOT NULL
              AND creative_id IS NOT NULL
        ) latest
        WHERE rn=1 AND is_delete=0
        """
    )
    inventory: list[dict[str, Any]] = []
    for row in frame.to_dict("records"):
        account_id = _as_int(row.get("account_id"))
        adgroup_id = _as_int(row.get("ad_id"))
        creative_id = _as_int(row.get("creative_id"))
        created_at = _as_shanghai_datetime(row.get("create_time"))
        if account_id is None or creative_id is None:
            continue
        inventory.append(
            {
                "account_id": account_id,
                "adgroup_id": adgroup_id,
                "creative_id": creative_id,
                "creative_name": str(row.get("creative_name") or ""),
                "creative_status": str(row.get("creative_status") or ""),
                "create_time": created_at,
                "update_time": _as_shanghai_datetime(row.get("update_time")),
            }
        )
    return inventory


def fetch_tencent_account_metadata(client) -> dict[int, dict[str, str]]:
    """从 ODPS 读取最新有效的账户名称和代理商昵称。"""

    frame = client.execute_sql(
        """
        SELECT account_id, account_name, agent_name
        FROM (
            SELECT id, account_id, account_name, agent_name, is_delete,
                   create_time, update_time,
                   ROW_NUMBER() OVER (
                       PARTITION BY account_id
                       ORDER BY COALESCE(update_time, create_time) DESC, id DESC
                   ) AS rn
            FROM loghubods.ad_put_tencent_account
            WHERE account_id IS NOT NULL
              AND TRIM(account_id) <> ''
        ) latest
        WHERE rn=1 AND is_delete=0
        """
    )
    metadata: dict[int, dict[str, str]] = {}
    for row in frame.to_dict("records"):
        account_id = _as_int(row.get("account_id"))
        if account_id is None:
            continue
        account_name_value = row.get("account_name")
        agent_name_value = row.get("agent_name")
        account_name = (
            ""
            if account_name_value is None or pd.isna(account_name_value)
            else str(account_name_value).strip()
        )
        agent_name = (
            ""
            if agent_name_value is None or pd.isna(agent_name_value)
            else str(agent_name_value).strip()
        )
        metadata[account_id] = {
            "account_name": account_name,
            "agent_name": agent_name,
        }
    return metadata


def prefetch_account_access_tokens(account_ids: list[int]) -> dict[int, str]:
    from tools.ad_api import prefetch_access_tokens

    return prefetch_access_tokens(account_ids)


def build_agency_context(daily: pd.DataFrame) -> dict[str, dict]:
    """构建最新创意映射，以及不存在歧义的账户级兜底映射。"""

    creative_agency_sets: dict[tuple[int, int], set[str]] = defaultdict(set)
    creative_agency_dates: dict[tuple[int, int], str] = {}
    account_agencies: dict[int, set[str]] = defaultdict(set)
    account_agency_dates: dict[int, str] = {}
    account_names: dict[int, str] = {}
    account_name_dates: dict[int, str] = {}
    if daily.empty:
        return {
            "creative_agencies": {},
            "account_agencies": {},
            "fallback_account_agencies": {},
            "account_names": account_names,
        }

    miniapp = daily[daily["entity_type"].eq("self")]
    for _, row in miniapp.iterrows():
        account_id = _as_int(row.get("账号id"))
        creative_id = _as_int(row.get("创意id"))
        agency = _agency_name(row.get("代理名称"))
        row_date = str(row.get("dt") or "")
        if account_id is None:
            continue
        account_name = str(row.get("账号名称") or "").strip()
        if account_name and row_date >= account_name_dates.get(account_id, ""):
            account_names[account_id] = account_name
            account_name_dates[account_id] = row_date
        if agency:
            account_date = account_agency_dates.get(account_id, "")
            if row_date > account_date:
                account_agencies[account_id] = {agency}
                account_agency_dates[account_id] = row_date
            elif row_date == account_date:
                account_agencies[account_id].add(agency)
            if creative_id is not None:
                key = (account_id, creative_id)
                current_date = creative_agency_dates.get(key, "")
                if row_date > current_date:
                    creative_agency_sets[key] = {agency}
                    creative_agency_dates[key] = row_date
                elif row_date == current_date:
                    creative_agency_sets[key].add(agency)

    return {
        "creative_agencies": {
            key: next(iter(agencies))
            for key, agencies in creative_agency_sets.items()
            if len(agencies) == 1
        },
        "account_agencies": {
            account_id: next(iter(agencies))
            for account_id, agencies in account_agencies.items()
            if len(agencies) == 1
        },
        "fallback_account_agencies": {},
        "account_names": account_names,
    }


def _resolve_agency(
    context: dict[str, dict],
    account_id: int,
    creative_id: int,
) -> str:
    return str(
        context["creative_agencies"].get((account_id, creative_id))
        or context["account_agencies"].get(account_id)
        or context.get("fallback_account_agencies", {}).get(account_id)
        or ""
    )


def upsert_cleanup_candidate(
    record: dict[str, Any],
    *,
    _connection=None,
    _fetch_result: bool = True,
) -> dict[str, Any] | None:
    """持久化一个清理候选。

    外部调用默认每条记录使用独立连接；内部批量写入器复用同一事务连接并跳过
    最后的 SELECT，在保持相同幂等 SQL 和终态保护的前提下降低连接与查询开销。
    """
    component_ids_json = _json(record.get("component_ids") or [])
    element_ids_json = _json(record.get("element_ids") or [])
    review_result_json = _json(record.get("review_result") or {})
    pre_state_json = _json(record.get("pre_state") or {})
    cleanup_status = (
        "ALERT_PENDING"
        if record["cleanup_action"] == ALERT_ONLY
        else "DISCOVERED"
    )
    cleanup_rule_type = (
        record.get("cleanup_rule_type") or REVIEW_DENIED_RULE
    )
    performance_suppression_at = (
        datetime.now(SHANGHAI).replace(tzinfo=None)
        if _is_performance_rule(cleanup_rule_type)
        else None
    )
    insert_values = (
        record["account_id"],
        record.get("account_name"),
        record.get("agent_name"),
        "" if performance_suppression_at else record.get("agency_name"),
        record["adgroup_id"],
        record.get("adgroup_name"),
        record["dynamic_creative_id"],
        record.get("dynamic_creative_name"),
        record["check_date"],
        record["cleanup_action"],
        cleanup_rule_type,
        component_ids_json,
        element_ids_json,
        record.get("recent_cost_fen"),
        record.get("cost_start_date"),
        record.get("cost_end_date"),
        record.get("action_reason") or record["reject_reason"],
        record["reject_reason"],
        review_result_json,
        pre_state_json,
        cleanup_status,
        performance_suppression_at,
    )
    owns_connection = _connection is None
    connection = _connection or get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO creative_rejection_cleanup_item
                    (account_id, account_name, agent_name, agency_name, adgroup_id,
                     adgroup_name, dynamic_creative_id, dynamic_creative_name,
                     check_date,
                     cleanup_action, cleanup_rule_type,
                     target_component_ids_json,
                     target_element_ids_json, recent_cost_fen,
                     cost_start_date, cost_end_date, action_reason, reject_reason,
                     review_result_json, pre_state_json, cleanup_status,
                     agency_notified_at)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                ON DUPLICATE KEY UPDATE
                    id=LAST_INSERT_ID(id),
                    agency_name=CASE
                        WHEN cleanup_status IN (
                            'DELETING','CREATIVE_DELETED','AD_DELETED'
                        ) THEN agency_name
                        WHEN LEFT(VALUES(cleanup_rule_type), 12)='PERFORMANCE_'
                            THEN '' ELSE agency_name
                    END,
                    agency_notified_at=CASE
                        WHEN cleanup_status IN (
                            'DELETING','CREATIVE_DELETED','AD_DELETED'
                        ) THEN agency_notified_at
                        WHEN LEFT(VALUES(cleanup_rule_type), 12)='PERFORMANCE_'
                            THEN COALESCE(
                                agency_notified_at,
                                VALUES(agency_notified_at),
                                NOW()
                            )
                        ELSE agency_notified_at
                    END
                """,
                insert_values,
            )
            cursor.execute(
                """
                UPDATE creative_rejection_cleanup_item AS item
                JOIN (
                    SELECT
                        %s AS account_id, %s AS account_name, %s AS agent_name,
                        %s AS agency_name, %s AS adgroup_id, %s AS adgroup_name,
                        %s AS dynamic_creative_id, %s AS dynamic_creative_name,
                        %s AS check_date, %s AS cleanup_action,
                        %s AS cleanup_rule_type,
                        %s AS target_component_ids_json,
                        %s AS target_element_ids_json, %s AS recent_cost_fen,
                        %s AS cost_start_date, %s AS cost_end_date,
                        %s AS action_reason, %s AS reject_reason,
                        %s AS review_result_json, %s AS pre_state_json,
                        %s AS cleanup_status,
                        %s AS performance_suppression_at
                ) AS incoming
                  ON incoming.account_id=item.account_id
                 AND incoming.dynamic_creative_id=item.dynamic_creative_id
                 AND incoming.check_date=item.check_date
                SET
                    item.agency_name=CASE
                        WHEN LEFT(incoming.cleanup_rule_type, 12)='PERFORMANCE_'
                            THEN ''
                        ELSE COALESCE(
                            NULLIF(incoming.agency_name,''), item.agency_name
                        )
                    END,
                    item.adgroup_id=incoming.adgroup_id,
                    item.adgroup_name=COALESCE(
                        NULLIF(incoming.adgroup_name,''), item.adgroup_name
                    ),
                    item.dynamic_creative_name=COALESCE(
                        NULLIF(incoming.dynamic_creative_name,''),
                        item.dynamic_creative_name
                    ),
                    item.action_reason=incoming.action_reason,
                    item.reject_reason=incoming.reject_reason,
                    item.review_result_json=incoming.review_result_json,
                    item.pre_state_json=incoming.pre_state_json,
                    notified_at=CASE
                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
                            THEN NULL ELSE item.notified_at
                    END,
                    item.agency_notified_at=CASE
                        WHEN LEFT(incoming.cleanup_rule_type, 12)='PERFORMANCE_'
                            THEN COALESCE(
                                item.agency_notified_at,
                                incoming.performance_suppression_at,
                                NOW()
                            )
                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
                            THEN NULL ELSE item.agency_notified_at
                    END,
                    item.operator_notified_at=CASE
                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
                            THEN NULL ELSE item.operator_notified_at
                    END,
                    item.deleted_at=CASE
                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
                            THEN NULL ELSE item.deleted_at
                    END,
                    item.readback_json=CASE
                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
                            THEN NULL ELSE item.readback_json
                    END,
                    item.cleanup_status=CASE
                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
                            THEN incoming.cleanup_status
                        WHEN item.cleanup_status='SKIPPED_REVIEW_NOT_RECONFIRMED'
                            THEN 'DISCOVERED'
                        ELSE item.cleanup_status
                    END,
                    item.target_component_ids_json=incoming.target_component_ids_json,
                    item.target_element_ids_json=incoming.target_element_ids_json,
                    item.recent_cost_fen=incoming.recent_cost_fen,
                    item.cost_start_date=incoming.cost_start_date,
                    item.cost_end_date=incoming.cost_end_date,
                    item.cleanup_action=incoming.cleanup_action
                WHERE item.cleanup_status NOT IN (
                    'DELETING','CREATIVE_DELETED','AD_DELETED'
                )
                """,
                insert_values,
            )
            extended_values = (
                cleanup_rule_type,
                record.get("account_name"),
                record.get("agent_name"),
                record.get("creative_created_at"),
                record.get("creative_age_days"),
                record.get("metric_impressions"),
                record.get("metric_daily_avg_impressions"),
                record.get("metric_window_days"),
                cleanup_status,
                record["account_id"],
                record["dynamic_creative_id"],
                record["check_date"],
                performance_suppression_at,
            )
            cursor.execute(
                """
                UPDATE creative_rejection_cleanup_item AS item
                JOIN (
                    SELECT %s AS cleanup_rule_type,
                           %s AS account_name,
                           %s AS agent_name,
                           %s AS creative_created_at,
                           %s AS creative_age_days,
                           %s AS metric_impressions,
                           %s AS metric_daily_avg_impressions,
                           %s AS metric_window_days,
                           %s AS cleanup_status,
                           %s AS account_id,
                           %s AS dynamic_creative_id,
                           %s AS check_date,
                           %s AS performance_suppression_at
                ) AS incoming
                  ON incoming.account_id=item.account_id
                 AND incoming.dynamic_creative_id=item.dynamic_creative_id
                 AND incoming.check_date=item.check_date
                SET item.notified_at=CASE
                        WHEN NOT (item.cleanup_rule_type <=> incoming.cleanup_rule_type)
                          OR NOT (item.account_name <=> COALESCE(NULLIF(incoming.account_name,''), item.account_name))
                          OR NOT (item.agent_name <=> COALESCE(NULLIF(incoming.agent_name,''), item.agent_name))
                          OR NOT (item.creative_created_at <=> incoming.creative_created_at)
                          OR NOT (item.creative_age_days <=> incoming.creative_age_days)
                          OR NOT (item.metric_impressions <=> incoming.metric_impressions)
                          OR NOT (item.metric_daily_avg_impressions <=> incoming.metric_daily_avg_impressions)
                            THEN NULL ELSE item.notified_at
                    END,
                    item.agency_notified_at=CASE
                        WHEN LEFT(incoming.cleanup_rule_type, 12)='PERFORMANCE_'
                            THEN COALESCE(
                                item.agency_notified_at,
                                incoming.performance_suppression_at,
                                NOW()
                            )
                        WHEN NOT (item.cleanup_rule_type <=> incoming.cleanup_rule_type)
                          OR NOT (item.account_name <=> COALESCE(NULLIF(incoming.account_name,''), item.account_name))
                          OR NOT (item.agent_name <=> COALESCE(NULLIF(incoming.agent_name,''), item.agent_name))
                          OR NOT (item.creative_created_at <=> incoming.creative_created_at)
                          OR NOT (item.creative_age_days <=> incoming.creative_age_days)
                          OR NOT (item.metric_impressions <=> incoming.metric_impressions)
                          OR NOT (item.metric_daily_avg_impressions <=> incoming.metric_daily_avg_impressions)
                            THEN NULL ELSE item.agency_notified_at
                    END,
                    item.operator_notified_at=CASE
                        WHEN NOT (item.cleanup_rule_type <=> incoming.cleanup_rule_type)
                          OR NOT (item.account_name <=> COALESCE(NULLIF(incoming.account_name,''), item.account_name))
                          OR NOT (item.agent_name <=> COALESCE(NULLIF(incoming.agent_name,''), item.agent_name))
                          OR NOT (item.creative_created_at <=> incoming.creative_created_at)
                          OR NOT (item.creative_age_days <=> incoming.creative_age_days)
                          OR NOT (item.metric_impressions <=> incoming.metric_impressions)
                          OR NOT (item.metric_daily_avg_impressions <=> incoming.metric_daily_avg_impressions)
                            THEN NULL ELSE item.operator_notified_at
                    END,
                    item.cleanup_status=CASE
                        WHEN NOT (item.cleanup_rule_type <=> incoming.cleanup_rule_type)
                          OR NOT (item.account_name <=> COALESCE(NULLIF(incoming.account_name,''), item.account_name))
                          OR NOT (item.agent_name <=> COALESCE(NULLIF(incoming.agent_name,''), item.agent_name))
                          OR NOT (item.creative_created_at <=> incoming.creative_created_at)
                          OR NOT (item.creative_age_days <=> incoming.creative_age_days)
                          OR NOT (item.metric_impressions <=> incoming.metric_impressions)
                          OR NOT (item.metric_daily_avg_impressions <=> incoming.metric_daily_avg_impressions)
                            THEN incoming.cleanup_status
                        ELSE item.cleanup_status
                    END,
                    item.cleanup_rule_type=incoming.cleanup_rule_type,
                    item.account_name=COALESCE(
                        NULLIF(incoming.account_name,''), item.account_name
                    ),
                    item.agent_name=COALESCE(
                        NULLIF(incoming.agent_name,''), item.agent_name
                    ),
                    item.creative_created_at=incoming.creative_created_at,
                    item.creative_age_days=incoming.creative_age_days,
                    item.metric_impressions=incoming.metric_impressions,
                    item.metric_daily_avg_impressions=incoming.metric_daily_avg_impressions,
                    item.metric_window_days=incoming.metric_window_days
                WHERE item.cleanup_status NOT IN (
                    'DELETING','CREATIVE_DELETED','AD_DELETED'
                )
                """,
                extended_values,
            )
            if _fetch_result:
                cursor.execute(
                    """
                    SELECT * FROM creative_rejection_cleanup_item
                    WHERE account_id=%s AND dynamic_creative_id=%s AND check_date=%s
                    """,
                    (
                        record["account_id"],
                        record["dynamic_creative_id"],
                        record["check_date"],
                    ),
                )
                return cursor.fetchone()
            return None
    finally:
        if owns_connection:
            connection.close()


def _positive_int_setting(name: str, default: int, *, maximum: int) -> int:
    try:
        value = int(os.getenv(name, str(default)))
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{name} must be an integer") from exc
    if value < 1 or value > maximum:
        raise ValueError(f"{name} must be between 1 and {maximum}")
    return value


def upsert_cleanup_candidates(
    records: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[tuple[dict[str, Any], str]]]:
    """按事务分块并发保存候选。

    每个分块复用一个 MySQL 连接并省略逐行结果读取。分块失败时先回滚，再通过
    既有幂等路径逐条重试，避免单条坏数据遮蔽同批次的其他有效候选。
    """

    if not records:
        return [], []

    # 同一幂等键保留最后一个值并维持确定顺序，避免重复键让并发分块互相争锁。
    unique_by_key: dict[tuple[int, int, str], dict[str, Any]] = {}
    for record in records:
        key = (
            int(record["account_id"]),
            int(record["dynamic_creative_id"]),
            str(record["check_date"]),
        )
        unique_by_key[key] = record
    unique_records = list(unique_by_key.values())

    batch_size = _positive_int_setting(
        "DAILY_CLEANUP_CANDIDATE_BATCH_SIZE",
        100,
        maximum=1000,
    )
    configured_workers = _positive_int_setting(
        "DAILY_CLEANUP_CANDIDATE_STORE_WORKERS",
        4,
        maximum=16,
    )
    chunks = [
        unique_records[index : index + batch_size]
        for index in range(0, len(unique_records), batch_size)
    ]
    workers = min(configured_workers, len(chunks))

    def store_chunk(
        chunk: list[dict[str, Any]],
    ) -> tuple[list[dict[str, Any]], list[tuple[dict[str, Any], str]]]:
        connection = None
        try:
            connection = get_connection()
            begin = getattr(connection, "begin", None)
            if callable(begin):
                begin()
            for record in chunk:
                upsert_cleanup_candidate(
                    record,
                    _connection=connection,
                    _fetch_result=False,
                )
            commit = getattr(connection, "commit", None)
            if callable(commit):
                commit()
            return chunk, []
        except Exception as batch_exc:
            rollback = getattr(connection, "rollback", None)
            if callable(rollback):
                try:
                    rollback()
                except Exception:
                    logger.exception("cleanup candidate batch rollback failed")
            logger.warning(
                "cleanup candidate batch failed; retrying individually "
                "size=%d error=%s",
                len(chunk),
                batch_exc,
            )
        finally:
            if connection is not None:
                connection.close()

        successful: list[dict[str, Any]] = []
        errors: list[tuple[dict[str, Any], str]] = []
        for record in chunk:
            try:
                upsert_cleanup_candidate(record)
            except Exception as exc:
                errors.append((record, str(exc)))
            else:
                successful.append(record)
        return successful, errors

    successful_records: list[dict[str, Any]] = []
    failed_records: list[tuple[dict[str, Any], str]] = []
    with ThreadPoolExecutor(
        max_workers=workers,
        thread_name_prefix="cleanup-candidate-store",
    ) as executor:
        futures = [executor.submit(store_chunk, chunk) for chunk in chunks]
        for future in as_completed(futures):
            stored, errors = future.result()
            successful_records.extend(stored)
            failed_records.extend(errors)
    return successful_records, failed_records


def load_retryable_cleanup_items() -> list[dict[str, Any]]:
    stale_minutes = int(
        os.getenv(
            "DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES",
            str(DEFAULT_DELETE_CLAIM_STALE_MINUTES),
        )
    )
    if stale_minutes <= 0:
        raise ValueError(
            "DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES must be positive"
        )
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT item.*
                FROM creative_rejection_cleanup_item item
                JOIN (
                    SELECT account_id, dynamic_creative_id, MAX(check_date) AS check_date
                    FROM creative_rejection_cleanup_item
                    GROUP BY account_id, dynamic_creative_id
                ) latest
                  ON latest.account_id=item.account_id
                 AND latest.dynamic_creative_id=item.dynamic_creative_id
                 AND latest.check_date=item.check_date
                WHERE (
                    item.cleanup_status IN
                        ('DISCOVERED','DEFERRED','FAILED','WRITE_OUTCOME_UNKNOWN')
                    OR (
                        item.cleanup_status='DELETING'
                        AND item.updated_at < DATE_SUB(NOW(), INTERVAL %s MINUTE)
                    )
                )
                  AND item.cleanup_action IN ('DELETE_CREATIVE','DELETE_AD')
                ORDER BY item.id
                """,
                (stale_minutes,),
            )
            return list(cursor.fetchall())
    finally:
        connection.close()


def claim_cleanup_item(item_id: int) -> bool:
    """为本轮执行原子认领一条可重试删除记录。"""

    stale_minutes = int(
        os.getenv(
            "DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES",
            str(DEFAULT_DELETE_CLAIM_STALE_MINUTES),
        )
    )
    if stale_minutes <= 0:
        raise ValueError(
            "DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES must be positive"
        )
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                UPDATE creative_rejection_cleanup_item
                SET cleanup_status='DELETING', error_message=NULL, updated_at=NOW()
                WHERE id=%s
                  AND cleanup_action IN ('DELETE_CREATIVE','DELETE_AD')
                  AND (
                      cleanup_status IN
                          ('DISCOVERED','DEFERRED','FAILED','WRITE_OUTCOME_UNKNOWN')
                      OR (
                          cleanup_status='DELETING'
                          AND updated_at < DATE_SUB(NOW(), INTERVAL %s MINUTE)
                      )
                  )
                """,
                (item_id, stale_minutes),
            )
            return cursor.rowcount == 1
    finally:
        connection.close()


def update_cleanup_item(item_id: int, **values: Any) -> bool:
    expected_cleanup_status = values.pop("_expected_cleanup_status", None)
    allowed = {
        "agency_name",
        "agent_name",
        "cleanup_action",
        "cleanup_rule_type",
        "target_component_ids_json",
        "target_element_ids_json",
        "recent_cost_fen",
        "cost_start_date",
        "cost_end_date",
        "action_reason",
        "reject_reason",
        "creative_created_at",
        "creative_age_days",
        "metric_impressions",
        "metric_daily_avg_impressions",
        "metric_window_days",
        "review_result_json",
        "cleanup_status",
        "error_message",
        "pre_state_json",
        "readback_json",
        "deleted_at",
        "agency_notified_at",
        "operator_notified_at",
        "notified_at",
    }
    unknown = set(values) - allowed
    if unknown:
        raise ValueError(f"Unsupported cleanup fields: {sorted(unknown)}")
    if not values:
        return False
    assignments = ", ".join(f"{name}=%s" for name in values)
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            where = "WHERE id=%s"
            params = [*values.values(), item_id]
            if expected_cleanup_status is not None:
                if isinstance(expected_cleanup_status, (tuple, list, set, frozenset)):
                    statuses = list(expected_cleanup_status)
                    if not statuses:
                        return False
                    placeholders = ",".join(["%s"] * len(statuses))
                    where += f" AND cleanup_status IN ({placeholders})"
                    params.extend(statuses)
                else:
                    where += " AND cleanup_status=%s"
                    params.append(expected_cleanup_status)
            cursor.execute(
                f"UPDATE creative_rejection_cleanup_item SET {assignments} {where}",
                params,
            )
            return cursor.rowcount > 0
    finally:
        connection.close()


def load_pending_notification_items(
    *,
    include_discovered: bool = False,
    check_date: date | None = None,
    performance_only: bool = False,
    include_notified: bool = False,
) -> list[dict[str, Any]]:
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            statuses = "'CREATIVE_DELETED','AD_DELETED','ALERT_PENDING'"
            if include_discovered:
                statuses += ",'DISCOVERED'"
            extra_where = ""
            params: list[Any] = []
            if check_date is not None:
                extra_where += " AND check_date=%s"
                params.append(check_date)
            if performance_only:
                extra_where += (
                    " AND LEFT(cleanup_rule_type, 12)='PERFORMANCE_'"
                )
            notification_where = ""
            if not include_notified:
                notification_where = """
                  AND (
                      operator_notified_at IS NULL
                      OR (
                          LEFT(cleanup_rule_type, 12) <> 'PERFORMANCE_'
                          AND agency_notified_at IS NULL
                      )
                  )
                """
            cursor.execute(
                f"""
                SELECT * FROM creative_rejection_cleanup_item
                WHERE cleanup_status IN ({statuses})
                  {notification_where}
                  {extra_where}
                ORDER BY check_date, agency_name, account_id, adgroup_id,
                         dynamic_creative_id
                """,
                params,
            )
            return list(cursor.fetchall())
    finally:
        connection.close()


def load_unnotified_deleted_items(
    *,
    include_discovered: bool = False,
    check_date: date | None = None,
    performance_only: bool = False,
    include_notified: bool = False,
) -> list[dict[str, Any]]:
    """待发送清理通知的兼容入口。"""

    return load_pending_notification_items(
        include_discovered=include_discovered,
        check_date=check_date,
        performance_only=performance_only,
        include_notified=include_notified,
    )


def _mark_cleanup_channel_notified(
    item_ids: list[int],
    notified_at: datetime,
    *,
    channel: str,
) -> None:
    if not item_ids:
        return
    placeholders = ",".join(["%s"] * len(item_ids))
    if channel not in {"agency", "operator"}:
        raise ValueError(f"Unsupported cleanup notification channel: {channel}")
    channel_column = f"{channel}_notified_at"
    other_column = (
        "operator_notified_at" if channel == "agency" else "agency_notified_at"
    )
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                f"""
                UPDATE creative_rejection_cleanup_item
                SET {channel_column}=%s,
                    notified_at=CASE
                        WHEN {other_column} IS NOT NULL THEN %s
                        ELSE NULL
                    END
                WHERE id IN ({placeholders}) AND {channel_column} IS NULL
                """,
                [notified_at, notified_at, *item_ids],
            )
    finally:
        connection.close()


def mark_cleanup_items_agency_notified(
    item_ids: list[int], notified_at: datetime
) -> None:
    _mark_cleanup_channel_notified(item_ids, notified_at, channel="agency")


def mark_cleanup_items_operator_notified(
    item_ids: list[int], notified_at: datetime
) -> None:
    _mark_cleanup_channel_notified(item_ids, notified_at, channel="operator")


def mark_cleanup_items_notified(
    item_ids: list[int], notified_at: datetime
) -> None:
    """代理商通知渠道的兼容入口。"""

    mark_cleanup_items_agency_notified(item_ids, notified_at)


def _update_owned_cleanup_item(item_id: int, **values: Any) -> bool:
    """仅当本轮仍持有 DELETING 认领状态时更新记录。"""

    return update_cleanup_item(
        item_id,
        _expected_cleanup_status="DELETING",
        **values,
    )


def upsert_cleanup_delivery(record: dict[str, Any]) -> dict[str, Any]:
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO creative_rejection_delivery
                    (run_id, agency_name, agency_report_version, file_path,
                     file_sha256, creative_rows, ad_rows,
                     route_fingerprint, status)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,'PENDING')
                ON DUPLICATE KEY UPDATE
                    status=CASE
                        WHEN status='SENT' THEN status
                        WHEN NOT (route_fingerprint <=> VALUES(route_fingerprint))
                            THEN 'PENDING'
                        ELSE status
                    END,
                    error_message=CASE
                        WHEN NOT (route_fingerprint <=> VALUES(route_fingerprint))
                            THEN NULL ELSE error_message
                    END,
                    file_path=IF(status='SENT', file_path, VALUES(file_path)),
                    file_sha256=IF(status='SENT', file_sha256, VALUES(file_sha256)),
                    creative_rows=VALUES(creative_rows),
                    ad_rows=VALUES(ad_rows),
                    route_fingerprint=VALUES(route_fingerprint)
                """,
                (
                    record["run_id"],
                    record["agency_name"],
                    record["agency_report_version"],
                    record["file_path"],
                    record["file_sha256"],
                    record["creative_rows"],
                    record["ad_rows"],
                    record.get("route_fingerprint"),
                ),
            )
            cursor.execute(
                """
                SELECT * FROM creative_rejection_delivery
                WHERE run_id=%s AND agency_name=%s AND agency_report_version=%s
                """,
                (
                    record["run_id"],
                    record["agency_name"],
                    record["agency_report_version"],
                ),
            )
            return cursor.fetchone()
    finally:
        connection.close()


def update_cleanup_delivery(delivery_id: int, **values: Any) -> None:
    allowed = {
        "status",
        "sheet_token",
        "sheet_url",
        "response_code",
        "error_message",
        "sent_at",
    }
    increment_attempt = bool(values.pop("increment_attempt", False))
    unknown = set(values) - allowed
    if unknown:
        raise ValueError(f"Unsupported delivery fields: {sorted(unknown)}")
    assignments = [f"{name}=%s" for name in values]
    params = list(values.values())
    if increment_attempt:
        assignments.append("attempt_count=attempt_count+1")
    if not assignments:
        return
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                f"UPDATE creative_rejection_delivery SET {', '.join(assignments)} WHERE id=%s",
                [*params, delivery_id],
            )
    finally:
        connection.close()


def _file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def publish_cleanup_operator_summary(
    *,
    run_id: str,
    report: dict[str, object],
    chat_id: str,
    publisher: RoiFeishuPublisher,
    now: datetime,
) -> dict[str, object]:
    """幂等上传全代理汇总报表并发送到内部运营群。"""

    path = Path(str(report["report"]))
    delivery_id: int | None = None
    is_performance_report = (
        report.get("notification_kind") == "performance_internal"
    )
    route_name = (
        PERFORMANCE_SUMMARY_ROUTE
        if is_performance_report
        else OPERATOR_SUMMARY_ROUTE
    )
    try:
        if not path.is_file():
            raise FileNotFoundError(path)
        target_chat_id = str(chat_id or "").strip()
        if not target_chat_id:
            raise RuntimeError("FEISHU_AD_PROJECT_CHAT_ID 未配置")
        delivery = upsert_cleanup_delivery(
            {
                "run_id": run_id,
                "agency_name": route_name,
                "agency_report_version": str(report["report_version"]),
                "file_path": str(path),
                "file_sha256": _file_sha256(path),
                "creative_rows": int(report.get("creative_rows") or 0),
                "ad_rows": int(report.get("ad_rows") or 0),
                "route_fingerprint": hashlib.sha256(
                    target_chat_id.encode("utf-8")
                ).hexdigest(),
            }
        )
        delivery_id = int(delivery["id"])
        if delivery.get("status") == "SENT":
            return {
                "route": route_name,
                "status": "SENT",
                "sheet_url": str(delivery.get("sheet_url") or ""),
                "reused": True,
            }
        sheet_url = str(delivery.get("sheet_url") or "")
        sheet_token = str(delivery.get("sheet_token") or "")
        if not sheet_url or not sheet_token:
            imported = publisher.upload_workbook(path)
            sheet_url = imported["url"]
            sheet_token = imported["sheet_token"]
            update_cleanup_delivery(
                delivery_id,
                status="UPLOADED",
                sheet_token=sheet_token,
                sheet_url=sheet_url,
            )
        entity_count_text = (
            f"**{int(report.get('creative_rows') or 0)}** 条创意、"
            f"**{int(report.get('ad_rows') or 0)}** 条广告"
            if is_performance_report
            else f"**{int(report.get('creative_rows') or 0)}** 条创意"
        )
        message_id = publisher.send_report_card(
            title=str(report.get("title") or path.stem),
            content=(
                (
                    f"本次预演共 {entity_count_text}，"
                    + (
                        "未执行任何删除；本消息仅发送内部群，不发送代理商群。"
                        if is_performance_report
                        else "未执行任何删除，包含建议删除项及需人工判断项。"
                    )
                )
                if report.get("dry_run")
                else (
                    f"本批次共 {entity_count_text}，"
                    f"其中 **{int(report.get('unexecuted_rows') or 0)}** 条仅建议、未执行；"
                    + (
                        "本消息仅发送内部群，不发送代理商群。"
                        if is_performance_report
                        else "其余为已执行或需人工判断项。"
                    )
                )
                if report.get("unexecuted_rows")
                else (
                    f"本批次共 {entity_count_text}，"
                    + (
                        "为长期未起量清理结果；本消息仅发送内部群，不发送代理商群。"
                        if is_performance_report
                        else "包含各代理自动删除及需人工判断的完整汇总。"
                    )
                )
            ),
            sheet_url=sheet_url,
            chat_id=target_chat_id,
            button_text="查看全部处理明细",
        )
        update_cleanup_delivery(
            delivery_id,
            status="SENT",
            response_code=message_id,
            sent_at=now,
            increment_attempt=True,
        )
        return {
            "route": route_name,
            "status": "SENT",
            "sheet_url": sheet_url,
            "reused": False,
        }
    except Exception as exc:
        if delivery_id is not None:
            try:
                update_cleanup_delivery(
                    delivery_id,
                    status="FAILED",
                    error_message=str(exc),
                    increment_attempt=True,
                )
            except Exception as audit_exc:
                logger.error("operator summary audit failed: %s", audit_exc)
        logger.error("operator summary delivery failed: %s", exc)
        return {
            "route": route_name,
            "status": "FAILED",
            "error": str(exc),
        }


def _reject_reason(raw_result: dict[str, Any] | None, system_status: str) -> str:
    if raw_result:
        parsed = parse_review_result(raw_result)
        reasons = list(parsed.reject_messages)
        reasons.extend(fact.reason for fact in parsed.rejection_facts)
        unique = list(dict.fromkeys(reason.strip() for reason in reasons if reason.strip()))
        if unique:
            return "；".join(unique)
    if system_status == DENIED_SYSTEM_STATUS:
        return "腾讯正式审核未通过（接口未返回具体原因）"
    return "腾讯正式审核未通过"


def _cleanup_reason(
    action: dict[str, Any],
    raw_result: dict[str, Any] | None,
    system_status: str,
) -> str:
    if action["cleanup_action"] == DELETE_CREATIVE:
        return "创意审核状态为 CREATIVE_SET_APPROVAL_STATUS_DENIED"
    return _reject_reason(raw_result, system_status)


def _json_object(value: Any) -> dict[str, Any]:
    if isinstance(value, dict):
        return value
    if not value:
        return {}
    try:
        parsed = json.loads(str(value))
    except (TypeError, ValueError, json.JSONDecodeError):
        return {}
    return parsed if isinstance(parsed, dict) else {}


def _display_date(value: Any) -> str:
    if isinstance(value, (date, datetime)):
        return value.strftime("%Y-%m-%d")
    return str(value or "")[:10]


def _cleanup_rule_label(rule_type: Any) -> str:
    return {
        REVIEW_DENIED_RULE: "创意审核拒绝",
        REVIEW_PARTIAL_RULE: "部分投放审核异常",
        PERFORMANCE_NEW_RULE: "新素材未起量",
        PERFORMANCE_OLD_LOW_RULE: "老素材低曝光零消耗",
        PERFORMANCE_OLD_HIGH_RULE: "老素材高曝光零消耗",
        PERFORMANCE_AD_ZERO_SPEND_RULE: "广告近3日及当日零消耗",
    }.get(str(rule_type or ""), str(rule_type or ""))


def _write_report(
    path: Path,
    rows: list[dict[str, Any]],
    *,
    columns: tuple[str, ...],
    sheet_title: str = "审核不通过创意清理",
) -> None:
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = sheet_title
    sheet.append(list(columns))
    for row in rows:
        raw_result = _json_object(
            row.get("review_result") or row.get("review_result_json")
        )
        pre_state = _json_object(row.get("pre_state") or row.get("pre_state_json"))
        granular = review_granularity_fields(raw_result)
        action = str(row.get("cleanup_action") or "")
        is_ad_cleanup = _is_ad_performance_rule(
            row.get("cleanup_rule_type")
        )
        if (
            action in {DELETE_CREATIVE, DELETE_AD}
            and row.get("cleanup_status") == "DISCOVERED"
        ):
            execution_action = (
                "建议删除广告（未执行）"
                if is_ad_cleanup else "建议删除创意（未执行）"
            )
        elif action == DELETE_AD:
            execution_action = "删除广告"
        elif action == DELETE_CREATIVE:
            execution_action = "删除创意"
        else:
            execution_action = "需人工判断"
        checked_at = (
            row.get("checked_at")
            or row.get("created_at")
            or row.get("updated_at")
            or row.get("deleted_at")
        )
        recent_cost_fen = row.get("recent_cost_fen")
        recent_cost_yuan = (
            ""
            if recent_cost_fen is None
            else f"{int(recent_cost_fen) / 100:.2f}"
        )
        rule_type = str(row.get("cleanup_rule_type") or REVIEW_DENIED_RULE)
        is_performance = _is_performance_rule(rule_type)
        cost_start = _display_date(row.get("cost_start_date"))
        cost_end = _display_date(row.get("cost_end_date"))
        metric_average = row.get("metric_daily_avg_impressions")
        created_at = _as_shanghai_datetime(row.get("creative_created_at"))
        values = {
            "清理对象": "广告" if is_ad_cleanup else "创意",
            "代理名称": row.get("agency_name") or "",
            "账户ID": str(row["account_id"]),
            "账户名称": row.get("account_name") or "",
            "代理商昵称": row.get("agent_name") or "",
            "广告ID": str(row["adgroup_id"]),
            "广告名称": row.get("adgroup_name") or "",
            "创意ID": (
                "" if is_ad_cleanup else str(row["dynamic_creative_id"])
            ),
            "创意名称": (
                "" if is_ad_cleanup
                else row.get("dynamic_creative_name") or ""
            ),
            "清理规则": _cleanup_rule_label(rule_type),
            "创意搭建时间": (
                created_at.strftime("%Y-%m-%d %H:%M:%S") if created_at else ""
            ),
            "创意年龄(天)": (
                "" if row.get("creative_age_days") is None
                else int(row["creative_age_days"])
            ),
            "对象创建时间": (
                created_at.strftime("%Y-%m-%d %H:%M:%S") if created_at else ""
            ),
            "对象年龄(天)": (
                "" if row.get("creative_age_days") is None
                else int(row["creative_age_days"])
            ),
            "规则窗口累计曝光": (
                "" if row.get("metric_impressions") is None
                else int(row["metric_impressions"])
            ),
            "规则窗口日均曝光": (
                "" if metric_average is None else f"{float(metric_average):.2f}"
            ),
            "规则窗口累计消耗(元)": (
                recent_cost_yuan if is_performance else ""
            ),
            "广告近3日日均消耗(元)": (
                f"{int(recent_cost_fen or 0) / 100 / int(row.get('metric_window_days') or DEFAULT_AD_PERFORMANCE_WINDOW_DAYS):.2f}"
                if is_ad_cleanup else ""
            ),
            "规则指标日期范围": (
                f"{cost_start} ~ {cost_end}"
                if is_performance and cost_start and cost_end else ""
            ),
            "近3天累计历史消耗(元)": (
                "" if is_performance else recent_cost_yuan
            ),
            "消耗日期范围": (
                f"{cost_start} ~ {cost_end}"
                if not is_performance and cost_start and cost_end else ""
            ),
            "执行操作": execution_action,
            "操作判断原因": row.get("action_reason") or "",
            "配置状态": status_desc(
                pre_state.get("configured_status")
                or row.get("configured_status")
            ),
            "创意审核状态": status_desc(
                pre_state.get("creative_set_approval_status")
                or row.get("creative_set_approval_status")
            ),
            "元素粒度审核状态": granular["element_review_status"],
            "元素粒度审核不通过原因": granular["element_reject_reason"],
            "版位粒度审核状态": granular["site_review_status"],
            "版位粒度审核不通过原因": granular["site_reject_reason"],
            "审核不通过原因": (
                "" if is_performance else row.get("reject_reason") or ""
            ),
            "检查时间": (
                checked_at.strftime("%Y-%m-%d %H:%M:%S")
                if isinstance(checked_at, (date, datetime))
                else str(checked_at or "")
            ),
        }
        sheet.append([values[column] for column in columns])
    header_fill = PatternFill("solid", fgColor="C65911")
    for cell in sheet[1]:
        cell.fill = header_fill
        cell.font = Font(color="FFFFFF", bold=True)
        cell.alignment = Alignment(horizontal="center", vertical="center")
    widths = {
        "清理对象": 12,
        "代理名称": 22,
        "账户ID": 14,
        "账户名称": 22,
        "代理商昵称": 22,
        "广告ID": 14,
        "广告名称": 30,
        "创意ID": 16,
        "创意名称": 30,
        "清理规则": 24,
        "创意搭建时间": 20,
        "创意年龄(天)": 14,
        "对象创建时间": 20,
        "对象年龄(天)": 14,
        "规则窗口累计曝光": 20,
        "规则窗口日均曝光": 20,
        "规则窗口累计消耗(元)": 22,
        "广告近3日日均消耗(元)": 24,
        "规则指标日期范围": 24,
        "近3天累计历史消耗(元)": 22,
        "消耗日期范围": 24,
        "执行操作": 16,
        "操作判断原因": 60,
        "配置状态": 20,
        "创意审核状态": 24,
        "元素粒度审核状态": 40,
        "元素粒度审核不通过原因": 60,
        "版位粒度审核状态": 40,
        "版位粒度审核不通过原因": 60,
        "审核不通过原因": 60,
        "检查时间": 20,
    }
    for index, column in enumerate(columns, start=1):
        sheet.column_dimensions[get_column_letter(index)].width = widths[column]
    for row in sheet.iter_rows(min_row=2):
        for cell in row:
            cell.alignment = Alignment(vertical="top", wrap_text=True)
        for id_column in ("账户ID", "广告ID", "创意ID"):
            row[columns.index(id_column)].number_format = "@"
    sheet.freeze_panes = "A2"
    sheet.auto_filter.ref = sheet.dimensions
    path.parent.mkdir(parents=True, exist_ok=True)
    workbook.save(path)


def write_cleanup_reports(
    rows: list[dict[str, Any]],
    output_dir: Path,
    report_date: str,
) -> tuple[str, list[dict[str, object]], dict[str, list[int]]]:
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        grouped[str(row["agency_name"])].append(row)
    def digest_for(report_rows: list[dict[str, Any]], destination: str) -> str:
        digest_input = "|".join(
            ":".join(
                [
                    str(row["id"]),
                    str(row.get("cleanup_action") or ""),
                    str(row.get("cleanup_rule_type") or ""),
                    str(row.get("cleanup_status") or ""),
                    str(row.get("recent_cost_fen")),
                    str(row.get("metric_impressions")),
                    str(row.get("metric_daily_avg_impressions")),
                    _display_date(row.get("cost_end_date")),
                    str(row.get("reject_reason") or ""),
                ]
            )
            for row in sorted(report_rows, key=lambda value: int(value["id"]))
        )
        return hashlib.sha256(
            f"{report_date}|{destination}|{digest_input}".encode("utf-8")
        ).hexdigest()[:12]

    digest = digest_for(rows, "all")
    run_id = f"reject_{report_date}_{digest}"
    reports: list[dict[str, object]] = []
    item_ids: dict[str, list[int]] = {}
    for agency, agency_rows in sorted(grouped.items()):
        agency_digest = digest_for(agency_rows, agency)
        safe_agency = agency.replace("/", "_").replace("\\", "_")
        path = output_dir / (
            f"{report_date}_{safe_agency}_创意审核异常处理_{agency_digest}.xlsx"
        )
        dry_run = any(
            row.get("cleanup_status") == "DISCOVERED"
            for row in agency_rows
        )
        _write_report(path, agency_rows, columns=AGENCY_REPORT_COLUMNS)
        reports.append(
            {
                "agency_name": agency,
                "report_version": REPORT_VERSION,
                "report": str(path),
                "title": (
                    f"{report_date}_{agency}_创意审核异常处理预演通知"
                    if dry_run
                    else f"{report_date}_{agency}_创意审核异常处理通知"
                ),
                "creative_rows": len(agency_rows),
                "ad_rows": 0,
                "notification_type": (
                    "creative_rejection_dry_run"
                    if dry_run
                    else "creative_rejection_cleanup"
                ),
                "run_id": f"reject_{report_date}_{agency_digest}",
            }
        )
        item_ids[agency] = [int(row["id"]) for row in agency_rows]
    return run_id, reports, item_ids


def write_cleanup_operator_summary(
    rows: list[dict[str, Any]],
    output_dir: Path,
    report_date: str,
    run_id: str,
) -> dict[str, object]:
    digest_input = "|".join(
        ":".join(
            [
                str(row["id"]),
                str(row.get("cleanup_action") or ""),
                str(row.get("cleanup_rule_type") or ""),
                str(row.get("cleanup_status") or ""),
                str(row.get("recent_cost_fen")),
                str(row.get("metric_impressions")),
                str(row.get("metric_daily_avg_impressions")),
                _display_date(row.get("cost_end_date")),
                str(row.get("reject_reason") or ""),
            ]
        )
        for row in sorted(rows, key=lambda value: int(value["id"]))
    )
    digest = hashlib.sha256(
        f"{report_date}|{OPERATOR_SUMMARY_ROUTE}|{digest_input}".encode("utf-8")
    ).hexdigest()[:12]
    path = output_dir / (
        f"{report_date}_投放调控_创意审核异常处理汇总_{digest}.xlsx"
    )
    unexecuted_rows = sum(
        row.get("cleanup_status") == "DISCOVERED" for row in rows
    )
    deleted_rows = sum(
        row.get("cleanup_status") == "CREATIVE_DELETED" for row in rows
    )
    dry_run = unexecuted_rows > 0 and deleted_rows == 0
    _write_report(path, rows, columns=OPERATOR_REPORT_COLUMNS)
    return {
        "report_version": f"{REPORT_VERSION}_operator_summary",
        "report": str(path),
        "title": (
            f"{report_date}_创意审核异常处理预演汇总通知"
            if dry_run
            else f"{report_date}_创意审核异常处理汇总通知"
        ),
        "creative_rows": len(rows),
        "run_id": f"reject_{report_date}_{digest}",
        "dry_run": dry_run,
        "unexecuted_rows": unexecuted_rows,
        "deleted_rows": deleted_rows,
    }


def write_performance_operator_summary(
    rows: list[dict[str, Any]],
    output_dir: Path,
    report_date: str,
) -> dict[str, object]:
    """生成只发内部群且独立发送的长期未起量报表。"""

    digest_input = "|".join(
        ":".join(
            [
                str(row["id"]),
                str(row.get("cleanup_rule_type") or ""),
                str(row.get("cleanup_status") or ""),
                str(row.get("account_name") or ""),
                str(row.get("agent_name") or ""),
                str(row.get("creative_age_days")),
                str(row.get("metric_impressions")),
                str(row.get("metric_daily_avg_impressions")),
                str(row.get("recent_cost_fen")),
                _display_date(row.get("cost_end_date")),
            ]
        )
        for row in sorted(rows, key=lambda value: int(value["id"]))
    )
    digest = hashlib.sha256(
        (
            f"{report_date}|{PERFORMANCE_SUMMARY_ROUTE}|"
            f"{PERFORMANCE_REPORT_VERSION}|{digest_input}"
        ).encode("utf-8")
    ).hexdigest()[:12]
    path = output_dir / (
        f"{report_date}_投放调控_长期未起量创意及广告清理汇总_{digest}.xlsx"
    )
    unexecuted_rows = sum(
        row.get("cleanup_status") == "DISCOVERED" for row in rows
    )
    deleted_rows = sum(
        row.get("cleanup_status") in {"CREATIVE_DELETED", "AD_DELETED"}
        for row in rows
    )
    dry_run = unexecuted_rows > 0 and deleted_rows == 0
    _write_report(
        path,
        rows,
        columns=PERFORMANCE_REPORT_COLUMNS,
        sheet_title="长期未起量清理",
    )
    ad_rows = sum(
        _is_ad_performance_rule(row.get("cleanup_rule_type")) for row in rows
    )
    creative_rows = len(rows) - ad_rows
    return {
        "report_version": PERFORMANCE_REPORT_VERSION,
        "report": str(path),
        "title": "长期未起量创意及广告清理汇总通知",
        "creative_rows": creative_rows,
        "ad_rows": ad_rows,
        "run_id": f"performance_{report_date}_{digest}",
        "dry_run": dry_run,
        "unexecuted_rows": unexecuted_rows,
        "deleted_rows": deleted_rows,
        "notification_kind": "performance_internal",
    }


def split_notification_rows(
    rows: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]:
    """拆分代理商、审核内部群和未起量内部群三类通知数据。"""

    agency_rows = [
        row
        for row in rows
        if row.get("agency_notified_at") is None
        and not _is_performance_rule(row.get("cleanup_rule_type"))
        and str(row.get("agency_name") or "").strip()
    ]
    review_internal_rows = [
        row
        for row in rows
        if row.get("operator_notified_at") is None
        and not _is_performance_rule(row.get("cleanup_rule_type"))
    ]
    performance_internal_rows = [
        row
        for row in rows
        if row.get("operator_notified_at") is None
        and _is_performance_rule(row.get("cleanup_rule_type"))
    ]
    return agency_rows, review_internal_rows, performance_internal_rows


def filter_preview_notification_rows(
    rows: list[dict[str, Any]],
    *,
    check_date: date,
    confirmed_actions: dict[tuple[int, int], dict[str, Any]],
    force_notification: bool,
) -> list[dict[str, Any]]:
    """预览通知只保留本轮重新确认并成功落库的未起量候选。

    同一天重复运行时，审计表里可能残留旧版本误判或本轮已不再满足规则的
    ``DISCOVERED`` 记录。仅按日期查询会把这些旧候选再次写进 Excel；这里同时
    校验本轮确认键和规则类型，避免过期时间、指标或账户信息进入新报表。
    """

    selected: list[dict[str, Any]] = []
    for row in rows:
        if (
            not _is_performance_rule(row.get("cleanup_rule_type"))
            or _as_date(row.get("check_date")) != check_date
        ):
            continue
        account_id = _as_int(row.get("account_id"))
        try:
            # 广告候选使用负广告 ID 作为审计目标键，不能调用只接受正数的
            # ``_as_int``，否则本轮确认的广告会被错误排除在预览通知之外。
            target_id = int(row.get("dynamic_creative_id"))
        except (TypeError, ValueError):
            target_id = 0
        if account_id is None or target_id == 0:
            continue
        confirmed = confirmed_actions.get((account_id, target_id))
        if (
            confirmed is None
            or confirmed.get("cleanup_rule_type")
            != row.get("cleanup_rule_type")
        ):
            continue
        selected.append(
            {**row, "operator_notified_at": None}
            if force_notification
            else row
        )
    return selected


def _chunks(values: list[int], size: int) -> list[list[int]]:
    return [values[offset : offset + size] for offset in range(0, len(values), size)]


def _scan_one_account(
    account: dict[str, Any],
    *,
    tencent,
    review_fetcher: Callable[[int, list[int]], list[dict]],
    spend_start_date: date,
    spend_end_date: date,
    review_enabled: bool = True,
    performance_enabled: bool = False,
    performance_source_creatives: dict[int, dict[str, Any]] | None = None,
    performance_start_date: date | None = None,
    performance_end_date: date | None = None,
    current_day_date: date | None = None,
    performance_as_of: datetime | None = None,
    performance_config: dict[str, int | float] | None = None,
    ad_cleanup_enabled: bool = False,
    ad_metric_start_date: date | None = None,
    ad_metric_end_date: date | None = None,
) -> tuple[
    list[dict],
    dict[int, dict],
    dict[int, dict],
    dict[int, int],
    str | None,
    int,
    str | None,
    dict[int, dict[str, Any]],
    str | None,
    dict[int, dict[str, int]],
    str | None,
    str | None,
    dict[str, int],
]:
    account_id = int(account["account_id"])
    empty_source_diagnostics = {
        "missing_tencent_creatives": 0,
        "missing_tencent_ads": 0,
        "ad_mismatches": 0,
        "missing_source_ad_ids": 0,
        "missing_create_time": 0,
    }
    try:
        creative_list_error = None
        try:
            creatives = [
                creative
                for creative in tencent.get_dynamic_creatives(account_id)
                if not _is_deleted_creative(creative)
            ]
        except Exception as exc:
            creatives = []
            creative_list_error = str(exc)
            logger.exception("creative list scan failed account=%d", account_id)

        ad_list_error = None
        try:
            ads = {
                int(ad["adgroup_id"]): ad
                for ad in tencent.get_ads(account_id)
                if _as_int(ad.get("adgroup_id")) is not None
            }
        except Exception as exc:
            ads = {}
            ad_list_error = str(exc)
            logger.exception("ad list scan failed account=%d", account_id)

        ids = [
            int(row["dynamic_creative_id"])
            for row in creatives
            if _as_int(row.get("dynamic_creative_id")) is not None
        ]
        creative_by_id = {
            int(row["dynamic_creative_id"]): row
            for row in creatives
            if _as_int(row.get("dynamic_creative_id")) is not None
        }
        raw_by_id: dict[int, dict] = {}
        review_error = creative_list_error if review_enabled else None
        if review_enabled and review_error is None:
            try:
                for batch in _chunks(ids, 100):
                    for raw in review_fetcher(account_id, batch):
                        creative_id = _as_int(raw.get("dynamic_creative_id"))
                        if creative_id is not None:
                            raw_by_id[creative_id] = raw
            except Exception as exc:
                review_error = str(exc)
                raw_by_id = {}
                logger.exception(
                    "creative review scan failed account=%d", account_id
                )
        cost_by_id: dict[int, int] = {}
        spend_error = None
        partial_ids = [
            int(row["dynamic_creative_id"])
            for row in creatives
            if _as_int(row.get("dynamic_creative_id")) is not None
            and row.get("creative_set_approval_status")
            == CREATIVE_PARTIAL_NORMAL_STATUS
        ]
        if review_enabled and review_error is None and partial_ids:
            try:
                cost_by_id = tencent.get_dynamic_creative_costs(
                    account_id,
                    partial_ids,
                    spend_start_date,
                    spend_end_date,
                )
            except Exception as exc:
                spend_error = str(exc)
                logger.exception(
                    "creative cost scan failed account=%d start=%s end=%s",
                    account_id,
                    spend_start_date,
                    spend_end_date,
                )
        performance_metrics: dict[int, dict[str, Any]] = {}
        performance_error = None
        source_creatives = performance_source_creatives or {}
        missing_tencent_creatives = (
            set(source_creatives) - set(creative_by_id)
            if creative_list_error is None
            else set()
        )
        missing_create_time = {
            creative_id
            for creative_id, source in source_creatives.items()
            if _as_shanghai_datetime(source.get("create_time")) is None
        }
        if performance_enabled:
            if creative_list_error is not None:
                performance_error = f"creative list failed: {creative_list_error}"
            elif ad_list_error is not None:
                performance_error = f"ad list failed: {ad_list_error}"
            elif review_enabled and review_error is not None:
                # 未起量创意必须在本账户审核判断成功后才能执行；广告链路不受此限制。
                performance_error = None
            else:
                try:
                    if performance_start_date is None or performance_end_date is None:
                        raise ValueError("performance metric date window is missing")
                    if current_day_date is None:
                        raise ValueError("current-day metric date is missing")
                    settings = performance_config or performance_cleanup_config()
                    as_of = _as_shanghai_datetime(performance_as_of)
                    if as_of is None:
                        as_of = datetime(
                            current_day_date.year,
                            current_day_date.month,
                            current_day_date.day,
                            23,
                            59,
                            59,
                            999999,
                            tzinfo=SHANGHAI,
                        )
                    min_age = int(settings["new_min_age_days"])
                    max_age = int(settings["new_max_age_days"])
                    new_ids_by_start: dict[date, list[int]] = defaultdict(list)
                    old_ids: list[int] = []
                    for creative_id, row in creative_by_id.items():
                        if (
                            creative_id not in source_creatives
                            or creative_id in missing_create_time
                            or row.get("configured_status") != "AD_STATUS_NORMAL"
                            or row.get("creative_set_approval_status")
                            not in {
                                CREATIVE_NORMAL_STATUS,
                                CREATIVE_PARTIAL_NORMAL_STATUS,
                            }
                        ):
                            continue
                        ad = ads.get(_as_int(row.get("adgroup_id")) or -1) or {}
                        if ad.get("configured_status") != "AD_STATUS_NORMAL":
                            continue
                        created_at = _as_shanghai_datetime(
                            source_creatives[creative_id].get("create_time")
                        )
                        if created_at is None:
                            continue
                        age_days = (as_of - created_at).days
                        if min_age < age_days <= max_age:
                            new_ids_by_start[created_at.date()].append(creative_id)
                        elif age_days > max_age:
                            old_ids.append(creative_id)

                    eligible_ids = [
                        creative_id
                        for ids_for_date in new_ids_by_start.values()
                        for creative_id in ids_for_date
                    ] + old_ids
                    if eligible_ids:
                        old_metrics: dict[int, dict[str, int]] = {}
                        if old_ids:
                            old_metrics = tencent.get_dynamic_creative_metrics(
                                account_id,
                                old_ids,
                                performance_start_date,
                                performance_end_date,
                            )
                            missing_old = set(old_ids) - set(old_metrics)
                            if missing_old:
                                raise RuntimeError(
                                    "Tencent historical creative metric response omitted "
                                    f"requested IDs: {sorted(missing_old)}"
                                )
                        new_metrics_by_start: dict[
                            date, dict[int, dict[str, int]]
                        ] = {}
                        for metric_start_date, new_ids in new_ids_by_start.items():
                            historical_metrics = (
                                tencent.get_dynamic_creative_metrics(
                                    account_id,
                                    new_ids,
                                    metric_start_date,
                                    performance_end_date,
                                )
                                if metric_start_date <= performance_end_date
                                else {
                                    creative_id: {
                                        "impressions": 0,
                                        "cost_fen": 0,
                                    }
                                    for creative_id in new_ids
                                }
                            )
                            missing_new = set(new_ids) - set(historical_metrics)
                            if missing_new:
                                raise RuntimeError(
                                    "Tencent cumulative creative metric response omitted "
                                    f"requested IDs: {sorted(missing_new)}"
                                )
                            new_metrics_by_start[
                                metric_start_date
                            ] = historical_metrics

                        current_day_metrics = tencent.get_dynamic_creative_metrics(
                            account_id,
                            eligible_ids,
                            current_day_date,
                            current_day_date,
                        )
                        missing_today = set(eligible_ids) - set(current_day_metrics)
                        if missing_today:
                            raise RuntimeError(
                                "Tencent current-day creative metric response omitted "
                                f"requested IDs: {sorted(missing_today)}"
                            )

                        for creative_id in old_ids:
                            performance_metrics[creative_id] = {
                                **old_metrics[creative_id],
                                "current_day_cost_fen": int(
                                    current_day_metrics[creative_id].get(
                                        "cost_fen", 0
                                    )
                                ),
                                "metric_start_date": performance_start_date,
                                "metric_end_date": performance_end_date,
                            }

                        for metric_start_date, new_ids in new_ids_by_start.items():
                            historical_metrics = new_metrics_by_start[
                                metric_start_date
                            ]
                            for creative_id in new_ids:
                                today = current_day_metrics[creative_id]
                                historical = historical_metrics[creative_id]
                                performance_metrics[creative_id] = {
                                    "impressions": int(
                                        historical.get("impressions", 0)
                                    )
                                    + int(today.get("impressions", 0)),
                                    "cost_fen": int(historical.get("cost_fen", 0))
                                    + int(today.get("cost_fen", 0)),
                                    "current_day_cost_fen": int(
                                        today.get("cost_fen", 0)
                                    ),
                                    "metric_start_date": metric_start_date,
                                    "metric_end_date": current_day_date,
                                }
                except Exception as exc:
                    performance_metrics = {}
                    performance_error = str(exc)
                    logger.exception(
                        "creative performance scan failed account=%d start=%s end=%s",
                        account_id,
                        performance_start_date,
                        performance_end_date,
                    )
        ad_metrics: dict[int, dict[str, int]] = {}
        ad_metric_error = None
        missing_tencent_ads: set[int] = set()
        ad_mismatches: set[int] = set()
        missing_source_ad_ids = {
            creative_id
            for creative_id, source in source_creatives.items()
            if _as_int(source.get("adgroup_id")) is None
        }
        if creative_list_error is None:
            for creative_id, source in source_creatives.items():
                creative = creative_by_id.get(creative_id)
                if creative is None:
                    continue
                source_adgroup_id = _as_int(source.get("adgroup_id"))
                tencent_adgroup_id = _as_int(creative.get("adgroup_id"))
                if (
                    source_adgroup_id is not None
                    and tencent_adgroup_id is not None
                    and source_adgroup_id != tencent_adgroup_id
                ):
                    # 只记录源映射漂移，不阻断广告按自身条件判断。
                    ad_mismatches.add(creative_id)
        if ad_cleanup_enabled:
            eligible_ad_ids: set[int] = set()
            if ad_list_error is not None:
                ad_metric_error = f"ad list failed: {ad_list_error}"
            else:
                source_ad_ids = {
                    adgroup_id
                    for source in source_creatives.values()
                    if (adgroup_id := _as_int(source.get("adgroup_id")))
                    is not None
                }
                missing_tencent_ads = source_ad_ids - set(ads)
                eligible_ad_ids = {
                    adgroup_id
                    for adgroup_id in source_ad_ids
                    if (ads.get(adgroup_id) or {}).get("configured_status")
                    == "AD_STATUS_NORMAL"
                }
            if ad_metric_error is None and eligible_ad_ids:
                try:
                    if ad_metric_start_date is None or ad_metric_end_date is None:
                        raise ValueError("ad metric date window is missing")
                    if current_day_date is None:
                        raise ValueError("current-day metric date is missing")
                    historical_ad_metrics = tencent.get_ad_metrics(
                        account_id,
                        sorted(eligible_ad_ids),
                        ad_metric_start_date,
                        ad_metric_end_date,
                    )
                    current_day_ad_metrics = tencent.get_ad_metrics(
                        account_id,
                        sorted(eligible_ad_ids),
                        current_day_date,
                        current_day_date,
                    )
                    missing_historical_ads = set(eligible_ad_ids) - set(
                        historical_ad_metrics
                    )
                    missing_today_ads = set(eligible_ad_ids) - set(
                        current_day_ad_metrics
                    )
                    if missing_historical_ads or missing_today_ads:
                        raise RuntimeError(
                            "Tencent ad metric response omitted requested IDs: "
                            f"historical={sorted(missing_historical_ads)} "
                            f"today={sorted(missing_today_ads)}"
                        )
                    ad_metrics = {
                        adgroup_id: {
                            **historical_ad_metrics[adgroup_id],
                            "current_day_cost_fen": int(
                                current_day_ad_metrics[adgroup_id].get(
                                    "cost_fen", 0
                                )
                            ),
                        }
                        for adgroup_id in eligible_ad_ids
                        if adgroup_id in historical_ad_metrics
                        and adgroup_id in current_day_ad_metrics
                    }
                except Exception as exc:
                    ad_metric_error = str(exc)
                    logger.exception(
                        "ad performance scan failed account=%d start=%s end=%s",
                        account_id,
                        ad_metric_start_date,
                        ad_metric_end_date,
                    )
        source_diagnostics = {
            "missing_tencent_creatives": len(missing_tencent_creatives),
            "missing_tencent_ads": len(missing_tencent_ads),
            "ad_mismatches": len(ad_mismatches),
            "missing_source_ad_ids": len(missing_source_ad_ids),
            "missing_create_time": len(missing_create_time),
        }
        return (
            creatives,
            ads,
            raw_by_id,
            cost_by_id,
            spend_error,
            len(ids),
            None,
            performance_metrics,
            performance_error,
            ad_metrics,
            ad_metric_error,
            review_error,
            source_diagnostics,
        )
    except Exception as exc:
        return (
            [],
            {},
            {},
            {},
            None,
            0,
            f"account={account_id} scan failed: {exc}",
            {},
            None,
            {},
            None,
            None,
            empty_source_diagnostics,
        )


def cleanup_precondition_failure(
    item: dict[str, Any],
    scanned_accounts: set[int],
    confirmed_actions: dict[tuple[int, int], dict[str, Any]],
    performance_scanned_accounts: set[int] | None = None,
    ad_scanned_accounts: set[int] | None = None,
    performance_scope_loaded: bool = True,
    performance_source_keys: set[tuple[int, int]] | None = None,
    performance_source_ad_keys: set[tuple[int, int]] | None = None,
) -> tuple[str, str] | None:
    """除非本轮再次确认同一清理动作，否则按保守策略拒绝执行。"""

    account_id = int(item["account_id"])
    creative_id = int(item["dynamic_creative_id"])
    is_performance_item = _is_performance_rule(item.get("cleanup_rule_type"))
    is_ad_item = _is_ad_performance_rule(item.get("cleanup_rule_type"))
    if is_ad_item:
        if not performance_scope_loaded:
            return "DEFERRED", "本轮有效创意范围读取失败，未执行广告删除"
        if (
            account_id,
            int(item["adgroup_id"]),
        ) not in (performance_source_ad_keys or set()):
            return (
                "SKIPPED_REVIEW_NOT_RECONFIRMED",
                "广告已不在当前有效创意派生范围内",
            )
        if account_id not in (ad_scanned_accounts or set()):
            return "DEFERRED", "本轮账户近3天广告消耗读取失败，未执行删除"
    elif is_performance_item:
        if not performance_scope_loaded:
            return "DEFERRED", "本轮有效创意范围读取失败，未执行创意删除"
        if (account_id, creative_id) not in (
            performance_source_keys or set()
        ):
            return (
                "SKIPPED_REVIEW_NOT_RECONFIRMED",
                "创意已不在当前 is_delete=0 有效范围内",
            )
        if account_id not in (performance_scanned_accounts or set()):
            return "DEFERRED", "本轮账户近7天创意指标读取失败，未执行删除"
    elif account_id not in scanned_accounts:
        return "DEFERRED", "本轮账户审核结果扫描失败，未执行删除"
    confirmed = confirmed_actions.get((account_id, creative_id))
    if not confirmed:
        return (
            "SKIPPED_REVIEW_NOT_RECONFIRMED",
            "本轮未按新规则再次确认清理动作",
        )
    item_action = str(item.get("cleanup_action") or "")
    allowed_action = DELETE_AD if is_ad_item else DELETE_CREATIVE
    if item_action != allowed_action:
        return "SKIPPED_REVIEW_NOT_RECONFIRMED", "当前规则不允许该删除动作"
    if item_action != confirmed["cleanup_action"]:
        return "SKIPPED_REVIEW_NOT_RECONFIRMED", "本轮清理动作与候选记录不一致"
    return None


def _same_cleanup_action(
    expected_action: str,
    actual: dict[str, Any] | None,
    *,
    expected_rule_type: str | None = None,
) -> bool:
    matches = bool(
        expected_action in {DELETE_CREATIVE, DELETE_AD}
        and actual
        and actual.get("cleanup_action") == expected_action
    )
    if not matches:
        return False
    if expected_rule_type and _is_performance_rule(expected_rule_type):
        return actual.get("cleanup_rule_type") == expected_rule_type
    return True


def run_rejected_creative_cleanup(
    *,
    output_dir: Path,
    now: datetime | None = None,
    tencent=None,
    odps=None,
    review_fetcher: Callable[[int, list[int]], list[dict]] | None = None,
    publisher: RoiFeishuPublisher | None = None,
    notifier=None,
    clock: Callable[[], datetime] | None = None,
    underperformance_preview_only: bool = False,
    force_notification: bool = False,
) -> dict[str, Any]:
    """先应用审核规则，再应用长期未起量规则。

    ``underperformance_preview_only`` 是人工预览模式：跳过审核规则发现，强制
    关闭两类腾讯删除开关，只写预览审计记录，并把当日未起量 Excel 发到内部群。
    ``force_notification`` 仅用于该预览模式，同日重发时创建新的通知审计批次。
    """

    if force_notification and not underperformance_preview_only:
        raise ValueError("force_notification 仅允许用于未起量预览模式")

    clock_was_provided = clock is not None
    clock_fn = clock or (lambda: datetime.now(SHANGHAI))
    effective_now = now or clock_fn()
    if effective_now.tzinfo is None:
        effective_now = effective_now.replace(tzinfo=SHANGHAI)
    if underperformance_preview_only:
        apply_enabled = False
        performance_enabled = True
        performance_apply_enabled = False
        ad_cleanup_enabled = True
        ad_apply_enabled = False
    else:
        apply_enabled = _env_flag("DAILY_REJECTED_CREATIVE_APPLY_ENABLED")
        performance_enabled = _env_flag(
            "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED"
        )
        performance_apply_enabled = _env_flag(
            "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED"
        )
        ad_cleanup_enabled = _env_flag(
            "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED"
        )
        ad_apply_enabled = _env_flag(
            "DAILY_UNDERPERFORMING_AD_APPLY_ENABLED"
        )
    if performance_apply_enabled and not performance_enabled:
        raise RuntimeError(
            "DAILY_UNDERPERFORMING_CREATIVE_APPLY_ENABLED=1 requires "
            "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED=1"
        )
    if ad_apply_enabled and not ad_cleanup_enabled:
        raise RuntimeError(
            "DAILY_UNDERPERFORMING_AD_APPLY_ENABLED=1 requires "
            "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED=1"
        )
    if ad_cleanup_enabled and not performance_enabled:
        raise RuntimeError(
            "DAILY_UNDERPERFORMING_AD_CLEANUP_ENABLED=1 requires "
            "DAILY_UNDERPERFORMING_CREATIVE_CLEANUP_ENABLED=1 so creative "
            "judgment always runs first"
        )
    performance_settings = (
        performance_cleanup_config() if performance_enabled else None
    )
    webhook_config = (
        AgencyWebhookConfig()
        if underperformance_preview_only
        else AgencyWebhookConfig.from_env()
    )
    if apply_enabled and not webhook_config.enabled:
        raise RuntimeError(
            "DAILY_REJECTED_CREATIVE_APPLY_ENABLED=1 requires "
            "ROI_AGENCY_WEBHOOK_ENABLED=1"
        )
    if (
        apply_enabled
        or performance_enabled
        or performance_apply_enabled
        or ad_cleanup_enabled
        or ad_apply_enabled
    ) and not _operator_summary_chat_id():
        raise RuntimeError(
            "creative cleanup internal notification requires "
            "FEISHU_AD_PROJECT_CHAT_ID"
        )
    initialize_schema()
    if odps is None:
        from roi_control.odps_client import ODPSClient

        odps_client = ODPSClient(project=os.getenv("ODPS_PROJECT", "loghubods"))
    else:
        odps_client = odps
    spend_end_date = effective_now.date() - timedelta(days=1)
    spend_start_date = spend_end_date - timedelta(days=2)
    performance_window_days = int(
        (performance_settings or {}).get(
            "window_days", DEFAULT_PERFORMANCE_WINDOW_DAYS
        )
    )
    performance_end_date = spend_end_date
    performance_start_date = performance_end_date - timedelta(
        days=performance_window_days - 1
    )
    current_day_date = effective_now.date()
    ad_window_days = DEFAULT_AD_PERFORMANCE_WINDOW_DAYS
    ad_min_age_days = DEFAULT_AD_PERFORMANCE_MIN_AGE_DAYS
    if ad_cleanup_enabled:
        ad_window_days = ad_performance_window_days()
        ad_min_age_days = ad_performance_min_age_days()
    ad_metric_end_date = spend_end_date
    ad_metric_start_date = ad_metric_end_date - timedelta(
        days=ad_window_days - 1
    )
    cost_threshold_fen = partial_creative_cost_threshold_fen()
    wechat_cost_threshold_fen = wechat_mini_program_cost_threshold_fen()

    # 审核范围/代理上下文与未起量 ODPS 范围相互隔离。审核数据异常时，
    # 审核删除和未起量创意均 fail closed，但广告仍可按自身条件继续判断。
    start_date: str | None = None
    end_date: str | None = None
    review_scope_error: str | None = None
    review_context_error: str | None = None
    review_accounts: list[dict[str, Any]] = []
    daily = pd.DataFrame()
    if not underperformance_preview_only:
        try:
            resolved_review_end_date = resolve_end_date(
                odps_client,
                now=effective_now,
            )
            start_date, end_date = date_window(resolved_review_end_date)
        except Exception as exc:
            review_scope_error = str(exc)
            review_context_error = str(exc)
            logger.exception("creative review date range resolution failed")
        else:
            try:
                daily = fetch_daily_data(odps_client, start_date, end_date)
            except Exception as exc:
                review_context_error = str(exc)
                logger.exception("creative review agency context query failed")
            try:
                review_accounts = fetch_recent_spend_accounts(
                    odps_client,
                    start_date,
                    end_date,
                )
            except Exception as exc:
                review_scope_error = str(exc)
                logger.exception("creative review account scope query failed")
    try:
        context = build_agency_context(daily)
    except Exception as exc:
        context = build_agency_context(pd.DataFrame())
        review_context_error = str(exc)
        logger.exception("creative review agency context build failed")
    review_account_ids = {
        int(account["account_id"]) for account in review_accounts
    }
    accounts_by_id = {
        int(account["account_id"]): account for account in review_accounts
    }
    performance_inventory: list[dict[str, Any]] = []
    performance_sources_by_account: dict[
        int, dict[int, dict[str, Any]]
    ] = defaultdict(dict)
    performance_scope_error: str | None = None
    performance_account_metadata: dict[int, dict[str, str]] = {}
    performance_account_metadata_error: str | None = None
    underperformance_enabled = performance_enabled or ad_cleanup_enabled
    if underperformance_enabled:
        try:
            performance_inventory = fetch_active_creative_inventory(odps_client)
        except Exception as exc:
            performance_scope_error = str(exc)
            logger.exception("active creative inventory query failed")
        else:
            for source in performance_inventory:
                account_id = int(source["account_id"])
                creative_id = int(source["creative_id"])
                performance_sources_by_account[account_id][creative_id] = source
                accounts_by_id.setdefault(
                    account_id,
                    {"account_id": account_id, "account_name": ""},
                )
        if performance_sources_by_account:
            try:
                performance_account_metadata = fetch_tencent_account_metadata(
                    odps_client
                )
            except Exception as exc:
                performance_account_metadata_error = str(exc)
                logger.exception("tencent account metadata query failed")
            else:
                for account_id, metadata in performance_account_metadata.items():
                    if account_id not in performance_sources_by_account:
                        continue
                    account = accounts_by_id.setdefault(
                        account_id,
                        {"account_id": account_id, "account_name": ""},
                    )
                    if metadata.get("account_name"):
                        account["account_name"] = metadata["account_name"]
    accounts = [accounts_by_id[key] for key in sorted(accounts_by_id)]
    account_ids = [int(account["account_id"]) for account in accounts]
    performance_account_ids = set(performance_sources_by_account)
    performance_creation_times = {
        (account_id, creative_id): created_at
        for account_id, sources in performance_sources_by_account.items()
        for creative_id, source in sources.items()
        if (created_at := _as_shanghai_datetime(source.get("create_time")))
        is not None
    }
    performance_source_keys = {
        (account_id, creative_id)
        for account_id, sources in performance_sources_by_account.items()
        for creative_id in sources
    }
    performance_source_ad_keys = {
        (account_id, adgroup_id)
        for account_id, sources in performance_sources_by_account.items()
        for source in sources.values()
        if (adgroup_id := _as_int(source.get("adgroup_id"))) is not None
    }
    unresolved_account_ids = [
        int(account["account_id"])
        for account in review_accounts
        if int(account["account_id"]) not in context["account_agencies"]
    ]
    if not underperformance_preview_only:
        try:
            context["fallback_account_agencies"] = fetch_account_agency_fallbacks(
                odps_client,
                unresolved_account_ids,
            )
        except Exception as exc:
            if review_context_error is None:
                review_context_error = str(exc)
            logger.exception(
                "account agency fallback query failed accounts=%d",
                len(unresolved_account_ids),
            )

    owned_tencent = tencent is None
    if tencent is None:
        from tencent_client import TencentClient

        client = TencentClient()
    else:
        client = tencent
    prefetched_tokens = prefetch_account_access_tokens(account_ids)
    seed_tokens = getattr(client, "seed_access_tokens", None)
    if callable(seed_tokens):
        seed_tokens(prefetched_tokens)
    fetch_reviews = review_fetcher or fetch_dynamic_creative_review_results
    discovered = 0
    review_discovered = 0
    performance_discovered = 0
    scanned = 0
    scanned_accounts: set[int] = set()
    performance_scanned_accounts: set[int] = set()
    ad_scanned_accounts: set[int] = set()
    confirmed_actions: dict[tuple[int, int], dict[str, Any]] = {}
    scan_errors: list[str] = []
    if review_scope_error:
        scan_errors.append(
            f"creative review account scope query failed: {review_scope_error}"
        )
    if review_context_error:
        scan_errors.append(
            f"creative review agency context query failed: {review_context_error}"
        )
    if performance_scope_error:
        scan_errors.append(
            "active creative inventory query failed: "
            f"{performance_scope_error}"
        )
    if performance_account_metadata_error:
        scan_errors.append(
            "tencent account metadata query failed: "
            f"{performance_account_metadata_error}"
        )
    source_diagnostic_totals = {
        "missing_tencent_creatives": 0,
        "missing_tencent_ads": 0,
        "ad_mismatches": 0,
        "missing_source_ad_ids": 0,
        "missing_create_time": 0,
    }
    try:
        scan_workers = int(os.getenv("TENCENT_AD_ACCOUNT_SCAN_WORKERS", "8"))
        if scan_workers < 1:
            raise ValueError("TENCENT_AD_ACCOUNT_SCAN_WORKERS must be at least 1")
        workers = min(scan_workers, len(accounts), 32) if accounts else 1

        def run_scan(account: dict[str, Any]):
            if owned_tencent:
                from tencent_client import TencentClient

                scan_client = TencentClient()
                scan_client.seed_access_tokens(prefetched_tokens)
            else:
                scan_client = client
            try:
                return _scan_one_account(
                    account,
                    tencent=scan_client,
                    review_fetcher=fetch_reviews,
                    spend_start_date=spend_start_date,
                    spend_end_date=spend_end_date,
                    review_enabled=(
                        int(account["account_id"]) in review_account_ids
                    ),
                    performance_enabled=(
                        performance_enabled
                        and review_scope_error is None
                        and int(account["account_id"])
                        in performance_account_ids
                    ),
                    performance_source_creatives=(
                        performance_sources_by_account.get(
                            int(account["account_id"]), {}
                        )
                    ),
                    performance_start_date=performance_start_date,
                    performance_end_date=performance_end_date,
                    current_day_date=current_day_date,
                    performance_as_of=effective_now,
                    performance_config=performance_settings,
                    ad_cleanup_enabled=(
                        ad_cleanup_enabled
                        and int(account["account_id"])
                        in performance_account_ids
                    ),
                    ad_metric_start_date=ad_metric_start_date,
                    ad_metric_end_date=ad_metric_end_date,
                )
            finally:
                if scan_client is not client:
                    scan_client.session.close()

        logger.info("account scan started accounts=%d workers=%d", len(accounts), workers)
        scan_results = []
        with ThreadPoolExecutor(
            max_workers=workers,
            thread_name_prefix="creative-scan",
        ) as executor:
            futures = {executor.submit(run_scan, account): account for account in accounts}
            for completed, future in enumerate(as_completed(futures), start=1):
                account = futures[future]
                account_id = int(account["account_id"])
                try:
                    (
                        creatives,
                        ads,
                        raw_by_id,
                        cost_by_id,
                        spend_error,
                        account_scanned,
                        error,
                        performance_metrics,
                        performance_error,
                        ad_metrics,
                        ad_metric_error,
                        review_error,
                        source_diagnostics,
                    ) = future.result()
                except Exception as exc:
                    creatives, ads, raw_by_id, cost_by_id = [], {}, {}, {}
                    spend_error, account_scanned = None, 0
                    error = f"account={account_id} scan failed: {exc}"
                    performance_metrics, performance_error = {}, None
                    ad_metrics, ad_metric_error = {}, None
                    review_error = None
                    source_diagnostics = {
                        key: 0 for key in source_diagnostic_totals
                    }
                scanned += account_scanned
                if error:
                    logger.error(error)
                    scan_errors.append(error)
                else:
                    if account_id in review_account_ids:
                        if review_error:
                            scan_errors.append(
                                f"account={account_id} review scan failed: "
                                f"{review_error}"
                            )
                        else:
                            scanned_accounts.add(account_id)
                    if (
                        performance_enabled
                        and review_scope_error is None
                        and account_id in performance_account_ids
                    ):
                        if review_error and account_id in review_account_ids:
                            pass
                        elif performance_error:
                            scan_errors.append(
                                "account="
                                f"{account_id} performance scan failed: "
                                f"{performance_error}"
                            )
                        else:
                            performance_scanned_accounts.add(account_id)
                    if (
                        ad_cleanup_enabled
                        and account_id in performance_account_ids
                    ):
                        if ad_metric_error:
                            scan_errors.append(
                                f"account={account_id} ad performance scan failed: "
                                f"{ad_metric_error}"
                            )
                        else:
                            ad_scanned_accounts.add(account_id)
                    for key in source_diagnostic_totals:
                        source_diagnostic_totals[key] += int(
                            source_diagnostics.get(key) or 0
                        )
                    scan_results.append(
                        (
                            account,
                            creatives,
                            ads,
                            raw_by_id,
                            cost_by_id,
                            spend_error,
                            performance_metrics,
                            ad_metrics,
                        )
                    )
                logger.info(
                    "account scan progress=%d/%d account=%d creatives=%d error=%s",
                    completed,
                    len(accounts),
                    account_id,
                    account_scanned,
                    bool(error),
                )

        if any(source_diagnostic_totals.values()):
            logger.warning(
                "active creative inventory mismatches=%s",
                source_diagnostic_totals,
            )

        review_candidate_keys: set[tuple[int, int]] = set()
        review_processing_failed_keys: set[tuple[int, int]] = set()

        def process_creative(task, *, phase: str):
            (
                account,
                account_id,
                creative,
                ads,
                raw_by_id,
                cost_by_id,
                spend_error,
                performance_metrics,
                _ad_metrics,
            ) = task
            creative_id = _as_int(creative.get("dynamic_creative_id"))
            adgroup_id = _as_int(creative.get("adgroup_id"))
            if creative_id is None or adgroup_id is None:
                return None
            raw_result = raw_by_id.get(creative_id)
            system_status = str(creative.get("system_status") or "")
            is_partial = (
                creative.get("creative_set_approval_status")
                == CREATIVE_PARTIAL_NORMAL_STATUS
            )
            key = (account_id, creative_id)
            action = None
            if phase == "review" and account_id in scanned_accounts:
                action = determine_cleanup_action(
                    creative,
                    raw_result,
                    recent_cost_fen=cost_by_id.get(creative_id),
                    cost_threshold_fen=cost_threshold_fen,
                    wechat_cost_threshold_fen=wechat_cost_threshold_fen,
                    spend_error=(spend_error if is_partial else None),
                )
            if phase == "performance":
                if (
                    key in review_candidate_keys
                    or key in review_processing_failed_keys
                    or _is_deleted_creative(creative)
                ):
                    return None
                if not performance_enabled or key not in performance_source_keys:
                    return None
                created_at = performance_creation_times.get(
                    (account_id, creative_id)
                )
                if created_at is None:
                    return None
                elif account_id in performance_scanned_accounts:
                    metrics = performance_metrics.get(creative_id)
                    if metrics is not None:
                        action = determine_performance_cleanup_action(
                            creative,
                            ads.get(adgroup_id) or {},
                            creative_created_at=created_at,
                            as_of_date=effective_now,
                            impressions=metrics.get("impressions", 0),
                            cost_fen=metrics.get("cost_fen", 0),
                            current_day_cost_fen=metrics.get(
                                "current_day_cost_fen", 0
                            ),
                            metric_start_date=metrics.get(
                                "metric_start_date", performance_start_date
                            ),
                            metric_end_date=metrics.get(
                                "metric_end_date", performance_end_date
                            ),
                            config=performance_settings,
                        )
            if action is None:
                return None
            ad = ads.get(adgroup_id) or {}
            account_metadata = performance_account_metadata.get(
                account_id, {}
            )
            if phase == "performance":
                candidate_account_name = (
                    account_metadata.get("account_name") or ""
                )
            else:
                candidate_account_name = (
                    context["account_names"].get(account_id)
                    or account.get("account_name")
                    or ""
                )
            record = {
                "account_id": account_id,
                "account_name": candidate_account_name,
                "agent_name": (
                    account_metadata.get("agent_name")
                    if phase == "performance"
                    else ""
                ),
                "agency_name": (
                    ""
                    if _is_performance_rule(action.get("cleanup_rule_type"))
                    else _resolve_agency(context, account_id, creative_id)
                ),
                "adgroup_id": adgroup_id,
                "adgroup_name": ad.get("adgroup_name") or "",
                "dynamic_creative_id": creative_id,
                "dynamic_creative_name": creative.get(
                    "dynamic_creative_name"
                )
                or "",
                "check_date": effective_now.date(),
                **action,
                "action_reason": action.get("action_reason")
                or _cleanup_reason(action, raw_result, system_status),
                "reject_reason": (
                    ""
                    if _is_performance_rule(action.get("cleanup_rule_type"))
                    else _reject_reason(raw_result, system_status)
                ),
                "cost_start_date": action.get("cost_start_date")
                or spend_start_date,
                "cost_end_date": action.get("cost_end_date")
                or spend_end_date,
                "review_result": raw_result or {},
                "pre_state": creative,
            }
            return account_id, creative_id, action, record

        tasks = [
            (
                account,
                int(account["account_id"]),
                creative,
                ads,
                raw_by_id,
                cost_by_id,
                spend_error,
                performance_metrics,
                ad_metrics,
            )
            for (
                account,
                creatives,
                ads,
                raw_by_id,
                cost_by_id,
                spend_error,
                performance_metrics,
                ad_metrics,
            )
            in scan_results
            for creative in creatives
            if not _is_deleted_creative(creative)
        ]
        process_workers = (
            min(
                int(os.getenv("TENCENT_AD_CREATIVE_PROCESS_WORKERS", "8")),
                len(tasks),
                32,
            )
            if tasks
            else 1
        )
        candidate_batch_min_size = _positive_int_setting(
            "DAILY_CLEANUP_CANDIDATE_BATCH_MIN_SIZE",
            20,
            maximum=10000,
        )

        def persist_candidate_results(
            candidate_results: list[
                tuple[int, int, dict[str, Any], dict[str, Any]]
            ],
            *,
            phase: str,
        ) -> None:
            nonlocal discovered, review_discovered, performance_discovered
            if not candidate_results:
                return
            started_at = time.monotonic()
            records = [result[3] for result in candidate_results]
            logger.info(
                "cleanup candidate storage phase=%s records=%d mode=%s",
                phase,
                len(records),
                (
                    "batch"
                    if len(records) >= candidate_batch_min_size
                    else "single"
                ),
            )
            if len(records) >= candidate_batch_min_size:
                stored_records, storage_errors = upsert_cleanup_candidates(records)
            else:
                stored_records = []
                storage_errors = []
                for record in records:
                    try:
                        upsert_cleanup_candidate(record)
                    except Exception as exc:
                        storage_errors.append((record, str(exc)))
                    else:
                        stored_records.append(record)

            result_by_key = {
                (account_id, creative_id): (action, record)
                for account_id, creative_id, action, record in candidate_results
            }
            stored_keys = {
                (int(record["account_id"]), int(record["dynamic_creative_id"]))
                for record in stored_records
            }
            for key in stored_keys:
                action, _record = result_by_key[key]
                confirmed_actions[key] = action
                discovered += 1
                if phase == "review":
                    review_candidate_keys.add(key)
                    review_discovered += 1
                else:
                    performance_discovered += 1
            for record, error_text in storage_errors:
                account_id = int(record["account_id"])
                creative_id = int(record["dynamic_creative_id"])
                error = (
                    f"cleanup candidate storage failed phase={phase} "
                    f"account={account_id} target={creative_id}: {error_text}"
                )
                scan_errors.append(error)
                if phase == "review":
                    review_processing_failed_keys.add((account_id, creative_id))
                logger.error(error)
            logger.info(
                "cleanup candidate storage completed phase=%s stored=%d "
                "failed=%d duration_ms=%d",
                phase,
                len(stored_records),
                len(storage_errors),
                int((time.monotonic() - started_at) * 1000),
            )

        for processing_phase in ("review", "performance"):
            if processing_phase == "review" and underperformance_preview_only:
                continue
            if processing_phase == "performance" and not performance_enabled:
                continue
            phase_started_at = time.monotonic()
            phase_candidates: list[
                tuple[int, int, dict[str, Any], dict[str, Any]]
            ] = []
            logger.info(
                "creative processing phase=%s creatives=%d workers=%d",
                processing_phase,
                len(tasks),
                process_workers,
            )
            with ThreadPoolExecutor(
                max_workers=process_workers,
                thread_name_prefix=f"creative-{processing_phase}",
            ) as executor:
                futures = {
                    executor.submit(
                        process_creative,
                        task,
                        phase=processing_phase,
                    ): task
                    for task in tasks
                }
                for completed, future in enumerate(as_completed(futures), start=1):
                    task = futures[future]
                    account_id = task[1]
                    creative_id = _as_int(task[2].get("dynamic_creative_id"))
                    try:
                        result = future.result()
                    except Exception as exc:
                        error = (
                            f"creative {processing_phase} processing failed "
                            f"account={account_id} creative={creative_id}: {exc}"
                        )
                        scan_errors.append(error)
                        if processing_phase == "review" and creative_id is not None:
                            review_processing_failed_keys.add(
                                (account_id, creative_id)
                            )
                        logger.exception(error)
                    else:
                        if result is not None:
                            phase_candidates.append(result)
                    if completed % 500 == 0 or completed == len(tasks):
                        logger.info(
                            "creative processing phase=%s progress=%d/%d confirmed=%d",
                            processing_phase,
                            completed,
                            len(tasks),
                            len(phase_candidates),
                        )
            logger.info(
                "creative processing completed phase=%s creatives=%d "
                "confirmed=%d duration_ms=%d",
                processing_phase,
                len(tasks),
                len(phase_candidates),
                int((time.monotonic() - phase_started_at) * 1000),
            )
            persist_candidate_results(
                phase_candidates,
                phase=processing_phase,
            )

        # 广告判断必须排在全部创意判断之后。审计表沿用同一幂等链路，
        # 用负的广告 ID 作为内部目标键，避免与真实创意 ID 冲突；报表不展示该值。
        ad_discovered = 0
        if ad_cleanup_enabled:
            ad_phase_started_at = time.monotonic()
            ad_tasks = [
                (account, int(account["account_id"]), adgroup_id, ad, ad_metrics)
                for (
                    account,
                    _creatives,
                    ads,
                    _raw_by_id,
                    _cost_by_id,
                    _spend_error,
                    _performance_metrics,
                    ad_metrics,
                ) in scan_results
                if int(account["account_id"]) in ad_scanned_accounts
                for adgroup_id, ad in ads.items()
            ]
            ad_workers = min(
                _positive_int_setting(
                    "TENCENT_AD_AD_PROCESS_WORKERS",
                    8,
                    maximum=32,
                ),
                len(ad_tasks),
            ) if ad_tasks else 1
            logger.info(
                "ad processing phase=performance ads=%d workers=%d",
                len(ad_tasks),
                ad_workers,
            )

            def process_ad(task):
                account, account_id, adgroup_id, ad, ad_metrics = task
                metrics = ad_metrics.get(adgroup_id)
                if metrics is None:
                    return None
                action = determine_ad_performance_cleanup_action(
                    ad,
                    as_of_date=effective_now,
                    cost_fen=metrics.get("cost_fen", 0),
                    current_day_cost_fen=metrics.get("current_day_cost_fen", 0),
                    metric_start_date=ad_metric_start_date,
                    metric_end_date=ad_metric_end_date,
                    window_days=ad_window_days,
                    min_age_days=ad_min_age_days,
                )
                if action is None:
                    return None
                target_id = -int(adgroup_id)
                account_metadata = performance_account_metadata.get(
                    account_id, {}
                )
                record = {
                    "account_id": account_id,
                    "account_name": account_metadata.get("account_name") or "",
                    "agent_name": account_metadata.get("agent_name") or "",
                    "agency_name": "",
                    "adgroup_id": adgroup_id,
                    "adgroup_name": ad.get("adgroup_name") or "",
                    "dynamic_creative_id": target_id,
                    "dynamic_creative_name": "",
                    "check_date": effective_now.date(),
                    **action,
                    "reject_reason": "",
                    "review_result": {},
                    "pre_state": ad,
                }
                return account_id, target_id, action, record

            ad_candidates: list[
                tuple[int, int, dict[str, Any], dict[str, Any]]
            ] = []
            with ThreadPoolExecutor(
                max_workers=ad_workers,
                thread_name_prefix="ad-performance",
            ) as executor:
                futures = {
                    executor.submit(process_ad, task): task for task in ad_tasks
                }
                for completed, future in enumerate(as_completed(futures), start=1):
                    task = futures[future]
                    try:
                        result = future.result()
                    except Exception as exc:
                        error = (
                            "ad performance processing failed "
                            f"account={task[1]} ad={task[2]}: {exc}"
                        )
                        scan_errors.append(error)
                        logger.exception(error)
                    else:
                        if result is not None:
                            ad_candidates.append(result)
                    if completed % 500 == 0 or completed == len(ad_tasks):
                        logger.info(
                            "ad processing phase=performance progress=%d/%d "
                            "confirmed=%d",
                            completed,
                            len(ad_tasks),
                            len(ad_candidates),
                        )
            logger.info(
                "ad processing completed ads=%d confirmed=%d duration_ms=%d",
                len(ad_tasks),
                len(ad_candidates),
                int((time.monotonic() - ad_phase_started_at) * 1000),
            )
            before_performance_discovered = performance_discovered
            persist_candidate_results(ad_candidates, phase="ad_performance")
            ad_discovered = (
                performance_discovered - before_performance_discovered
            )

        deleted = 0
        deferred = 0
        delete_errors: list[str] = []
        write_lock_name = os.getenv(
            "RTC_DB_LOCK_NAME", "tencent_realtime_control"
        )
        retryable_load_started_at = time.monotonic()
        retryable_items = load_retryable_cleanup_items()
        logger.info(
            "cleanup retryable items loaded rows=%d duration_ms=%d",
            len(retryable_items),
            int((time.monotonic() - retryable_load_started_at) * 1000),
        )

        def item_apply_enabled(item: dict[str, Any]) -> bool:
            if _is_ad_performance_rule(item.get("cleanup_rule_type")):
                return ad_apply_enabled
            if _is_performance_rule(item.get("cleanup_rule_type")):
                return performance_apply_enabled
            return apply_enabled

        # 阶段一（锁外）：纯前置判断，无腾讯写。通过者进入 deletable_items。
        deletable_items: list[dict[str, Any]] = []
        for item in retryable_items:
            if not item_apply_enabled(item):
                continue
            item_id = int(item["id"])
            account_id = int(item["account_id"])
            creative_id = int(item["dynamic_creative_id"])
            snapshot_status = str(item.get("cleanup_status") or "")
            if snapshot_status == "DELETING":
                # 只有取得腾讯全局写锁后才恢复超时认领，避免改动仍由其他实例执行的记录。
                deletable_items.append(item)
                continue
            if item.get("cleanup_action") not in {DELETE_CREATIVE, DELETE_AD}:
                update_cleanup_item(
                    item_id,
                    _expected_cleanup_status=snapshot_status,
                    cleanup_status="SKIPPED_REVIEW_NOT_RECONFIRMED",
                    error_message="当前规则仅允许删除整条创意",
                )
                continue
            is_performance_item = _is_performance_rule(
                item.get("cleanup_rule_type")
            )
            agency = (
                ""
                if is_performance_item
                else str(item.get("agency_name") or "")
                or _resolve_agency(context, account_id, creative_id)
            )
            if is_performance_item and item.get("agency_name"):
                update_cleanup_item(
                    item_id,
                    _expected_cleanup_status=snapshot_status,
                    agency_name="",
                    agency_notified_at=(
                        item.get("agency_notified_at") or effective_now
                    ),
                )
                item["agency_name"] = ""
            elif agency and agency != item.get("agency_name"):
                update_cleanup_item(
                    item_id,
                    _expected_cleanup_status=snapshot_status,
                    agency_name=agency,
                )
                item["agency_name"] = agency
            webhook_url = resolve_agency_webhook(
                agency, webhook_config.webhooks or {}
            )
            if not is_performance_item and (not agency or not webhook_url):
                reason = (
                    "代理商归属为空，禁止自动删除"
                    if not agency
                    else f"代理商 {agency} 未配置通知群，禁止自动删除"
                )
                updated = update_cleanup_item(
                    item_id,
                    _expected_cleanup_status=snapshot_status,
                    cleanup_status="DEFERRED",
                    error_message=reason,
                )
                if updated is not False:
                    deferred += 1
                continue
            if item.get("cleanup_status") not in {
                "WRITE_OUTCOME_UNKNOWN",
                "DELETING",
            }:
                precondition_failure = cleanup_precondition_failure(
                    item,
                    scanned_accounts,
                    confirmed_actions,
                    performance_scanned_accounts,
                    ad_scanned_accounts,
                    performance_scope_loaded=(
                        performance_scope_error is None
                    ),
                    performance_source_keys=performance_source_keys,
                    performance_source_ad_keys=performance_source_ad_keys,
                )
                if precondition_failure:
                    status, reason = precondition_failure
                    updated = update_cleanup_item(
                        item_id,
                        _expected_cleanup_status=snapshot_status,
                        cleanup_status=status,
                        error_message=reason,
                    )
                    if status == "DEFERRED" and updated is not False:
                        deferred += 1
                    continue
            deletable_items.append(item)

        # 阶段二（锁内并发）：整个删除批次持锁一次，锁内并发回读/复审/删除。
        # 每 worker 用独立 TencentClient（requests.Session 非线程安全），
        # 避免共享 session 并发导致不可预测行为（与扫描阶段一致）。
        if deletable_items:
            configured_delete_workers = int(
                os.getenv("TENCENT_AD_DELETE_WORKERS", "4")
            )
            if configured_delete_workers < 1:
                raise ValueError("TENCENT_AD_DELETE_WORKERS must be at least 1")
            delete_workers = min(
                configured_delete_workers,
                len(deletable_items),
                16,
            )
            if not owned_tencent:
                # 调用方传入的客户端可能封装 requests.Session，不能假定它线程安全。
                delete_workers = 1
            delete_worker_local = threading.local()
            delete_worker_clients = []
            delete_worker_clients_lock = threading.Lock()

            def delete_one(item, delete_client):
                """锁内单条创意：回读 → 复审 → 删除；返回 (deleted, deferred, error)。"""
                item_id = int(item["id"])
                account_id = int(item["account_id"])
                creative_id = int(item["dynamic_creative_id"])
                adgroup_id = int(item["adgroup_id"])
                is_ad_item = _is_ad_performance_rule(
                    item.get("cleanup_rule_type")
                )
                try:
                    if now is None or clock_was_provided:
                        if not _is_current_cleanup_day(
                            current_day_date,
                            clock_fn(),
                        ):
                            _update_owned_cleanup_item(
                                item_id,
                                cleanup_status="DEFERRED",
                                error_message=(
                                    "任务已跨上海自然日，旧指标窗口禁止执行删除"
                                ),
                            )
                            return 0, 1, None
                    try:
                        before = (
                            delete_client.get_ad(account_id, adgroup_id)
                            if is_ad_item
                            else delete_client.get_dynamic_creative(
                                account_id, creative_id
                            )
                        )
                    except Exception as read_exc:
                        if is_ad_item and str(read_exc).startswith(
                            "Ad not found after update:"
                        ):
                            _update_owned_cleanup_item(
                                item_id,
                                cleanup_status="DEFERRED",
                                error_message=(
                                    "按广告ID查询（含已删除范围）仍未返回广告，"
                                    "无法确认 is_deleted/system_status，禁止推断为已删除"
                                ),
                            )
                            return 0, 1, None
                        if (
                            not is_ad_item
                            and item.get("cleanup_action") == DELETE_CREATIVE
                            and str(read_exc).startswith(
                                "Dynamic creative not found:"
                            )
                        ):
                            updated = _update_owned_cleanup_item(
                                item_id,
                                cleanup_status="CREATIVE_DELETED",
                                error_message=None,
                                readback_json=_json({"deleted_from_listing": True}),
                                deleted_at=effective_now,
                            )
                            if updated is not False:
                                return 1, 0, None
                            return 0, 0, (
                                f"account={account_id} creative={creative_id}: "
                                "delete result ignored because claim ownership was lost"
                            )
                        raise
                    already_deleted = (
                        _is_deleted_ad(before)
                        if is_ad_item
                        else _is_deleted_creative(before)
                    )
                    if already_deleted:
                        updated = _update_owned_cleanup_item(
                            item_id,
                            cleanup_status=(
                                "AD_DELETED" if is_ad_item
                                else "CREATIVE_DELETED"
                            ),
                            error_message=None,
                            pre_state_json=_json(before),
                            readback_json=_json(before),
                            deleted_at=effective_now,
                        )
                        if updated is not False:
                            return 1, 0, None
                        return 0, 0, (
                            f"account={account_id} creative={creative_id}: "
                            "delete result ignored because claim ownership was lost"
                        )
                    if item.get("cleanup_status") in {
                        "WRITE_OUTCOME_UNKNOWN",
                        "DELETING",
                    }:
                        precondition_failure = cleanup_precondition_failure(
                            item,
                            scanned_accounts,
                            confirmed_actions,
                            performance_scanned_accounts,
                            ad_scanned_accounts,
                            performance_scope_loaded=(
                                performance_scope_error is None
                            ),
                            performance_source_keys=performance_source_keys,
                            performance_source_ad_keys=(
                                performance_source_ad_keys
                            ),
                        )
                        if precondition_failure:
                            status, reason = precondition_failure
                            _update_owned_cleanup_item(
                                item_id,
                                cleanup_status=status,
                                error_message=reason,
                            )
                            if status == "DEFERRED":
                                return 0, 1, None
                            return 0, 0, None
                    action = str(item.get("cleanup_action") or "")
                    expected_rule_type = str(
                        item.get("cleanup_rule_type") or REVIEW_DENIED_RULE
                    )
                    fresh_raw = None
                    if _is_ad_performance_rule(expected_rule_type):
                        fresh_metrics = delete_client.get_ad_metrics(
                            account_id,
                            [adgroup_id],
                            ad_metric_start_date,
                            ad_metric_end_date,
                        ).get(adgroup_id)
                        if fresh_metrics is None:
                            raise RuntimeError(
                                "Tencent ad metric response omitted requested ID"
                            )
                        fresh_today_metrics = delete_client.get_ad_metrics(
                            account_id,
                            [adgroup_id],
                            current_day_date,
                            current_day_date,
                        ).get(adgroup_id)
                        if fresh_today_metrics is None:
                            raise RuntimeError(
                                "Tencent current-day ad metric response omitted "
                                "requested ID"
                            )
                        fresh_action = determine_ad_performance_cleanup_action(
                            before,
                            as_of_date=effective_now,
                            cost_fen=fresh_metrics.get("cost_fen", 0),
                            current_day_cost_fen=fresh_today_metrics.get(
                                "cost_fen", 0
                            ),
                            metric_start_date=ad_metric_start_date,
                            metric_end_date=ad_metric_end_date,
                            window_days=ad_window_days,
                            min_age_days=ad_min_age_days,
                        )
                    elif _is_performance_rule(expected_rule_type):
                        before_ad = delete_client.get_ad(
                            account_id,
                            int(item["adgroup_id"]),
                        )
                        created_at = performance_creation_times.get(
                            (account_id, creative_id)
                        )
                        if created_at is None:
                            raise RuntimeError(
                                "active creative inventory omitted creation time"
                            )
                        fresh_metric_start_date = performance_start_date
                        fresh_metric_end_date = performance_end_date
                        if expected_rule_type == PERFORMANCE_NEW_RULE:
                            fresh_metric_start_date = created_at.date()
                        fresh_metrics = delete_client.get_dynamic_creative_metrics(
                            account_id,
                            [creative_id],
                            fresh_metric_start_date,
                            fresh_metric_end_date,
                        ).get(creative_id)
                        if fresh_metrics is None:
                            raise RuntimeError(
                                "Tencent creative metric response omitted requested ID"
                            )
                        fresh_today_metrics = (
                            delete_client.get_dynamic_creative_metrics(
                                account_id,
                                [creative_id],
                                current_day_date,
                                current_day_date,
                            ).get(creative_id)
                        )
                        if fresh_today_metrics is None:
                            raise RuntimeError(
                                "Tencent current-day creative metric response "
                                "omitted requested ID"
                            )
                        fresh_impressions = int(
                            fresh_metrics.get("impressions", 0)
                        )
                        fresh_cost_fen = int(fresh_metrics.get("cost_fen", 0))
                        if expected_rule_type == PERFORMANCE_NEW_RULE:
                            fresh_impressions += int(
                                fresh_today_metrics.get("impressions", 0)
                            )
                            fresh_cost_fen += int(
                                fresh_today_metrics.get("cost_fen", 0)
                            )
                            fresh_metric_end_date = current_day_date
                        fresh_action = determine_performance_cleanup_action(
                            before,
                            before_ad,
                            creative_created_at=created_at,
                            as_of_date=effective_now,
                            impressions=fresh_impressions,
                            cost_fen=fresh_cost_fen,
                            current_day_cost_fen=fresh_today_metrics.get(
                                "cost_fen", 0
                            ),
                            metric_start_date=fresh_metric_start_date,
                            metric_end_date=fresh_metric_end_date,
                            config=performance_settings,
                        )
                    else:
                        approval_status = str(
                            before.get("creative_set_approval_status") or ""
                        )
                        if approval_status == CREATIVE_DENIED_STATUS:
                            fresh_action = determine_cleanup_action(before, None)
                        else:
                            fresh_results = fetch_reviews(account_id, [creative_id])
                            fresh_raw = next(
                                (
                                    result
                                    for result in fresh_results
                                    if _as_int(result.get("dynamic_creative_id"))
                                    == creative_id
                                ),
                                None,
                            )
                            fresh_cost_fen = None
                            fresh_spend_error = None
                            if approval_status == CREATIVE_PARTIAL_NORMAL_STATUS:
                                try:
                                    fresh_cost_fen = (
                                        delete_client.get_dynamic_creative_costs(
                                            account_id,
                                            [creative_id],
                                            spend_start_date,
                                            spend_end_date,
                                        ).get(creative_id, 0)
                                    )
                                except Exception as spend_exc:
                                    fresh_spend_error = str(spend_exc)
                            fresh_action = determine_cleanup_action(
                                before,
                                fresh_raw,
                                recent_cost_fen=fresh_cost_fen,
                                cost_threshold_fen=cost_threshold_fen,
                                wechat_cost_threshold_fen=wechat_cost_threshold_fen,
                                spend_error=fresh_spend_error,
                            )
                    if not _same_cleanup_action(
                        action,
                        fresh_action,
                        expected_rule_type=expected_rule_type,
                    ):
                        if (
                            fresh_action
                            and fresh_action.get("cleanup_action") == ALERT_ONLY
                        ):
                            _update_owned_cleanup_item(
                                item_id,
                                cleanup_action=ALERT_ONLY,
                                cleanup_rule_type=fresh_action.get(
                                    "cleanup_rule_type"
                                ),
                                target_component_ids_json="[]",
                                target_element_ids_json="[]",
                                recent_cost_fen=fresh_action.get("recent_cost_fen"),
                                cost_start_date=spend_start_date,
                                cost_end_date=spend_end_date,
                                action_reason=fresh_action["action_reason"],
                                reject_reason=_reject_reason(
                                    fresh_raw,
                                    str(before.get("system_status") or ""),
                                ),
                                review_result_json=_json(fresh_raw or {}),
                                cleanup_status="ALERT_PENDING",
                                error_message=None,
                                pre_state_json=_json(before),
                                readback_json=None,
                                deleted_at=None,
                                notified_at=None,
                            )
                            return 0, 0, None
                        updated = _update_owned_cleanup_item(
                            item_id,
                            cleanup_status="SKIPPED_REVIEW_NOT_RECONFIRMED",
                            error_message=(
                                "写锁内回读发现广告删除条件已变化"
                                if is_ad_item
                                else "写锁内回读发现整创意删除条件已变化"
                            ),
                            pre_state_json=_json(before),
                        )
                        return 0, 0, None
                    if (
                        (now is None or clock_was_provided)
                        and not _is_current_cleanup_day(
                            current_day_date,
                            clock_fn(),
                        )
                    ):
                        _update_owned_cleanup_item(
                            item_id,
                            cleanup_status="DEFERRED",
                            error_message=(
                                "腾讯写前已跨上海自然日，旧指标窗口禁止执行删除"
                            ),
                        )
                        return 0, 1, None
                    if action == DELETE_CREATIVE:
                        readback = delete_client.delete_dynamic_creative(
                            account_id, creative_id
                        )
                        updated = _update_owned_cleanup_item(
                            item_id,
                            cleanup_status="CREATIVE_DELETED",
                            error_message=None,
                            pre_state_json=_json(before),
                            readback_json=_json(readback),
                            deleted_at=effective_now,
                        )
                        if updated is not False:
                            return 1, 0, None
                        return 0, 0, (
                            f"account={account_id} creative={creative_id}: "
                            "delete result ignored because claim ownership was lost"
                        )
                    if action == DELETE_AD:
                        readback = delete_client.delete_ad(account_id, adgroup_id)
                        updated = _update_owned_cleanup_item(
                            item_id,
                            cleanup_status="AD_DELETED",
                            error_message=None,
                            pre_state_json=_json(before),
                            readback_json=_json(readback),
                            deleted_at=effective_now,
                        )
                        if updated is not False:
                            return 1, 0, None
                        return 0, 0, (
                            f"account={account_id} ad={adgroup_id}: "
                            "delete result ignored because claim ownership was lost"
                        )
                    return 0, 0, None
                except Exception as exc:
                    from tencent_client import (
                        PostWriteVerificationError,
                        TencentWriteOutcomeUnknownError,
                        TencentWriteRateLimitedError,
                    )

                    outcome_unknown = isinstance(
                        exc,
                        (
                            TencentWriteOutcomeUnknownError,
                            PostWriteVerificationError,
                        ),
                    )
                    rate_limited = isinstance(exc, TencentWriteRateLimitedError)
                    error = f"account={account_id} creative={creative_id}: {exc}"
                    try:
                        _update_owned_cleanup_item(
                            item_id,
                            cleanup_status=(
                                "DEFERRED"
                                if rate_limited
                                else (
                                    "WRITE_OUTCOME_UNKNOWN"
                                    if outcome_unknown
                                    else "FAILED"
                                )
                            ),
                            error_message=str(exc)[:4000],
                        )
                    except Exception as update_exc:
                        logger.exception(
                            "creative delete failure status update failed "
                            "account=%d creative=%d",
                            account_id,
                            creative_id,
                        )
                        error += f"; status_update_failed={update_exc}"
                    return 0, int(rate_limited), error

            def run_delete(item):
                if owned_tencent:
                    delete_client = getattr(delete_worker_local, "client", None)
                    if delete_client is None:
                        from tencent_client import TencentClient

                        delete_client = TencentClient()
                        delete_client.seed_access_tokens(prefetched_tokens)
                        delete_worker_local.client = delete_client
                        with delete_worker_clients_lock:
                            delete_worker_clients.append(delete_client)
                else:
                    delete_client = client
                return delete_one(item, delete_client)

            with advisory_lock(write_lock_name) as acquired:
                if not acquired:
                    for item in deletable_items:
                        snapshot_status = str(item.get("cleanup_status") or "")
                        if snapshot_status == "DELETING":
                            continue
                        updated = update_cleanup_item(
                            int(item["id"]),
                            _expected_cleanup_status=snapshot_status,
                            cleanup_status="DEFERRED",
                            error_message="腾讯写锁被实时调控占用",
                        )
                        if updated is not False:
                            deferred += 1
                else:
                    claimed_items = []
                    for item in deletable_items:
                        try:
                            if claim_cleanup_item(int(item["id"])):
                                claimed_items.append(item)
                        except Exception as exc:
                            error = (
                                "creative delete claim failed "
                                f"account={item['account_id']} "
                                f"creative={item['dynamic_creative_id']}: {exc}"
                            )
                            delete_errors.append(error)
                            logger.exception(error)
                    executable_items = []
                    for item in claimed_items:
                        if _is_performance_rule(item.get("cleanup_rule_type")):
                            executable_items.append(item)
                            continue
                        agency = str(item.get("agency_name") or "") or _resolve_agency(
                            context,
                            int(item["account_id"]),
                            int(item["dynamic_creative_id"]),
                        )
                        webhook_url = resolve_agency_webhook(
                            agency, webhook_config.webhooks or {}
                        )
                        if agency and webhook_url:
                            if agency != item.get("agency_name"):
                                _update_owned_cleanup_item(
                                    int(item["id"]),
                                    agency_name=agency,
                                )
                                item["agency_name"] = agency
                            executable_items.append(item)
                            continue
                        reason = (
                            "代理商归属为空，禁止自动删除"
                            if not agency
                            else f"代理商 {agency} 未配置通知群，禁止自动删除"
                        )
                        updated = _update_owned_cleanup_item(
                            int(item["id"]),
                            cleanup_status="DEFERRED",
                            error_message=reason,
                        )
                        if updated is not False:
                            deferred += 1
                    logger.info(
                        "cleanup delete started candidates=%d claimed=%d workers=%d",
                        len(deletable_items),
                        len(executable_items),
                        delete_workers,
                    )
                    if executable_items:
                        try:
                            delete_phases = (
                                (
                                    "review",
                                    [
                                        item for item in executable_items
                                        if not _is_performance_rule(
                                            item.get("cleanup_rule_type")
                                        )
                                    ],
                                ),
                                (
                                    "underperforming-creative",
                                    [
                                        item for item in executable_items
                                        if _is_performance_rule(
                                            item.get("cleanup_rule_type")
                                        )
                                        and not _is_ad_performance_rule(
                                            item.get("cleanup_rule_type")
                                        )
                                    ],
                                ),
                                (
                                    "ad",
                                    [
                                        item for item in executable_items
                                        if _is_ad_performance_rule(
                                            item.get("cleanup_rule_type")
                                        )
                                    ],
                                ),
                            )
                            for phase_name, phase_items in delete_phases:
                                if not phase_items:
                                    continue
                                logger.info(
                                    "cleanup delete phase=%s items=%d",
                                    phase_name,
                                    len(phase_items),
                                )
                                with ThreadPoolExecutor(
                                    max_workers=min(delete_workers, len(phase_items)),
                                    thread_name_prefix=f"{phase_name}-delete",
                                ) as executor:
                                    futures = {
                                        executor.submit(run_delete, item): item
                                        for item in phase_items
                                    }
                                    for future in as_completed(futures):
                                        item = futures[future]
                                        try:
                                            d_deleted, d_deferred, d_error = future.result()
                                        except Exception as exc:
                                            d_deleted = 0
                                            d_deferred = 0
                                            d_error = (
                                                "cleanup delete worker failed "
                                                f"account={item['account_id']} "
                                                f"target={item['dynamic_creative_id']}: {exc}"
                                            )
                                            logger.exception(d_error)
                                            try:
                                                _update_owned_cleanup_item(
                                                    int(item["id"]),
                                                    cleanup_status="FAILED",
                                                    error_message=str(exc)[:4000],
                                                )
                                            except Exception as update_exc:
                                                logger.exception(
                                                    "cleanup delete worker failure status "
                                                    "update failed account=%s target=%s",
                                                    item["account_id"],
                                                    item["dynamic_creative_id"],
                                                )
                                                d_error += (
                                                    f"; status_update_failed={update_exc}"
                                                )
                                        deleted += d_deleted
                                        deferred += d_deferred
                                        if d_error:
                                            delete_errors.append(d_error)
                        finally:
                            for delete_worker_client in delete_worker_clients:
                                delete_worker_client.session.close()

        deliveries: list[dict[str, object]] = []
        operator_deliveries: list[dict[str, object]] = []
        notification_errors: list[str] = []

        def publish_pending_notifications(
            pending_notifications: list[dict[str, Any]],
        ) -> None:
            owned_publisher = publisher is None
            sheet_publisher = publisher or RoiFeishuPublisher(require_chat_ids=False)
            try:
                rows_by_date: dict[str, list[dict[str, Any]]] = defaultdict(list)
                for row in pending_notifications:
                    rows_by_date[_display_date(row.get("check_date"))].append(row)
                for check_date, daily_rows in sorted(rows_by_date.items()):
                    daily_started_at = time.monotonic()
                    report_date = check_date.replace("-", "")
                    report_dir = output_dir / report_date
                    (
                        agency_rows,
                        operator_rows,
                        performance_rows,
                    ) = split_notification_rows(daily_rows)
                    logger.info(
                        "cleanup notification preparing date=%s agency_rows=%d "
                        "operator_rows=%d performance_rows=%d",
                        check_date,
                        len(agency_rows),
                        len(operator_rows),
                        len(performance_rows),
                    )
                    run_id = f"reject_{report_date}_{REPORT_VERSION}"
                    if agency_rows and webhook_config.enabled:
                        run_id, reports, report_item_ids = write_cleanup_reports(
                            agency_rows,
                            report_dir,
                            report_date,
                        )
                        for report in reports:
                            daily_deliveries = publish_agency_reports(
                                run_id=str(report.get("run_id") or run_id),
                                reports=[report],
                                config=webhook_config,
                                publisher=sheet_publisher,
                                notifier=notifier,
                                now=effective_now,
                                upsert_delivery=upsert_cleanup_delivery,
                                update_delivery=update_cleanup_delivery,
                            )
                            deliveries.extend(daily_deliveries)
                            for outcome in daily_deliveries:
                                agency_name = str(outcome["agency_name"])
                                if outcome.get("status") == "SENT":
                                    mark_cleanup_items_notified(
                                        report_item_ids.get(agency_name, []),
                                        effective_now,
                                    )
                                else:
                                    notification_errors.append(
                                        f"agency={agency_name}: "
                                        f"{outcome.get('error') or outcome.get('reason') or outcome.get('status')}"
                                    )
                    if operator_rows:
                        operator_report = write_cleanup_operator_summary(
                            operator_rows,
                            report_dir,
                            report_date,
                            run_id,
                        )
                        operator_outcome = publish_cleanup_operator_summary(
                            run_id=str(operator_report["run_id"]),
                            report=operator_report,
                            chat_id=_operator_summary_chat_id(),
                            publisher=sheet_publisher,
                            now=effective_now,
                        )
                        operator_deliveries.append(operator_outcome)
                        if operator_outcome.get("status") == "SENT":
                            mark_cleanup_items_operator_notified(
                                [int(row["id"]) for row in operator_rows],
                                effective_now,
                            )
                        else:
                            notification_errors.append(
                                f"operator={check_date}: "
                                f"{operator_outcome.get('error') or operator_outcome.get('status')}"
                            )
                    if performance_rows:
                        report_started_at = time.monotonic()
                        performance_report = write_performance_operator_summary(
                            performance_rows,
                            report_dir,
                            report_date,
                        )
                        if force_notification:
                            force_suffix = effective_now.strftime(
                                "%H%M%S%f"
                            )
                            performance_report["run_id"] = (
                                f"{performance_report['run_id']}_force_"
                                f"{force_suffix}"
                            )[:64]
                            performance_report["title"] = (
                                f"{performance_report['title']}（手动重发）"
                            )
                        logger.info(
                            "performance notification excel generated rows=%d "
                            "duration_ms=%d path=%s",
                            len(performance_rows),
                            int((time.monotonic() - report_started_at) * 1000),
                            performance_report.get("report"),
                        )
                        publish_started_at = time.monotonic()
                        performance_outcome = publish_cleanup_operator_summary(
                            run_id=str(performance_report["run_id"]),
                            report=performance_report,
                            chat_id=_operator_summary_chat_id(),
                            publisher=sheet_publisher,
                            now=effective_now,
                        )
                        logger.info(
                            "performance notification published rows=%d status=%s "
                            "duration_ms=%d",
                            len(performance_rows),
                            performance_outcome.get("status"),
                            int((time.monotonic() - publish_started_at) * 1000),
                        )
                        operator_deliveries.append(performance_outcome)
                        if performance_outcome.get("status") == "SENT":
                            mark_cleanup_items_operator_notified(
                                [int(row["id"]) for row in performance_rows],
                                effective_now,
                            )
                        else:
                            notification_errors.append(
                                f"performance_operator={check_date}: "
                                f"{performance_outcome.get('error') or performance_outcome.get('status')}"
                            )
                    logger.info(
                        "cleanup notification completed date=%s rows=%d "
                        "duration_ms=%d",
                        check_date,
                        len(daily_rows),
                        int((time.monotonic() - daily_started_at) * 1000),
                    )
            finally:
                if owned_publisher:
                    sheet_publisher.close()

        include_discovered = (
            not apply_enabled
            or (performance_enabled and not performance_apply_enabled)
            or (ad_cleanup_enabled and not ad_apply_enabled)
        )

        def notification_rows_for_this_run(
            rows: list[dict[str, Any]],
        ) -> list[dict[str, Any]]:
            if not underperformance_preview_only:
                return rows
            selected_rows = filter_preview_notification_rows(
                rows,
                check_date=effective_now.date(),
                confirmed_actions=confirmed_actions,
                force_notification=force_notification,
            )
            logger.info(
                "preview notification selection queried=%d "
                "current_confirmed=%d selected=%d stale_excluded=%d "
                "force_notification=%s",
                len(rows),
                sum(
                    _is_performance_rule(action.get("cleanup_rule_type"))
                    for action in confirmed_actions.values()
                ),
                len(selected_rows),
                len(rows) - len(selected_rows),
                force_notification,
            )
            return selected_rows

        notification_query_filters = (
            {
                "check_date": effective_now.date(),
                "performance_only": True,
                "include_notified": force_notification,
            }
            if underperformance_preview_only
            else {}
        )
        notification_probe_started_at = time.monotonic()
        pending_notification_probe = notification_rows_for_this_run(
            load_unnotified_deleted_items(
                include_discovered=include_discovered,
                **notification_query_filters,
            )
        )
        logger.info(
            "cleanup notification probe rows=%d duration_ms=%d",
            len(pending_notification_probe),
            int((time.monotonic() - notification_probe_started_at) * 1000),
        )
        if pending_notification_probe and (
            webhook_config.enabled
            or any(
                _is_performance_rule(row.get("cleanup_rule_type"))
                for row in pending_notification_probe
            )
        ):
            notification_lock_name = os.getenv(
                "DAILY_REJECTED_CREATIVE_NOTIFICATION_LOCK_NAME",
                "ad_rejected_creative_notification",
            )
            with advisory_lock(notification_lock_name) as acquired:
                if not acquired:
                    logger.info(
                        "creative cleanup notification skipped: lock busy name=%s",
                        notification_lock_name,
                    )
                else:
                    notification_reload_started_at = time.monotonic()
                    pending_notifications = notification_rows_for_this_run(
                        load_unnotified_deleted_items(
                            include_discovered=include_discovered,
                            **notification_query_filters,
                        )
                    )
                    logger.info(
                        "cleanup notification rows reloaded rows=%d duration_ms=%d",
                        len(pending_notifications),
                        int(
                            (time.monotonic() - notification_reload_started_at)
                            * 1000
                        ),
                    )
                    if pending_notifications:
                        publish_pending_notifications(pending_notifications)

        return {
            "underperformance_preview_only": underperformance_preview_only,
            "force_notification": force_notification,
            "apply_enabled": apply_enabled,
            "performance_cleanup_enabled": performance_enabled,
            "performance_apply_enabled": performance_apply_enabled,
            "ad_cleanup_enabled": ad_cleanup_enabled,
            "ad_apply_enabled": ad_apply_enabled,
            "performance_metric_start_date": performance_start_date,
            "performance_metric_end_date": performance_end_date,
            "current_day_metric_date": current_day_date,
            "ad_metric_start_date": ad_metric_start_date,
            "ad_metric_end_date": ad_metric_end_date,
            "performance_missing_creation_time": source_diagnostic_totals[
                "missing_create_time"
            ],
            "account_scope": (
                "opengid_recent_3d_spend_union_odps_active_creatives"
                if underperformance_enabled
                else "opengid_recent_3d_spend"
            ),
            "account_scope_start_date": start_date,
            "account_scope_end_date": end_date,
            "review_scope_error": review_scope_error,
            "review_context_error": review_context_error,
            "review_account_ids": sorted(review_account_ids),
            "performance_account_ids": sorted(performance_account_ids),
            "performance_active_creatives": len(performance_inventory),
            "performance_account_metadata_count": len(
                performance_account_metadata
            ),
            "performance_account_metadata_error": (
                performance_account_metadata_error
            ),
            "performance_source_missing_tencent_creatives": (
                source_diagnostic_totals["missing_tencent_creatives"]
            ),
            "performance_source_missing_tencent_ads": (
                source_diagnostic_totals["missing_tencent_ads"]
            ),
            "performance_source_ad_mismatches": (
                source_diagnostic_totals["ad_mismatches"]
            ),
            "performance_source_missing_ad_ids": (
                source_diagnostic_totals["missing_source_ad_ids"]
            ),
            "accounts": len(accounts),
            "account_ids": account_ids,
            "tokens_prefetched": len(prefetched_tokens),
            "creatives_scanned": scanned,
            "cleanup_discovered": discovered,
            "rejected_discovered": review_discovered,
            "performance_discovered": performance_discovered,
            "ad_discovered": ad_discovered,
            "pending_cleanup": sum(
                1 for item in retryable_items if not item_apply_enabled(item)
            ),
            "deleted": deleted,
            "deferred": deferred,
            "scan_errors": scan_errors,
            "delete_errors": delete_errors,
            "notification_errors": notification_errors,
            "deliveries": deliveries,
            "operator_deliveries": operator_deliveries,
        }
    finally:
        if owned_tencent:
            client.session.close()