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 concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, Iterable, 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 = int(os.getenv("TENCENT_AD_TIMEOUT_SECONDS", "30"))

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

# Token 缓存：避免每次 API 调用都重新获取
_token_cache: Dict[int, Dict[str, Any]] = {}
_user_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 prefetch_access_tokens(
    account_ids: Iterable[int],
    max_workers: Optional[int] = None,
) -> Dict[int, str]:
    """Fetch distinct account access tokens concurrently and warm the cache."""

    account_set = set()
    for raw_value in account_ids:
        try:
            account_id = int(raw_value)
        except (TypeError, ValueError):
            continue
        if account_id > 0:
            account_set.add(account_id)
    accounts = sorted(account_set)
    if not accounts:
        return {}
    if max_workers is None:
        max_workers = int(os.getenv("TENCENT_AD_TOKEN_PREFETCH_WORKERS", "8"))
    if max_workers < 1:
        raise ValueError("TENCENT_AD_TOKEN_PREFETCH_WORKERS must be at least 1")
    workers = min(max_workers, len(accounts), 32)

    tokens: Dict[int, str] = {}
    errors: Dict[int, str] = {}
    with ThreadPoolExecutor(
        max_workers=workers,
        thread_name_prefix="tencent-token",
    ) as executor:
        futures = {
            executor.submit(_get_access_token, account_id): account_id
            for account_id in accounts
        }
        for future in as_completed(futures):
            account_id = futures[future]
            try:
                token = future.result()
                tokens[account_id] = token
                _token_cache[account_id] = {"token": token, "ts": time.time()}
            except Exception as exc:
                errors[account_id] = str(exc)

    if errors:
        logger.warning(
            "[TokenAPI] 并发预取部分失败 success=%d failed=%d accounts=%s",
            len(tokens),
            len(errors),
            sorted(errors),
        )
    logger.info(
        "[TokenAPI] 并发预取完成 accounts=%d success=%d workers=%d",
        len(accounts),
        len(tokens),
        workers,
    )
    return tokens


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 参数传递。
    """
    attempts = int(os.getenv("TENCENT_AD_GET_RETRY_ATTEMPTS", "3"))
    backoff_seconds = float(
        os.getenv("TENCENT_AD_GET_RETRY_BACKOFF_SECONDS", "1")
    )
    if attempts < 1:
        raise ValueError("TENCENT_AD_GET_RETRY_ATTEMPTS must be at least 1")
    if backoff_seconds < 0:
        raise ValueError(
            "TENCENT_AD_GET_RETRY_BACKOFF_SECONDS must not be negative"
        )

    account_id = params.get("account_id", 0)
    for attempt in range(1, attempts + 1):
        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)
        try:
            resp = httpx.get(url, timeout=TIMEOUT)
            resp.raise_for_status()
            return resp.json()
        except httpx.RequestError as exc:
            retryable = True
            error = exc
        except httpx.HTTPStatusError as exc:
            status_code = exc.response.status_code
            retryable = status_code in {408, 429} or status_code >= 500
            error = exc

        if not retryable or attempt >= attempts:
            raise error
        delay = backoff_seconds * (2 ** (attempt - 1))
        logger.warning(
            "[TencentAPI] GET 临时失败，准备重试 "
            "account=%s path=%s attempt=%d/%d delay=%.1fs error=%s",
            account_id,
            path,
            attempt,
            attempts,
            delay,
            error,
        )
        if delay > 0:
            time.sleep(delay)

    raise RuntimeError("Tencent GET retry loop exited unexpectedly")


def _get_user_token_for_account(account_id: int) -> str:
    """动态获取 user_token,接口不可用时 fallback 到 DB/环境变量。

    DB 设计原因(2026-06-05):每个账户可能有不同的经办人授权 token,
    全局环境变量无法覆盖多账户场景。
    """
    cached = _user_token_cache.get(account_id)
    if cached and time.time() - cached["ts"] < _TOKEN_CACHE_TTL:
        return cached["token"]

    try:
        resp = httpx.get(
            USER_TOKEN_API_URL,
            params={"accountId": account_id},
            timeout=10,
        )
        resp.raise_for_status()
        token = resp.text.strip()
        if token and len(token) > 10:
            _user_token_cache[account_id] = {"token": token, "ts": time.time()}
            logger.info(
                "[UserTokenAPI] 已获取 account=%s 的 user_token (缓存30分钟)",
                account_id,
            )
            return token
        logger.warning("[UserTokenAPI] 返回内容异常,降级使用 DB/env")
    except Exception as e:
        logger.warning("[UserTokenAPI] 请求失败,降级使用 DB/env: %s", e)

    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（实名认证令牌）
    优先级:动态 user token API > DB account_whitelist.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))