"""模块 A · 广告新建(基于 SOP 固定参数 + 唯一性枚举) 设计原则(参考 CLAUDE.md no-guessing-rule): - 不猜测字段值;所有 SOP 字段从 config.py 读 - conversion_id **不传**:文档说可选,且不支持朋友圈,我们朋友圈版位有 - 用 optimization_goal=PROMOTION_VIEW_KEY_PAGE 替代 conversion_id 本模块只生成 request body,不直接调腾讯 API; 真实执行由 execution_engine + ad_api.ad_create() 走。 """ import hashlib import json import logging from dataclasses import dataclass, field, asdict from datetime import datetime from typing import Optional # SOP 固定参数 + 业务配置全部从 config.py 取 from config import ( # 营销内容 MARKETING_GOAL, MARKETING_SUB_GOAL, MARKETING_CARRIER_TYPE, MARKETING_TARGET_TYPE, MARKETING_ASSET_OUTER_SPEC, # 优化目标 OPTIMIZATION_GOAL, # 出价 / 计费 BID_MODE, SMART_BID_TYPE, BID_STRATEGY, AUTO_ACQUISITION_ENABLED, AUTO_ACQUISITION_BUDGET_FEN, AUTO_DERIVED_CREATIVE_ENABLED, AIM_SMART_TARGETING_ENABLED, AIM_SMART_SITE_ENABLED, # AIM 智能定向(2026-06-11 修正:write 接口字段是 smart_targeting_mode,值 SMART_TARGETING_MANUAL) SMART_TARGETING_MODE, # 版位定投场景(2026-06-09 新增,1 账户 2 广告差异化) WECHAT_POSITION_TARGETED_PRESET, ADS_PER_ACCOUNT, # 转化 DEFAULT_CONVERSION_ID, # 搜索场景扩量 · 定向拓展开关 SEARCH_EXPAND_TARGETING_SWITCH, # 定向 / 日期 DEFAULT_END_DATE, # 账户创建配置(DB 单一真相源) get_account_creation_config, # 监测链接 / 反馈 ID get_account_feedback_id, ) from tools.delivery_config import BID_MODE_MAX_CONVERSION logger = logging.getLogger(__name__) # ═══════════════════════════════════════════ # 候选数据结构 # ═══════════════════════════════════════════ @dataclass class AdCandidate: """一条新广告候选 — 唯一性枚举阶段产出,审批后才转 API request""" account_id: int adgroup_name: str site_set: list # ["SITE_SET_MOMENTS", ...] custom_audience: Optional[list] # [audience_id] 或 None(不传) bid_amount_fen: int # 出价(分) audience_tier_label: str # 用于 reason / 审批表展示 age: list # [{"min":45,"max":66}] location_types: list # ["LIVE_IN"] region_ids: list # 地域 region_id daily_budget_fen: int time_series: str delivery_version: str fingerprint: str # 营销内容指纹(本地判重) # 版位定投场景 wechat_position(2026-06-09 1 账户 N 广告差异化机制) # None = 无定投(走 site_set 默认全场景)/ list[int] = 勾选具体场景 ID wechat_position: Optional[list] = None bid_scene: str = "average_cost" custom_cost_cap_fen: Optional[int] = None automatic_site_enabled: bool = False # ═══════════════════════════════════════════ # 营销内容指纹 + 出价取值 # ═══════════════════════════════════════════ def compute_fingerprint( account_id: int, site_set: list, custom_audience: Optional[list], age: list, geo_regions: list, wechat_position: Optional[list] = None, automatic_site_enabled: bool = False, bid_scene: str = "average_cost", ) -> str: """计算"营销内容指纹"用于本地唯一性预校验。 注意:这里只是本地预筛(避免无效 API 调用)。 腾讯实际判重还看 marketing_goal/carrier_type/asset/opt_goal/smart_bid_type/site_set 其中 marketing_goal 等 5 个对本业务都固定,所以 site_set + targeting + wechat_position 决定 unique。 """ payload = { "account_id": account_id, "site_set": sorted(site_set), "custom_audience": sorted(custom_audience) if custom_audience else None, "age": age, "geo_regions": sorted(geo_regions), "wechat_position": sorted(wechat_position) if wechat_position else None, "automatic_site_enabled": automatic_site_enabled, "bid_scene": bid_scene, } return hashlib.md5( json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8") ).hexdigest() # ═══════════════════════════════════════════ # 命名 / 摘要 # ═══════════════════════════════════════════ _SITE_SHORT_MAP = { "SITE_SET_MOMENTS": "MOMT", "SITE_SET_WECHAT": "WCHT", "SITE_SET_MINI_PROGRAM_WECHAT": "MINI", "SITE_SET_WECHAT_PLUGIN": "PLGN", "SITE_SET_SEARCH_SCENE": "SRCH", } def _add_years(date_str: str, years: int) -> str: """date_str 是 YYYY-MM-DD,返回 N 年后的同一天(不处理 2/29 等边界,默认无 leap day 输入)""" d = datetime.strptime(date_str, "%Y-%m-%d") return d.replace(year=d.year + years).strftime("%Y-%m-%d") # tier 名 → 中文短名(命名用) _TIER_NAME_MAP = { "no_audience_pack": "泛人群", } # 转化目标 → 中文短名(命名用) _OPT_GOAL_NAME_MAP = { "OPTIMIZATIONGOAL_PROMOTION_VIEW_KEY_PAGE": "关键页面", "OPTIMIZATIONGOAL_CLICK": "点击", "OPTIMIZATIONGOAL_PAGE_VIEW": "页面浏览", } def build_adgroup_name( account_id: int, site_set: list, audience_tier: str, seq: int, ) -> str: """命名规则(用户 2026-06-05 确认):{人群包}-{日期}-{转化目标} 例: 83846793(no_audience_pack) → 泛人群-20260605-关键页面 83846804(R330+) → R330+-20260605-关键页面 """ date_str = datetime.now().strftime("%Y%m%d") tier_name = _TIER_NAME_MAP.get(audience_tier, audience_tier) opt_name = _OPT_GOAL_NAME_MAP.get(OPTIMIZATION_GOAL, OPTIMIZATION_GOAL) return f"{tier_name}-{date_str}-{opt_name}" # ═══════════════════════════════════════════ # 候选枚举 # ═══════════════════════════════════════════ def enumerate_new_ad_candidates( account_id: int, count: int = 3, existing_fingerprints: Optional[set] = None, ) -> list[AdCandidate]: """一账一包 + 固定定向 模式下,枚举 N 条 unique 候选广告。 差异化维度:site_set 组合(本业务有 3 种)。 其他维度(audience pack / age / geo / 优化目标 / 出价类型)都固定。 Args: account_id: 广告账户 ID(从 DB account_whitelist 取 pack) count: 期望产出条数(上限 = 可用 site_set 组合数 - 已存在指纹) existing_fingerprints: 已存在的指纹 set,用于跳过 Returns: list[AdCandidate] """ existing_fingerprints = existing_fingerprints or set() cfg = get_account_creation_config(account_id) pack_id = cfg["audience_pack_id"] tier_label = cfg["audience_tier_label"] custom_audience = [pack_id] if pack_id else None bid_amount_fen = cfg["bid_amount_fen"] automatic_site_enabled = bool(cfg.get("automatic_site_enabled")) bid_scene = cfg.get("bid_scene") or "average_cost" # 差异化策略(2026-06-09 用户确认 + 2026-06-11 缩减): # 广告 #1 wechat_position=None (无定投,走 site_set 默认全场景) # 广告 #2 wechat_position=WECHAT_POSITION_TARGETED_PRESET(公众号内容 + 小程序位置) # 同 site_set 同 targeting 但 wechat_position 不同 → 腾讯唯一性绕过(参考 77868332 实测) wechat_position_variants = [None] if automatic_site_enabled else [None, WECHAT_POSITION_TARGETED_PRESET] candidates: list[AdCandidate] = [] site_set = cfg["site_set"] for seq, wp in enumerate(wechat_position_variants, start=1): if len(candidates) >= count: break fingerprint = compute_fingerprint( account_id=account_id, site_set=site_set, custom_audience=custom_audience, age=cfg["age"], geo_regions=cfg["region_ids"], wechat_position=wp, automatic_site_enabled=automatic_site_enabled, bid_scene=bid_scene, ) if fingerprint in existing_fingerprints: logger.info( "[enumerate] skip: fingerprint %s 已存在 (account=%d wp=%s)", fingerprint[:8], account_id, "targeted" if wp else "default", ) continue wp_label = "targeted" if wp else "default" adgroup_name = build_adgroup_name(account_id, site_set, tier_label, seq) # 加 wp 后缀,让两条广告名可识别 adgroup_name = f"{adgroup_name}-{wp_label}" candidates.append( AdCandidate( account_id=account_id, adgroup_name=adgroup_name, site_set=site_set, custom_audience=custom_audience, bid_amount_fen=bid_amount_fen, bid_scene=bid_scene, custom_cost_cap_fen=cfg.get("custom_cost_cap_fen"), automatic_site_enabled=automatic_site_enabled, audience_tier_label=tier_label, age=cfg["age"], location_types=cfg["location_types"], region_ids=cfg["region_ids"], daily_budget_fen=cfg["daily_budget_fen"], time_series=cfg["time_series"], delivery_version=cfg["delivery_version"], fingerprint=fingerprint, wechat_position=wp, ) ) logger.info( "[enumerate] account=%d 产出 %d 条候选(目标 %d 条) — 差异化:wechat_position 有/无", account_id, len(candidates), count, ) return candidates # ═══════════════════════════════════════════ # 构造腾讯 adgroups/add 请求 body # ═══════════════════════════════════════════ def build_ad_request_body( candidate: AdCandidate, begin_date: Optional[str] = None, configured_status: str = "AD_STATUS_NORMAL", ) -> dict: """生成 /v3.0/adgroups/add 的完整请求 body。 关键点: - configured_status 默认 NORMAL:新广告先开启,实际投放受创意审核/账户/预算控制 如需暂停创建,调用方显式传 AD_STATUS_SUSPEND - 不传 conversion_id(可选 + 不支持朋友圈,我们有朋友圈版位) - 不传顶层 marketing_target_type,只通过 marketing_asset_outer_spec 传 (待 dry run 验证;若腾讯要求,加上) - gender 不传 = 不限性别 """ begin_date = begin_date or datetime.now().strftime("%Y-%m-%d") # 定向 — 账户级年龄/地域 + 可选人群包。 targeting: dict = {"age": candidate.age} if candidate.location_types or candidate.region_ids: targeting["geo_location"] = { "location_types": candidate.location_types, "regions": candidate.region_ids, } if candidate.custom_audience: targeting["custom_audience"] = candidate.custom_audience # 监测链接 ID(账户级)— 不能传 None feedback_id = get_account_feedback_id(candidate.account_id) if feedback_id is None: raise ValueError( f"account_id {candidate.account_id} 的 feedback_id 未配置。" f"请在 config.ACCOUNT_FEEDBACK_ID_MAPPING 中补充,或等运营提供。" ) body: dict = { # === 基础 === "account_id": candidate.account_id, "adgroup_name": candidate.adgroup_name, "configured_status": configured_status, "feedback_id": feedback_id, # === 营销内容 === "marketing_goal": MARKETING_GOAL, "marketing_sub_goal": MARKETING_SUB_GOAL, "marketing_carrier_type": MARKETING_CARRIER_TYPE, "marketing_target_type": MARKETING_TARGET_TYPE, # 小程序投流必传(用户 2026-06-05 确认) "marketing_asset_outer_spec": MARKETING_ASSET_OUTER_SPEC, # === 优化目标 + 转化 === # conversion_id 关联完整"平台转化"包(含优化目标 + 数据上报 + 归因方式) # 用户 2026-06-05 确认:两个测试账户都用 1007(样本一致) "optimization_goal": OPTIMIZATION_GOAL, "conversion_id": DEFAULT_CONVERSION_ID, # === 出价 / 计费(SOP 稳定拿量)=== "bid_mode": BID_MODE, "bid_strategy": BID_STRATEGY, "smart_bid_type": ( "SMART_BID_TYPE_SYSTEMATIC" if candidate.bid_scene == BID_MODE_MAX_CONVERSION else SMART_BID_TYPE ), "bid_amount": ( 0 if candidate.bid_scene == BID_MODE_MAX_CONVERSION else candidate.bid_amount_fen ), "daily_budget": candidate.daily_budget_fen, "auto_acquisition_enabled": AUTO_ACQUISITION_ENABLED, "auto_derived_creative_enabled": AUTO_DERIVED_CREATIVE_ENABLED, # === 时段 / 日期 === # end_date 必填,且不接受 "0"。默认设 begin_date + 1 年表示长期 "begin_date": begin_date, "end_date": _add_years(begin_date, 1), "time_series": candidate.time_series, # === 版位 === "automatic_site_enabled": candidate.automatic_site_enabled, # 搜索场景扩量 · 定向拓展(用户 2026-06-05 确认:关) "search_expand_targeting_switch": SEARCH_EXPAND_TARGETING_SWITCH, # === 定向 === # AIM 智能定向(2026-06-11 实测修正:字段名是 smart_targeting_mode 不是 status) # 之前用错字段名 smart_targeting_status,腾讯静默忽略,创建出来全 AUTO # 正确写法:smart_targeting_mode=SMART_TARGETING_MANUAL(手动定向 = AIM 关闭) "smart_targeting_mode": SMART_TARGETING_MODE, "targeting": targeting, } if not candidate.automatic_site_enabled: body["site_set"] = candidate.site_set if candidate.bid_scene == BID_MODE_MAX_CONVERSION: if candidate.custom_cost_cap_fen is None: raise ValueError( f"account_id {candidate.account_id} 最大转化量缺少 custom_cost_cap_fen" ) body["cost_constraint_scene"] = "COST_CONSTRAINT_SCENE_OPEN" body["custom_cost_cap"] = candidate.custom_cost_cap_fen # 版位定投场景 wechat_position(2026-06-10 实测修正:POST 时 scene_spec 是顶层字段) # 之前误把 scene_spec 放 targeting 内 → 腾讯 reject code 12813 "包含不识别的参数 scene_spec" # WebFetch /adgroups/add 文档确认:scene_spec 是顶层 struct,wechat_position 是它的子字段 if candidate.wechat_position and not candidate.automatic_site_enabled: body["scene_spec"] = {"wechat_position": list(candidate.wechat_position)} if AUTO_ACQUISITION_ENABLED and AUTO_ACQUISITION_BUDGET_FEN: body["auto_acquisition_budget"] = AUTO_ACQUISITION_BUDGET_FEN return body # ═══════════════════════════════════════════ # 调试 / 演示入口(用于"先打印 body,再决定是否真调") # ═══════════════════════════════════════════ def post_ad_with_prepared_body(account_id: int, body: dict) -> Optional[int]: """Phase 0 POST:用 build_ad_request_body 构造的 body 调腾讯 /adgroups/add。 对偶 creative_creation.post_creative_with_prepared_body,Phase 0 模块 A 用。 Returns: 成功:adgroup_id;失败:None(记 error log) """ from tools.ad_api import _check, _post try: logger.info( "[post_ad] account=%d adgroup_name=%s site_set=%s wp=%s", account_id, body.get("adgroup_name"), body.get("site_set"), "targeted" if (body.get("targeting") or {}).get("scene_spec") else "default", ) resp = _post("/adgroups/add", body) data = _check(resp, "ad_create") adgroup_id = data.get("adgroup_id") if adgroup_id: logger.info("[post_ad] 成功 adgroup_id=%s", adgroup_id) return int(adgroup_id) logger.error("[post_ad] 返回 data 缺 adgroup_id: %s", data) return None except RuntimeError as e: logger.error( "[post_ad] account=%d 失败: %s", account_id, str(e)[:200], ) return None def preview_candidates(account_id: int, count: int = 3) -> dict: """生成候选 + 完整 request body,只打印不调 API。 用法:在 REPL / 脚本里调用,看 body 后再决定是否真 dry run。 """ candidates = enumerate_new_ad_candidates(account_id, count=count) bodies = [build_ad_request_body(c) for c in candidates] return { "account_id": account_id, "candidate_count": len(candidates), "candidates": [asdict(c) for c in candidates], "request_bodies": bodies, }