Agent/examples/auto_put_ad/tools/audience_tools.py

"""
人群定向工具 — 定向策略与人群包管理
"""

import logging
from typing import Any, Dict, List, Optional

from agent.tools import tool
from agent.tools.models import ToolResult

from examples.auto_put_ad.tools.ad_api import _post, _check, DEFAULT_ACCOUNT_ID

logger = logging.getLogger(__name__)

# 本业务常用的年龄区间定向配置
AGE_TARGETING_PRESETS = {
    "18-24": [{"min": 18, "max": 24}],
    "25-30": [{"min": 25, "max": 30}],
    "25-35": [{"min": 25, "max": 35}],
    "31-40": [{"min": 31, "max": 40}],
    "35-45": [{"min": 35, "max": 45}],
    "18-35": [{"min": 18, "max": 35}],
    "25-45": [{"min": 25, "max": 35}, {"min": 35, "max": 45}],
}


@tool(description="生成定向设置（targeting结构体），根据人群包和年龄段组合生成可直接传入ad_create的targeting参数")
async def audience_build_targeting(
    custom_audience_ids: Optional[List[int]] = None,
    excluded_audience_ids: Optional[List[int]] = None,
    age_preset: Optional[str] = None,
    age_ranges: Optional[List[Dict[str, int]]] = None,
    gender: Optional[str] = None,
    geo_regions: Optional[List[int]] = None,
    user_os: Optional[List[str]] = None,
) -> ToolResult:
    """生成腾讯广告 targeting 结构体，直接用于 ad_create 的 targeting 参数。

    Args:
        custom_audience_ids: 自有人群包ID列表（定向投放给这些人群）
        excluded_audience_ids: 排除人群包ID列表（不投给这些人）
        age_preset: 预设年龄区间名称，可选：
            "18-24", "25-30", "25-35", "31-40", "35-45", "18-35", "25-45"
        age_ranges: 自定义年龄区间，如 [{"min": 25, "max": 35}]
            （age_preset 和 age_ranges 二选一，age_ranges 优先）
        gender: 性别定向 "MALE" / "FEMALE"，不传则不限
        geo_regions: 地域ID列表（省市区县，不传则全国）
        user_os: 操作系统 ["IOS"] / ["ANDROID"] / ["IOS", "ANDROID"]，不传则不限

    Returns:
        targeting 字典，可直接传给 ad_create(targeting=...)
    """
    targeting: Dict[str, Any] = {}

    if custom_audience_ids:
        targeting["custom_audience"] = custom_audience_ids
    if excluded_audience_ids:
        targeting["excluded_custom_audience"] = excluded_audience_ids

    # 年龄定向
    if age_ranges:
        targeting["age"] = age_ranges
    elif age_preset:
        if age_preset not in AGE_TARGETING_PRESETS:
            return ToolResult(
                title="audience_build_targeting 失败",
                output=f"不支持的 age_preset: {age_preset}，可选：{list(AGE_TARGETING_PRESETS.keys())}"
            )
        targeting["age"] = AGE_TARGETING_PRESETS[age_preset]

    if gender:
        targeting["gender"] = gender
    if geo_regions:
        targeting["geo_location"] = {"regions": geo_regions}
    if user_os:
        targeting["user_os"] = user_os

    if not targeting:
        return ToolResult(title="audience_build_targeting", output="警告：targeting 为空（宽泛定向），将投放给全量用户")

    # 生成可读描述
    desc_parts = []
    if custom_audience_ids:
        desc_parts.append(f"人群包: {custom_audience_ids}")
    if excluded_audience_ids:
        desc_parts.append(f"排除: {excluded_audience_ids}")
    if targeting.get("age"):
        ages = targeting["age"]
        desc_parts.append(f"年龄: {'-'.join(str(a.get('min', '')) + '~' + str(a.get('max', '')) for a in ages)}")
    if gender:
        desc_parts.append(f"性别: {'男' if gender == 'MALE' else '女'}")
    if geo_regions:
        desc_parts.append(f"地域: {len(geo_regions)}个地区")
    if user_os:
        desc_parts.append(f"系统: {'/'.join(user_os)}")

    return ToolResult(
        title="定向设置已生成",
        output="定向参数：" + "，".join(desc_parts),
        metadata={"targeting": targeting},
    )


@tool(description="查询可用人群包并推荐最优定向组合（基于历史效果数据）")
async def audience_recommend_targeting(
    optimization_goal: str = "OPTIMIZATIONGOAL_PAGE_VIEW",
    account_id: int = 0,
) -> ToolResult:
    """根据优化目标，推荐效果最好的人群包和定向组合方案。

    Args:
        optimization_goal: 优化目标，影响推荐策略
            OPTIMIZATIONGOAL_PAGE_VIEW（关键页面访问）
            OPTIMIZATIONGOAL_CLICK（点击）
        account_id: 广告主账号ID
    """
    from examples.auto_put_ad.tools.ad_api import audience_get_list

    # 查询账户下所有可用人群包
    result = await audience_get_list(account_id=account_id)

    if "失败" in result.title:
        return result

    audiences = (result.metadata or {}).get("list", [])
    if not audiences:
        return ToolResult(
            title="人群推荐",
            output="账户下暂无自定义人群包，建议先上传人群包后再进行精准定向。\n"
                   "过渡期可使用宽泛定向（仅年龄+地域），让系统自动探索最优人群。"
        )

    # 按人数排序，推荐人数较大的人群包
    audiences_sorted = sorted(audiences, key=lambda x: x.get("user_count", 0), reverse=True)

    recommendations = []
    for ag in audiences_sorted[:5]:
        recommendations.append(
            f"- [{ag['audience_id']}] {ag.get('name', '未命名')} "
            f"| 覆盖人数: {ag.get('user_count', 0):,}"
        )

    age_presets = ["25-35", "18-35"] if optimization_goal == "OPTIMIZATIONGOAL_CLICK" else ["25-45", "31-40"]

    output = (
        f"推荐定向方案（优化目标: {optimization_goal}）：\n\n"
        f"📦 推荐人群包（按覆盖量排序）：\n" + "\n".join(recommendations) + "\n\n"
        f"📅 推荐年龄区间：{' / '.join(age_presets)}\n\n"
        f"💡 建议：先用 1-2 个大人群包测试，确认效果后再细分"
    )
    return ToolResult(title="人群定向推荐", output=output)