长期未起量创意及广告清理 告警fix

examples/auto_put_ad_mini/.env.example

@@ -126,6 +126,9 @@ ROI_FAILURE_FEISHU_CHAT_ID=
 # 创意审核异常/长期未起量清理任务失败告警；为空时回退ROI_FAILURE_FEISHU_CHAT_ID /
 # FEISHU_OPERATOR_CHAT_ID。
 CREATIVE_CLEANUP_FAILURE_FEISHU_CHAT_ID=
+# 逗号分隔、精确匹配；这些归属只发 FEISHU_AD_PROJECT_CHAT_ID 内部汇总，
+# 不生成代理商报表或代理机器人通知。显式配置为空可关闭默认名单。
+CREATIVE_CLEANUP_INTERNAL_ONLY_AGENCIES=小程序-自动化
 # 代理表由飞书应用上传后，通过各代理自定义机器人发送在线表卡片。
 # 默认关闭；JSON 的 key 使用代理简称（如“棱镜”），“内部”为内部测试专用路由。
 ROI_AGENCY_WEBHOOK_ENABLED=0

examples/auto_put_ad_mini/cleanup_rejected_creatives.py

@@ -29,11 +29,47 @@ while str(HERE) in sys.path:
 sys.path.insert(0, str(HERE))
 
 from tools.creative_rejection_cleanup import run_rejected_creative_cleanup  # noqa: E402
+from roi_control.feishu import RoiFeishuPublisher  # noqa: E402
 
 
 logger = logging.getLogger("auto_put_ad_mini.creative_rejection_cleanup")
 
 
+def _notify_delete_failures(delete_errors: list[str]) -> None:
+    """删除失败单独告警，但不改变清理任务的成功退出状态。"""
+
+    if not delete_errors:
+        return
+    publisher: RoiFeishuPublisher | None = None
+    try:
+        publisher = RoiFeishuPublisher(require_chat_ids=False)
+        displayed = [
+            str(error).replace("`", "'")[:500]
+            for error in delete_errors[:10]
+        ]
+        details = "\n".join(f"- `{error}`" for error in displayed)
+        if len(delete_errors) > len(displayed):
+            details += f"\n- 另有 {len(delete_errors) - len(displayed)} 条，详见任务日志"
+        publisher.send_service_alert(
+            title="创意审核异常及未起量清理删除失败",
+            content=(
+                f"本次腾讯删除失败 **{len(delete_errors)}** 条，已保留数据库审计，"
+                "不影响任务成功状态。\n"
+                f"{details}"
+            ),
+            chat_id=os.getenv("CREATIVE_CLEANUP_FAILURE_FEISHU_CHAT_ID", ""),
+        )
+    except Exception:
+        logger.exception(
+            "event=delete_failure_alert task=creative_cleanup result=failed "
+            "delete_error_count=%d",
+            len(delete_errors),
+        )
+    finally:
+        if publisher is not None:
+            publisher.close()
+
+
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
@@ -102,8 +138,10 @@ def main(argv: list[str] | None = None) -> int:
         "delete": list(summary.get("delete_errors") or []),
         "notification": list(summary.get("notification_errors") or []),
     }
-    has_errors = any(error_groups.values())
-    if has_errors:
+    blocking_errors = bool(error_groups["scan"] or error_groups["notification"])
+    if error_groups["delete"]:
+        _notify_delete_failures(error_groups["delete"])
+    if blocking_errors:
         bounded_details = {
             group: [str(error)[:1000] for error in errors[:20]]
             for group, errors in error_groups.items()
@@ -121,13 +159,30 @@ def main(argv: list[str] | None = None) -> int:
             len(error_groups["notification"]),
             json.dumps(bounded_details, ensure_ascii=False, default=str),
         )
+    elif error_groups["delete"]:
+        logger.warning(
+            "event=creative_rejection_cleanup stage=run "
+            "result=completed_with_delete_failures deleted=%s deferred=%s "
+            "delete_error_count=%d",
+            summary.get("deleted", 0),
+            summary.get("deferred", 0),
+            len(error_groups["delete"]),
+        )
     logger.info(
         "event=creative_rejection_cleanup stage=run result=%s duration_ms=%d summary=%s",
-        "partial" if has_errors else "succeeded",
+        (
+            "partial"
+            if blocking_errors
+            else (
+                "completed_with_delete_failures"
+                if error_groups["delete"]
+                else "succeeded"
+            )
+        ),
         int((time.monotonic() - started) * 1000),
         json.dumps(summary, ensure_ascii=False, default=str),
     )
-    return 1 if has_errors else 0
+    return 1 if blocking_errors else 0
 
 
 if __name__ == "__main__":

examples/auto_put_ad_mini/test_creative_performance_cleanup.py

@@ -922,6 +922,23 @@ class AdPerformanceRuleTests(unittest.TestCase):
 
 
 class CreativePerformanceNotificationTests(unittest.TestCase):
+    def test_internal_only_agency_uses_internal_route_without_webhook(self):
+        from tools import creative_rejection_cleanup as cleanup
+
+        with patch.dict(
+            os.environ,
+            {
+                "CREATIVE_CLEANUP_INTERNAL_ONLY_AGENCIES": "小程序-自动化",
+            },
+            clear=False,
+        ):
+            self.assertTrue(
+                cleanup._has_cleanup_notification_route("小程序-自动化", {})
+            )
+            self.assertFalse(
+                cleanup._has_cleanup_notification_route("其他代理", {})
+            )
+
     def test_preview_notification_excludes_candidates_not_confirmed_this_run(self):
         from tools import creative_rejection_cleanup as cleanup
 
@@ -997,13 +1014,27 @@ class CreativePerformanceNotificationTests(unittest.TestCase):
             "agency_notified_at": datetime(2026, 8, 19, 10),
             "operator_notified_at": None,
         }
+        internal_only_review_row = {
+            "id": 4,
+            "agency_name": "小程序-自动化",
+            "cleanup_rule_type": REVIEW_DENIED_RULE,
+            "agency_notified_at": None,
+            "operator_notified_at": None,
+        }
 
-        agency, review_internal, performance_internal = split_notification_rows(
-            [review_row, performance_row, ad_row]
-        )
+        with patch.dict(
+            os.environ,
+            {
+                "CREATIVE_CLEANUP_INTERNAL_ONLY_AGENCIES": "小程序-自动化",
+            },
+            clear=False,
+        ):
+            agency, review_internal, performance_internal = split_notification_rows(
+                [review_row, performance_row, ad_row, internal_only_review_row]
+            )
 
         self.assertEqual(agency, [review_row])
-        self.assertEqual(review_internal, [review_row])
+        self.assertEqual(review_internal, [review_row, internal_only_review_row])
         self.assertEqual(performance_internal, [performance_row, ad_row])
         legacy_agency_rows = [
             row

examples/auto_put_ad_mini/test_creative_review_scan.py

@@ -971,6 +971,33 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         review_fetcher.assert_not_called()
         tencent.get_dynamic_creative_costs.assert_not_called()
 
+    def test_invalid_access_token_skips_all_remaining_account_scans(self):
+        from tools import creative_rejection_cleanup as cleanup
+
+        tencent = Mock()
+        tencent.get_dynamic_creatives.side_effect = RuntimeError(
+            "get_dynamic_creatives failed: code=11002 "
+            "message=您的 access_token 无效，请通过 oauth/authorize 接口获取 token。"
+        )
+        review_fetcher = Mock()
+
+        result = cleanup._scan_one_account(
+            {"account_id": 1},
+            tencent=tencent,
+            review_fetcher=review_fetcher,
+            spend_start_date=date(2026, 8, 14),
+            spend_end_date=date(2026, 8, 16),
+            performance_enabled=True,
+            ad_cleanup_enabled=True,
+        )
+
+        self.assertTrue(cleanup._is_token_skipped_scan_error(result[6]))
+        tencent.get_ads.assert_not_called()
+        review_fetcher.assert_not_called()
+        tencent.get_dynamic_creative_costs.assert_not_called()
+        tencent.get_dynamic_creative_metrics.assert_not_called()
+        tencent.get_ad_metrics.assert_not_called()
+
     def test_tencent_creative_delete_retries_explicit_rate_limit_only(self):
         import tencent_client
 
@@ -2851,6 +2878,48 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         self.assertEqual(values[:5], [3, 1, 0, 0, 1])
         self.assertIn("operator failed", values[5])
 
+    def test_cleanup_entry_delete_failure_notifies_but_returns_zero(self):
+        from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry
+
+        delete_errors = [
+            "account=1 creative=2: delete_dynamic_creative failed: code=30000"
+        ]
+        summary = {
+            "deleted": 3,
+            "deferred": 1,
+            "scan_errors": [],
+            "delete_errors": delete_errors,
+            "notification_errors": [],
+        }
+        with patch("logging_setup.setup_logging"), patch.object(
+            entry, "run_rejected_creative_cleanup", return_value=summary
+        ), patch.object(
+            entry, "_notify_delete_failures"
+        ) as notify_delete_failures:
+            self.assertEqual(entry.main(), 0)
+
+        notify_delete_failures.assert_called_once_with(delete_errors)
+
+    def test_delete_failure_notification_reuses_cleanup_failure_destination(self):
+        from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry
+
+        publisher = Mock()
+        with patch.dict(
+            os.environ,
+            {"CREATIVE_CLEANUP_FAILURE_FEISHU_CHAT_ID": "chat-cleanup-failure"},
+            clear=False,
+        ), patch.object(
+            entry, "RoiFeishuPublisher", return_value=publisher
+        ):
+            entry._notify_delete_failures(["account=1 creative=2: 腾讯系统繁忙"])
+
+        publisher.send_service_alert.assert_called_once()
+        alert = publisher.send_service_alert.call_args.kwargs
+        self.assertEqual(alert["chat_id"], "chat-cleanup-failure")
+        self.assertIn("删除失败", alert["title"])
+        self.assertIn("不影响任务成功状态", alert["content"])
+        publisher.close.assert_called_once_with()
+
     def test_cleanup_entry_exposes_underperformance_preview_only_mode(self):
         from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry
 

examples/auto_put_ad_mini/tools/creative_rejection_cleanup.py

@@ -68,6 +68,8 @@ 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
+DEFAULT_INTERNAL_ONLY_AGENCIES = ("小程序-自动化",)
+TOKEN_SKIPPED_SCAN_PREFIX = "account_scan_skipped reason=access_token_unavailable"
 AGENCY_REPORT_COLUMNS = (
     "代理名称",
     "账户ID",
@@ -753,6 +755,66 @@ def _env_flag(name: str, default: bool = False) -> bool:
     return raw.strip().lower() in {"1", "true", "yes", "on"}
 
 
+def _internal_only_agencies() -> set[str]:
+    raw = os.getenv("CREATIVE_CLEANUP_INTERNAL_ONLY_AGENCIES")
+    if raw is None:
+        values = DEFAULT_INTERNAL_ONLY_AGENCIES
+    else:
+        values = tuple(raw.split(","))
+    return {name for value in values if (name := _agency_name(value))}
+
+
+def _is_internal_only_agency(value: Any) -> bool:
+    return _agency_name(value) in _internal_only_agencies()
+
+
+def _has_cleanup_notification_route(agency: Any, routes: Any) -> bool:
+    normalized = _agency_name(agency)
+    return bool(normalized) and (
+        _is_internal_only_agency(normalized)
+        or bool(resolve_agency_webhook(normalized, routes or {}))
+    )
+
+
+def _is_access_token_unavailable_error(error: Any) -> bool:
+    message = str(error or "").lower()
+    return (
+        "code=11002" in message
+        or "invalid access token" in message
+        or "access_token 无效" in message
+        or "getaccesstoken" in message
+        or "token api 请求失败" in message
+    )
+
+
+def _token_skipped_scan_result(account_id: int, error: Any):
+    return (
+        [],
+        {},
+        {},
+        {},
+        None,
+        0,
+        f"{TOKEN_SKIPPED_SCAN_PREFIX} account={account_id} error={error}",
+        {},
+        None,
+        {},
+        None,
+        None,
+        {
+            "missing_tencent_creatives": 0,
+            "missing_tencent_ads": 0,
+            "ad_mismatches": 0,
+            "missing_source_ad_ids": 0,
+            "missing_create_time": 0,
+        },
+    )
+
+
+def _is_token_skipped_scan_error(error: Any) -> bool:
+    return str(error or "").startswith(TOKEN_SKIPPED_SCAN_PREFIX)
+
+
 def resolve_end_date(client, requested=None, *, now=None):
     from roi_control.data_source import resolve_end_date as resolve
 
@@ -2311,6 +2373,7 @@ def split_notification_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()
+        and not _is_internal_only_agency(row.get("agency_name"))
     ]
     review_internal_rows = [
         row
@@ -2428,6 +2491,12 @@ def _scan_one_account(
         except Exception as exc:
             creatives = []
             creative_list_error = str(exc)
+            if _is_access_token_unavailable_error(exc):
+                logger.warning(
+                    "account scan skipped: access token unavailable account=%d",
+                    account_id,
+                )
+                return _token_skipped_scan_result(account_id, exc)
             logger.exception("creative list scan failed account=%d", account_id)
 
         ad_list_error = None
@@ -2440,6 +2509,12 @@ def _scan_one_account(
         except Exception as exc:
             ads = {}
             ad_list_error = str(exc)
+            if _is_access_token_unavailable_error(exc):
+                logger.warning(
+                    "account scan skipped: access token unavailable account=%d",
+                    account_id,
+                )
+                return _token_skipped_scan_result(account_id, exc)
             logger.exception("ad list scan failed account=%d", account_id)
 
         ids = [
@@ -2464,6 +2539,12 @@ def _scan_one_account(
             except Exception as exc:
                 review_error = str(exc)
                 raw_by_id = {}
+                if _is_access_token_unavailable_error(exc):
+                    logger.warning(
+                        "account scan skipped: access token unavailable account=%d",
+                        account_id,
+                    )
+                    return _token_skipped_scan_result(account_id, exc)
                 logger.exception(
                     "creative review scan failed account=%d", account_id
                 )
@@ -2486,6 +2567,12 @@ def _scan_one_account(
                 )
             except Exception as exc:
                 spend_error = str(exc)
+                if _is_access_token_unavailable_error(exc):
+                    logger.warning(
+                        "account scan skipped: access token unavailable account=%d",
+                        account_id,
+                    )
+                    return _token_skipped_scan_result(account_id, exc)
                 logger.exception(
                     "creative cost scan failed account=%d start=%s end=%s",
                     account_id,
@@ -2660,6 +2747,12 @@ def _scan_one_account(
                 except Exception as exc:
                     performance_metrics = {}
                     performance_error = str(exc)
+                    if _is_access_token_unavailable_error(exc):
+                        logger.warning(
+                            "account scan skipped: access token unavailable account=%d",
+                            account_id,
+                        )
+                        return _token_skipped_scan_result(account_id, exc)
                     logger.exception(
                         "creative performance scan failed account=%d start=%s end=%s",
                         account_id,
@@ -2752,6 +2845,12 @@ def _scan_one_account(
                     }
                 except Exception as exc:
                     ad_metric_error = str(exc)
+                    if _is_access_token_unavailable_error(exc):
+                        logger.warning(
+                            "account scan skipped: access token unavailable account=%d",
+                            account_id,
+                        )
+                        return _token_skipped_scan_result(account_id, exc)
                     logger.exception(
                         "ad performance scan failed account=%d start=%s end=%s",
                         account_id,
@@ -2946,10 +3045,15 @@ def run_rejected_creative_cleanup(
         if underperformance_preview_only
         else AgencyWebhookConfig.from_env()
     )
-    if apply_enabled and not webhook_config.enabled:
+    if (
+        apply_enabled
+        and not webhook_config.enabled
+        and not _internal_only_agencies()
+    ):
         raise RuntimeError(
             "DAILY_REJECTED_CREATIVE_APPLY_ENABLED=1 requires "
-            "ROI_AGENCY_WEBHOOK_ENABLED=1"
+            "ROI_AGENCY_WEBHOOK_ENABLED=1 or a non-empty "
+            "CREATIVE_CLEANUP_INTERNAL_ONLY_AGENCIES"
         )
     if (
         apply_enabled
@@ -3128,6 +3232,11 @@ def run_rejected_creative_cleanup(
     else:
         client = tencent
     prefetched_tokens = prefetch_account_access_tokens(account_ids)
+    token_prefetch_skipped_accounts = (
+        set(account_ids) - set(prefetched_tokens)
+        if owned_tencent
+        else set()
+    )
     seed_tokens = getattr(client, "seed_access_tokens", None)
     if callable(seed_tokens):
         seed_tokens(prefetched_tokens)
@@ -3139,6 +3248,7 @@ def run_rejected_creative_cleanup(
     scanned_accounts: set[int] = set()
     performance_scanned_accounts: set[int] = set()
     ad_scanned_accounts: set[int] = set()
+    token_skipped_accounts: set[int] = set()
     confirmed_actions: dict[tuple[int, int], dict[str, Any]] = {}
     scan_errors: list[str] = []
     if review_scope_error:
@@ -3173,6 +3283,12 @@ def run_rejected_creative_cleanup(
         workers = min(scan_workers, len(accounts), 32) if accounts else 1
 
         def run_scan(account: dict[str, Any]):
+            account_id = int(account["account_id"])
+            if account_id in token_prefetch_skipped_accounts:
+                return _token_skipped_scan_result(
+                    account_id,
+                    "access token prefetch failed",
+                )
             if owned_tencent:
                 from tencent_client import TencentClient
 
@@ -3256,8 +3372,12 @@ def run_rejected_creative_cleanup(
                     }
                 scanned += account_scanned
                 if error:
-                    logger.error(error)
-                    scan_errors.append(error)
+                    if _is_token_skipped_scan_error(error):
+                        token_skipped_accounts.add(account_id)
+                        logger.warning(error)
+                    else:
+                        logger.error(error)
+                        scan_errors.append(error)
                 else:
                     if account_id in review_account_ids:
                         if review_error:
@@ -3310,12 +3430,14 @@ def run_rejected_creative_cleanup(
                         )
                     )
                 logger.info(
-                    "account scan progress=%d/%d account=%d creatives=%d error=%s",
+                    "account scan progress=%d/%d account=%d creatives=%d "
+                    "error=%s token_skipped=%s",
                     completed,
                     len(accounts),
                     account_id,
                     account_scanned,
-                    bool(error),
+                    bool(error and not _is_token_skipped_scan_error(error)),
+                    _is_token_skipped_scan_error(error),
                 )
 
         if any(source_diagnostic_totals.values()):
@@ -3814,10 +3936,11 @@ def run_rejected_creative_cleanup(
                     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):
+            if not is_performance_item and (
+                not _has_cleanup_notification_route(
+                    agency, webhook_config.webhooks or {}
+                )
+            ):
                 reason = (
                     "代理商归属为空，禁止自动删除"
                     if not agency
@@ -4329,10 +4452,9 @@ def run_rejected_creative_cleanup(
                             int(item["account_id"]),
                             int(item["dynamic_creative_id"]),
                         )
-                        webhook_url = resolve_agency_webhook(
+                        if _has_cleanup_notification_route(
                             agency, webhook_config.webhooks or {}
-                        )
-                        if agency and webhook_url:
+                        ):
                             if agency != item.get("agency_name"):
                                 _update_owned_cleanup_item(
                                     int(item["id"]),
@@ -4456,6 +4578,19 @@ def run_rejected_creative_cleanup(
             owned_publisher = publisher is None
             sheet_publisher = publisher or RoiFeishuPublisher(require_chat_ids=False)
             try:
+                # 兼容升级前已成功发送内部汇总、但仍残留代理待通知标记的记录。
+                # 内部通知已完成时直接关闭被明确抑制的代理渠道，避免每日重复加载。
+                suppressed_item_ids = [
+                    int(row["id"])
+                    for row in pending_notifications
+                    if _is_internal_only_agency(row.get("agency_name"))
+                    and row.get("agency_notified_at") is None
+                    and row.get("operator_notified_at") is not None
+                ]
+                if suppressed_item_ids:
+                    mark_cleanup_items_notified(
+                        suppressed_item_ids, effective_now
+                    )
                 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)
@@ -4523,10 +4658,25 @@ def run_rejected_creative_cleanup(
                         )
                         operator_deliveries.append(operator_outcome)
                         if operator_outcome.get("status") == "SENT":
+                            operator_item_ids = [
+                                int(row["id"]) for row in operator_rows
+                            ]
                             mark_cleanup_items_operator_notified(
-                                [int(row["id"]) for row in operator_rows],
-                                effective_now,
+                                operator_item_ids, effective_now
                             )
+                            internal_only_item_ids = [
+                                int(row["id"])
+                                for row in operator_rows
+                                if _is_internal_only_agency(
+                                    row.get("agency_name")
+                                )
+                            ]
+                            # 这类记录明确只发内部群，代理渠道标记为已完成，
+                            # 避免后续任务持续把它当成待发代理通知。
+                            if internal_only_item_ids:
+                                mark_cleanup_items_notified(
+                                    internal_only_item_ids, effective_now
+                                )
                         else:
                             notification_errors.append(
                                 f"operator={check_date}: "
@@ -4653,6 +4803,10 @@ def run_rejected_creative_cleanup(
                 _is_performance_rule(row.get("cleanup_rule_type"))
                 for row in pending_notification_probe
             )
+            or any(
+                _is_internal_only_agency(row.get("agency_name"))
+                for row in pending_notification_probe
+            )
         ):
             notification_lock_name = os.getenv(
                 "DAILY_REJECTED_CREATIVE_NOTIFICATION_LOCK_NAME",
@@ -4732,6 +4886,8 @@ def run_rejected_creative_cleanup(
             "accounts": len(accounts),
             "account_ids": account_ids,
             "tokens_prefetched": len(prefetched_tokens),
+            "token_skipped_account_count": len(token_skipped_accounts),
+            "token_skipped_accounts": sorted(token_skipped_accounts),
             "creatives_scanned": scanned,
             "cleanup_discovered": discovered,
             "rejected_discovered": review_discovered,

runtime.env.example

@@ -195,6 +195,10 @@ ROI_STOP_QUANTILE=0.20
 ROI_STOP_WEIGHT_CAP_QUANTILE=0.95
 ROI_FEISHU_CHAT_ID=
 ROI_FAILURE_FEISHU_CHAT_ID=
+# 清理任务阻断失败和删除失败独立通知使用同一位置；为空时按失败告警默认路由回退。
+CREATIVE_CLEANUP_FAILURE_FEISHU_CHAT_ID=
+# 逗号分隔、精确匹配；默认“小程序-自动化”只通知内部群。
+CREATIVE_CLEANUP_INTERNAL_ONLY_AGENCIES=小程序-自动化
 ROI_AGENCY_WEBHOOK_ENABLED=0
 ROI_AGENCY_WEBHOOKS_JSON={}