Refine creative preparation safeguards

AGENTS.md

@@ -15,7 +15,7 @@
 ## 模块 B 创意创建规则
 
 - 目标是单广告最终合格创意数,不是单次生成的 pending 行数。
-- 当前目标:每条广告最终有效创意不少于 12 个。
+- 当前目标:每条广告最终有效创意不少于 8 个。
 - `DENIED` 创意不计入有效创意。
 - 每次创意准备中,每个视频来源最多获取并尝试 100 条视频。
 - 先尝试 primary 视频来源。
@@ -54,6 +54,7 @@
 - 同一广告内同一 `landing_video_id` 仍保留限频保护,默认最多 1 次。
 - 素材需要跨账户/跨天排重,默认按同人群包下近期使用过的 `material_id` 排除。
 - 当前运行内的审批候选需要做轻量展示去重:同一 `crowd_package + material_id` 只进入本轮审批表一次;该去重只存在内存,不写入历史排重库。
+- 当前 `prepare_one_creative_for_ad` 会真实上传图片并创建 `xcx/save` 落地计划,默认不能直接并行执行;如需并行,必须先拆分为纯候选选择和副作用串行确认两段。
 - `videoContentList` 每个 source 默认最多读取 3 页,每页 100 条,按 `video_id` 去重合并。
 
 ## 视频召回人群包映射

examples/auto_put_ad_mini/PRODUCTION_AUTOMATION.md

@@ -42,17 +42,29 @@
 当前单广告最终有效创意目标:
 
 ```text
-TARGET_CREATIVES_PER_AD = 12
+TARGET_CREATIVES_PER_AD = 8
 ```
 
 含义:
 
-- 每条广告最终有效创意数不少于 12 个
+- 每条广告最终有效创意数不少于 8 个
 - 已有有效创意会计入目标
 - `DENIED` 创意不计入有效创意
 - 差几个补几个
 - 补充创意会先进入飞书审批,审批通过后才调用腾讯创建接口
 
+Phase 1 创意准备默认串行:
+
+```text
+CREATIVE_PREPARE_MAX_WORKERS = 1
+CREATIVE_PREPARE_TASK_BUFFER = 16
+```
+
+- `prepare_one_creative_for_ad` 当前会真实上传图片并调用 `xcx/save` 创建落地计划,不是纯候选选择函数。
+- 直接并行该函数会导致多个 worker 基于同一排重快照选中同一 landing/material,产生重复落地计划。
+- 因此生产默认保持串行。后续要并行时,应先拆成“纯候选召回/筛选并行 + 图片上传/落地计划创建串行确认”。
+- 主线程统一接收 pending record 并做最终 landing/material 排重。
+
 ## 承接视频选择
 
 每次为一条广告准备一条创意时,视频最多扫描:
@@ -264,6 +276,9 @@ VIDEO_RECALL_CROWD_PACKAGE_MAP={"cell*year*商业":"wx*商业","回流330以上
 MAX_SAME_LANDING_PER_AD_IN_RUN=1
 PIAOQUANTV_VIDEO_MAX_PAGES=3
 CREATIVE_LANDING_DEDUPE_LOOKBACK_DAYS=7
+TARGET_CREATIVES_PER_AD=8
+CREATIVE_PREPARE_MAX_WORKERS=1
+CREATIVE_PREPARE_TASK_BUFFER=16
 TENCENT_AUDIENCE_SOURCE_ACCOUNT_ID=55615440
 TENCENT_AUDIENCE_GRANT_BUSINESS_ID=12312
 OPENROUTER_API_KEY=...

examples/auto_put_ad_mini/config.py

@@ -942,10 +942,12 @@ AUDIENCE_TIER_PATTERNS = [
 # 数据流:find_ads_needing_creatives → 关联点过滤 → 召回素材 → POST 创意。
 
 # --- 创意补量目标(单广告期望创意数)---
-# 2026-07-08:为提升冷启动素材供给,用户确认最终合格创意目标提升到 12 个
+# 2026-07-09:用户确认最终合格创意目标调整为 8 个
 # 生产阶段可继续测试 15-30(腾讯经验下限,MIN_CREATIVES_PER_AD)
 # 这是 find_ads_needing_creatives 阈值 + 补量目标的**同一个语义变量**,不要拆
-TARGET_CREATIVES_PER_AD = int(os.getenv("TARGET_CREATIVES_PER_AD", "12"))
+TARGET_CREATIVES_PER_AD = int(os.getenv("TARGET_CREATIVES_PER_AD", "8"))
+CREATIVE_PREPARE_MAX_WORKERS = int(os.getenv("CREATIVE_PREPARE_MAX_WORKERS", "1"))
+CREATIVE_PREPARE_TASK_BUFFER = int(os.getenv("CREATIVE_PREPARE_TASK_BUFFER", "16"))
 
 # --- 主循环 try-fallback 限额(防无限召回)---
 # 单广告最多尝试 N 条 landing,超过即放弃此条创意(不影响广告剩余 to_add)

examples/auto_put_ad_mini/execute_creation_once.py

@@ -26,6 +26,7 @@ import logging
 import os
 import sys
 import time
+from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
 from pathlib import Path
 
 _HERE = Path(__file__).parent
@@ -39,6 +40,8 @@ from config import (  # noqa: E402
     ADS_PER_ACCOUNT,
     CREATION_APPROVAL_REQUIRED,
     CREATION_APPROVAL_TIMEOUT_MINUTES,
+    CREATIVE_PREPARE_MAX_WORKERS,
+    CREATIVE_PREPARE_TASK_BUFFER,
     MAX_SAME_LANDING_PER_AD_IN_RUN,
     TARGET_CREATIVES_PER_AD,
     WHITELIST_ACCOUNTS,
@@ -468,10 +471,43 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
         display_excluded_material_ids: set[str],
         crowd_landing_counts: dict[str, dict[int, int]],
         landing_counts_for_ad: dict[int, int],
+        landing_max_uses: int,
         recovered: bool = False,
-    ) -> None:
-        pending_records.append(rec)
+    ) -> bool:
         material_id = rec.get("_material_id")
+        if material_id and str(material_id) in display_excluded_material_ids:
+            logger.info(
+                "[phase1] pending 接收排重丢弃 crowd=%r material=%s%s",
+                crowd_package, material_id,
+                " recovered" if recovered else "",
+            )
+            return False
+        landing_video_id = rec.get("landing_video_id")
+        actual_material_source = (
+            MATERIAL_SOURCE_AI_GENERATED
+            if str(rec.get("material_source") or "") == MATERIAL_SOURCE_AI_GENERATED
+            or str(rec.get("_material_id") or "").startswith("ai:")
+            else MATERIAL_SOURCE_HISTORY
+        )
+        if landing_video_id is not None:
+            landing_video_id = int(landing_video_id)
+            if landing_counts_for_ad.get(landing_video_id, 0) >= MAX_SAME_LANDING_PER_AD_IN_RUN:
+                logger.info(
+                    "[phase1] pending 接收排重丢弃 adgroup=%s landing=%d 同广告已达上限%s",
+                    rec.get("adgroup_id"), landing_video_id,
+                    " recovered" if recovered else "",
+                )
+                return False
+            source_counts = crowd_landing_counts.setdefault(actual_material_source, {})
+            if source_counts.get(landing_video_id, 0) >= landing_max_uses:
+                logger.info(
+                    "[phase1] pending 接收排重丢弃 crowd=%r material_source=%s landing=%d 同人群包已达上限%s",
+                    crowd_package, actual_material_source, landing_video_id,
+                    " recovered" if recovered else "",
+                )
+                return False
+
+        pending_records.append(rec)
         if material_id:
             display_excluded_material_ids.add(str(material_id))
             logger.info(
@@ -480,25 +516,17 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
                 len(display_excluded_material_ids),
                 " recovered" if recovered else "",
             )
-        landing_video_id = rec.get("landing_video_id")
         if landing_video_id is not None:
-            landing_video_id = int(landing_video_id)
             landing_counts_for_ad[landing_video_id] = (
                 landing_counts_for_ad.get(landing_video_id, 0) + 1
             )
-            actual_material_source = (
-                MATERIAL_SOURCE_AI_GENERATED
-                if str(rec.get("material_source") or "") == MATERIAL_SOURCE_AI_GENERATED
-                or str(rec.get("_material_id") or "").startswith("ai:")
-                else MATERIAL_SOURCE_HISTORY
-            )
             source_counts = crowd_landing_counts.setdefault(actual_material_source, {})
             source_counts[landing_video_id] = source_counts.get(landing_video_id, 0) + 1
             logger.info(
                 "[phase1] 同人群包 landing 使用登记 crowd=%r material_source=%s landing=%d count=%d max=%d%s",
                 crowd_package, actual_material_source, landing_video_id,
                 source_counts[landing_video_id],
-                MAX_SAME_LANDING_PER_AD_IN_RUN,
+                landing_max_uses,
                 " recovered" if recovered else "",
             )
             logger.info(
@@ -517,6 +545,7 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
                     rec.get("account_id"), rec.get("adgroup_id"),
                     rec.get("_material_id"), e,
                 )
+        return True
 
     for account_id in creation_accounts:
         logger.info("=" * 60)
@@ -612,23 +641,46 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
                 material_source=requested_material_source,
                 limit=to_add,
             )
+            recovered_accepted = 0
             for rec in recovered_records:
-                accept_pending_record(
+                if accept_pending_record(
                     rec,
                     crowd_package=crowd_package,
                     display_excluded_material_ids=display_excluded_material_ids,
                     crowd_landing_counts=crowd_landing_counts,
                     landing_counts_for_ad=landing_counts_for_ad,
+                    landing_max_uses=landing_max_uses,
                     recovered=True,
-                )
+                ):
+                    recovered_accepted += 1
             if recovered_records:
-                prepared_for_ad += len(recovered_records)
+                prepared_for_ad += recovered_accepted
                 logger.info(
-                    "[phase1]   adgroup=%d 恢复未提交 prepared records=%d remaining=%d",
-                    adgroup_id, len(recovered_records), max(0, to_add - prepared_for_ad),
+                    "[phase1]   adgroup=%d 恢复未提交 prepared records=%d accepted=%d remaining=%d",
+                    adgroup_id, len(recovered_records), recovered_accepted,
+                    max(0, to_add - prepared_for_ad),
                 )
 
-            for _ in range(max(0, to_add - prepared_for_ad)):
+            def prepare_attempt(
+                *,
+                attempt_no: int,
+                excluded_material_ids_snapshot: set[str],
+                excluded_landing_ids_snapshot: set[int],
+            ) -> dict | None:
+                try:
+                    return prepare_one_creative_for_ad(
+                        account_id, adgroup_id,
+                        excluded_material_ids=excluded_material_ids_snapshot,
+                        excluded_landing_ids=excluded_landing_ids_snapshot,
+                    )
+                except Exception as e:
+                    logger.exception(
+                        "[phase1] adgroup=%d attempt=%d 准备失败:%s",
+                        adgroup_id, attempt_no, e,
+                    )
+                    return None
+
+            def current_excluded_landing_ids() -> set[int]:
                 landing_excluded_for_ad = {
                     vid
                     for vid, count in landing_counts_for_ad.items()
@@ -641,38 +693,132 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
                     .items()
                     if count >= landing_max_uses
                 }
-                effective_excluded_landing_ids = (
-                    landing_excluded_for_source | landing_excluded_for_ad
+                return landing_excluded_for_source | landing_excluded_for_ad
+
+            remaining_to_add = max(0, to_add - prepared_for_ad)
+            max_workers = max(1, int(CREATIVE_PREPARE_MAX_WORKERS))
+            if max_workers > 1:
+                logger.warning(
+                    "[phase1] CREATIVE_PREPARE_MAX_WORKERS=%d 已配置但当前 prepare_one_creative_for_ad "
+                    "包含图片上传/xcx-save 等外部副作用,为避免重复创建落地计划,本轮强制串行执行",
+                    max_workers,
                 )
-                try:
-                    rec = prepare_one_creative_for_ad(
-                        account_id, adgroup_id,
-                        excluded_material_ids=display_excluded_material_ids,
-                        excluded_landing_ids=effective_excluded_landing_ids,
-                    )
-                except Exception as e:
-                    logger.exception(
-                        "[phase1] adgroup=%d 准备失败:%s", adgroup_id, e,
-                    )
-                    rec = None
-
-                if rec:
-                    prepared_for_ad += 1
-                    accept_pending_record(
-                        rec,
-                        crowd_package=crowd_package,
-                        display_excluded_material_ids=display_excluded_material_ids,
-                        crowd_landing_counts=crowd_landing_counts,
-                        landing_counts_for_ad=landing_counts_for_ad,
-                    )
-                else:
-                    failed_prepare_for_ad += 1
-                    # 2026-06-10 用户要求:单条 prepare 失败 → continue 不 break
-                    # 同广告剩余 to_add 创意还能继续试,不被一次失败拖累
-                    logger.info(
-                        "[phase1]   adgroup=%d 本条创意 prepare 失败,试下一条",
-                        adgroup_id,
+                max_workers = 1
+            task_buffer = max(remaining_to_add, int(CREATIVE_PREPARE_TASK_BUFFER))
+            task_limit = min(max(0, remaining_to_add * 2), task_buffer)
+            if remaining_to_add > 0 and max_workers > 1:
+                logger.info(
+                    "[phase1]   adgroup=%d 并行准备 start remaining=%d workers=%d task_limit=%d",
+                    adgroup_id, remaining_to_add, max_workers, task_limit,
+                )
+                submitted = 0
+                accepted_parallel = 0
+                dropped_parallel = 0
+                failed_parallel = 0
+                started_at = time.monotonic()
+                with ThreadPoolExecutor(max_workers=max_workers) as executor:
+                    futures = {}
+
+                    def submit_next() -> None:
+                        nonlocal submitted
+                        # 已接收 + 运行中任务不超过缺口,避免并发准备产出无法追踪的多余副作用。
+                        if (
+                            submitted >= task_limit
+                            or prepared_for_ad + len(futures) >= to_add
+                        ):
+                            return
+                        submitted += 1
+                        attempt_no = submitted
+                        future = executor.submit(
+                            prepare_attempt,
+                            attempt_no=attempt_no,
+                            excluded_material_ids_snapshot=set(display_excluded_material_ids),
+                            excluded_landing_ids_snapshot=current_excluded_landing_ids(),
+                        )
+                        futures[future] = attempt_no
+
+                    for _ in range(min(max_workers, task_limit)):
+                        submit_next()
+
+                    while futures and prepared_for_ad < to_add:
+                        done, _ = wait(futures, return_when=FIRST_COMPLETED)
+                        for future in done:
+                            futures.pop(future, None)
+                            rec = future.result()
+                            if not rec:
+                                failed_prepare_for_ad += 1
+                                failed_parallel += 1
+                            elif accept_pending_record(
+                                rec,
+                                crowd_package=crowd_package,
+                                display_excluded_material_ids=display_excluded_material_ids,
+                                crowd_landing_counts=crowd_landing_counts,
+                                landing_counts_for_ad=landing_counts_for_ad,
+                                landing_max_uses=landing_max_uses,
+                            ):
+                                prepared_for_ad += 1
+                                accepted_parallel += 1
+                            else:
+                                dropped_parallel += 1
+                            if prepared_for_ad < to_add:
+                                submit_next()
+                            else:
+                                break
+                    if prepared_for_ad >= to_add:
+                        logger.info(
+                            "[phase1]   adgroup=%d 并行准备 target reached accepted=%d",
+                            adgroup_id, accepted_parallel,
+                        )
+                        cancelled = 0
+                        for pending_future in futures:
+                            if pending_future.cancel():
+                                cancelled += 1
+                        if cancelled:
+                            logger.info(
+                                "[phase1]   adgroup=%d 并行准备 cancelled_pending=%d",
+                                adgroup_id, cancelled,
+                            )
+                    for future in futures:
+                        if future.cancelled():
+                            continue
+                        rec = future.result()
+                        if rec:
+                            dropped_parallel += 1
+                            logger.info(
+                                "[phase1]   adgroup=%d 并行结果丢弃:target reached material=%s landing=%s",
+                                adgroup_id, rec.get("_material_id"), rec.get("landing_video_id"),
+                            )
+                logger.info(
+                    "[phase1]   adgroup=%d 并行准备 done submitted=%d accepted=%d dropped=%d failed=%d elapsed=%.1fs",
+                    adgroup_id, submitted, accepted_parallel, dropped_parallel,
+                    failed_parallel, time.monotonic() - started_at,
+                )
+            else:
+                for attempt_no in range(1, remaining_to_add + 1):
+                    rec = prepare_attempt(
+                        attempt_no=attempt_no,
+                        excluded_material_ids_snapshot=set(display_excluded_material_ids),
+                        excluded_landing_ids_snapshot=current_excluded_landing_ids(),
                     )
+
+                    if rec:
+                        if accept_pending_record(
+                            rec,
+                            crowd_package=crowd_package,
+                            display_excluded_material_ids=display_excluded_material_ids,
+                            crowd_landing_counts=crowd_landing_counts,
+                            landing_counts_for_ad=landing_counts_for_ad,
+                            landing_max_uses=landing_max_uses,
+                        ):
+                            prepared_for_ad += 1
+                    else:
+                        failed_prepare_for_ad += 1
+                        # 2026-06-10 用户要求:单条 prepare 失败 → continue 不 break
+                        # 同广告剩余 to_add 创意还能继续试,不被一次失败拖累
+                        logger.info(
+                            "[phase1]   adgroup=%d 本条创意 prepare 失败,试下一条",
+                            adgroup_id,
+                        )
             logger.info(
                 "[phase1]   adgroup=%d 补创意完成 target=%d have_before=%d "
                 "planned=%d prepared=%d failed_prepare=%d",