|
|
@@ -0,0 +1,702 @@
|
|
|
+"""
|
|
|
+腾讯广告 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 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 # 秒
|
|
|
+
|
|
|
+
|
|
|
+def _get_access_token() -> str:
|
|
|
+ token = os.getenv("TENCENT_AD_ACCESS_TOKEN", "")
|
|
|
+ if not token:
|
|
|
+ raise ValueError("未配置 TENCENT_AD_ACCESS_TOKEN 环境变量")
|
|
|
+ return token
|
|
|
+
|
|
|
+
|
|
|
+def _common_params() -> Dict[str, str]:
|
|
|
+ """公共查询参数:access_token / timestamp / nonce"""
|
|
|
+ return {
|
|
|
+ "access_token": _get_access_token(),
|
|
|
+ "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 参数传递。
|
|
|
+ """
|
|
|
+ query = dict(_common_params())
|
|
|
+ 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 _post(path: str, body: Dict[str, Any]) -> Dict[str, Any]:
|
|
|
+ """
|
|
|
+ 发送 POST 请求。
|
|
|
+ 公共参数在 URL query,业务参数在 JSON body。
|
|
|
+ """
|
|
|
+ query = urlencode(_common_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 {}
|
|
|
+
|
|
|
+
|
|
|
+# ===== 广告(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))
|