""" 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")