Blame for src/criteo_api.py

"""
Criteo Marketing API 数据拉取引擎。
通过 Client Credentials 认证，拉取 campaign 级日度数据，
转换为与 Google Sheets 相同的列结构。
"""
import logging
import os
import re
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")

TOKEN_URL = "https://api.criteo.com/oauth2/token"
API_VERSION = "2026-01"
REPORT_URL = f"https://api.criteo.com/{API_VERSION}/statistics/report"

REPORT_DIMENSIONS = ["CampaignId", "Campaign", "Day"]
REPORT_METRICS = [
    "Displays",
    "Clicks",
    "AdvertiserCost",
    "RevenueGeneratedClientAttribution",
    "SalesClientAttribution",
]

CAMPAIGN_TYPE_KEYWORDS = {
    "Awareness": ["awareness", "branding", "reach"],
    "Consideration": ["consideration", "pageview", "traffic", "lal", "lookalike"],
    "Conversion": ["conversion", "purchase", "retarget", "rt", "retention", "asc"],
}


def _infer_campaign_type(name: str) -> str:
    lower = name.lower()
    for ctype, keywords in CAMPAIGN_TYPE_KEYWORDS.items():
        if any(kw in lower for kw in keywords):
            return ctype
    return "Conversion"


def _get_token() -> str:
    client_id = os.getenv("CRITEO_CLIENT_ID")
    client_secret = os.getenv("CRITEO_CLIENT_SECRET")
    if not client_id or not client_secret:
        raise RuntimeError("未配置 CRITEO_CLIENT_ID / CRITEO_CLIENT_SECRET")

    resp = requests.post(TOKEN_URL, data={
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
    })
    if resp.status_code != 200:
        raise RuntimeError(f"Criteo token 获取失败 ({resp.status_code}): {resp.text[:300]}")
    return resp.json()["access_token"]


def _fetch_report(
    token: str, advertiser_id: str, since: str, until: str,
) -> list[dict]:
    resp = requests.post(
        REPORT_URL,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        },
        json={
            "advertiserIds": advertiser_id,
            "startDate": since,
            "endDate": until,
            "dimensions": REPORT_DIMENSIONS,
            "metrics": REPORT_METRICS,
            "format": "JSON",
            "currency": "USD",
            "timezone": "UTC",
        },
    )
    if resp.status_code != 200:
        raise RuntimeError(f"Criteo report 请求失败 ({resp.status_code}): {resp.text[:500]}")

    data = resp.json()
    rows = data.get("Rows") or data if isinstance(data, list) else data.get("Rows", [])
    return rows


def _parse_row(row: dict, country: str) -> dict[str, Any]:
    campaign_name = row.get("Campaign", "")
    cost = float(row.get("AdvertiserCost") or 0)
    impressions = int(float(row.get("Displays") or 0))
    clicks = int(float(row.get("Clicks") or 0))
    revenue = float(row.get("RevenueGeneratedClientAttribution") or 0)
    conversions = int(float(row.get("SalesClientAttribution") or 0))

    ctr = clicks / impressions if impressions > 0 else None
    cpc = cost / clicks if clicks > 0 else None
    cpm = (cost / impressions * 1000) if impressions > 0 else None
    roas = revenue / cost if cost > 0 else None
    cvr = conversions / clicks if clicks > 0 else None

    return {
        "country": country,
        "platform": "Criteo",
        "campaign_type": _infer_campaign_type(campaign_name),
        "campaign_name": campaign_name,
        "date": row.get("Day", ""),
        "Cost": cost,
        "Ads.Sales": revenue,
        "Ads.Sales(1d+7d)": None,
        "SPF.Sales": None,
        "ROAS": roas,
        "GA4.Sessions": None,
        "SPF.Sessions": None,
        "LPV": None,
        "CTR": ctr,
        "CVR": cvr,
        "CPC": cpc,
        "CPSE": None,
        "CPLPV": None,
        "CPM": cpm,
        "Impressions": impressions,
        "Bounce Rate": None,
        "Average Duration": None,
        "ATC": None,
        "CP-ATC": None,
        "Email": None,
        "CP-Email": None,
        "Video Views": None,
        "CP-Video Views": None,
        "50% Play Rate": None,
        "100% Play Rate": None,
        "Hook Rate": None,
    }


def fetch_criteo_data(
    since: date | None = None,
    until: date | None = None,
    days_back: int = 90,
) -> pd.DataFrame:
    """
    从所有配置的 Criteo 广告主拉取数据，返回标准化 DataFrame。
    环境变量 CRITEO_ADVERTISER_IDS 格式: advertiser_id:country,...
    """
    accounts_str = os.getenv("CRITEO_ADVERTISER_IDS", "")
    if not accounts_str:
        raise RuntimeError("未配置 CRITEO_ADVERTISER_IDS (格式: id:country,...)")

    accounts = []
    for pair in accounts_str.split(","):
        pair = pair.strip()
        if ":" in pair:
            aid, country = pair.split(":", 1)
            accounts.append((aid.strip(), country.strip()))

    if not accounts:
        raise RuntimeError("CRITEO_ADVERTISER_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")

    token = _get_token()

    all_parsed: list[dict] = []
    for advertiser_id, country in accounts:
        log.info(f"拉取 Criteo 数据: advertiser={advertiser_id} country={country} {since_str}~{until_str}")
        try:
            rows = _fetch_report(token, advertiser_id, 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_criteo_data(days_back=7)
    print(f"\n拉取到 {len(df)} 条数据")
    print(f"日期范围: {df['date'].min()} ~ {df['date'].max()}")
    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")