"""创意搭建主入口(模块 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.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 load_excluded_ad_ids_from_adjustment( date_str: Optional[str] = None, output_dir: str = "outputs/reports", ) -> set[int]: """读调控当日决策 CSV,提取被 pause 的 adgroup_id(P0-E 关联点过滤,2026-06-09)。 新建子系统启动时调用,排除调控当日已决策 pause 的广告,避免"调控刚关、新建又补创意"的冲突。 Args: date_str: 形如 '20260609';None → 用昨天日期(调控 cron 02:00 UTC 跑完后) output_dir: 调控产物目录,相对当前工作目录 Returns: 被 pause 的 adgroup_id 集合;文件不存在 / CSV 字段缺失时返回空集(降级,不阻塞) """ import csv import glob import os from datetime import datetime, timedelta if date_str is None: date_str = (datetime.utcnow() - timedelta(days=1)).strftime("%Y%m%d") pattern = os.path.join(output_dir, f"llm_decisions_{date_str}*.csv") matches = sorted(glob.glob(pattern)) if not matches: logger.warning( "[load_excluded_ad_ids] 未找到调控决策 CSV: %s,返回空排除集(不阻塞主循环)", pattern, ) return set() latest = matches[-1] excluded: set[int] = set() with open(latest, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: action = (row.get("action") or "").strip().lower() ad_id_raw = (row.get("ad_id") or row.get("adgroup_id") or "").strip() if action == "pause" and ad_id_raw.isdigit(): excluded.add(int(ad_id_raw)) logger.info( "[load_excluded_ad_ids] 从 %s 读出 %d 个 pause 广告(排除)", os.path.basename(latest), len(excluded), ) return excluded 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 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, "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_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 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 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] 选中 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 未产出可用创意 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:") if is_ai_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_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,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_ai_generated_material else "", "_ai_generated_material_id": chosen_ai_generated_material_id, "_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, ) 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