Spaces:
Runtime error
Runtime error
| """ | |
| Meta Marketing API 数据拉取引擎。 | |
| 从 Meta Ads API 拉取 ad（素材）级日度数据，转换为统一的列结构。 | |
| 支持 campaign / adset / ad 三个粒度，默认拉取 ad 级别。 | |
| 注意：部分云环境（如 HF Spaces）可能无法直连 graph.facebook.com， | |
| 此时 Meta 数据将依赖 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 = "v21.0" | |
| BASE_URL = f"https://graph.facebook.com/{API_VERSION}" | |
| OBJECTIVE_MAP = { | |
| "OUTCOME_AWARENESS": "Awareness", | |
| "BRAND_AWARENESS": "Awareness", | |
| "REACH": "Awareness", | |
| "OUTCOME_ENGAGEMENT": "Consideration", | |
| "OUTCOME_TRAFFIC": "Consideration", | |
| "TRAFFIC": "Consideration", | |
| "POST_ENGAGEMENT": "Consideration", | |
| "VIDEO_VIEWS": "Consideration", | |
| "LINK_CLICKS": "Consideration", | |
| "OUTCOME_SALES": "Conversion", | |
| "OUTCOME_LEADS": "Conversion", | |
| "CONVERSIONS": "Conversion", | |
| "PRODUCT_CATALOG_SALES": "Conversion", | |
| } | |
| INSIGHTS_FIELDS = [ | |
| "campaign_name", | |
| "campaign_id", | |
| "adset_name", | |
| "adset_id", | |
| "ad_name", | |
| "ad_id", | |
| "objective", | |
| "spend", | |
| "impressions", | |
| "clicks", | |
| "cpc", | |
| "cpm", | |
| "ctr", | |
| "actions", | |
| "action_values", | |
| "video_30_sec_watched_actions", | |
| "video_p50_watched_actions", | |
| "video_p100_watched_actions", | |
| ] | |
| def _get_action_value(actions: list[dict], action_type: str) -> float | None: | |
| if not actions: | |
| return None | |
| for a in actions: | |
| if a["action_type"] == action_type: | |
| return float(a["value"]) | |
| return None | |
| def _fetch_insights( | |
| account_id: str, token: str, since: str, until: str, | |
| level: str = "ad", | |
| ) -> list[dict]: | |
| """分页拉取一个广告账户的日度数据。level 可选 campaign / adset / ad。""" | |
| url = f"{BASE_URL}/act_{account_id}/insights" | |
| params = { | |
| "access_token": token, | |
| "fields": ",".join(INSIGHTS_FIELDS), | |
| "level": level, | |
| "time_range": json.dumps({"since": since, "until": until}), | |
| "time_increment": 1, | |
| "limit": 500, | |
| } | |
| all_rows: list[dict] = [] | |
| while url: | |
| try: | |
| resp = requests.get(url, params=params, timeout=60) | |
| except requests.exceptions.ConnectionError as e: | |
| if "NameResolutionError" in str(e) or "getaddrinfo" in str(e): | |
| raise RuntimeError( | |
| "无法连接 graph.facebook.com（DNS 解析失败）。" | |
| "当前云环境可能限制了 Facebook 域名访问。" | |
| "Meta 数据请通过 Google Sheets 手动更新，或在无网络限制的环境中运行同步。" | |
| ) from e | |
| raise | |
| if resp.status_code != 200: | |
| log.error(f"Meta API error ({resp.status_code}): {resp.text[:300]}") | |
| raise RuntimeError(f"Meta API 请求失败: {resp.status_code}") | |
| data = resp.json() | |
| all_rows.extend(data.get("data", [])) | |
| paging = data.get("paging", {}) | |
| url = paging.get("next") | |
| params = {} | |
| return all_rows | |
| def _parse_row(row: dict, country: str) -> dict[str, Any]: | |
| """将一条 Meta API 返回数据转换为与 Google Sheets 相同的列结构。""" | |
| actions = row.get("actions") or [] | |
| action_vals = row.get("action_values") or [] | |
| spend = float(row.get("spend") or 0) | |
| impressions = int(row.get("impressions") or 0) | |
| clicks = int(row.get("clicks") or 0) | |
| ctr_raw = float(row.get("ctr") or 0) | |
| purchase_revenue = _get_action_value(action_vals, "omni_purchase") or _get_action_value(action_vals, "purchase") or 0 | |
| purchase_count = _get_action_value(actions, "omni_purchase") or _get_action_value(actions, "purchase") or 0 | |
| atc = _get_action_value(actions, "omni_add_to_cart") or _get_action_value(actions, "add_to_cart") or 0 | |
| lpv = _get_action_value(actions, "landing_page_view") or 0 | |
| video_views = _get_action_value(actions, "video_view") or 0 | |
| link_clicks = _get_action_value(actions, "link_click") or 0 | |
| roas = purchase_revenue / spend if spend > 0 else None | |
| cpc = float(row.get("cpc") or 0) | |
| cpm = float(row.get("cpm") or 0) | |
| ctr = ctr_raw / 100 # Meta returns CTR as percentage, convert to decimal | |
| cvr = purchase_count / link_clicks if link_clicks > 0 else None | |
| cp_atc = spend / atc if atc > 0 else None | |
| cplpv = spend / lpv if lpv > 0 else None | |
| cp_video = spend / video_views if video_views > 0 else None | |
| video_p50 = row.get("video_p50_watched_actions") | |
| video_p100 = row.get("video_p100_watched_actions") | |
| p50_count = float(video_p50[0]["value"]) if video_p50 else 0 | |
| p100_count = float(video_p100[0]["value"]) if video_p100 else 0 | |
| play_50_rate = p50_count / impressions if impressions > 0 and p50_count else None | |
| play_100_rate = p100_count / impressions if impressions > 0 and p100_count else None | |
| hook_rate = (video_views / impressions) if impressions > 0 and video_views > 0 else None | |
| campaign_name = row.get("campaign_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: | |
| objective = row.get("objective", "") | |
| campaign_type = OBJECTIVE_MAP.get(objective, "Conversion") | |
| return { | |
| "country": country, | |
| "platform": "Meta", | |
| "campaign_type": campaign_type, | |
| "campaign_name": campaign_name, | |
| "adset_name": row.get("adset_name", ""), | |
| "ad_name": row.get("ad_name", ""), | |
| "ad_id": row.get("ad_id", ""), | |
| "date": row.get("date_start", ""), | |
| "Cost": spend, | |
| "Ads.Sales": purchase_revenue, | |
| "Ads.Sales(1d+7d)": None, | |
| "SPF.Sales": None, | |
| "ROAS": roas, | |
| "GA4.Sessions": None, | |
| "SPF.Sessions": None, | |
| "LPV": lpv, | |
| "CTR": ctr, | |
| "CVR": cvr, | |
| "CPC": cpc, | |
| "CPSE": None, | |
| "CPLPV": cplpv, | |
| "CPM": cpm, | |
| "Impressions": impressions, | |
| "Bounce Rate": None, | |
| "Average Duration": None, | |
| "ATC": atc, | |
| "CP-ATC": cp_atc, | |
| "Email": None, | |
| "CP-Email": None, | |
| "Video Views": video_views, | |
| "CP-Video Views": cp_video, | |
| "50% Play Rate": play_50_rate, | |
| "100% Play Rate": play_100_rate, | |
| "Hook Rate": hook_rate, | |
| } | |
| def fetch_meta_data( | |
| since: date | None = None, | |
| until: date | None = None, | |
| days_back: int = 90, | |
| level: str = "ad", | |
| ) -> pd.DataFrame: | |
| """ | |
| 从所有配置的 Meta 广告账户拉取数据，返回统一结构的 DataFrame。 | |
| level: "campaign" / "adset" / "ad"（默认 ad，素材级别）。 | |
| 默认拉取最近 90 天。 | |
| """ | |
| token = os.getenv("META_ACCESS_TOKEN") | |
| if not token: | |
| raise RuntimeError("未配置 META_ACCESS_TOKEN") | |
| accounts_str = os.getenv("META_AD_ACCOUNTS", "") | |
| if not accounts_str: | |
| raise RuntimeError("未配置 META_AD_ACCOUNTS (格式: account_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("META_AD_ACCOUNTS 格式错误") | |
| 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") | |
| all_parsed: list[dict] = [] | |
| for account_id, country in accounts: | |
| log.info(f"拉取 Meta 数据: account={account_id} country={country} level={level} {since_str}~{until_str}") | |
| try: | |
| rows = _fetch_insights(account_id, token, since_str, until_str, level=level) | |
| 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__": | |
| import argparse | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--days", type=int, default=7) | |
| parser.add_argument("--level", default="ad", choices=["campaign", "adset", "ad"]) | |
| args = parser.parse_args() | |
| logging.basicConfig(level=logging.INFO) | |
| df = fetch_meta_data(days_back=args.days, level=args.level) | |
| print(f"\n拉取到 {len(df)} 条数据 (level={args.level})") | |
| print(f"日期范围: {df['date'].min()} ~ {df['date'].max()}") | |
| print(f"国家: {df['country'].unique()}") | |
| print(f"Campaign Types: {df['campaign_type'].unique()}") | |
| if "ad_name" in df.columns: | |
| print(f"唯一 Ad 数: {df['ad_name'].nunique()}") | |
| 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") | |
| if args.level == "ad" and len(df) > 0: | |
| print(f"\n前 5 条 Ad 数据:") | |
| cols = ["ad_name", "campaign_name", "Cost", "Ads.Sales", "ROAS", "CTR", "Hook Rate"] | |
| print(df[cols].head().to_string()) | |