Optimize creative candidate preparation

examples/auto_put_ad_mini/execute_creation_once.py

@@ -63,6 +63,7 @@ from tools.ad_creation import (  # noqa: E402
 )
 from tools.audience_grant import ensure_account_audience_grant  # noqa: E402
 from tools.creative_creation import (  # noqa: E402
+    build_landing_candidate_pool,
     find_ads_needing_creatives,
     load_excluded_ad_ids_from_adjustment,
     prepare_one_creative_for_ad,
@@ -618,6 +619,7 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
             else MATERIAL_SOURCE_HISTORY
         )
         landing_max_uses = MAX_SAME_LANDING_PER_AD_IN_RUN
+        landing_candidate_pool = None
         logger.info(
             "[phase1] account=%d landing 排重池=%s max_uses=%d",
             account_id, requested_material_source, landing_max_uses,
@@ -629,6 +631,7 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
             to_add = max(0, target_creatives - already_have)
             prepared_for_ad = 0
             failed_prepare_for_ad = 0
+            failed_landing_ids_for_ad: set[int] = set()
             landing_counts_for_ad: dict[int, int] = {}
             logger.info(
                 "[phase1]   adgroup=%d(have=%d need=%d)",
@@ -667,11 +670,16 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
                 excluded_material_ids_snapshot: set[str],
                 excluded_landing_ids_snapshot: set[int],
             ) -> dict | None:
+                nonlocal landing_candidate_pool
                 try:
+                    if landing_candidate_pool is None:
+                        landing_candidate_pool = build_landing_candidate_pool(account_id)
                     return prepare_one_creative_for_ad(
                         account_id, adgroup_id,
                         excluded_material_ids=excluded_material_ids_snapshot,
                         excluded_landing_ids=excluded_landing_ids_snapshot,
+                        landing_candidates=landing_candidate_pool,
+                        failed_landing_ids=failed_landing_ids_for_ad,
                     )
                 except Exception as e:
                     logger.exception(

examples/auto_put_ad_mini/test_landing_video_dedupe.py

@@ -21,7 +21,7 @@ class LandingVideoDedupeTest(unittest.TestCase):
     def test_same_crowd_package_excludes_landing_video_across_accounts_in_one_run(self):
         calls = []
 
-        def fake_prepare(account_id, adgroup_id, *, excluded_material_ids, excluded_landing_ids):
+        def fake_prepare(account_id, adgroup_id, *, excluded_material_ids, excluded_landing_ids, **kwargs):
             calls.append((account_id, adgroup_id, set(excluded_landing_ids)))
             return {
                 "account_id": account_id,
@@ -41,6 +41,7 @@ class LandingVideoDedupeTest(unittest.TestCase):
                 [{"adgroup_id": 101, "creative_count": 3}],
                 [{"adgroup_id": 201, "creative_count": 3}],
             ]), \
+            patch.object(execute_creation_once, "build_landing_candidate_pool", return_value=object()), \
             patch.object(execute_creation_once, "prepare_one_creative_for_ad", side_effect=fake_prepare), \
             patch.object(execute_creation_once, "record_prepared_material_usage"):
             records = execute_creation_once.phase1_prepare(target_creatives=4)
@@ -52,7 +53,7 @@ class LandingVideoDedupeTest(unittest.TestCase):
     def test_ai_landing_dedupe_is_independent_from_history_and_limits_once_per_ai_pool(self):
         calls = []
 
-        def fake_prepare(account_id, adgroup_id, *, excluded_material_ids, excluded_landing_ids):
+        def fake_prepare(account_id, adgroup_id, *, excluded_material_ids, excluded_landing_ids, **kwargs):
             calls.append(set(excluded_landing_ids))
             return {
                 "account_id": account_id,
@@ -76,14 +77,101 @@ class LandingVideoDedupeTest(unittest.TestCase):
                 [{"adgroup_id": 101, "creative_count": 11}],
                 [{"adgroup_id": 201, "creative_count": 11}],
             ]), \
+            patch.object(execute_creation_once, "build_landing_candidate_pool", return_value=object()), \
             patch.object(execute_creation_once, "prepare_one_creative_for_ad", side_effect=fake_prepare), \
             patch.object(execute_creation_once, "record_prepared_material_usage"):
             records = execute_creation_once.phase1_prepare(target_creatives=12)
 
-        self.assertEqual(2, len(records))
+        self.assertEqual(1, len(records))
         self.assertEqual(set(), calls[0])
         self.assertEqual({777}, calls[1])
 
+    def test_phase1_reuses_landing_candidate_pool_for_multiple_creative_attempts(self):
+        pool = object()
+        build_calls = []
+        prepare_calls = []
+
+        def fake_build_pool(account_id):
+            build_calls.append(account_id)
+            return pool
+
+        def fake_prepare(
+            account_id,
+            adgroup_id,
+            *,
+            excluded_material_ids,
+            excluded_landing_ids,
+            landing_candidates,
+            failed_landing_ids,
+        ):
+            prepare_calls.append((landing_candidates, failed_landing_ids))
+            return {
+                "account_id": account_id,
+                "adgroup_id": adgroup_id,
+                "audience_tier": "R330",
+                "landing_video_id": 1000 + len(prepare_calls),
+                "_material_id": f"ai:{len(prepare_calls)}",
+                "material_source": "ai_generated",
+            }
+
+        with patch.object(execute_creation_once, "get_creation_account_ids", return_value=[1]), \
+            patch.object(execute_creation_once, "load_excluded_ad_ids_from_adjustment", return_value=set()), \
+            patch.object(execute_creation_once, "get_account_crowd_package", return_value="R330"), \
+            patch.object(execute_creation_once, "load_recent_landing_usage_counts", return_value={"history": {}, "ai_generated": {}}), \
+            patch.object(execute_creation_once, "load_account_material_strategy", return_value=_Strategy("ai_generated")), \
+            patch.object(execute_creation_once, "find_ads_needing_creatives", return_value=[
+                {"adgroup_id": 101, "creative_count": 0},
+            ]), \
+            patch.object(execute_creation_once, "build_landing_candidate_pool", side_effect=fake_build_pool), \
+            patch.object(execute_creation_once, "prepare_one_creative_for_ad", side_effect=fake_prepare), \
+            patch.object(execute_creation_once, "record_prepared_material_usage"):
+            records = execute_creation_once.phase1_prepare(target_creatives=2)
+
+        self.assertEqual(2, len(records))
+        self.assertEqual([1], build_calls)
+        self.assertEqual([(pool, set()), (pool, set())], prepare_calls)
+
+    def test_phase1_carries_failed_landing_ids_between_attempts(self):
+        calls = []
+
+        def fake_prepare(
+            account_id,
+            adgroup_id,
+            *,
+            excluded_material_ids,
+            excluded_landing_ids,
+            landing_candidates,
+            failed_landing_ids,
+        ):
+            calls.append(set(failed_landing_ids))
+            if len(calls) == 1:
+                failed_landing_ids.add(777)
+                return None
+            return {
+                "account_id": account_id,
+                "adgroup_id": adgroup_id,
+                "audience_tier": "R330",
+                "landing_video_id": 888,
+                "_material_id": "ai:ok",
+                "material_source": "ai_generated",
+            }
+
+        with patch.object(execute_creation_once, "get_creation_account_ids", return_value=[1]), \
+            patch.object(execute_creation_once, "load_excluded_ad_ids_from_adjustment", return_value=set()), \
+            patch.object(execute_creation_once, "get_account_crowd_package", return_value="R330"), \
+            patch.object(execute_creation_once, "load_recent_landing_usage_counts", return_value={"history": {}, "ai_generated": {}}), \
+            patch.object(execute_creation_once, "load_account_material_strategy", return_value=_Strategy("ai_generated")), \
+            patch.object(execute_creation_once, "find_ads_needing_creatives", return_value=[
+                {"adgroup_id": 101, "creative_count": 0},
+            ]), \
+            patch.object(execute_creation_once, "build_landing_candidate_pool", return_value=object()), \
+            patch.object(execute_creation_once, "prepare_one_creative_for_ad", side_effect=fake_prepare), \
+            patch.object(execute_creation_once, "record_prepared_material_usage"):
+            records = execute_creation_once.phase1_prepare(target_creatives=2)
+
+        self.assertEqual(1, len(records))
+        self.assertEqual([set(), {777}], calls)
+
 
 if __name__ == "__main__":
     unittest.main()

examples/auto_put_ad_mini/tools/creative_creation.py

@@ -23,8 +23,8 @@
 
 import logging
 import random
-from dataclasses import asdict
-from typing import Optional
+from dataclasses import asdict, dataclass
+from typing import Iterator, Optional
 
 from config import (
     CREATIVE_DESCRIPTION_COUNT_PER_AD,
@@ -55,7 +55,7 @@ from tools.video_recall import (
     get_account_crowd_package,
     map_crowd_package_for_video_recall,
 )
-from tools.video_feature_query import fetch_video_element_features
+from tools.video_feature_query import VideoElementFeature, fetch_video_element_features
 from tools.video_risk import VideoRiskResult, check_video_risk
 
 logger = logging.getLogger(__name__)
@@ -90,6 +90,121 @@ def _pick_image_url(material: Material) -> str:
     return images[0] if images else ""
 
 
+@dataclass
+class _LandingSourceState:
+    videos: list[LandingVideo]
+    features_by_vid: dict[int, list[VideoElementFeature]]
+    stats: dict
+
+
+class LandingCandidatePool:
+    """一次运行内复用的承接视频候选池。
+
+    只缓存无外部副作用的步骤:视频拉取、品类过滤、ODPS 特征读取。
+    风险审核按实际遍历到的 landing 懒执行并缓存,避免一次性审核用不到的视频。
+    图片上传、AI 生图、xcx/save 仍由 prepare_one_creative_for_ad 串行执行。
+    """
+
+    def __init__(self, account_id: int, max_landings: int = MAX_LANDING_ATTEMPTS_PER_AD):
+        self.account_id = account_id
+        self.max_landings = max_landings
+        self.crowd_package = get_account_crowd_package(account_id)
+        self.video_crowd_package = map_crowd_package_for_video_recall(self.crowd_package)
+        self._source_state_by_label: dict[str, _LandingSourceState] = {}
+        self._risk_by_vid: dict[int, VideoRiskResult] = {}
+        self._no_features_logged: set[tuple[str, int]] = set()
+
+    @property
+    def source_plan(self) -> list[tuple[str, str]]:
+        plan = [("primary", PIAOQUANTV_VIDEO_SOURCE)]
+        if PIAOQUANTV_HOT_FALLBACK_SOURCE != PIAOQUANTV_VIDEO_SOURCE:
+            plan.append(("hot", PIAOQUANTV_HOT_FALLBACK_SOURCE))
+        return plan
+
+    def iter_featured_landings(self) -> Iterator[tuple[str, LandingVideo, list[VideoElementFeature]]]:
+        for source_label, source in self.source_plan:
+            state = self._load_source(source_label, source)
+            for v in state.videos:
+                element_features = state.features_by_vid.get(v.video_id) or []
+                if not element_features:
+                    if (source_label, v.video_id) not in self._no_features_logged:
+                        state.stats["no_features"] += 1
+                        self._no_features_logged.add((source_label, v.video_id))
+                        logger.info(
+                            "[landing_candidate_pool]   landing=%d 无 ODPS 召回特征,跳过",
+                            v.video_id,
+                        )
+                    continue
+
+                state.stats["feature_candidates_yielded"] += 1
+                yield source_label, v, element_features
+
+    def check_risk(self, source_label: str, landing: LandingVideo) -> VideoRiskResult:
+        risk = self._risk_by_vid.get(landing.video_id)
+        if risk is not None:
+            return risk
+
+        state = self._source_state_by_label[source_label]
+        risk = check_video_risk(landing.video_id)
+        self._risk_by_vid[landing.video_id] = risk
+        state.stats["risk_checked"] += 1
+        if not risk.passed:
+            state.stats["risk_blocked"] += 1
+            state.stats["risk_blocked_by_vid"][landing.video_id] = risk.reason
+            logger.warning(
+                "[landing_candidate_pool]   landing=%d 风险拦截:%s",
+                landing.video_id, risk.reason,
+            )
+        return risk
+
+    def _load_source(self, source_label: str, source: str) -> _LandingSourceState:
+        if source_label in self._source_state_by_label:
+            return self._source_state_by_label[source_label]
+
+        videos = fetch_landing_videos_for_account(
+            self.account_id,
+            page_size=100,
+            source=source,
+            enable_hot_fallback=False,
+        )
+        valid = [v for v in videos if _is_landing_candidate(v)]
+        features_by_vid = fetch_video_element_features(v.video_id for v in valid)
+        stats = {
+            "fetched": len(videos),
+            "valid": len(valid),
+            "category_filtered": max(len(videos) - len(valid), 0),
+            "risk_checked": 0,
+            "risk_blocked": 0,
+            "risk_blocked_by_vid": {},
+            "no_features": 0,
+            "feature_candidates_yielded": 0,
+        }
+        logger.info(
+            "[landing_candidate_pool] account=%d source=%s valid=%d/%d feature_videos=%d feature_rows=%d",
+            self.account_id, source_label, len(valid),
+            len(videos), len(features_by_vid), sum(len(v) for v in features_by_vid.values()),
+        )
+
+        state = _LandingSourceState(
+            videos=valid,
+            features_by_vid=features_by_vid,
+            stats=stats,
+        )
+        self._source_state_by_label[source_label] = state
+        logger.info(
+            "[landing_candidate_pool] account=%d source=%s stats=%s",
+            self.account_id, source_label, stats,
+        )
+        return state
+
+
+def build_landing_candidate_pool(
+    account_id: int,
+    max_landings: int = MAX_LANDING_ATTEMPTS_PER_AD,
+) -> LandingCandidatePool:
+    return LandingCandidatePool(account_id, max_landings=max_landings)
+
+
 def find_ads_needing_creatives(
     account_id: int,
     min_creatives: int = TARGET_CREATIVES_PER_AD,
@@ -522,6 +637,8 @@ def prepare_one_creative_for_ad(
     adgroup_id: int,
     excluded_material_ids: Optional[set] = None,
     excluded_landing_ids: Optional[set] = None,
+    landing_candidates: Optional[LandingCandidatePool] = None,
+    failed_landing_ids: Optional[set[int]] = None,
     max_landings: int = MAX_LANDING_ATTEMPTS_PER_AD,
     max_materials_per_landing: int = MAX_MATERIAL_PER_LANDING,
 ) -> Optional[dict]:
@@ -539,6 +656,8 @@ def prepare_one_creative_for_ad(
                                召回后过滤掉这些,避免同广告挂重复素材被腾讯模型降权曝光。
         excluded_landing_ids: 本轮/近期已使用的 landing_video_id 集合。
                               用于同人群包跨广告、跨轮 landing 排重。
+        landing_candidates: 本轮复用的承接视频候选池。传入后不重复拉视频/查特征/风险。
+        failed_landing_ids: 本广告本轮已尝试失败的 landing_video_id,避免 AI 拒审/无素材后重复尝试。
 
     Returns:
         pending record dict(飞书表格字段 + Phase 3 POST 用的完整 body + 元数据);
@@ -548,9 +667,14 @@ def prepare_one_creative_for_ad(
 
     excluded_material_ids = excluded_material_ids or set()
     excluded_landing_ids = excluded_landing_ids or set()
+    failed_landing_ids = failed_landing_ids if failed_landing_ids is not None else set()
     crowd_package = get_account_crowd_package(account_id)
     material_strategy = load_account_material_strategy(account_id)
     video_crowd_package = map_crowd_package_for_video_recall(crowd_package)
+    landing_candidates = landing_candidates or build_landing_candidate_pool(
+        account_id,
+        max_landings=max_landings,
+    )
     if material_strategy.use_ai_generated and not material_strategy.ai_fallback_to_history:
         recent_material_ids = set()
         logger.info(
@@ -588,177 +712,146 @@ def prepare_one_creative_for_ad(
     chosen_material_source = material_strategy.material_source
     chosen_ai_generated_material_id = None
 
-    # 每个来源最多拉 100 条视频。先主池,主池产不出可用创意时再用同人群包 hot 兜底。
-    source_plan = [("primary", PIAOQUANTV_VIDEO_SOURCE)]
-    if PIAOQUANTV_HOT_FALLBACK_SOURCE != PIAOQUANTV_VIDEO_SOURCE:
-        source_plan.append(("hot", PIAOQUANTV_HOT_FALLBACK_SOURCE))
+    source_stats_by_label: dict[str, dict] = {}
 
-    for source_label, source in source_plan:
-        videos = fetch_landing_videos_for_account(
-            account_id,
-            page_size=100,
-            source=source,
-            enable_hot_fallback=False,
-        )
-        valid = [v for v in videos if _is_landing_candidate(v)]
-        features_by_vid = fetch_video_element_features(v.video_id for v in valid)
-        source_stats = {
-            "fetched": len(videos),
-            "valid": len(valid),
-            "category_filtered": max(len(videos) - len(valid), 0),
+    for source_label, v, element_features in landing_candidates.iter_featured_landings():
+        source_stats = source_stats_by_label.setdefault(source_label, {
+            "pool_candidates": 0,
             "landing_dedupe": 0,
-            "risk_blocked": 0,
-            "no_features": 0,
+            "failed_landing_dedupe": 0,
             "ai_attempts": 0,
             "ai_failed": 0,
             "ai_empty": 0,
             "history_recall_empty": 0,
             "all_excluded": 0,
-            "max_landing_limit": 0,
             "selected": 0,
-        }
-        logger.info(
-            "[prepare_one_creative] account=%d adgroup=%d source=%s material_source=%s "
-            "valid landing=%d/100 feature_videos=%d feature_rows=%d excl_mat=%d",
-            account_id, adgroup_id, source_label, material_strategy.material_source,
-            len(valid),
-            len(features_by_vid), sum(len(v) for v in features_by_vid.values()),
-            len(effective_excluded_material_ids),
-        )
+        })
+        source_stats["pool_candidates"] += 1
 
-        attempts = 0
-        for v in valid:
-            if v.video_id in excluded_landing_ids:
-                source_stats["landing_dedupe"] += 1
-                logger.info(
-                    "[prepare_one_creative]   landing=%d landing 排重命中,跳过",
-                    v.video_id,
-                )
-                continue
-            attempts += 1
-            if attempts > max_landings:
-                source_stats["max_landing_limit"] += 1
-                break
-
-            # 2026-06-29:承接视频风险审核。必须在素材召回 / xcx-save 前完成,
-            # 避免高风险 landing 继续产生 plan/rootSourceId 等外部副作用。
-            risk = check_video_risk(v.video_id)
-            if not risk.passed:
-                source_stats["risk_blocked"] += 1
-                logger.warning(
-                    "[prepare_one_creative]   landing=%d 风险拦截:%s",
-                    v.video_id, risk.reason,
-                )
-                continue
+        if v.video_id in excluded_landing_ids:
+            source_stats["landing_dedupe"] += 1
+            logger.info(
+                "[prepare_one_creative]   landing=%d landing 排重命中,跳过",
+                v.video_id,
+            )
+            continue
+        if v.video_id in failed_landing_ids:
+            source_stats["failed_landing_dedupe"] += 1
+            logger.info(
+                "[prepare_one_creative]   landing=%d 本广告本轮已失败,跳过",
+                v.video_id,
+            )
+            continue
 
-            element_features = features_by_vid.get(v.video_id) or []
-            if not element_features:
-                source_stats["no_features"] += 1
+        risk = landing_candidates.check_risk(source_label, v)
+        if not risk.passed:
+            continue
+        source_stats = source_stats_by_label[source_label]
+        materials = []
+        candidate_material_source = material_strategy.material_source
+        if material_strategy.use_ai_generated:
+            source_stats["ai_attempts"] += 1
+            try:
+                assets = get_or_generate_assets_for_landing(
+                    account_id=account_id,
+                    adgroup_id=adgroup_id,
+                    crowd_package=crowd_package,
+                    landing=v,
+                )
+                materials = [asset.to_material() for asset in assets]
                 logger.info(
-                    "[prepare_one_creative]   landing=%d 无 ODPS 召回特征,跳过",
-                    v.video_id,
+                    "[prepare_one_creative]   landing=%d AI生成素材候选=%d fallback_history=%s",
+                    v.video_id, len(materials), material_strategy.ai_fallback_to_history,
                 )
-                continue
+            except Exception as e:
+                source_stats["ai_failed"] += 1
+                failed_landing_ids.add(v.video_id)
+                logger.exception(
+                    "[prepare_one_creative]   landing=%d AI生成素材失败:%s",
+                    v.video_id, e,
+                )
+                if not material_strategy.ai_fallback_to_history:
+                    continue
+
+        if (
+            material_strategy.use_ai_generated
+            and not materials
+            and not material_strategy.ai_fallback_to_history
+        ):
+            source_stats["ai_empty"] += 1
+            failed_landing_ids.add(v.video_id)
+            logger.info(
+                "[prepare_one_creative]   landing=%d AI无可用素材且不允许回退历史素材,跳过",
+                v.video_id,
+            )
+            continue
 
-            materials = []
-            candidate_material_source = material_strategy.material_source
+        if (not materials) and (
+            not material_strategy.use_ai_generated
+            or material_strategy.ai_fallback_to_history
+        ):
             if material_strategy.use_ai_generated:
-                source_stats["ai_attempts"] += 1
-                try:
-                    assets = get_or_generate_assets_for_landing(
-                        account_id=account_id,
-                        adgroup_id=adgroup_id,
-                        crowd_package=crowd_package,
-                        landing=v,
-                    )
-                    materials = [asset.to_material() for asset in assets]
-                    logger.info(
-                        "[prepare_one_creative]   landing=%d AI生成素材候选=%d fallback_history=%s",
-                        v.video_id, len(materials), material_strategy.ai_fallback_to_history,
-                    )
-                except Exception as e:
-                    source_stats["ai_failed"] += 1
-                    logger.exception(
-                        "[prepare_one_creative]   landing=%d AI生成素材失败:%s",
-                        v.video_id, e,
-                    )
-                    if not material_strategy.ai_fallback_to_history:
-                        continue
-
-            if (
-                material_strategy.use_ai_generated
-                and not materials
-                and not material_strategy.ai_fallback_to_history
-            ):
-                source_stats["ai_empty"] += 1
                 logger.info(
-                    "[prepare_one_creative]   landing=%d AI无可用素材且不允许回退历史素材,跳过",
+                    "[prepare_one_creative]   landing=%d AI无可用素材,按账户配置回退历史素材",
                     v.video_id,
                 )
-                continue
-
-            if (not materials) and (
-                not material_strategy.use_ai_generated
-                or material_strategy.ai_fallback_to_history
-            ):
-                if material_strategy.use_ai_generated:
-                    logger.info(
-                        "[prepare_one_creative]   landing=%d AI无可用素材,按账户配置回退历史素材",
-                        v.video_id,
-                    )
-                    candidate_material_source = "history_fallback"
-                materials = recall_materials_for_video(
-                    v,
-                    final_top_n=max_materials_per_landing,
-                    element_features=element_features,
-                )
-                if not materials:
-                    source_stats["history_recall_empty"] += 1
-            # material_id 去重(2026-06-09):跳过已用素材(账户层 set,跨广告也共享)
-            fresh = [
-                m for m in materials
-                if m.material_id not in effective_excluded_material_ids
-            ]
-            if fresh:
-                chosen_landing = v
-                chosen_material = fresh[0]
-                chosen_risk = risk
-                chosen_landing_source = source_label
-                chosen_material_source = candidate_material_source
-                if chosen_material.material_id.startswith("ai:"):
-                    chosen_ai_generated_material_id = chosen_material.raw.get("ai_generated_material_id")
-                source_stats["selected"] += 1
-                logger.info(
-                    "[prepare_one_creative]   选中 landing=%d source=%s category=%r material_source=%s material=%s recall=%s/%s/%s cost=%s roi=%s ctr=%s imp=%s score=%s policy=%s",
-                    v.video_id, source_label, v.category,
-                    chosen_material_source,
-                    chosen_material.material_id,
-                    chosen_material.recall_element_dimension,
-                    chosen_material.recall_point_type,
-                    chosen_material.recall_standard_element,
-                    chosen_material.cost,
-                    chosen_material.roi,
-                    chosen_material.ctr,
-                    chosen_material.impressions,
-                    chosen_material.score,
-                    "ai_generated" if chosen_material.material_id.startswith("ai:") else "score>=0.8,cost_desc",
-                )
-                break
-            if materials:
-                source_stats["all_excluded"] += 1
-                logger.info(
-                    "[prepare_one_creative]   landing=%d 召回 %d 全在 excluded,试下一条",
-                    v.video_id, len(materials),
-                )
-        if chosen_landing and chosen_material:
+                candidate_material_source = "history_fallback"
+            materials = recall_materials_for_video(
+                v,
+                final_top_n=max_materials_per_landing,
+                element_features=element_features,
+            )
+            if not materials:
+                source_stats["history_recall_empty"] += 1
+                failed_landing_ids.add(v.video_id)
+        # material_id 去重(2026-06-09):跳过已用素材(账户层 set,跨广告也共享)
+        fresh = [
+            m for m in materials
+            if m.material_id not in effective_excluded_material_ids
+        ]
+        if fresh:
+            chosen_landing = v
+            chosen_material = fresh[0]
+            chosen_risk = risk
+            chosen_landing_source = source_label
+            chosen_material_source = candidate_material_source
+            if chosen_material.material_id.startswith("ai:"):
+                chosen_ai_generated_material_id = chosen_material.raw.get("ai_generated_material_id")
+            source_stats["selected"] += 1
             logger.info(
-                "[prepare_one_creative] source_summary account=%d adgroup=%d source=%s stats=%s",
-                account_id, adgroup_id, source_label, source_stats,
+                "[prepare_one_creative]   选中 landing=%d source=%s category=%r material_source=%s material=%s recall=%s/%s/%s cost=%s roi=%s ctr=%s imp=%s score=%s policy=%s",
+                v.video_id, source_label, v.category,
+                chosen_material_source,
+                chosen_material.material_id,
+                chosen_material.recall_element_dimension,
+                chosen_material.recall_point_type,
+                chosen_material.recall_standard_element,
+                chosen_material.cost,
+                chosen_material.roi,
+                chosen_material.ctr,
+                chosen_material.impressions,
+                chosen_material.score,
+                "ai_generated" if chosen_material.material_id.startswith("ai:") else "score>=0.8,cost_desc",
             )
             break
+        if materials:
+            source_stats["all_excluded"] += 1
+            failed_landing_ids.add(v.video_id)
+            logger.info(
+                "[prepare_one_creative]   landing=%d 召回 %d 全在 excluded,试下一条",
+                v.video_id, len(materials),
+            )
+
+    if chosen_landing and chosen_material:
+        logger.info(
+            "[prepare_one_creative] source_summary account=%d adgroup=%d source=%s stats=%s",
+            account_id, adgroup_id, chosen_landing_source,
+            source_stats_by_label.get(chosen_landing_source, {}),
+        )
+    else:
         logger.info(
-            "[prepare_one_creative] account=%d adgroup=%d source=%s 未产出可用创意 stats=%s",
-            account_id, adgroup_id, source_label, source_stats,
+            "[prepare_one_creative] account=%d adgroup=%d 未产出可用创意 stats=%s",
+            account_id, adgroup_id, source_stats_by_label,
         )
 
     if not chosen_landing or not chosen_material:
@@ -776,6 +869,7 @@ def prepare_one_creative_for_ad(
             "[prepare_one_creative] material=%s 既无 cover 也无 imageList,放弃",
             chosen_material.material_id,
         )
+        failed_landing_ids.add(chosen_landing.video_id)
         return None
 
     is_ai_generated_material = chosen_material.material_id.startswith("ai:")