""" 腾讯广告 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 访问令牌 TENCENT_AD_ACCOUNT_ID 默认广告账户 ID（可被参数覆盖） TENCENT_AD_BASE_URL API 基础地址，默认 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]: """并发读取去重后的账户 access token，并预热缓存。""" 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))