Agent/examples/auto_put_ad_mini/tools/ad_api.py

"""
腾讯广告 Marketing API v3.0 封装工具

层级结构（3.0，仅2层）：
  广告（Ad） → 创意（Dynamic Creative）

⚠️ 重要：
- 业务概念是"广告"，但 API 端点技术上仍叫 adgroups
- POST 请求：公共参数（access_token/timestamp/nonce）在 URL query，业务参数在 JSON body
- GET 请求：所有参数（含公共参数）在 URL query，复杂对象需 JSON 序列化后 URL 编码

环境变量：
  TENCENT_AD_ACCESS_TOKEN   OAuth2 access token
  TENCENT_AD_ACCOUNT_ID     默认广告账户 ID（可被参数覆盖）
  TENCENT_AD_BASE_URL       API base，默认 https://api.e.qq.com/v3.0
"""

import hashlib
import json
import logging
import os
import time
import uuid
from typing import Any, Dict, List, Optional
from urllib.parse import urlencode

import httpx

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

logger = logging.getLogger(__name__)

# ===== 基础配置 =====

BASE_URL = os.getenv("TENCENT_AD_BASE_URL", "https://api.e.qq.com/v3.0")
DEFAULT_ACCOUNT_ID = int(os.getenv("TENCENT_AD_ACCOUNT_ID", "0") or 0)
TIMEOUT = 30  # 秒

# Token 获取 API（内部服务，根据 accountId 返回最新 access_token）
TOKEN_API_URL = os.getenv(
    "TENCENT_AD_TOKEN_API",
    "https://api.piaoquantv.com/ad/put/tencent/getAccessToken",
)

# Token 缓存：避免每次 API 调用都重新获取
_token_cache: Dict[int, Dict[str, Any]] = {}
_TOKEN_CACHE_TTL = 1800  # 缓存有效期 30 分钟


def _get_access_token(account_id: int = 0) -> str:
    """动态获取 access_token。

    优先通过内部 token API 获取（自动刷新），缓存 30 分钟。
    若 token API 不可用，降级使用环境变量中的静态 token。
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    if not acct:
        raise ValueError("未配置 TENCENT_AD_ACCOUNT_ID 环境变量")

    # 检查缓存是否有效
    cached = _token_cache.get(acct)
    if cached and time.time() - cached["ts"] < _TOKEN_CACHE_TTL:
        return cached["token"]

    # 尝试从 token API 动态获取
    try:
        resp = httpx.get(
            TOKEN_API_URL,
            params={"accountId": acct},
            timeout=10,
        )
        resp.raise_for_status()
        token = resp.text.strip()
        if token and len(token) > 10:
            _token_cache[acct] = {"token": token, "ts": time.time()}
            logger.info("[TokenAPI] 已获取 account=%s 的 access_token (缓存30分钟)", acct)
            return token
        else:
            logger.warning("[TokenAPI] 返回内容异常: %s，降级使用环境变量", token[:50])
    except Exception as e:
        logger.warning("[TokenAPI] 请求失败: %s，降级使用环境变量", e)

    # 降级：使用环境变量中的静态 token
    static_token = os.getenv("TENCENT_AD_ACCESS_TOKEN", "")
    if not static_token:
        raise ValueError(
            f"Token API 请求失败且未配置 TENCENT_AD_ACCESS_TOKEN 环境变量 (account={acct})"
        )
    return static_token


def _common_params(account_id: int = 0) -> Dict[str, str]:
    """公共查询参数：access_token / timestamp / nonce"""
    return {
        "access_token": _get_access_token(account_id),
        "timestamp": str(int(time.time())),
        "nonce": uuid.uuid4().hex,
    }


def _get(path: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """
    发送 GET 请求。
    复杂对象（list/dict）自动 JSON 序列化后作为 query string 参数传递。
    """
    account_id = params.get("account_id", 0)
    query = dict(_common_params(account_id))
    for k, v in params.items():
        if v is None:
            continue
        if isinstance(v, (dict, list)):
            query[k] = json.dumps(v, ensure_ascii=False)
        else:
            query[k] = str(v)

    url = f"{BASE_URL}{path}?{urlencode(query)}"
    logger.debug("[TencentAPI] GET %s", url)

    resp = httpx.get(url, timeout=TIMEOUT)
    resp.raise_for_status()
    return resp.json()


def _get_user_token_for_account(account_id: int) -> str:
    """获取 user_token。优先 DB(account_whitelist.user_token),fallback 到环境变量。

    DB 设计原因(2026-06-05):每个账户可能有不同的经办人授权 token,
    全局环境变量无法覆盖多账户场景。
    """
    if account_id:
        try:
            from db.connection import get_connection
            conn = get_connection()
            try:
                with conn.cursor() as cur:
                    cur.execute(
                        "SELECT user_token FROM account_whitelist WHERE account_id=%s",
                        (account_id,),
                    )
                    row = cur.fetchone()
                    if row and row.get("user_token"):
                        return row["user_token"]
            finally:
                conn.close()
        except Exception as e:
            logger.warning(
                "[user_token] account=%d DB 查询失败,fallback 到 env: %s",
                account_id, e,
            )
    return os.getenv("TENCENT_AD_USER_TOKEN", "")


def _post(path: str, body: Dict[str, Any]) -> Dict[str, Any]:
    """
    发送 POST 请求。
    公共参数在 URL query，业务参数在 JSON body。

    ⚠️ 重要：腾讯广告写操作需要 user_token（实名认证令牌）
    优先级:DB account_whitelist.user_token > 环境变量 TENCENT_AD_USER_TOKEN
    """
    account_id = body.get("account_id", 0)
    params = _common_params(account_id)

    # 写操作需要额外的 user_token（读操作不需要）
    user_token = _get_user_token_for_account(account_id)
    if user_token:
        params["user_token"] = user_token
        logger.debug("[TencentAPI] user_token 长度=%d (account=%d)", len(user_token), account_id)
    else:
        logger.warning(
            "[TencentAPI] 未配置 TENCENT_AD_USER_TOKEN，"
            "写操作可能失败（错误码 11101）"
        )

    query = urlencode(params)
    url = f"{BASE_URL}{path}?{query}"
    logger.debug("[TencentAPI] POST %s body=%s", url, json.dumps(body, ensure_ascii=False)[:200])

    resp = httpx.post(url, json=body, timeout=TIMEOUT)
    resp.raise_for_status()
    return resp.json()


def _check(resp: Dict[str, Any], op: str) -> Dict[str, Any]:
    """统一检查 API 响应，code != 0 时抛异常"""
    code = resp.get("code", -1)
    if code != 0:
        msg = resp.get("message_cn") or resp.get("message", "未知错误")
        raise RuntimeError(f"[{op}] 腾讯广告 API 错误 code={code}: {msg}")
    return resp.get("data") or {}


# ===== 素材库（Image）— 公共能力,供创意/广告/品牌 共用 =====

def images_add(account_id: int, image_url: str, timeout: int = 60) -> str:
    """下载图片 → MD5 → POST /v3.0/images/add (multipart) → 返回 image_id (str)。

    腾讯 MD5 幂等(已实测):同账户重复上传同图返回相同 image_id,**无需本地缓存**。
    跨账户不复用:同图在不同账户下 image_id 不同。

    Args:
        account_id: 腾讯广告主账号 ID
        image_url: 图片公网可下载 URL

    Returns:
        image_id (str) — 可直接填入 creative_components.image[].value.image_id
                       或 creative_components.brand[].value.brand_image_id

    Raises:
        RuntimeError: 下载失败 / 腾讯 reject(含 code + message_cn)
    """
    resp = httpx.get(image_url, timeout=TIMEOUT)
    resp.raise_for_status()
    img_bytes = resp.content
    img_md5 = hashlib.md5(img_bytes).hexdigest()

    params = dict(_common_params(account_id))
    user_token = _get_user_token_for_account(account_id)
    if user_token:
        params["user_token"] = user_token
    url = f"{BASE_URL}/images/add?{urlencode(params)}"

    files = {"file": ("image.jpg", img_bytes, "image/jpeg")}
    data = {
        "account_id": str(account_id),
        "signature": img_md5,
        "upload_type": "UPLOAD_TYPE_FILE",
    }
    logger.info(
        "[images_add] account=%d url=%s md5=%s size=%d",
        account_id, image_url[:60], img_md5, len(img_bytes),
    )
    api_resp = httpx.post(url, files=files, data=data, timeout=timeout).json()
    if api_resp.get("code") != 0:
        raise RuntimeError(
            f"images_add 失败 account={account_id} url={image_url} "
            f"code={api_resp.get('code')} msg={api_resp.get('message_cn') or api_resp.get('message')}"
        )
    return str(api_resp["data"]["image_id"])


# ===== 广告（Ad）— 3.0 顶层单位 =====

@tool(description="创建广告（腾讯广告3.0顶层单位，含营销目标/定向/出价/预算，对应API: /adgroups/add）")
async def ad_create(
    adgroup_name: str,
    marketing_goal: str = "MARKETING_GOAL_USER_GROWTH",
    marketing_carrier_type: str = "MARKETING_CARRIER_TYPE_MINI_PROGRAM_WECHAT",
    marketing_carrier_id: str = "",
    begin_date: str = "",
    end_date: str = "",
    time_series: str = "1" * 336,
    bid_mode: str = "BID_MODE_OCPM",
    optimization_goal: str = "OPTIMIZATIONGOAL_PAGE_VIEW",
    bid_amount: int = 0,
    daily_budget: int = 0,
    automatic_site_enabled: bool = True,
    targeting: Optional[Dict[str, Any]] = None,
    configured_status: str = "AD_STATUS_NORMAL",
    account_id: int = 0,
) -> ToolResult:
    """创建广告（3.0 顶层单位，API 端点: /v3.0/adgroups/add）

    本业务固定参数：
    - marketing_goal: MARKETING_GOAL_USER_GROWTH（用户增长）
    - bid_mode: BID_MODE_OCPM（oCPM 出价，固定）
    - optimization_goal: OPTIMIZATIONGOAL_PAGE_VIEW 或 OPTIMIZATIONGOAL_CLICK

    targeting 结构示例：
    {
        "age": [{"min": 25, "max": 35}],
        "custom_audience": [人群包ID列表],
        "excluded_custom_audience": [排除人群包ID列表],
        "geo_location": {"regions": [省市区县ID列表]},
        "gender": "MALE",  // 可选，不传则不限性别
        "user_os": ["IOS", "ANDROID"]  // 可选
    }

    Args:
        adgroup_name: 广告名称（1-60个等宽字符）
        marketing_goal: 营销目的，固定 MARKETING_GOAL_USER_GROWTH
        marketing_carrier_type: 推广载体，MARKETING_CARRIER_TYPE_MINI_PROGRAM_WECHAT 或 MARKETING_CARRIER_TYPE_WECHAT_OFFICIAL_ACCOUNT
        marketing_carrier_id: 载体ID（小程序AppID或公众号ID）
        begin_date: 投放开始日期，格式 YYYY-MM-DD
        end_date: 投放结束日期，格式 YYYY-MM-DD
        time_series: 投放时段，336位字符串（48段×7天），"1"=投放，"0"=不投，全1表示全时段
        bid_mode: 出价方式，固定 BID_MODE_OCPM
        optimization_goal: 优化目标，OPTIMIZATIONGOAL_PAGE_VIEW 或 OPTIMIZATIONGOAL_CLICK
        bid_amount: 出价（单位：分），如 5000 = 50元
        daily_budget: 日预算（单位：分），0=不限
        automatic_site_enabled: 是否开启智能版位（建议 True）
        targeting: 定向设置（见上方说明）
        configured_status: AD_STATUS_NORMAL（投放中）或 AD_STATUS_SUSPEND（暂停）
        account_id: 广告主账号ID，0则使用环境变量
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    if not acct:
        return ToolResult(title="ad_create 失败", output="account_id 未指定且未配置 TENCENT_AD_ACCOUNT_ID")

    body: Dict[str, Any] = {
        "account_id": acct,
        "adgroup_name": adgroup_name,
        "marketing_goal": marketing_goal,
        "marketing_carrier_type": marketing_carrier_type,
        "bid_mode": bid_mode,
        "optimization_goal": optimization_goal,
        "configured_status": configured_status,
        "automatic_site_enabled": automatic_site_enabled,
    }
    if marketing_carrier_id:
        body["marketing_carrier_detail"] = {"marketing_carrier_id": marketing_carrier_id}
    if begin_date:
        body["begin_date"] = begin_date
    if end_date:
        body["end_date"] = end_date
    if time_series:
        body["time_series"] = time_series
    if bid_amount:
        body["bid_amount"] = bid_amount
    if daily_budget:
        body["daily_budget"] = daily_budget
    if targeting:
        body["targeting"] = targeting

    try:
        resp = _post("/adgroups/add", body)
        data = _check(resp, "ad_create")
        adgroup_id = data.get("adgroup_id")
        return ToolResult(
            title=f"广告创建成功",
            output=f"广告已创建，adgroup_id={adgroup_id}，名称：{adgroup_name}",
            metadata={"adgroup_id": adgroup_id, "adgroup_name": adgroup_name},
        )
    except Exception as e:
        logger.error("ad_create 失败: %s", e)
        return ToolResult(title="ad_create 失败", output=str(e))


@tool(description="更新广告设置（出价/预算/定向/状态/名称），对应API: /adgroups/update")
async def ad_update(
    adgroup_id: int,
    adgroup_name: Optional[str] = None,
    bid_amount: Optional[int] = None,
    daily_budget: Optional[int] = None,
    targeting: Optional[Dict[str, Any]] = None,
    configured_status: Optional[str] = None,
    account_id: int = 0,
) -> ToolResult:
    """更新广告设置。只传需要修改的字段，未传字段保持不变。

    Args:
        adgroup_id: 广告ID（API字段名，实际是3.0的广告ID）
        adgroup_name: 新名称（可选）
        bid_amount: 新出价，单位分（可选）
        daily_budget: 新日预算，单位分，0=不限（可选）
        targeting: 新定向设置（可选）
        configured_status: 新状态 AD_STATUS_NORMAL / AD_STATUS_SUSPEND（可选）
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    body: Dict[str, Any] = {"account_id": acct, "adgroup_id": adgroup_id}
    if adgroup_name is not None:
        body["adgroup_name"] = adgroup_name
    if bid_amount is not None:
        body["bid_amount"] = bid_amount
    if daily_budget is not None:
        body["daily_budget"] = daily_budget
    if targeting is not None:
        body["targeting"] = targeting
    if configured_status is not None:
        body["configured_status"] = configured_status

    try:
        resp = _post("/adgroups/update", body)
        _check(resp, "ad_update")
        changes = [k for k in ["adgroup_name", "bid_amount", "daily_budget", "targeting", "configured_status"] if k in body]
        return ToolResult(
            title="广告更新成功",
            output=f"广告 {adgroup_id} 已更新字段：{', '.join(changes)}",
        )
    except Exception as e:
        return ToolResult(title="ad_update 失败", output=str(e))


@tool(description="批量修改广告状态（开启/暂停），一次最多50个广告")
async def ad_batch_update_status(
    adgroup_ids: List[int],
    configured_status: str,
    account_id: int = 0,
) -> ToolResult:
    """批量开启或暂停广告，单次最多50个。

    Args:
        adgroup_ids: 广告ID列表，最多50个
        configured_status: AD_STATUS_NORMAL（开启）或 AD_STATUS_SUSPEND（暂停）
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    if len(adgroup_ids) > 50:
        return ToolResult(title="ad_batch_update_status 失败", output="单次最多操作50个广告（API限制）")

    results = []
    errors = []
    for adgroup_id in adgroup_ids:
        try:
            body = {"account_id": acct, "adgroup_id": adgroup_id, "configured_status": configured_status}
            resp = _post("/adgroups/update", body)
            _check(resp, "ad_batch_update_status")
            results.append(adgroup_id)
        except Exception as e:
            errors.append(f"{adgroup_id}: {e}")

    status_label = "开启" if configured_status == "AD_STATUS_NORMAL" else "暂停"
    summary = f"成功{status_label} {len(results)} 个广告"
    if errors:
        summary += f"，失败 {len(errors)} 个：{'; '.join(errors)}"
    return ToolResult(title=f"批量{status_label}广告", output=summary)


@tool(description="查询广告列表，支持按ID/状态/营销目标过滤")
async def ad_get_list(
    adgroup_ids: Optional[List[int]] = None,
    configured_status: Optional[List[str]] = None,
    marketing_goal: Optional[str] = None,
    page: int = 1,
    page_size: int = 20,
    account_id: int = 0,
) -> ToolResult:
    """查询广告列表。

    Args:
        adgroup_ids: 按广告ID过滤（可选）
        configured_status: 按状态过滤，如 ["AD_STATUS_NORMAL", "AD_STATUS_SUSPEND"]
        marketing_goal: 按营销目标过滤（可选）
        page: 页码，从1开始
        page_size: 每页数量，最大100
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    params: Dict[str, Any] = {"account_id": acct, "page": page, "page_size": page_size}

    filtering: Dict[str, Any] = {}
    if adgroup_ids:
        filtering["adgroup_id_list"] = adgroup_ids
    if configured_status:
        filtering["configured_status_list"] = configured_status
    if marketing_goal:
        filtering["marketing_goal"] = marketing_goal
    if filtering:
        params["filtering"] = filtering

    try:
        resp = _get("/adgroups/get", params)
        data = _check(resp, "ad_get_list")
        items = data.get("list", [])
        page_info = data.get("page_info", {})

        summary_lines = []
        for item in items:
            summary_lines.append(
                f"- [{item.get('adgroup_id')}] {item.get('adgroup_name')} "
                f"| 状态:{item.get('configured_status')} "
                f"| 出价:{item.get('bid_amount', 0)/100:.2f}元 "
                f"| 日预算:{item.get('daily_budget', 0)/100:.0f}元"
            )

        output = f"共 {page_info.get('total_number', len(items))} 个广告，当前第{page}页：\n" + "\n".join(summary_lines)
        return ToolResult(title=f"查询广告列表（{len(items)}条）", output=output, metadata={"list": items, "page_info": page_info})
    except Exception as e:
        return ToolResult(title="ad_get_list 失败", output=str(e))


# ===== 创意（Dynamic Creative）=====

@tool(description="创建动态创意（绑定素材组件到广告），对应API: /dynamic_creatives/add")
async def creative_create(
    adgroup_id: int,
    creative_name: str,
    page_id: Optional[int] = None,
    title_list: Optional[List[str]] = None,
    description_list: Optional[List[str]] = None,
    image_id_list: Optional[List[str]] = None,
    video_id: Optional[str] = None,
    call_to_action: Optional[str] = None,
    configured_status: str = "AD_STATUS_NORMAL",
    account_id: int = 0,
) -> ToolResult:
    """创建动态创意，系统自动组合素材组件并优化投放。

    Args:
        adgroup_id: 广告ID（绑定到哪个广告）
        creative_name: 创意名称
        page_id: 落地页ID（小程序页面或H5）
        title_list: 标题列表，系统从中优选（≤30字/条）
        description_list: 描述列表（≤60字/条）
        image_id_list: 图片素材ID列表（从素材库获取）
        video_id: 视频素材ID
        call_to_action: 行动号召按钮文案，如"立即体验"
        configured_status: AD_STATUS_NORMAL 或 AD_STATUS_SUSPEND
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    body: Dict[str, Any] = {
        "account_id": acct,
        "adgroup_id": adgroup_id,
        "dynamic_creative_name": creative_name,
        "configured_status": configured_status,
    }
    if page_id:
        body["page_id"] = page_id
    if title_list:
        body["title_list"] = title_list
    if description_list:
        body["description_list"] = description_list
    if image_id_list:
        body["image_id_list"] = image_id_list
    if video_id:
        body["video_id"] = video_id
    if call_to_action:
        body["call_to_action"] = call_to_action

    try:
        resp = _post("/dynamic_creatives/add", body)
        data = _check(resp, "creative_create")
        creative_id = data.get("dynamic_creative_id")
        return ToolResult(
            title="创意创建成功",
            output=f"创意已创建，dynamic_creative_id={creative_id}，绑定广告 {adgroup_id}",
            metadata={"dynamic_creative_id": creative_id},
        )
    except Exception as e:
        return ToolResult(title="creative_create 失败", output=str(e))


@tool(description="查询创意列表，支持按广告ID或状态过滤")
async def creative_get_list(
    adgroup_id: Optional[int] = None,
    creative_ids: Optional[List[int]] = None,
    configured_status: Optional[List[str]] = None,
    page: int = 1,
    page_size: int = 20,
    account_id: int = 0,
) -> ToolResult:
    """查询动态创意列表。

    Args:
        adgroup_id: 按广告ID过滤
        creative_ids: 按创意ID过滤
        configured_status: 按状态过滤
        page: 页码
        page_size: 每页数量
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    params: Dict[str, Any] = {"account_id": acct, "page": page, "page_size": page_size}

    filtering: Dict[str, Any] = {}
    if adgroup_id:
        filtering["adgroup_id"] = adgroup_id
    if creative_ids:
        filtering["dynamic_creative_id_list"] = creative_ids
    if configured_status:
        filtering["configured_status_list"] = configured_status
    if filtering:
        params["filtering"] = filtering

    try:
        resp = _get("/dynamic_creatives/get", params)
        data = _check(resp, "creative_get_list")
        items = data.get("list", [])
        page_info = data.get("page_info", {})
        output = f"共 {page_info.get('total_number', len(items))} 个创意"
        return ToolResult(title=f"查询创意列表（{len(items)}条）", output=output, metadata={"list": items})
    except Exception as e:
        return ToolResult(title="creative_get_list 失败", output=str(e))


@tool(description="更新创意状态或素材（对应API: /dynamic_creatives/update）")
async def creative_update(
    creative_id: int,
    creative_name: Optional[str] = None,
    configured_status: Optional[str] = None,
    title_list: Optional[List[str]] = None,
    image_id_list: Optional[List[str]] = None,
    account_id: int = 0,
) -> ToolResult:
    """更新动态创意。只传需要修改的字段。"""
    acct = account_id or DEFAULT_ACCOUNT_ID
    body: Dict[str, Any] = {"account_id": acct, "dynamic_creative_id": creative_id}
    if creative_name is not None:
        body["dynamic_creative_name"] = creative_name
    if configured_status is not None:
        body["configured_status"] = configured_status
    if title_list is not None:
        body["title_list"] = title_list
    if image_id_list is not None:
        body["image_id_list"] = image_id_list

    try:
        resp = _post("/dynamic_creatives/update", body)
        _check(resp, "creative_update")
        return ToolResult(title="创意更新成功", output=f"创意 {creative_id} 已更新")
    except Exception as e:
        return ToolResult(title="creative_update 失败", output=str(e))


# ===== 数据报表 =====

@tool(description="获取广告数据报表（消耗/点击/转化/CTR等），支持广告和创意两个维度")
async def ad_get_report(
    date_range: Dict[str, str],
    level: str = "adgroup",
    fields: Optional[List[str]] = None,
    adgroup_ids: Optional[List[int]] = None,
    group_by: Optional[List[str]] = None,
    page: int = 1,
    page_size: int = 100,
    account_id: int = 0,
) -> ToolResult:
    """查询广告数据报表。

    Args:
        date_range: {"start_date": "2026-04-01", "end_date": "2026-04-07"}
        level: 报表维度，"adgroup"（广告级）或 "dynamic_creative"（创意级）
        fields: 指标字段列表，默认 ["cost", "impression", "click", "ctr", "cpc", "cpm", "conversion", "cvr", "cpa"]
        adgroup_ids: 按广告ID过滤（可选）
        group_by: 额外分组维度，如 ["date", "adgroup_id"]
        page: 页码
        page_size: 每页数量
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    default_fields = ["cost", "impression", "click", "ctr", "cpc", "cpm", "conversion", "cvr", "cpa"]
    report_fields = fields or default_fields

    params: Dict[str, Any] = {
        "account_id": acct,
        "level": level.upper() if level == "adgroup" else "DYNAMIC_CREATIVE",
        "date_range": date_range,
        "fields": report_fields,
        "page": page,
        "page_size": page_size,
    }
    if group_by:
        params["group_by"] = group_by
    filtering: Dict[str, Any] = {}
    if adgroup_ids:
        filtering["adgroup_id_list"] = adgroup_ids
    if filtering:
        params["filtering"] = filtering

    # 报表 API 路径根据 level 不同
    path_map = {"adgroup": "/daily_reports/adgroups/get", "dynamic_creative": "/daily_reports/dynamic_creatives/get"}
    path = path_map.get(level, "/daily_reports/adgroups/get")

    try:
        resp = _get(path, params)
        data = _check(resp, "ad_get_report")
        items = data.get("list", [])

        if not items:
            return ToolResult(title="广告报表（无数据）", output="该时间段内无数据")

        # 格式化输出
        lines = [f"报表维度: {level}，时间: {date_range['start_date']} ~ {date_range['end_date']}"]
        for item in items[:10]:  # 最多显示10条
            cost = item.get("cost", 0)
            lines.append(
                f"- 广告{item.get('adgroup_id', '')}: "
                f"消耗{cost/100:.2f}元 "
                f"| 展示{item.get('impression', 0):,} "
                f"| 点击{item.get('click', 0):,} "
                f"| CTR{item.get('ctr', 0):.2%} "
                f"| 转化{item.get('conversion', 0)} "
                f"| CPA{item.get('cpa', 0)/100:.2f}元"
            )
        if len(items) > 10:
            lines.append(f"...共 {len(items)} 条，仅显示前10条")

        return ToolResult(title=f"广告报表（{len(items)}条）", output="\n".join(lines), metadata={"list": items})
    except Exception as e:
        return ToolResult(title="ad_get_report 失败", output=str(e))


@tool(description="获取单个创意的效果报表（CTR/CVR/消耗/转化），按日汇总")
async def creative_get_report(
    adcreative_id: int,
    date_range: Dict[str, str],
    fields: Optional[List[str]] = None,
    account_id: int = 0,
) -> ToolResult:
    """获取创意效果报告，用于素材衰退检测和优化决策。

    Args:
        adcreative_id: 创意ID（dynamic_creative_id）
        date_range: {"start_date": "2026-04-01", "end_date": "2026-04-07"}
        fields: 指标字段列表，默认 ["cost", "impression", "click", "ctr", "conversion", "cvr", "cpa"]
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    default_fields = ["cost", "impression", "click", "ctr", "conversion", "cvr", "cpa"]
    report_fields = fields or default_fields

    params: Dict[str, Any] = {
        "account_id": acct,
        "level": "DYNAMIC_CREATIVE",
        "date_range": date_range,
        "fields": report_fields,
        "filtering": {"dynamic_creative_id_list": [adcreative_id]},
        "group_by": ["date"],
    }

    try:
        resp = _get("/daily_reports/dynamic_creatives/get", params)
        data = _check(resp, "creative_get_report")
        items = data.get("list", [])

        if not items:
            return ToolResult(title="创意报表（无数据）", output=f"创意 {adcreative_id} 在该时间段无数据")

        lines = [f"创意 {adcreative_id} 报表：{date_range['start_date']} ~ {date_range['end_date']}"]
        total_cost = sum(r.get("cost", 0) for r in items)
        total_click = sum(r.get("click", 0) for r in items)
        total_conv = sum(r.get("conversion", 0) for r in items)
        avg_ctr = (total_click / max(sum(r.get("impression", 0) for r in items), 1))
        lines.append(
            f"汇总: 消耗{total_cost/100:.2f}元 | 点击{total_click:,} | 转化{total_conv} "
            f"| 均CTR{avg_ctr:.2%} | 均CPA{(total_cost/max(total_conv,1))/100:.2f}元"
        )
        for item in items:
            lines.append(
                f"  {item.get('date', '-')}: 消耗{item.get('cost', 0)/100:.2f}元"
                f" | CTR{item.get('ctr', 0):.2%}"
                f" | 转化{item.get('conversion', 0)}"
            )

        return ToolResult(
            title=f"创意报表（{len(items)}天）",
            output="\n".join(lines),
            metadata={"list": items, "adcreative_id": adcreative_id},
        )
    except Exception as e:
        return ToolResult(title="creative_get_report 失败", output=str(e))


# ===== 素材库 =====

@tool(description="查询账户素材库列表（图片/视频）")
async def asset_get_list(
    material_type: Optional[str] = None,
    page: int = 1,
    page_size: int = 20,
    account_id: int = 0,
) -> ToolResult:
    """查询账户下的素材库。

    Args:
        material_type: "IMAGE" 或 "VIDEO"，不传则查全部
        page: 页码
        page_size: 每页数量
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    params: Dict[str, Any] = {"account_id": acct, "page": page, "page_size": page_size}
    if material_type:
        params["material_type"] = material_type

    try:
        resp = _get("/material_infos/get", params)
        data = _check(resp, "asset_get_list")
        items = data.get("list", [])
        output = f"素材库共 {len(items)} 条：\n" + "\n".join(
            f"- [{m.get('material_id')}] {m.get('material_type')} {m.get('material_name', '')}"
            for m in items
        )
        return ToolResult(title=f"素材库（{len(items)}条）", output=output, metadata={"list": items})
    except Exception as e:
        return ToolResult(title="asset_get_list 失败", output=str(e))


# ===== 人群包 =====

@tool(description="查询账户下可用的自定义人群包列表")
async def audience_get_list(
    page: int = 1,
    page_size: int = 50,
    account_id: int = 0,
) -> ToolResult:
    """查询账户下的自定义人群包（用于 targeting.custom_audience 字段）。

    Args:
        page: 页码
        page_size: 每页数量
        account_id: 广告主账号ID
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    params: Dict[str, Any] = {"account_id": acct, "page": page, "page_size": page_size}

    try:
        resp = _get("/custom_audiences/get", params)
        data = _check(resp, "audience_get_list")
        items = data.get("list", [])
        output = f"共 {len(items)} 个人群包：\n" + "\n".join(
            f"- [{a.get('audience_id')}] {a.get('name')} "
            f"| 状态:{a.get('status')} "
            f"| 人数:{a.get('user_count', 0):,}"
            for a in items
        )
        return ToolResult(title=f"人群包列表（{len(items)}个）", output=output, metadata={"list": items})
    except Exception as e:
        return ToolResult(title="audience_get_list 失败", output=str(e))


# ===== 账户信息 =====

@tool(description="获取广告账户基本信息（余额、日限额、账户状态等）")
async def account_get_info(account_id: int = 0) -> ToolResult:
    """获取广告账户基本信息。

    Args:
        account_id: 广告主账号ID，0则使用环境变量
    """
    acct = account_id or DEFAULT_ACCOUNT_ID
    params: Dict[str, Any] = {
        "account_id": acct,
        "fields": ["balance", "daily_budget", "configured_status"],
    }

    try:
        resp = _get("/accounts/get", params)
        data = _check(resp, "account_get_info")
        items = data.get("list", [data])
        info = items[0] if items else {}
        balance = info.get("balance", 0)
        output = (
            f"账户 {acct} 信息：\n"
            f"- 余额：{balance/100:.2f} 元\n"
            f"- 日限额：{info.get('daily_budget', 0)/100:.0f} 元\n"
            f"- 状态：{info.get('configured_status', '未知')}"
        )
        return ToolResult(title="账户信息", output=output, metadata=info)
    except Exception as e:
        return ToolResult(title="account_get_info 失败", output=str(e))