#!/usr/bin/env python3 """ActionKey read-only starter. Python 3.9+, no third-party dependencies. Run `python3 actionkey_client.py --demo accounts` without a credential or network. Then discover a plan with `--demo campaigns --ad-account-id 7`. Live calls read ACTIONKEY_TOKEN or ask for it without echoing. No write API exists in this example. The report endpoint does not document a date-range parameter. """ import argparse from datetime import datetime, timezone import getpass import json import os from pathlib import Path import re import socket import sys from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import HTTPRedirectHandler, Request, build_opener BASE_URL = "https://baiz.ai/api/mcp" MAX_RESPONSE_BYTES = 2 * 1024 * 1024 ACCOUNT_FIELDS = ("id", "ad_account_id", "name", "account_status", "currency", "timezone_name", "balance", "spend") CAMPAIGN_FIELDS = ("id", "name", "status", "budget_type", "limit_amount", "limit_daily_amount") CAMPAIGN_LIST_FIELDS = ("id", "name", "status") METRIC_FIELDS = ("total_ads", "total_spend", "total_impressions", "total_clicks", "total_link_clicks", "total_reach", "total_purchase", "cpm", "cpc", "ctr", "cpa") class APIError(Exception): """A request or API response cannot be accepted as a successful result.""" class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise APIError("接口发生重定向，已停止；请核对服务地址。") class ActionKeyClient: def __init__(self, token, opener=None): if not token or "\n" in token or "\r" in token: raise APIError("需要有效的 ActionKey 服务 Token。") self.token = token self.opener = opener or build_opener(NoRedirect()) def get(self, path, params=None): # Restrict this starter to documented read-only paths on one fixed host. if not (path in {"/ad-accounts", "/campaigns"} or re.fullmatch(r"/campaigns/[1-9][0-9]*/ads", path)): raise APIError("示例只支持广告账户、计划列表和计划效果查询。") url = BASE_URL + path if params: url += "?" + urlencode(params) request = Request(url, method="GET", headers={ "Authorization": "Bearer " + self.token, "Accept": "application/json", }) try: with self.opener.open(request, timeout=20) as response: raw = response.read(MAX_RESPONSE_BYTES + 1) except HTTPError as error: # Do not echo response bodies that may include private request data. raise APIError(f"HTTP {error.code}；请核对权限、服务状态或接口限制。") from None except (URLError, TimeoutError, socket.timeout): raise APIError("网络失败或超时；本次没有取得可核对的查询结果。") from None if len(raw) > MAX_RESPONSE_BYTES: raise APIError("响应超过示例读取上限；请缩小查询条数。") try: payload = json.loads(raw) except (ValueError, UnicodeDecodeError): raise APIError("返回内容不是有效 JSON。") from None return accept_payload(payload) def accept_payload(payload): if not isinstance(payload, dict) or payload.get("status") != "success": raise APIError("接口未返回 status=success；请在受控后台排查业务错误。") if "data" not in payload: raise APIError("成功响应缺少 data，不能当作有效查询结果。") return payload["data"] def select_fields(value, fields): if not isinstance(value, dict): raise APIError("数据结构不符；请按实际接口响应更新解析器。") return {field: value[field] for field in fields if field in value} def scrub(value, token=""): """Avoid printing credential keys and the credential used for this call.""" if isinstance(value, dict): return {key: "[已隐藏]" if key.lower() in {"token", "access_token", "authorization", "password"} else scrub(item, token) for key, item in value.items()} if isinstance(value, list): return [scrub(item, token) for item in value] if isinstance(value, str) and token: return value.replace(token, "[已隐藏]") return value def format_accounts(data): # Array and an object containing a `data` array are accepted explicitly. rows = data.get("data") if isinstance(data, dict) else data if not isinstance(rows, list): raise APIError("账户列表结构不符；请核对实际返回的列表字段。") return {"accounts": [select_fields(row, ACCOUNT_FIELDS) for row in rows], "scope": "本次返回的一页账户，不能据此宣称已列出全部账户。"}, False def format_campaigns(data): rows = data.get("data") if isinstance(data, dict) else data if not isinstance(rows, list): raise APIError("计划列表结构不符；请核对实际返回的列表字段。") campaigns = [] for row in rows: selected = select_fields(row, CAMPAIGN_LIST_FIELDS) if not selected.get("id"): raise APIError("计划列表项缺少内部 ID，不能继续查询该计划。") campaigns.append(selected) return {"campaigns": campaigns, "id_note": "id 是 ActionKey 计划内部 ID，用于 report --campaign-id。", "scope": "本次返回的一页计划，不能据此宣称已列出全部计划。"}, False def format_report(data): if not isinstance(data, dict) or not isinstance(data.get("metrics"), dict): raise APIError("响应缺少 metrics，不能生成效果报告。") ads = data.get("ads") if not isinstance(ads, dict) or not isinstance(ads.get("data"), list): raise APIError("响应缺少广告列表或分页信息。") errors = data.get("errors") report = { "campaign": select_fields(data.get("campaign"), CAMPAIGN_FIELDS), "metrics": select_fields(data["metrics"], METRIC_FIELDS), "spend_currency": "USD（公开接口 total_spend 的单位）", "reporting_window": "此端点未公开日期过滤参数；请向团队确认统计口径。", "pagination": select_fields(ads, ("total", "per_page", "current_page")), "returned_ads": len(ads["data"]), "partial_failure": bool(errors), "errors": errors or [], "budget_note": "计划字段不能代替关联广告户实际预算的独立回读。", } return report, bool(errors) def positive_int(value): number = int(value) if number <= 0: raise argparse.ArgumentTypeError("需要大于 0 的整数。") return number def main(argv=None): parser = argparse.ArgumentParser(description="ActionKey 只读查询示例") parser.add_argument("--demo", action="store_true", help="使用旁边的示意数据，不联网") sub = parser.add_subparsers(dest="command", required=True) accounts = sub.add_parser("accounts", help="查询一页可访问的广告账户") accounts.add_argument("--name", help="按账户名称模糊搜索") accounts.add_argument("--per-page", type=positive_int, default=20) campaigns = sub.add_parser("campaigns", help="按广告户内部 ID 查询一页计划") campaigns.add_argument("--ad-account-id", type=positive_int, required=True, help="账户查询返回的 id，非 Facebook 广告户 ID") campaigns.add_argument("--per-page", type=positive_int, default=20) report = sub.add_parser("report", help="查询一个 ActionKey 计划的效果") report.add_argument("--campaign-id", type=positive_int, required=True, help="ActionKey 计划内部 ID，非 Facebook Campaign ID") report.add_argument("--page", type=positive_int, default=1) report.add_argument("--per-page", type=positive_int, default=50) args = parser.parse_args(argv) token = "" try: paths = {"accounts": "/ad-accounts", "campaigns": "/campaigns"} path = paths.get(args.command) if args.command == "report": path = f"/campaigns/{args.campaign_id}/ads" params = {"per_page": args.per_page} if args.command == "accounts" and args.name: params["name"] = args.name if args.command == "campaigns": params["ad_account_id"] = args.ad_account_id if args.command == "report": params["page"] = args.page if args.demo: if args.command == "campaigns" and args.ad_account_id != 7: raise APIError("示意数据仅包含广告户内部 ID 7 的计划。") if args.command == "report" and (args.campaign_id != 42 or args.page != 1): raise APIError("示意数据仅包含计划内部 ID 42 的第 1 页。") fixture = Path(__file__).with_name("demo-data.json") data = accept_payload(json.loads(fixture.read_text(encoding="utf-8"))[args.command]) else: token = os.environ.get("ACTIONKEY_TOKEN", "") or getpass.getpass("ActionKey 服务 Token（输入不回显）：") data = ActionKeyClient(token).get(path, params) formatters = {"accounts": format_accounts, "campaigns": format_campaigns, "report": format_report} result, partial = formatters[args.command](data) output = {"demo": args.demo, "queried_at": datetime.now(timezone.utc).isoformat(), "endpoint": path, "query": params, "result": scrub(result, token)} print(json.dumps(output, ensure_ascii=False, indent=2)) return 2 if partial else 0 except (APIError, OSError, ValueError) as error: message = str(error).replace(token, "[已隐藏]") if token else str(error) print("查询失败：" + message, file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())