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,
    # 转化
    DEFAULT_CONVERSION_ID,
    # 搜索场景扩量 · 定向拓展开关
    SEARCH_EXPAND_TARGETING_SWITCH,
    # 版位
    AVAILABLE_SITE_SETS,
    SITE_SET_COMBINATIONS,
    # 定向
    FIXED_TARGETING_AGE,
    FIXED_TARGETING_GENDER,
    FIXED_TARGETING_LOCATION_TYPES,
    FIXED_TARGETING_REGION_IDS,
    # 时段 / 日期 / 预算
    TIME_SERIES_DEFAULT,
    DEFAULT_END_DATE,
    DEFAULT_DAILY_BUDGET_FEN,
    # 一账一包
    ACCOUNT_AUDIENCE_PACK_MAPPING,
    # 监测链接 / 反馈 ID
    get_account_feedback_id,
    # 出价区间策略
    AUDIENCE_BID_RANGES,
    BID_PICK_STRATEGY,
)

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 / 审批表展示
    fingerprint: str                        # 营销内容指纹(本地判重)


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


def compute_fingerprint(
    account_id: int,
    site_set: list,
    custom_audience: Optional[list],
    age: list,
    geo_regions: list,
) -> str:
    """计算"营销内容指纹"用于本地唯一性预校验。

    注意:这里只是本地预筛(避免无效 API 调用)。
    腾讯实际判重还看 marketing_goal/carrier_type/asset/opt_goal/smart_bid_type/site_set
    其中 marketing_goal 等 5 个对本业务都固定,所以 site_set + targeting 决定 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),
    }
    return hashlib.md5(
        json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
    ).hexdigest()


def pick_bid_amount_fen(tier_label: str, strategy: Optional[str] = None) -> int:
    """从 SOP 出价区间内取一个具体出价(单位:分)。"""
    strategy = strategy or BID_PICK_STRATEGY
    bid_range = AUDIENCE_BID_RANGES.get(tier_label)
    if not bid_range:
        raise ValueError(
            f"tier_label '{tier_label}' 不在 AUDIENCE_BID_RANGES 中。"
            f"可用 keys: {list(AUDIENCE_BID_RANGES.keys())}"
        )
    min_yuan, max_yuan = bid_range
    if strategy == "midpoint":
        yuan = (min_yuan + max_yuan) / 2
    elif strategy == "max":
        yuan = max_yuan
    elif strategy == "min":
        yuan = min_yuan
    else:
        raise ValueError(
            f"未知出价策略 '{strategy}',支持:midpoint / max / min"
        )
    return int(round(yuan * 100))  # 元 → 分


# ═══════════════════════════════════════════
# 命名 / 摘要
# ═══════════════════════════════════════════


_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(从 ACCOUNT_AUDIENCE_PACK_MAPPING 取 pack)
        count: 期望产出条数(上限 = 可用 site_set 组合数 - 已存在指纹)
        existing_fingerprints: 已存在的指纹 set,用于跳过

    Returns:
        list[AdCandidate]
    """
    existing_fingerprints = existing_fingerprints or set()

    if account_id not in ACCOUNT_AUDIENCE_PACK_MAPPING:
        raise ValueError(
            f"account_id {account_id} 不在 ACCOUNT_AUDIENCE_PACK_MAPPING 中。"
            f"请先在 config.py 配置该账户的 audience pack。"
        )

    pack_id, tier_label = ACCOUNT_AUDIENCE_PACK_MAPPING[account_id]
    custom_audience = [pack_id] if pack_id else None

    bid_amount_fen = pick_bid_amount_fen(tier_label)

    candidates: list[AdCandidate] = []
    for seq, site_set in enumerate(SITE_SET_COMBINATIONS, start=1):
        if len(candidates) >= count:
            break

        fingerprint = compute_fingerprint(
            account_id=account_id,
            site_set=site_set,
            custom_audience=custom_audience,
            age=FIXED_TARGETING_AGE,
            geo_regions=FIXED_TARGETING_REGION_IDS,
        )
        if fingerprint in existing_fingerprints:
            logger.info(
                "[enumerate] skip: fingerprint %s 已存在 (account=%d site_set=%s)",
                fingerprint[:8], account_id, site_set,
            )
            continue

        adgroup_name = build_adgroup_name(account_id, site_set, tier_label, seq)
        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,
                fingerprint=fingerprint,
            )
        )

    logger.info(
        "[enumerate] account=%d 产出 %d 条候选(目标 %d 条)",
        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": FIXED_TARGETING_LOCATION_TYPES,
            "regions": FIXED_TARGETING_REGION_IDS,
        },
        "age": FIXED_TARGETING_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": DEFAULT_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": TIME_SERIES_DEFAULT,
        # === 版位 ===
        "automatic_site_enabled": False,
        "site_set": candidate.site_set,
        # 搜索场景扩量 · 定向拓展(用户 2026-06-05 确认:关)
        "search_expand_targeting_switch": SEARCH_EXPAND_TARGETING_SWITCH,
        # === 定向 ===
        "targeting": targeting,
    }

    if AUTO_ACQUISITION_ENABLED and AUTO_ACQUISITION_BUDGET_FEN:
        body["auto_acquisition_budget"] = AUTO_ACQUISITION_BUDGET_FEN

    return body


# ═══════════════════════════════════════════
# 调试 / 演示入口(用于"先打印 body,再决定是否真调")
# ═══════════════════════════════════════════


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,
    }