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