Spaces:
Runtime error
Runtime error
| """ | |
| 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") | |