Spaces:
Runtime error
Runtime error
| """ | |
| Google Ads API 数据拉取引擎。 | |
| 通过 REST API 拉取 campaign 级日度数据，转换为与 Google Sheets 相同的列结构。 | |
| """ | |
| import json | |
| import logging | |
| import os | |
| from datetime import date, timedelta | |
| from typing import Any | |
| import pandas as pd | |
| import requests | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| log = logging.getLogger("ad-agent") | |
| API_VERSION = "v23" | |
| BASE_URL = f"https://googleads.googleapis.com/{API_VERSION}" | |
| CHANNEL_TYPE_MAP = { | |
| "SEARCH": "Conversion", | |
| "SHOPPING": "Conversion", | |
| "PERFORMANCE_MAX": "Conversion", | |
| "DISPLAY": "Awareness", | |
| "VIDEO": "Consideration", | |
| "MULTI_CHANNEL": "Conversion", | |
| "DEMAND_GEN": "Consideration", | |
| "LOCAL": "Conversion", | |
| "SMART": "Conversion", | |
| "DISCOVERY": "Consideration", | |
| } | |
| GAQL_QUERY = """ | |
| SELECT | |
| campaign.name, | |
| campaign.id, | |
| campaign.advertising_channel_type, | |
| metrics.cost_micros, | |
| metrics.impressions, | |
| metrics.clicks, | |
| metrics.ctr, | |
| metrics.average_cpc, | |
| metrics.average_cpm, | |
| metrics.conversions, | |
| metrics.conversions_value, | |
| segments.date | |
| FROM campaign | |
| WHERE segments.date BETWEEN '{since}' AND '{until}' | |
| AND metrics.impressions > 0 | |
| ORDER BY segments.date DESC, campaign.name | |
| """ | |
| def _get_access_token() -> str: | |
| resp = requests.post("https://oauth2.googleapis.com/token", data={ | |
| "client_id": os.getenv("GOOGLE_ADS_CLIENT_ID"), | |
| "client_secret": os.getenv("GOOGLE_ADS_CLIENT_SECRET"), | |
| "refresh_token": os.getenv("GOOGLE_ADS_REFRESH_TOKEN"), | |
| "grant_type": "refresh_token", | |
| }) | |
| token = resp.json().get("access_token") | |
| if not token: | |
| raise RuntimeError(f"Google OAuth token 刷新失败: {resp.text[:200]}") | |
| return token | |
| def _fetch_campaigns( | |
| customer_id: str, access_token: str, since: str, until: str | |
| ) -> list[dict]: | |
| """分页拉取一个广告账户的 campaign 级日度数据。""" | |
| login_id = os.getenv("GOOGLE_ADS_LOGIN_CUSTOMER_ID", "") | |
| dev_token = os.getenv("GOOGLE_ADS_DEVELOPER_TOKEN", "") | |
| headers = { | |
| "Authorization": f"Bearer {access_token}", | |
| "developer-token": dev_token, | |
| "Content-Type": "application/json", | |
| } | |
| if login_id: | |
| headers["login-customer-id"] = login_id | |
| query = GAQL_QUERY.format(since=since, until=until) | |
| all_results: list[dict] = [] | |
| page_token = None | |
| while True: | |
| body: dict[str, Any] = {"query": query} | |
| if page_token: | |
| body["pageToken"] = page_token | |
| url = f"{BASE_URL}/customers/{customer_id}/googleAds:search" | |
| resp = requests.post(url, headers=headers, json=body) | |
| if resp.status_code != 200: | |
| detail = resp.text[:800] | |
| log.error(f"Google Ads API error ({resp.status_code}): {detail}") | |
| raise RuntimeError(f"Google Ads API 请求失败 ({resp.status_code}): {detail}") | |
| data = resp.json() | |
| all_results.extend(data.get("results", [])) | |
| page_token = data.get("nextPageToken") | |
| if not page_token: | |
| break | |
| return all_results | |
| def _parse_row(row: dict, country: str) -> dict[str, Any]: | |
| """将一条 Google Ads API 返回数据转换为与 Google Sheets 相同的列结构。""" | |
| campaign = row.get("campaign", {}) | |
| metrics = row.get("metrics", {}) | |
| segments = row.get("segments", {}) | |
| cost = int(metrics.get("costMicros", 0)) / 1_000_000 | |
| impressions = int(metrics.get("impressions", 0)) | |
| clicks = int(metrics.get("clicks", 0)) | |
| ctr = float(metrics.get("ctr", 0)) | |
| avg_cpc = int(metrics.get("averageCpc", 0)) / 1_000_000 | |
| avg_cpm = int(metrics.get("averageCpm", 0)) / 1_000_000 | |
| conversions = float(metrics.get("conversions", 0)) | |
| conv_value = float(metrics.get("conversionsValue", 0)) | |
| video_views = int(metrics.get("videoViews", 0)) | |
| bounce_rate = None | |
| roas = conv_value / cost if cost > 0 else None | |
| cvr = conversions / clicks if clicks > 0 else None | |
| cplpv = cost / clicks if clicks > 0 else None | |
| campaign_name = campaign.get("name", "") | |
| name_parts = campaign_name.split("_") | |
| name_funnel = name_parts[1] if len(name_parts) > 1 else "" | |
| if name_funnel in ("Awareness", "Consideration", "Conversion"): | |
| campaign_type = name_funnel | |
| else: | |
| channel_type = campaign.get("advertisingChannelType", "") | |
| campaign_type = CHANNEL_TYPE_MAP.get(channel_type, "Conversion") | |
| name_lower = campaign_name.lower() | |
| if "awareness" in name_lower or "aw_" in name_lower: | |
| campaign_type = "Awareness" | |
| elif "consideration" in name_lower or "traffic" in name_lower: | |
| campaign_type = "Consideration" | |
| elif "conversion" in name_lower or "purchase" in name_lower or "pmax" in name_lower: | |
| campaign_type = "Conversion" | |
| return { | |
| "country": country, | |
| "platform": "Google", | |
| "campaign_type": campaign_type, | |
| "campaign_name": campaign_name, | |
| "date": segments.get("date", ""), | |
| "Cost": cost, | |
| "Ads.Sales": conv_value, | |
| "Ads.Sales(1d+7d)": None, | |
| "SPF.Sales": None, | |
| "ROAS": roas, | |
| "GA4.Sessions": clicks, | |
| "SPF.Sessions": None, | |
| "LPV": clicks, | |
| "CTR": ctr, | |
| "CVR": cvr, | |
| "CPC": avg_cpc, | |
| "CPSE": None, | |
| "CPLPV": cplpv, | |
| "CPM": avg_cpm, | |
| "Impressions": impressions, | |
| "Bounce Rate": bounce_rate, | |
| "Average Duration": None, | |
| "ATC": None, | |
| "CP-ATC": None, | |
| "Email": None, | |
| "CP-Email": None, | |
| "Video Views": video_views if video_views > 0 else None, | |
| "CP-Video Views": (cost / video_views) if video_views > 0 else None, | |
| "50% Play Rate": None, | |
| "100% Play Rate": None, | |
| "Hook Rate": None, | |
| } | |
| def fetch_google_data( | |
| since: date | None = None, | |
| until: date | None = None, | |
| days_back: int = 90, | |
| ) -> pd.DataFrame: | |
| """ | |
| 从所有配置的 Google Ads 账户拉取数据，返回与 Google Sheets 相同结构的 DataFrame。 | |
| """ | |
| refresh_token = os.getenv("GOOGLE_ADS_REFRESH_TOKEN") | |
| if not refresh_token: | |
| raise RuntimeError("未配置 GOOGLE_ADS_REFRESH_TOKEN") | |
| accounts_str = os.getenv("GOOGLE_ADS_CUSTOMER_IDS", "") | |
| if not accounts_str: | |
| raise RuntimeError("未配置 GOOGLE_ADS_CUSTOMER_IDS (格式: customer_id:country,...)") | |
| accounts = [] | |
| for pair in accounts_str.split(","): | |
| pair = pair.strip() | |
| if ":" in pair: | |
| cid, country = pair.split(":", 1) | |
| accounts.append((cid.strip(), country.strip())) | |
| if not accounts: | |
| raise RuntimeError("GOOGLE_ADS_CUSTOMER_IDS 格式错误") | |
| if until is None: | |
| until = date.today() - timedelta(days=1) | |
| if since is None: | |
| since = until - timedelta(days=days_back) | |
| since_str = since.strftime("%Y-%m-%d") | |
| until_str = until.strftime("%Y-%m-%d") | |
| access_token = _get_access_token() | |
| all_parsed: list[dict] = [] | |
| for customer_id, country in accounts: | |
| log.info(f"拉取 Google Ads 数据: customer={customer_id} country={country} {since_str}~{until_str}") | |
| try: | |
| rows = _fetch_campaigns(customer_id, access_token, since_str, until_str) | |
| log.info(f" 获取 {len(rows)} 条记录") | |
| for row in rows: | |
| parsed = _parse_row(row, country) | |
| all_parsed.append(parsed) | |
| except Exception as e: | |
| log.error(f" 拉取失败: {e}") | |
| raise | |
| if not all_parsed: | |
| return pd.DataFrame() | |
| df = pd.DataFrame(all_parsed) | |
| df["date"] = pd.to_datetime(df["date"]) | |
| return df | |
| if __name__ == "__main__": | |
| logging.basicConfig(level=logging.INFO) | |
| df = fetch_google_data(days_back=7) | |
| print(f"\n拉取到 {len(df)} 条数据") | |
| print(f"日期范围: {df['date'].min()} ~ {df['date'].max()}") | |
| print(f"国家: {df['country'].unique()}") | |
| print(f"Campaign Types: {df['campaign_type'].unique()}") | |
| print(f"\n汇总:") | |
| print(f" 总花费: ${df['Cost'].sum():,.2f}") | |
| print(f" 总销售: ${df['Ads.Sales'].sum():,.2f}") | |
| total_cost = df["Cost"].sum() | |
| total_sales = df["Ads.Sales"].sum() | |
| print(f" ROAS: {total_sales / total_cost:.2f}" if total_cost > 0 else " ROAS: N/A") | |