Agent/examples/auto_put_ad_mini/tools/ad_creation.py

"""模块 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,
)

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


# ═══════════════════════════════════════════
# 营销内容指纹 + 出价取值
# ═══════════════════════════════════════════


def compute_fingerprint(
    account_id: int,
    site_set: list,
    custom_audience: Optional[list],
    age: list,
    geo_regions: list,
    wechat_position: Optional[list] = None,
) -> 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,
    }
    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"]

    # 差异化策略(2026-06-09 用户确认 + 2026-06-11 缩减):
    #   广告 #1 wechat_position=None   (无定投,走 site_set 默认全场景)
    #   广告 #2 wechat_position=WECHAT_POSITION_TARGETED_PRESET(本期 4 项小程序版位)
    # 同 site_set 同 targeting 但 wechat_position 不同 → 腾讯唯一性绕过(参考 77868332 实测)
    wechat_position_variants = [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,
        )
        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,
                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_SUSPEND",
) -> dict:
    """生成 /v3.0/adgroups/add 的完整请求 body。

    关键点:
    - configured_status 默认 SUSPEND(创建后默认暂停,运营 review 后启用)
    - 不传 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 = {
        "geo_location": {
            "location_types": candidate.location_types,
            "regions": candidate.region_ids,
        },
        "age": candidate.age,
    }
    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,
        "smart_bid_type": SMART_BID_TYPE,
        "bid_strategy": BID_STRATEGY,
        "bid_amount": 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": False,
        "site_set": candidate.site_set,
        # 搜索场景扩量 · 定向拓展(用户 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,
    }
    # 版位定投场景 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:
        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,
    }