Blame for src/google_api.py

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