创意审核检查删除通知

examples/auto_put_ad_mini/.env.example

@@ -120,6 +120,9 @@ ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2
 ROI_FEISHU_CHAT_ID=
 # 日级ROI定时任务失败告警群；为空时回退FEISHU_OPERATOR_CHAT_ID。
 ROI_FAILURE_FEISHU_CHAT_ID=
+# 拒审创意自动删除任务失败告警；为空时回退ROI_FAILURE_FEISHU_CHAT_ID /
+# FEISHU_OPERATOR_CHAT_ID。
+CREATIVE_CLEANUP_FAILURE_FEISHU_CHAT_ID=
 # 代理表由飞书应用上传后，通过各代理自定义机器人发送在线表卡片。
 # 默认关闭；JSON 的 key 使用代理简称（如“棱镜”），“内部”为内部测试专用路由。
 ROI_AGENCY_WEBHOOK_ENABLED=0

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -135,7 +135,8 @@ ROI_AGENCY_WEBHOOKS_JSON={}
 
 `DAILY_REJECTED_CREATIVE_CLEANUP_ENABLED=1` 会在默认 11:00 注册一个与日级 ROI
 并行的独立调度线程。`DAILY_REJECTED_CREATIVE_APPLY_ENABLED` 默认关闭；关闭时任务只枚举、
-判断和保存待清理审计，不调用腾讯写接口。任务从 `loghubods.opengid_base_data` 查询
+判断和保存待清理审计，不调用腾讯写接口，同时将待删除项标记为“建议删除创意（未执行）”并与
+需人工判断项一起发送预演通知。任务从 `loghubods.opengid_base_data` 查询
 T-3 至 T-1 三个完整日内 `SUM(成本)>0` 的 `账号id`，枚举这些账户的全部未删除创意；
 账户范围不附加 `usersharedepth`、`videoid` 或 `hotsencetype` 条件。
 账户范围不再依赖 MySQL 白名单或自动创建配置。任务批量读取腾讯正式审核结果：
@@ -161,9 +162,9 @@ T-3 至 T-1 三个完整日内 `SUM(成本)>0` 的 `账号id`，枚举这些账
 
 自动清理成功项和待人工判断报警项按日级 ROI 数据中的代理归属生成一代理一份
 `YYYYMMDD_代理名称_创意审核异常处理_批次摘要.xlsx`。代理报表仅展示代理、账户、广告、创意、
-配置状态、创意审核状态 (`creative_set_approval_status`)、元素粒度审核状态/原因、版位粒度审核状态/原因、
-审核不通过原因、检查时间和执行操作；近 3 天消耗、消耗日期范围和操作判断原因只保留在数据库内部审计中，
-不向代理展示。代理归属优先使用日级 ROI 数据中的创意级归属，其次使用账户级唯一归属；仍为空时按账户 ID
+近 3 天历史消耗、执行操作、配置状态、创意审核状态 (`creative_set_approval_status`) 和审核不通过原因。
+投放调控内部汇总表额外展示消耗日期范围、操作判断原因、元素和版位粒度审核状态/原因及检查时间。
+代理归属优先使用日级 ROI 数据中的创意级归属，其次使用账户级唯一归属；仍为空时按账户 ID
 从 `loghubods.ad_put_tencent_account` 按账户读取 `id` 最大的一条记录；只有该最新记录满足
 `is_delete=0`、`status=1` 且 `agent_name` 非空时才作为兜底，不会使用更旧记录，也不会覆盖已有归属。
 Excel 中

examples/auto_put_ad_mini/roi_control/agency_delivery.py

@@ -59,7 +59,14 @@ class AgencyWebhookNotifier:
         ad_rows: int,
         notification_type: str = "roi_advice",
     ) -> str:
-        if notification_type == "creative_rejection_cleanup":
+        if notification_type == "creative_rejection_dry_run":
+            content = (
+                f"创意审核异常处理预演：**{creative_rows}** 条\n"
+                "本次未执行任何删除，请查看建议删除项及需人工判断项。"
+            )
+            button_text = "查看预演明细"
+            header_template = "orange"
+        elif notification_type == "creative_rejection_cleanup":
             content = (
                 f"创意审核异常处理及人工报警：**{creative_rows}** 条\n"
                 "请查看已自动删除项及需要人工判断的部分投放中创意。"

examples/auto_put_ad_mini/run_daily_service.py

@@ -87,6 +87,7 @@ def _notify_creative_cleanup_failure(error: str) -> None:
                 f"错误：`{error[:500]}`\n"
                 "请检查审核扫描、腾讯回读、数据库审计和代理通知日志。"
             ),
+            chat_id=os.getenv("CREATIVE_CLEANUP_FAILURE_FEISHU_CHAT_ID", ""),
         )
     except Exception as exc:
         logger.error("Failed to send creative cleanup failure alert: %s", exc)

examples/auto_put_ad_mini/test_creative_review_scan.py

@@ -742,6 +742,67 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         self.assertIn("cleanup_action='DELETE_CREATIVE'", sql)
         self.assertNotIn("COMPONENTS_PARTIAL", sql)
 
+    def test_dry_run_notification_query_includes_discovered_candidates(self):
+        from tools import creative_rejection_cleanup as cleanup
+
+        cursor = Mock()
+        cursor.__enter__ = Mock(return_value=cursor)
+        cursor.__exit__ = Mock(return_value=None)
+        cursor.fetchall.return_value = []
+        connection = Mock()
+        connection.cursor.return_value = cursor
+
+        with patch.object(cleanup, "get_connection", return_value=connection):
+            cleanup.load_pending_notification_items(include_discovered=True)
+
+        sql = cursor.execute.call_args.args[0]
+        self.assertIn("'DISCOVERED'", sql)
+
+    def test_dry_run_reports_mark_delete_as_not_executed(self):
+        from tools.creative_rejection_cleanup import (
+            write_cleanup_operator_summary,
+            write_cleanup_reports,
+        )
+
+        row = {
+            "id": 7,
+            "agency_name": "棱镜",
+            "account_id": 1,
+            "adgroup_id": 2,
+            "dynamic_creative_id": 3,
+            "cleanup_action": "DELETE_CREATIVE",
+            "cleanup_status": "DISCOVERED",
+            "reject_reason": "审核拒绝",
+            "review_result_json": "{}",
+            "pre_state_json": "{}",
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            output_dir = Path(directory)
+            _, reports, _ = write_cleanup_reports(
+                [row], output_dir, "20260813"
+            )
+            operator = write_cleanup_operator_summary(
+                [row], output_dir, "20260813", "unused"
+            )
+            agency_sheet = load_workbook(reports[0]["report"])[
+                "审核不通过创意清理"
+            ]
+            operator_sheet = load_workbook(operator["report"])[
+                "审核不通过创意清理"
+            ]
+
+        for sheet in (agency_sheet, operator_sheet):
+            columns = {cell.value: cell.column for cell in sheet[1]}
+            self.assertEqual(
+                sheet.cell(2, columns["执行操作"]).value,
+                "建议删除创意（未执行）",
+            )
+        self.assertEqual(
+            reports[0]["notification_type"],
+            "creative_rejection_dry_run",
+        )
+        self.assertTrue(operator["dry_run"])
+
     def test_cleanup_action_does_not_delete_for_overall_or_site_denial(self):
         from tools.creative_rejection_cleanup import determine_cleanup_action
 
@@ -989,7 +1050,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
 
     def test_cleanup_report_has_required_business_columns(self):
         from tools.creative_rejection_cleanup import (
-            REPORT_COLUMNS,
+            AGENCY_REPORT_COLUMNS,
             write_cleanup_reports,
         )
 
@@ -1077,18 +1138,14 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
                 "广告名称",
                 "创意ID",
                 "创意名称",
+                "近3天累计历史消耗(元)",
                 "配置状态",
                 "创意审核状态",
-                "元素粒度审核状态",
-                "元素粒度审核不通过原因",
-                "版位粒度审核状态",
-                "版位粒度审核不通过原因",
                 "审核不通过原因",
-                "检查时间",
                 "执行操作",
             ],
         )
-        self.assertEqual(headers, list(REPORT_COLUMNS))
+        self.assertEqual(headers, list(AGENCY_REPORT_COLUMNS))
         for column_index, expected in (
             (2, "1000000000001"),
             (4, "2000000000002"),
@@ -1105,20 +1162,8 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             "审核拒绝",
         )
         self.assertEqual(
-            sheet.cell(2, by_header["元素粒度审核状态"]).value,
-            "主图(id=10001): 审核拒绝",
-        )
-        self.assertEqual(
-            sheet.cell(2, by_header["元素粒度审核不通过原因"]).value,
-            "主图(id=10001): 图片违规",
-        )
-        self.assertEqual(
-            sheet.cell(2, by_header["版位粒度审核状态"]).value,
-            "SITE_SET_MOMENTS(id=40004): 审核拒绝",
-        )
-        self.assertEqual(
-            sheet.cell(2, by_header["版位粒度审核不通过原因"]).value,
-            "SITE_SET_MOMENTS(id=40004): 朋友圈版位拒绝",
+            sheet.cell(2, by_header["近3天累计历史消耗(元)"]).value,
+            "542.61",
         )
         self.assertEqual(
             sheet.cell(2, by_header["执行操作"]).value,
@@ -1129,14 +1174,14 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             sheet.cell(2, by_header["审核不通过原因"]).value,
             "图片违规",
         )
-        self.assertEqual(
-            sheet.cell(2, by_header["检查时间"]).value,
-            "2026-08-11 11:00:00",
-        )
         for removed_header in (
-            "近3天历史消耗(元)",
             "消耗日期范围",
             "操作判断原因",
+            "元素粒度审核状态",
+            "元素粒度审核不通过原因",
+            "版位粒度审核状态",
+            "版位粒度审核不通过原因",
+            "检查时间",
             "元素ID",
             "组件ID",
             "处理结果",
@@ -1188,11 +1233,15 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         columns = {cell.value: cell.column for cell in sheet[1]}
         self.assertEqual(sheet.cell(2, columns["执行操作"]).value, "需人工判断")
         self.assertEqual(sheet.cell(2, columns["创意审核状态"]).value, "部分投放中")
-        self.assertNotIn("近3天历史消耗(元)", columns)
+        self.assertEqual(
+            sheet.cell(2, columns["近3天累计历史消耗(元)"]).value,
+            "30.00",
+        )
         self.assertNotIn("操作判断原因", columns)
 
     def test_operator_summary_report_contains_all_agencies(self):
         from tools.creative_rejection_cleanup import (
+            OPERATOR_REPORT_COLUMNS,
             write_cleanup_operator_summary,
         )
 
@@ -1207,6 +1256,10 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
                 "dynamic_creative_id": index + 20,
                 "dynamic_creative_name": f"创意{index}",
                 "cleanup_action": "DELETE_CREATIVE",
+                "recent_cost_fen": index * 100,
+                "cost_start_date": "2026-08-09",
+                "cost_end_date": "2026-08-11",
+                "action_reason": f"判断原因{index}",
                 "reject_reason": "图片违规",
                 "review_result_json": "{}",
                 "pre_state_json": "{}",
@@ -1224,7 +1277,20 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             sheet = load_workbook(report["report"])["审核不通过创意清理"]
 
         agencies = [sheet.cell(row, 1).value or "" for row in range(2, 5)]
+        headers = [cell.value for cell in sheet[1]]
+        columns = {cell.value: cell.column for cell in sheet[1]}
         self.assertEqual(agencies, ["代理A", "代理B", ""])
+        self.assertEqual(headers, list(OPERATOR_REPORT_COLUMNS))
+        self.assertEqual(
+            sheet.cell(2, columns["近3天累计历史消耗(元)"]).value,
+            "1.00",
+        )
+        self.assertEqual(
+            sheet.cell(2, columns["消耗日期范围"]).value,
+            "2026-08-09 ~ 2026-08-11",
+        )
+        self.assertEqual(sheet.cell(2, columns["操作判断原因"]).value, "判断原因1")
+        self.assertEqual(headers[-2:], ["执行操作", "操作判断原因"])
         self.assertEqual(report["creative_rows"], 3)
         self.assertIn("投放调控", Path(report["report"]).name)
 
@@ -1349,7 +1415,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             {
                 "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                 "ROI_AGENCY_WEBHOOK_ENABLED": "1",
-                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
+                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                 "ROI_AGENCY_WEBHOOKS_JSON": json.dumps(
                     {
                         "棱镜": (
@@ -1519,7 +1585,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
                 "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                 "DAILY_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN": "30",
                 "ROI_AGENCY_WEBHOOK_ENABLED": "1",
-                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
+                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                 "ROI_AGENCY_WEBHOOKS_JSON": json.dumps(
                     {
                         "棱镜": (
@@ -1652,7 +1718,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             {
                 "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                 "ROI_AGENCY_WEBHOOK_ENABLED": "1",
-                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
+                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                 "ROI_AGENCY_WEBHOOKS_JSON": json.dumps(
                     {
                         "棱镜": (
@@ -1781,7 +1847,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             {
                 "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                 "ROI_AGENCY_WEBHOOK_ENABLED": "1",
-                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
+                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                 "ROI_AGENCY_WEBHOOKS_JSON": "{}",
             },
             clear=False,
@@ -1867,7 +1933,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             {
                 "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
                 "ROI_AGENCY_WEBHOOK_ENABLED": "1",
-                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
+                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                 "ROI_AGENCY_WEBHOOKS_JSON": "{}",
             },
             clear=False,
@@ -2022,11 +2088,11 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
                         )
                     }
                 ),
-                "FEISHU_OPERATOR_CHAT_ID": "",
+                "FEISHU_AD_PROJECT_CHAT_ID": "",
             },
             clear=False,
         ), patch.object(cleanup, "initialize_schema") as initialize:
-            with self.assertRaisesRegex(RuntimeError, "FEISHU_OPERATOR_CHAT_ID"):
+            with self.assertRaisesRegex(RuntimeError, "FEISHU_AD_PROJECT_CHAT_ID"):
                 cleanup.run_rejected_creative_cleanup(output_dir=Path("unused"))
 
         initialize.assert_not_called()
@@ -2062,7 +2128,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
                         "test-cleanup-route"
                     )
                 }),
-                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
+                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
             },
             clear=False,
         ), patch.object(cleanup, "initialize_schema"), patch.object(
@@ -2131,7 +2197,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
                         "test-cleanup-route"
                     )
                 }),
-                "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
+                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
             },
             clear=False,
         ), patch.object(cleanup, "initialize_schema"), patch.object(
@@ -2190,7 +2256,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
                             "test-cleanup-route"
                         )
                     }),
-                    "FEISHU_OPERATOR_CHAT_ID": "chat-operator",
+                    "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
                 },
                 clear=False,
             ), patch.object(cleanup, "initialize_schema"), patch.object(

examples/auto_put_ad_mini/tools/creative_rejection_cleanup.py

@@ -33,7 +33,7 @@ from tools.creative_review import (
 
 logger = logging.getLogger(__name__)
 SHANGHAI = ZoneInfo("Asia/Shanghai")
-REPORT_VERSION = "creative_rejection_cleanup_v7"
+REPORT_VERSION = "creative_rejection_cleanup_v11"
 OPERATOR_SUMMARY_ROUTE = "投放调控汇总"
 DENIED_SYSTEM_STATUS = "DYNAMIC_CREATIVE_STATUS_DENIED"
 DELETED_STATUS = "AD_STATUS_DELETED"
@@ -43,7 +43,7 @@ DELETE_CREATIVE = "DELETE_CREATIVE"
 ALERT_ONLY = "ALERT_ONLY"
 DEFAULT_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN = 50.0
 DEFAULT_WECHAT_MINI_PROGRAM_COST_THRESHOLD_YUAN = 100.0
-REPORT_COLUMNS = (
+AGENCY_REPORT_COLUMNS = (
     "代理名称",
     "账户ID",
     "账户名称",
@@ -51,6 +51,22 @@ REPORT_COLUMNS = (
     "广告名称",
     "创意ID",
     "创意名称",
+    "近3天累计历史消耗(元)",
+    "配置状态",
+    "创意审核状态",
+    "审核不通过原因",
+    "执行操作",
+)
+OPERATOR_REPORT_COLUMNS = (
+    "代理名称",
+    "账户ID",
+    "账户名称",
+    "广告ID",
+    "广告名称",
+    "创意ID",
+    "创意名称",
+    "近3天累计历史消耗(元)",
+    "消耗日期范围",
     "配置状态",
     "创意审核状态",
     "元素粒度审核状态",
@@ -60,7 +76,10 @@ REPORT_COLUMNS = (
     "审核不通过原因",
     "检查时间",
     "执行操作",
+    "操作判断原因",
 )
+# Compatibility for callers that treat the agency report as the default report.
+REPORT_COLUMNS = AGENCY_REPORT_COLUMNS
 
 
 def _json(value: Any) -> str:
@@ -542,16 +561,20 @@ def update_cleanup_item(item_id: int, **values: Any) -> None:
         connection.close()
 
 
-def load_pending_notification_items() -> list[dict[str, Any]]:
+def load_pending_notification_items(
+    *,
+    include_discovered: bool = False,
+) -> list[dict[str, Any]]:
     connection = get_connection()
     try:
         with connection.cursor() as cursor:
+            statuses = "'CREATIVE_DELETED','ALERT_PENDING'"
+            if include_discovered:
+                statuses += ",'DISCOVERED'"
             cursor.execute(
-                """
+                f"""
                 SELECT * FROM creative_rejection_cleanup_item
-                WHERE cleanup_status IN (
-                    'CREATIVE_DELETED','ALERT_PENDING'
-                )
+                WHERE cleanup_status IN ({statuses})
                   AND (
                       agency_notified_at IS NULL
                       OR operator_notified_at IS NULL
@@ -565,10 +588,13 @@ def load_pending_notification_items() -> list[dict[str, Any]]:
         connection.close()
 
 
-def load_unnotified_deleted_items() -> list[dict[str, Any]]:
+def load_unnotified_deleted_items(
+    *,
+    include_discovered: bool = False,
+) -> list[dict[str, Any]]:
     """Compatibility entry point for pending cleanup notifications."""
 
-    return load_pending_notification_items()
+    return load_pending_notification_items(include_discovered=include_discovered)
 
 
 def _mark_cleanup_channel_notified(
@@ -735,7 +761,7 @@ def publish_cleanup_operator_summary(
             raise FileNotFoundError(path)
         target_chat_id = str(chat_id or "").strip()
         if not target_chat_id:
-            raise RuntimeError("FEISHU_OPERATOR_CHAT_ID 未配置")
+            raise RuntimeError("FEISHU_AD_PROJECT_CHAT_ID 未配置")
         delivery = upsert_cleanup_delivery(
             {
                 "run_id": run_id,
@@ -773,8 +799,15 @@ def publish_cleanup_operator_summary(
         message_id = publisher.send_report_card(
             title=str(report.get("title") or path.stem),
             content=(
-                f"本批次共 **{int(report.get('creative_rows') or 0)}** 条创意，"
-                "包含各代理自动删除及需人工判断的完整汇总。"
+                (
+                    f"本次预演共 **{int(report.get('creative_rows') or 0)}** 条创意，"
+                    "未执行任何删除，包含建议删除项及需人工判断项。"
+                )
+                if report.get("dry_run")
+                else (
+                    f"本批次共 **{int(report.get('creative_rows') or 0)}** 条创意，"
+                    "包含各代理自动删除及需人工判断的完整汇总。"
+                )
             ),
             sheet_url=sheet_url,
             chat_id=target_chat_id,
@@ -853,11 +886,16 @@ def _display_date(value: Any) -> str:
     return str(value or "")[:10]
 
 
-def _write_report(path: Path, rows: list[dict[str, Any]]) -> None:
+def _write_report(
+    path: Path,
+    rows: list[dict[str, Any]],
+    *,
+    columns: tuple[str, ...],
+) -> None:
     workbook = Workbook()
     sheet = workbook.active
     sheet.title = "审核不通过创意清理"
-    sheet.append(list(REPORT_COLUMNS))
+    sheet.append(list(columns))
     for row in rows:
         raw_result = _json_object(
             row.get("review_result") or row.get("review_result_json")
@@ -865,7 +903,12 @@ def _write_report(path: Path, rows: list[dict[str, Any]]) -> None:
         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 "")
-        if action == DELETE_CREATIVE:
+        if (
+            action == DELETE_CREATIVE
+            and row.get("cleanup_status") == "DISCOVERED"
+        ):
+            execution_action = "建议删除创意（未执行）"
+        elif action == DELETE_CREATIVE:
             execution_action = "删除创意"
         else:
             execution_action = "需人工判断"
@@ -875,49 +918,81 @@ def _write_report(path: Path, rows: list[dict[str, Any]]) -> None:
             or row.get("updated_at")
             or row.get("deleted_at")
         )
-        sheet.append(
-            [
-                row.get("agency_name") or "",
-                str(row["account_id"]),
-                row.get("account_name") or "",
-                str(row["adgroup_id"]),
-                row.get("adgroup_name") or "",
-                str(row["dynamic_creative_id"]),
-                row.get("dynamic_creative_name") 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"],
-                row["reject_reason"],
-                (
-                    checked_at.strftime("%Y-%m-%d %H:%M:%S")
-                    if isinstance(checked_at, (date, datetime))
-                    else str(checked_at or "")
-                ),
-                execution_action,
-            ]
+        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}"
         )
+        cost_start = _display_date(row.get("cost_start_date"))
+        cost_end = _display_date(row.get("cost_end_date"))
+        values = {
+            "代理名称": row.get("agency_name") or "",
+            "账户ID": str(row["account_id"]),
+            "账户名称": row.get("account_name") or "",
+            "广告ID": str(row["adgroup_id"]),
+            "广告名称": row.get("adgroup_name") or "",
+            "创意ID": str(row["dynamic_creative_id"]),
+            "创意名称": row.get("dynamic_creative_name") or "",
+            "近3天累计历史消耗(元)": recent_cost_yuan,
+            "消耗日期范围": (
+                f"{cost_start} ~ {cost_end}" if 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"],
+            "审核不通过原因": row["reject_reason"],
+            "检查时间": (
+                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 = [22, 14, 22, 14, 30, 16, 30, 20, 24, 40, 60, 40, 60, 60, 20, 16]
-    for index, width in enumerate(widths, start=1):
-        sheet.column_dimensions[get_column_letter(index)].width = width
+    widths = {
+        "代理名称": 22,
+        "账户ID": 14,
+        "账户名称": 22,
+        "广告ID": 14,
+        "广告名称": 30,
+        "创意ID": 16,
+        "创意名称": 30,
+        "近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 column_index in (2, 4, 6):
-            row[column_index - 1].number_format = "@"
+        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)
@@ -960,16 +1035,28 @@ def write_cleanup_reports(
         path = output_dir / (
             f"{report_date}_{safe_agency}_创意审核异常处理_{agency_digest}.xlsx"
         )
-        _write_report(path, agency_rows)
+        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}_创意审核异常处理通知",
+                "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_cleanup",
+                "notification_type": (
+                    "creative_rejection_dry_run"
+                    if dry_run
+                    else "creative_rejection_cleanup"
+                ),
                 "run_id": f"reject_{report_date}_{agency_digest}",
             }
         )
@@ -1002,13 +1089,19 @@ def write_cleanup_operator_summary(
     path = output_dir / (
         f"{report_date}_投放调控_创意审核异常处理汇总_{digest}.xlsx"
     )
-    _write_report(path, rows)
+    dry_run = any(row.get("cleanup_status") == "DISCOVERED" for row in rows)
+    _write_report(path, rows, columns=OPERATOR_REPORT_COLUMNS)
     return {
         "report_version": f"{REPORT_VERSION}_operator_summary",
         "report": str(path),
-        "title": f"{report_date}_创意审核异常处理汇总通知",
+        "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,
     }
 
 
@@ -1141,10 +1234,10 @@ def run_rejected_creative_cleanup(
             "DAILY_REJECTED_CREATIVE_APPLY_ENABLED=1 requires "
             "ROI_AGENCY_WEBHOOK_ENABLED=1"
         )
-    if apply_enabled and not os.getenv("FEISHU_OPERATOR_CHAT_ID", "").strip():
+    if apply_enabled and not os.getenv("FEISHU_AD_PROJECT_CHAT_ID", "").strip():
         raise RuntimeError(
             "DAILY_REJECTED_CREATIVE_APPLY_ENABLED=1 requires "
-            "FEISHU_OPERATOR_CHAT_ID"
+            "FEISHU_AD_PROJECT_CHAT_ID"
         )
     initialize_schema()
     if odps is None:
@@ -1271,66 +1364,112 @@ def run_rejected_creative_cleanup(
                     bool(error),
                 )
 
-        for (
-            account,
-            creatives,
-            ads,
-            raw_by_id,
-            cost_by_id,
-            spend_error,
-        ) in scan_results:
-            account_id = int(account["account_id"])
-            for creative in creatives:
-                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:
-                    continue
-                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
-                )
-                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 action is None:
+        def process_creative(task):
+            (
+                account,
+                account_id,
+                creative,
+                ads,
+                raw_by_id,
+                cost_by_id,
+                spend_error,
+            ) = 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
+            )
+            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 action is None:
+                return None
+            ad = ads.get(adgroup_id) or {}
+            upsert_cleanup_candidate(
+                {
+                    "account_id": account_id,
+                    "account_name": context["account_names"].get(account_id)
+                    or account.get("account_name")
+                    or "",
+                    "agency_name": _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": _reject_reason(raw_result, system_status),
+                    "cost_start_date": spend_start_date,
+                    "cost_end_date": spend_end_date,
+                    "review_result": raw_result or {},
+                    "pre_state": creative,
+                }
+            )
+            return account_id, creative_id, action
+
+        tasks = [
+            (
+                account,
+                int(account["account_id"]),
+                creative,
+                ads,
+                raw_by_id,
+                cost_by_id,
+                spend_error,
+            )
+            for account, creatives, ads, raw_by_id, cost_by_id, spend_error
+            in scan_results
+            for creative in creatives
+        ]
+        process_workers = (
+            min(
+                int(os.getenv("TENCENT_AD_CREATIVE_PROCESS_WORKERS", "8")),
+                len(tasks),
+                32,
+            )
+            if tasks
+            else 1
+        )
+        logger.info(
+            "creative processing started creatives=%d workers=%d",
+            len(tasks),
+            process_workers,
+        )
+        with ThreadPoolExecutor(
+            max_workers=process_workers,
+            thread_name_prefix="creative-process",
+        ) as executor:
+            results = executor.map(process_creative, tasks)
+            for completed, result in enumerate(results, start=1):
+                if result is None:
                     continue
+                account_id, creative_id, action = result
                 confirmed_actions[(account_id, creative_id)] = action
-                ad = ads.get(adgroup_id) or {}
-                upsert_cleanup_candidate(
-                    {
-                        "account_id": account_id,
-                        "account_name": context["account_names"].get(account_id)
-                        or account.get("account_name")
-                        or "",
-                        "agency_name": _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": _reject_reason(raw_result, system_status),
-                        "cost_start_date": spend_start_date,
-                        "cost_end_date": spend_end_date,
-                        "review_result": raw_result or {},
-                        "pre_state": creative,
-                    }
-                )
                 discovered += 1
+                if completed % 500 == 0:
+                    logger.info(
+                        "creative processing progress=%d/%d confirmed=%d",
+                        completed,
+                        len(tasks),
+                        discovered,
+                    )
 
         deleted = 0
         deferred = 0
@@ -1547,7 +1686,9 @@ def run_rejected_creative_cleanup(
                         f"account={account_id} creative={creative_id}: {exc}"
                     )
 
-        pending_notifications = load_unnotified_deleted_items()
+        pending_notifications = load_unnotified_deleted_items(
+            include_discovered=not apply_enabled,
+        )
         deliveries: list[dict[str, object]] = []
         operator_deliveries: list[dict[str, object]] = []
         notification_errors: list[str] = []
@@ -1613,7 +1754,7 @@ def run_rejected_creative_cleanup(
                         operator_outcome = publish_cleanup_operator_summary(
                             run_id=str(operator_report["run_id"]),
                             report=operator_report,
-                            chat_id=os.getenv("FEISHU_OPERATOR_CHAT_ID", ""),
+                            chat_id=os.getenv("FEISHU_AD_PROJECT_CHAT_ID", ""),
                             publisher=sheet_publisher,
                             now=effective_now,
                         )