Agent/examples/auto_put_ad_mini/tools/creative_creation.py

"""创意搭建主入口(模块 B)。

数据流(2026-06-08 用户确认 + 端到端打通):
  承接视频(piaoquantv) → 多路召回素材(vector) → top 1 → 上传素材图(MD5 幂等)
  → 调 xcx/save 注册落地计划(拿 page_url + creative_name)→ 读账户 brand
  → 构造 body → POST /dynamic_creatives/add → 绑到广告

决策落地(已验证):
- image 组件:**必须用 image_id**(image_url 会 reject code=18001)
  → 实现:下载 cover URL → MD5 → POST /v3.0/images/add multipart → 拿 image_id
  → MD5 幂等:同账户重复上传返回同 ID(无需本地缓存)
- brand 组件:**必填**(漏传会 reject code=1800269)
  → brand_image_id 跨账户不可复用(reject code=1530003)
  → 优先从 DB account_whitelist 读账户级 brand_name/brand_image_id
  → 若缺失,按 ad_creation_account_config.brand_key 读取 brand_asset_template,上传并写回账户
- dynamic_creative_type: DYNAMIC_CREATIVE_TYPE_PROGRAM(参考样本 9744753978)
- description: 素材 title 一条(MVP)
- jump_info / 命名:**piaoquantv xcx/save 接口统一管**
  → page_url(mini_program_path)和 root_source_id(creative_name)都来自服务侧
  → 客户端不本地拼接、不本地命名,保证归因锚点跨系统一致
- 配置 NORMAL 状态,挂上后腾讯进入 PENDING(普通素材审核 2-4 小时)
"""

import logging
import random
from dataclasses import asdict, dataclass
from typing import Iterator, Optional

from config import (
    CREATIVE_DESCRIPTION_COUNT_PER_AD,
    CREATIVE_DESCRIPTION_POOL,
    LANDING_EXCLUDED_CATEGORIES,
    MARKETING_CARRIER_GH_ID,
    MAX_LANDING_ATTEMPTS_PER_AD,
    MAX_MATERIAL_PER_LANDING,
    TARGET_CREATIVES_PER_AD,
)
from tools.ad_api import _check, _get, _post, images_add
from tools.account_material_strategy import load_account_material_strategy
from tools.ai_generated_material import (
    get_or_generate_assets_for_landing,
    update_generated_material_status,
)
from tools.external_recalled_material import (
    prepare_external_material_for_landing,
    update_external_material_status,
)
from tools.landing_plan import LandingPlanResult, create_landing_plan
from tools.material_recall import Material, recall_materials_for_video
from tools.creative_material_usage import (
    load_recent_used_material_ids,
    merge_used_material_ids,
)
from tools.video_recall import (
    LandingVideo,
    PIAOQUANTV_HOT_FALLBACK_SOURCE,
    PIAOQUANTV_VIDEO_SOURCE,
    fetch_landing_videos_for_account,
    get_account_crowd_package,
    map_crowd_package_for_video_recall,
)
from tools.video_feature_query import VideoElementFeature, fetch_video_element_features
from tools.video_risk import VideoRiskResult, check_video_risk

logger = logging.getLogger(__name__)


def _landing_category_values(v: LandingVideo) -> set[str]:
    """解析承接视频的内容品类，返回去重后的品类标签集合。"""
    raw = v.category or ""
    return {part.strip() for part in raw.replace("，", ",").split(",") if part.strip()}


def _is_landing_candidate(v: LandingVideo) -> bool:
    """承接视频基础预筛:内容品类不在黑名单。

    素材召回特征统一通过 video_id 查 ODPS,不再依赖 videoContentList 返回的
    pointType / standardElement。
    """
    excluded = _landing_category_values(v) & LANDING_EXCLUDED_CATEGORIES
    if excluded:
        logger.info(
            "[creative_creation] landing=%d category=%r 命中排除品类,跳过",
            v.video_id, v.category,
        )
        return False
    return True


def _pick_image_url(material: Material) -> str:
    """从素材里取 image URL — material.cover 优先,fallback 到 raw.imageList[0]"""
    if material.cover:
        return material.cover
    images = (material.raw or {}).get("imageList") or []
    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,
) -> list[dict]:
    """扫描账户下"需要补创意"的广告(P0-B,2026-06-09)。

    口径(2026-06-08 用户确认 = C):
      configured_status = AD_STATUS_NORMAL  AND  creative_count < min_creatives

    实测发现:腾讯 /adgroups/get 不返回 creative_count 字段,
    所以分两条路径取创意数:
      - 快路径:system_status = ADGROUP_STATUS_CREATIVE_EMPTY → creative_count = 0
      - 慢路径:其他状态 → 单独调 /dynamic_creatives/get 数总数(N+1,目前广告 ≤ 10 可接受)

    Args:
        account_id: 腾讯广告主账号 ID
        min_creatives: 默认读 config.TARGET_CREATIVES_PER_AD(测试=1,生产应回 15)
                       — find 阈值 + 补量目标的同一变量,无矛盾

    Returns:
        [{adgroup_id, adgroup_name, system_status, creative_count}, ...]
    """
    # 注意:腾讯 filtering 对 configured_status IN 静默拒绝(实测 2026-06-09 加了 filtering 返回 0)
    # 改为不加 filtering 拉全部 + Python 内存过滤(单账户 ≤ 10 条广告,代价≈0)
    resp = _get("/adgroups/get", {
        "account_id": account_id, "page": 1, "page_size": 100,
        "fields": ["adgroup_id", "adgroup_name", "system_status", "configured_status"],
    })
    all_ads_raw = (resp.get("data") or {}).get("list") or []
    all_ads = [a for a in all_ads_raw if a.get("configured_status") == "AD_STATUS_NORMAL"]
    logger.info(
        "[find_ads_needing_creatives] account=%d 共 %d 条广告,%d 条 NORMAL",
        account_id, len(all_ads_raw), len(all_ads),
    )

    out = []
    for ad in all_ads:
        adgroup_id = ad["adgroup_id"]
        system_status = ad.get("system_status", "")
        adgroup_name = ad.get("adgroup_name", "")

        if system_status == "ADGROUP_STATUS_CREATIVE_EMPTY":
            creative_count = 0
        else:
            cresp = _get("/dynamic_creatives/get", {
                "account_id": account_id, "page": 1, "page_size": 100,
                "filtering": [{
                    "field": "adgroup_id", "operator": "IN",
                    "values": [str(adgroup_id)],
                }],
                "fields": ["dynamic_creative_id", "system_status"],
            })
            all_creatives = (cresp.get("data") or {}).get("list") or []
            # 2026-06-09:忽略 DENIED 状态(审核拒绝),让 find_ads 能重补
            # 其他状态(PENDING/ACTIVE/SUSPEND 等)都算"已挂"
            denied_count = sum(
                1 for c in all_creatives
                if c.get("system_status") == "DYNAMIC_CREATIVE_STATUS_DENIED"
            )
            creative_count = len(all_creatives) - denied_count
            if denied_count:
                logger.info(
                    "[find_ads_needing_creatives]   adgroup=%d 总创意 %d 减去 %d 条 DENIED → 有效 %d",
                    adgroup_id, len(all_creatives), denied_count, creative_count,
                )

        if creative_count < min_creatives:
            out.append({
                "adgroup_id": adgroup_id,
                "adgroup_name": adgroup_name,
                "system_status": system_status,
                "creative_count": creative_count,
            })
            logger.info(
                "[find_ads_needing_creatives]   ✓ adgroup=%d name=%r creative_count=%d < %d",
                adgroup_id, adgroup_name, creative_count, min_creatives,
            )

    logger.info(
        "[find_ads_needing_creatives] account=%d 命中 %d 条需补创意",
        account_id, len(out),
    )
    return out


def find_existing_creative_by_image(
    account_id: int, adgroup_id: int, material_image_id: str,
) -> Optional[int]:
    """查 adgroup 下是否已有创意挂了这个素材 image_id(幂等检查)。

    判等键(2026-06-08 决策):**同 adgroup + 同 material_image_id**。
    - 不按 creative_name 判等:name = root_source_id,每次 xcx/save 都新,等于没做
    - 按 image_id 判等:同 image 在同 adgroup 下挂多条会被腾讯模型降权曝光

    Returns:
        已有创意的 dynamic_creative_id;无则 None。
    """
    resp = _get("/dynamic_creatives/get", {
        "account_id": account_id, "page": 1, "page_size": 100,
        "filtering": [{
            "field": "adgroup_id", "operator": "IN", "values": [str(adgroup_id)],
        }],
        "fields": ["dynamic_creative_id", "creative_components"],
    })
    items = (resp.get("data") or {}).get("list") or []
    for c in items:
        images = (c.get("creative_components") or {}).get("image") or []
        for img in images:
            existing_id = ((img.get("value") or {}).get("image_id")) or ""
            if str(existing_id) == str(material_image_id):
                cid = int(c.get("dynamic_creative_id"))
                logger.info(
                    "[idempotency] adgroup=%d image_id=%s 已挂创意 creative_id=%d,skip 新建",
                    adgroup_id, material_image_id, cid,
                )
                return cid
    return None


def get_account_brand(account_id: int) -> dict:
    """读取或初始化账户级 brand 资产。

    返回 {"brand_name": str, "brand_image_id": str}。
    跨账户 brand_image_id 不可复用,所以一定按 account_id 取。
    如果 account_whitelist 还没有 brand_image_id,则按待投放配置的 brand_key
    找统一品牌模板,上传模板 URL 到当前账户,并把新 image_id 写回 DB。
    """
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT brand_name, brand_image_id FROM account_whitelist WHERE account_id=%s",
                (account_id,),
            )
            row = cur.fetchone()
            if row and row.get("brand_name") and row.get("brand_image_id"):
                return {
                    "brand_name": row["brand_name"],
                    "brand_image_id": str(row["brand_image_id"]),
                }

            cur.execute(
                """
                SELECT b.brand_key, b.brand_name, b.brand_image_url
                FROM ad_creation_account_config c
                JOIN brand_asset_template b
                  ON b.brand_key = c.brand_key
                 AND b.enabled = TRUE
                WHERE c.account_id=%s
                  AND c.enabled = TRUE
                """,
                (account_id,),
            )
            tpl = cur.fetchone()
    finally:
        conn.close()

    if not tpl or not tpl.get("brand_name") or not tpl.get("brand_image_url"):
        raise RuntimeError(
            f"account {account_id} 未配置可用品牌资产模板。"
            "请在 ad_creation_account_config.brand_key 关联 brand_asset_template"
        )

    brand_name = tpl["brand_name"]
    brand_image_url = tpl["brand_image_url"]
    logger.info(
        "[brand] account=%d 缺 brand_image_id,按模板 %s 上传品牌图",
        account_id, tpl.get("brand_key"),
    )
    brand_image_id = images_add(account_id, brand_image_url)

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                UPDATE account_whitelist
                SET brand_name=%s,
                    brand_image_id=%s,
                    updated_by=%s,
                    updated_at=CURRENT_TIMESTAMP
                WHERE account_id=%s
                """,
                (brand_name, brand_image_id, "auto-brand-init", account_id),
            )
        conn.commit()
    finally:
        conn.close()

    logger.info(
        "[brand] account=%d 已初始化 brand_name=%s brand_image_id=%s",
        account_id, brand_name, brand_image_id,
    )
    return {
        "brand_name": brand_name,
        "brand_image_id": str(brand_image_id),
    }


def build_creative_request_body(
    account_id: int,
    adgroup_id: int,
    landing: LandingVideo,
    material: Material,
    material_image_id: str,
    brand_name: str,
    brand_image_id: str,
    creative_name: str,
    jump_path: str,
    description_contents: Optional[list] = None,
) -> dict:
    """生成 /v3.0/dynamic_creatives/add 的请求 body(纯函数,无 I/O)。

    Args:
        material_image_id: 已上传到当前账户的素材图 image_id
        brand_name / brand_image_id: 账户级 brand 资产(已上传)
        creative_name: 由 xcx/save 返回的 root_source_id(归因锚点)
        jump_path: 由 xcx/save 返回的 page_url(小程序跳转路径)
        description_contents: 文案列表(可选);None 时自动 random.sample(用于审批表回显)
    """
    # 默认文案由配置控制;生产当前统一使用"打开看看"。
    if description_contents is None:
        description_count = min(CREATIVE_DESCRIPTION_COUNT_PER_AD, len(CREATIVE_DESCRIPTION_POOL))
        description_contents = CREATIVE_DESCRIPTION_POOL[:description_count]
    jump_spec = {
        "page_type": "PAGE_TYPE_WECHAT_MINI_PROGRAM",
        "page_spec": {
            "wechat_mini_program_spec": {
                "mini_program_id": MARKETING_CARRIER_GH_ID,
                "mini_program_path": jump_path,
            }
        },
    }

    return {
        "account_id": account_id,
        "adgroup_id": adgroup_id,
        "dynamic_creative_name": creative_name,
        "delivery_mode": "DELIVERY_MODE_COMPONENT",
        "dynamic_creative_type": "DYNAMIC_CREATIVE_TYPE_PROGRAM",
        "configured_status": "AD_STATUS_NORMAL",
        "creative_components": {
            "description": [{"value": {"content": d}} for d in description_contents],
            "image": [{"value": {"image_id": material_image_id}}],
            "brand": [{"value": {
                "brand_name": brand_name,
                "brand_image_id": brand_image_id,
            }}],
            "action_button": [{"value": {"button_text": "查看详情"}}],
            "jump_info": [{"value": jump_spec}],
            "main_jump_info": [{"value": jump_spec}],
        },
    }


def _pick_landing_and_materials(account_id: int) -> tuple[LandingVideo, list[Material]]:
    """召回链:承接视频 → 多路素材召回 → 取第一个有素材命中的 landing 及其 top 素材。

    出口:确保 landing.point_type+standard_element 都有值(否则多路召回会全 skip)。
    """
    videos = fetch_landing_videos_for_account(account_id, page_size=50)
    valid = [v for v in videos if _is_landing_candidate(v)]
    if not valid:
        raise RuntimeError(
            f"account={account_id} 无 pointType+standardElement 都有值的承接视频"
        )

    for v in valid:
        risk = check_video_risk(v.video_id)
        if not risk.passed:
            logger.warning(
                "[creative_creation] landing=%d 风险拦截:%s",
                v.video_id, risk.reason,
            )
            continue
        materials = recall_materials_for_video(v, final_top_n=5)
        if materials:
            return v, materials

    raise RuntimeError(
        f"account={account_id} 前 {len(valid)} 条承接视频都召回 0 素材"
    )


def preview_for_account(account_id: int, adgroup_id: int) -> dict:
    """承接视频 → 召回素材 → top 1 → 上传素材图 → 注册落地计划 → 读 brand → 构造 body。

    **会真实调:**
    - 腾讯 /images/add(MD5 幂等)
    - piaoquantv xcx/save(每次新建一条计划)

    body 是真实可投放的(只差 POST /dynamic_creatives/add)。
    """
    landing, materials = _pick_landing_and_materials(account_id)
    top_material = materials[0]

    image_url = _pick_image_url(top_material)
    if not image_url:
        raise ValueError(
            f"素材 {top_material.material_id} 既无 cover 也无 imageList,无法构造 image 组件"
        )
    material_image_id = images_add(account_id, image_url)

    crowd_package = get_account_crowd_package(account_id)
    plan = create_landing_plan(crowd_package, landing)

    brand = get_account_brand(account_id)

    body = build_creative_request_body(
        account_id=account_id, adgroup_id=adgroup_id,
        landing=landing, material=top_material,
        material_image_id=material_image_id,
        brand_name=brand["brand_name"],
        brand_image_id=brand["brand_image_id"],
        creative_name=plan.root_source_id,
        jump_path=plan.page_url,
    )
    return {
        "landing": asdict(landing),
        "material": asdict(top_material),
        "material_image_id": material_image_id,
        "brand": brand,
        "landing_plan": asdict(plan),
        "body": body,
    }


def create_creative_for_ad(
    account_id: int, adgroup_id: int,
    landing: LandingVideo, material: Material,
    skip_if_exists: bool = True,
) -> int:
    """编排:上传素材图 → 幂等检查 → 注册落地计划 → 读 brand → build body → POST。

    Args:
        skip_if_exists: True(默认) — 同 adgroup + 同 material_image_id 已有创意时直接返回已有 ID,
                        跳过 xcx/save 注册和腾讯 POST,节省审核额度 + 避免模型降权。
                        False — 强制新建(用于 A/B 测试不同归因锚点的场景)。

    Returns:
        dynamic_creative_id(新建或已有)
    """
    image_url = _pick_image_url(material)
    if not image_url:
        raise ValueError(
            f"素材 {material.material_id} 既无 cover 也无 imageList,无法构造 image 组件"
        )
    material_image_id = images_add(account_id, image_url)

    if skip_if_exists:
        existing_cid = find_existing_creative_by_image(account_id, adgroup_id, material_image_id)
        if existing_cid:
            return existing_cid

    crowd_package = get_account_crowd_package(account_id)
    plan = create_landing_plan(crowd_package, landing)

    brand = get_account_brand(account_id)

    body = build_creative_request_body(
        account_id, adgroup_id, landing, material,
        material_image_id=material_image_id,
        brand_name=brand["brand_name"],
        brand_image_id=brand["brand_image_id"],
        creative_name=plan.root_source_id,
        jump_path=plan.page_url,
    )
    logger.info(
        "[creative_creation] POST /dynamic_creatives/add adgroup=%d name=%s image_id=%s plan_id=%d",
        adgroup_id, body["dynamic_creative_name"], material_image_id, plan.plan_id,
    )
    resp = _post("/dynamic_creatives/add", body)
    data = _check(resp, "creative_create")
    return data.get("dynamic_creative_id")


def prepare_one_creative_for_ad(
    account_id: int,
    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]:
    """Phase 1 准备:召回 + 上传图 + xcx/save + build body → 返回 pending record。

    **不 POST 腾讯**。POST 行为留给 Phase 3(供"先审后挂"流程审批后才挂)。

    跟 try_create_one_creative_with_fallback 的差异:
    - 不做 POST-based fallback(因为不 POST)
    - 信任黑名单 + 选 top 1 material(`EXCLUDED_COVER_URL_PATTERNS` 已经截掉尺寸不符的)
    - 如果 Phase 3 POST 失败(腾讯尺寸 reject 等),由 Phase 3 记 error,下轮主循环重试

    Args:
        excluded_material_ids: 同广告内已挂的 material_id 集合(2026-06-09 N=3 时去重必需)。
                               召回后过滤掉这些,避免同广告挂重复素材被腾讯模型降权曝光。
        excluded_landing_ids: 本轮/近期已使用的 landing_video_id 集合。
                              用于同人群包跨广告、跨轮 landing 排重。
        landing_candidates: 本轮复用的承接视频候选池。传入后不重复拉视频/查特征/风险。
        failed_landing_ids: 本广告本轮已尝试失败的 landing_video_id,避免 AI 拒审/无素材后重复尝试。

    Returns:
        pending record dict(飞书表格字段 + Phase 3 POST 用的完整 body + 元数据);
        所有 landing 都召回 0 素材时返回 None。
    """
    from datetime import datetime, timezone

    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(
            "[prepare_one_creative] account=%d adgroup=%d material_source=%s fallback_history=%s; "
            "AI素材不回退历史,跳过历史素材排重库",
            account_id, adgroup_id, material_strategy.material_source,
            material_strategy.ai_fallback_to_history,
        )
    else:
        try:
            recent_material_ids = load_recent_used_material_ids(crowd_package)
        except Exception as e:
            logger.warning(
                "[prepare_one_creative] account=%d crowd=%r 读取素材使用历史失败,仅使用本轮排重:%s",
                account_id, crowd_package, e,
            )
            recent_material_ids = set()
    effective_excluded_material_ids = merge_used_material_ids(
        excluded_material_ids,
        recent_material_ids,
    )
    logger.info(
        "[prepare_one_creative] account=%d adgroup=%d crowd=%r video_crowd=%r "
        "material_source=%s fallback_history=%s material_dedupe run=%d recent=%d effective=%d",
        account_id, adgroup_id, crowd_package, video_crowd_package,
        material_strategy.material_source, material_strategy.ai_fallback_to_history,
        len(excluded_material_ids), len(recent_material_ids),
        len(effective_excluded_material_ids),
    )

    chosen_landing = None
    chosen_material = None
    chosen_risk: Optional[VideoRiskResult] = None
    chosen_landing_source = ""
    chosen_material_source = material_strategy.material_source
    chosen_ai_generated_material_id = None
    chosen_external_recalled_material_id = None

    source_stats_by_label: dict[str, dict] = {}

    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,
            "failed_landing_dedupe": 0,
            "ai_attempts": 0,
            "ai_failed": 0,
            "ai_empty": 0,
            "external_attempts": 0,
            "external_failed": 0,
            "external_empty": 0,
            "history_recall_empty": 0,
            "all_excluded": 0,
            "selected": 0,
        })
        source_stats["pool_candidates"] += 1

        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

        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_external_recall:
            source_stats["external_attempts"] += 1
            try:
                processed = prepare_external_material_for_landing(
                    account_id=account_id,
                    adgroup_id=adgroup_id,
                    crowd_package=crowd_package,
                    landing=v,
                    element_features=element_features,
                    excluded_material_ids=effective_excluded_material_ids,
                )
                materials = [processed] if processed is not None else []
            except Exception as e:
                source_stats["external_failed"] += 1
                logger.exception(
                    "[prepare_one_creative]   landing=%d 外部素材处理失败:%s",
                    v.video_id, e,
                )
            if not materials:
                source_stats["external_empty"] += 1
                failed_landing_ids.add(v.video_id)
                continue
        elif 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
                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

        if (not materials) and (
            (
                not material_strategy.use_ai_generated
                and not material_strategy.use_external_recall
            )
            or (
                material_strategy.use_ai_generated
                and 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
                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")
            if chosen_material.material_id.startswith("external:"):
                chosen_external_recalled_material_id = chosen_material.raw.get(
                    "external_recalled_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,visit_uv_window_desc,ai_cleanup"
                    if chosen_material.material_id.startswith("external:")
                    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 未产出可用创意 stats=%s",
            account_id, adgroup_id, source_stats_by_label,
        )

    if not chosen_landing or not chosen_material:
        logger.error(
            "[prepare_one_creative] account=%d adgroup=%d material_source=%s fallback_history=%s "
            "穷尽 landing 后无可用素材(excluded=%d)",
            account_id, adgroup_id, material_strategy.material_source,
            material_strategy.ai_fallback_to_history, len(effective_excluded_material_ids),
        )
        return None

    image_url = _pick_image_url(chosen_material)
    if not image_url:
        logger.error(
            "[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:")
    is_external_recalled_material = chosen_material.material_id.startswith("external:")
    is_deferred_generated_material = (
        is_ai_generated_material or is_external_recalled_material
    )
    if is_deferred_generated_material:
        # AI生成图与外部素材派生图先走人工审批,Phase 3 才上传腾讯。
        material_image_id = ""
    else:
        # 历史素材保持原有行为:Phase 1 上传腾讯图片(MD5 幂等)。
        material_image_id = images_add(account_id, image_url)

    # 注册落地计划(xcx/save)
    plan = create_landing_plan(crowd_package, chosen_landing)

    # 读账户 brand
    brand = get_account_brand(account_id)

    # 先确定文案,再传给 build_body — 这样 record 跟 body 文案一致。
    desc_count = min(CREATIVE_DESCRIPTION_COUNT_PER_AD, len(CREATIVE_DESCRIPTION_POOL))
    description_contents = CREATIVE_DESCRIPTION_POOL[:desc_count]

    # 构造完整 POST body(Phase 3 直接用)
    body = build_creative_request_body(
        account_id, adgroup_id, chosen_landing, chosen_material,
        material_image_id=material_image_id,
        brand_name=brand["brand_name"],
        brand_image_id=brand["brand_image_id"],
        creative_name=plan.root_source_id,
        jump_path=plan.page_url,
        description_contents=description_contents,
    )

    # 反查广告 metadata(飞书表格 B 组用)
    ad_info = _fetch_ad_metadata_for_approval(account_id, adgroup_id)

    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    rec = {
        # === 飞书表格字段(im_approval_creation 用)===
        "approval_date": today,
        "account_id": account_id,
        "audience_tier": crowd_package,
        "video_crowd_package": video_crowd_package,
        "adgroup_id": adgroup_id,
        "adgroup_name": ad_info.get("adgroup_name", ""),
        "bid_amount_yuan": ad_info.get("bid_amount_yuan", ""),
        "site_set": ad_info.get("site_set", ""),
        "age_range": ad_info.get("age_range", ""),
        "landing_video_id": chosen_landing.video_id,
        "landing_video_url": chosen_landing.video_url,
        "landing_title": chosen_landing.title,
        "landing_source": chosen_landing_source,
        "landing_category": chosen_landing.category,
        "material_source": chosen_material_source,
        "material_cover_url": chosen_material.cover,
        # 素材质量字段(2026-06-10 batchByText 升级 — 给 Task 25 飞书表展示用)
        "material_ctr": chosen_material.ctr,
        "material_cost": chosen_material.cost,
        "material_roi": chosen_material.roi,
        "material_impressions": chosen_material.impressions,
        "material_visit_uv_30d": chosen_material.visit_uv_30d,
        "material_quality_score": chosen_material.quality_score,
        "material_score": chosen_material.score,
        "material_selection_policy": (
            "ai_generated_manual_review"
            if is_ai_generated_material
            else "score>=0.8,visit_uv_window_desc,ai_cleanup_manual_review"
            if is_external_recalled_material
            else "score>=0.8,cost_desc"
        ),
        "material_recall_strategy": chosen_material.recall_strategy,
        "material_recall_query_text": chosen_material.recall_query_text,
        "material_recall_config_code": chosen_material.recall_config_code,
        "material_recall_element_dimension": chosen_material.recall_element_dimension,
        "material_recall_point_type": chosen_material.recall_point_type,
        "material_recall_standard_element": chosen_material.recall_standard_element,
        "material_recall_hit_queries": chosen_material.recall_hit_queries,
        "material_dedupe_recent_count": len(recent_material_ids),
        "material_dedupe_run_count": len(excluded_material_ids),
        "material_dedupe_effective_count": len(effective_excluded_material_ids),
        "landing_dedupe_run_count": len(excluded_landing_ids),
        "creative_name": plan.root_source_id,
        # === Phase 3 POST 用 ===
        "_request_body": body,
        # === 追溯元数据(写 summary JSON)===
        "_material_id": chosen_material.material_id,
        "_material_image_id": material_image_id,
        "_pending_image_url": image_url if is_deferred_generated_material else "",
        "_ai_generated_material_id": chosen_ai_generated_material_id,
        "_external_recalled_material_id": chosen_external_recalled_material_id,
        "_external_source_material_id": (
            chosen_material.raw.get("source_material_id")
            if is_external_recalled_material else ""
        ),
        "_external_source_image_url": (
            chosen_material.raw.get("source_image_url")
            if is_external_recalled_material else ""
        ),
        "_plan_id": plan.plan_id,
        "_experiment_id": chosen_landing.experiment_id,
        "_brand_image_id": brand["brand_image_id"],
        # 文案选择回显(飞书表格"创意文案"列要展示)
        "_description_contents": description_contents,
    }
    if chosen_risk is not None:
        rec.update(chosen_risk.to_record_fields())
    if chosen_ai_generated_material_id:
        try:
            update_generated_material_status(
                rec,
                "prepared",
                tencent_image_id=material_image_id,
            )
        except Exception as e:
            logger.warning(
                "[prepare_one_creative] AI素材状态更新失败 id=%s:%s",
                chosen_ai_generated_material_id, e,
            )
    if chosen_external_recalled_material_id:
        try:
            update_external_material_status(rec, "prepared")
        except Exception as e:
            logger.warning(
                "[prepare_one_creative] 外部素材状态更新失败 id=%s:%s",
                chosen_external_recalled_material_id, e,
            )
    return rec


def _fetch_ad_metadata_for_approval(account_id: int, adgroup_id: int) -> dict:
    """反查广告,为飞书审批表组装 B 组(广告维度)字段。"""
    resp = _get("/adgroups/get", {
        "account_id": account_id, "page": 1, "page_size": 1,
        "filtering": [{
            "field": "adgroup_id", "operator": "IN", "values": [str(adgroup_id)],
        }],
        "fields": [
            "adgroup_id", "adgroup_name",
            "bid_amount", "site_set", "targeting",
        ],
    })
    items = (resp.get("data") or {}).get("list") or []
    if not items:
        return {}
    ad = items[0]

    bid_fen = ad.get("bid_amount")
    bid_yuan = f"{int(bid_fen) / 100:.2f}" if bid_fen is not None else ""

    site_set_cn_map = {
        "SITE_SET_WECHAT": "微信公众号",
        "SITE_SET_WECHAT_PLUGIN": "微信插件",
        "SITE_SET_SEARCH_SCENE": "搜索场景",
        "SITE_SET_MOMENTS": "朋友圈",
        "SITE_SET_MINI_GAME_WECHAT": "小游戏",
        "SITE_SET_MINI_PROGRAM_WECHAT": "小程序",
    }
    site_raw = ad.get("site_set") or []
    if isinstance(site_raw, str):
        site_str = site_set_cn_map.get(site_raw, site_raw)
    else:
        site_str = ",".join(site_set_cn_map.get(s, s) for s in site_raw)

    targeting = ad.get("targeting") or {}
    age_list = targeting.get("age") or []
    if age_list:
        age_str = ",".join(f"{a.get('min', '?')}-{a.get('max', '?')}" for a in age_list)
    else:
        age_str = "不限"

    return {
        "adgroup_name": ad.get("adgroup_name", ""),
        "bid_amount_yuan": bid_yuan,
        "site_set": site_str,
        "age_range": age_str,
    }


# Task 28:永久业务错误码(重试无意义,同一 body 永远报同样错)
# 实测/文档来源:
#   1801159 — 图片尺寸/格式不符
#   1801143 — 创意素材审核拒绝
#   1800269 — 品牌字段缺失/不合规
#   1801118 — 视频 / 图片资产 ID 无效
#   1901634 — 唯一性 reject(同营销内容广告组已存在相同创意)
#   1901589 — 相似性 reject
#   33001   — 参数校验失败
_PERMANENT_ERROR_CODES = ("1801159", "1801143", "1800269", "1801118", "1901634", "1901589", "33001")


def post_creative_with_prepared_body(
    account_id: int,
    body: dict,
    skip_if_exists: bool = True,
    max_retries: int = 2,
) -> Optional[int]:
    """Phase 3 POST:用 Phase 1 准备好的 body 调腾讯 /dynamic_creatives/add。

    Task 28(2026-06-11):加重试机制,区分瞬时 / 永久错误。
      - 永久业务错误(图尺寸 / 品牌缺失 / 唯一性 / 33001 等)→ 直接返回 None,不重试
      - 瞬时错误(网络超时 / 5xx / QPS limit)→ 最多重试 max_retries 次,指数退避

    Args:
        account_id: 腾讯账户 ID(用于幂等检查)
        body: Phase 1 prepare_one_creative_for_ad 返回的 _request_body
        skip_if_exists: True(默认)POST 前做幂等检查,命中返回已有 cid
        max_retries: 瞬时错误的最大重试次数(默认 2;首次 + 2 次重试 = 最坏 3 次总尝试)

    Returns:
        成功 dynamic_creative_id;失败 None(记 error log)
    """
    import time

    adgroup_id = body.get("adgroup_id")
    image_comp = body.get("creative_components", {}).get("image") or []
    material_image_id = ""
    if image_comp:
        material_image_id = (image_comp[0].get("value") or {}).get("image_id", "")

    if skip_if_exists and material_image_id and adgroup_id:
        existing_cid = find_existing_creative_by_image(
            account_id, int(adgroup_id), str(material_image_id),
        )
        if existing_cid:
            return existing_cid

    for attempt in range(max_retries + 1):  # 首次 + 重试
        try:
            logger.info(
                "[post_creative] account=%d adgroup=%s name=%s image_id=%s attempt=%d/%d",
                account_id, adgroup_id,
                body.get("dynamic_creative_name", ""), material_image_id,
                attempt + 1, max_retries + 1,
            )
            resp = _post("/dynamic_creatives/add", body)
            data = _check(resp, "creative_create")
            cid = data.get("dynamic_creative_id")
            if cid:
                logger.info("[post_creative] 成功 cid=%s", cid)
                return cid
            logger.error("[post_creative] 返回 data 缺 dynamic_creative_id: %s", data)
            return None
        except RuntimeError as e:
            err = str(e)
            # 永久错误 → 不重试,直接放弃(节省 quota,避免无意义打)
            is_permanent = any(code in err for code in _PERMANENT_ERROR_CODES)
            if is_permanent:
                logger.error(
                    "[post_creative] 永久错误,不重试 account=%d adgroup=%s: %s",
                    account_id, adgroup_id, err[:200],
                )
                return None
            # 瞬时错误:重试 quota 没用完 → backoff 后重试
            if attempt < max_retries:
                backoff_s = 5 * (2 ** attempt)  # 5s, 10s
                logger.warning(
                    "[post_creative] 瞬时错误 attempt=%d/%d,sleep %ds 后重试 account=%d adgroup=%s: %s",
                    attempt + 1, max_retries + 1, backoff_s,
                    account_id, adgroup_id, err[:150],
                )
                time.sleep(backoff_s)
                continue
            # 重试用完 → 放弃
            logger.error(
                "[post_creative] 重试 %d 次后仍失败 account=%d adgroup=%s: %s",
                max_retries, account_id, adgroup_id, err[:200],
            )
            return None
    return None


def try_create_one_creative_with_fallback(
    account_id: int,
    adgroup_id: int,
    max_landings: int = MAX_LANDING_ATTEMPTS_PER_AD,
    max_materials_per_landing: int = MAX_MATERIAL_PER_LANDING,
) -> Optional[int]:
    """对一条广告挂 1 条创意,带 try-fallback 鲁棒性(P0-A 核心 helper,2026-06-09)。

    召回的素材不一定都能挂(实测 code=1801159 尺寸不符,1530003 等),
    所以遍历 landing × material 尝试 POST,失败就试下一条,直到成功或穷尽。

    Args:
        account_id: 腾讯广告主账号 ID
        adgroup_id: 目标广告 ID
        max_landings: 最多尝试的 landing 数
        max_materials_per_landing: 每条 landing 召回 top N 素材尝试

    Returns:
        成功:dynamic_creative_id;穷尽:None(记 error log,主循环继续下一广告)
    """
    videos = fetch_landing_videos_for_account(account_id, page_size=max_landings * 2)
    valid = [v for v in videos if _is_landing_candidate(v)][:max_landings]
    logger.info(
        "[try_create_one_creative] account=%d adgroup=%d 候选 landing=%d",
        account_id, adgroup_id, len(valid),
    )

    attempts = []
    for v in valid:
        materials = recall_materials_for_video(v, final_top_n=max_materials_per_landing)
        if not materials:
            attempts.append(f"  · landing={v.video_id} 召回 0 素材,跳过")
            continue
        for m in materials:
            try:
                cid = create_creative_for_ad(account_id, adgroup_id, v, m)
                attempts.append(
                    f"  · landing={v.video_id} material={m.material_id[:12]} → 成功 cid={cid}"
                )
                logger.info("[try_create_one_creative] 成功 cid=%s\n%s", cid, "\n".join(attempts))
                return cid
            except RuntimeError as e:
                err = str(e)[:100]
                attempts.append(f"  · landing={v.video_id} material={m.material_id[:12]} → 失败 {err}")
                continue

    logger.error(
        "[try_create_one_creative] account=%d adgroup=%d 穷尽所有 landing×material 仍失败:\n%s",
        account_id, adgroup_id, "\n".join(attempts),
    )
    return None