"""Pulls Lead Ads and campaign spend from the Facebook Marketing API.""" import os import sys from datetime import date, timedelta from typing import Any, Optional from facebook_business.adobjects.adaccount import AdAccount from facebook_business.api import FacebookAdsApi from facebook_business.exceptions import FacebookRequestError # Facebook renamed campaign objectives in 2024; both forms can appear in older accounts. LEAD_GEN_OBJECTIVES = {"LEAD_GENERATION", "OUTCOME_LEADS"} def fetch_leads(config: dict) -> list[dict[str, Any]]: """Return Lead Ad submissions with email and campaign metadata. Walks campaigns -> ads -> leads. We only call get_leads() for ads in lead-gen campaigns (objective LEAD_GENERATION or OUTCOME_LEADS) to avoid wasting Marketing API quota on non-lead ads. Skips leads with no email or whose email domain is in `filters.internal_domains`. """ token = os.environ.get("FACEBOOK_ACCESS_TOKEN") if not token: print( "Facebook leads: skipping (FACEBOOK_ACCESS_TOKEN not set).", file=sys.stderr, flush=True, ) return [] fb_config = config.get("facebook", {}) ad_accounts = fb_config.get("ad_accounts", []) if not ad_accounts or any("REPLACE" in str(a) for a in ad_accounts): print( "Facebook leads: skipping (facebook.ad_accounts not configured).", file=sys.stderr, flush=True, ) return [] internal_domains = { d.lower() for d in config.get("filters", {}).get("internal_domains", []) } FacebookAdsApi.init(access_token=token) leads: list[dict[str, Any]] = [] skipped_no_email = 0 skipped_internal = 0 for account_id in ad_accounts: print( f"Fetching Facebook leads for {account_id}...", file=sys.stderr, flush=True, ) account = AdAccount(account_id) try: campaigns = list( account.get_campaigns(fields=["id", "name", "objective"]) ) except FacebookRequestError as e: print( f" failed to list campaigns: {e.api_error_message()}", file=sys.stderr, ) continue lead_campaign_meta: dict[str, dict[str, Any]] = { c["id"]: c for c in campaigns if c.get("objective") in LEAD_GEN_OBJECTIVES } print( f" {len(lead_campaign_meta)} of {len(campaigns)} campaigns are lead-gen.", file=sys.stderr, flush=True, ) if not lead_campaign_meta: continue try: ads = list( account.get_ads(fields=["id", "name", "campaign_id"]) ) except FacebookRequestError as e: print( f" failed to list ads: {e.api_error_message()}", file=sys.stderr, ) continue lead_ads = [a for a in ads if a.get("campaign_id") in lead_campaign_meta] print( f" {len(lead_ads)} ads in lead-gen campaigns; querying leads for each...", file=sys.stderr, flush=True, ) for ad in lead_ads: cid = ad.get("campaign_id") campaign = lead_campaign_meta.get(cid, {}) try: ad_leads = ad.get_leads( fields=[ "id", "created_time", "ad_id", "campaign_id", "field_data", ] ) except FacebookRequestError as e: print( f" ad {ad.get('id')}: get_leads failed ({e.api_error_message()}); skipping", file=sys.stderr, ) continue for lead in ad_leads: email = _extract_email(lead.get("field_data", [])) if not email: skipped_no_email += 1 continue domain = email.rsplit("@", 1)[-1].strip().lower() if domain in internal_domains: skipped_internal += 1 continue leads.append( { "id": lead["id"], "email": email, "created_time": lead.get("created_time"), "ad_id": lead.get("ad_id") or ad.get("id"), "campaign_id": lead.get("campaign_id") or cid, "campaign_name": campaign.get("name"), "ad_name": ad.get("name"), } ) print( f" Total: {len(leads)} leads, " f"{skipped_no_email} skipped (no email), " f"{skipped_internal} skipped (internal domain).", file=sys.stderr, flush=True, ) return leads def fetch_campaign_spend(config: dict) -> dict[str, dict[str, Any]]: """Return campaign_id -> {name, spend_usd} for the lookback window.""" token = os.environ.get("FACEBOOK_ACCESS_TOKEN") if not token: print( "Facebook campaign spend: skipping (FACEBOOK_ACCESS_TOKEN not set).", file=sys.stderr, flush=True, ) return {} fb_config = config.get("facebook", {}) ad_accounts = fb_config.get("ad_accounts", []) if not ad_accounts or any("REPLACE" in str(a) for a in ad_accounts): print( "Facebook campaign spend: skipping (facebook.ad_accounts not configured).", file=sys.stderr, flush=True, ) return {} lookback_days = fb_config.get("lookback_days", 365) end = date.today() start = end - timedelta(days=lookback_days) FacebookAdsApi.init(access_token=token) campaigns: dict[str, dict[str, Any]] = {} for account_id in ad_accounts: print( f"Fetching campaign spend for {account_id}, " f"{start.isoformat()} -> {end.isoformat()}...", file=sys.stderr, flush=True, ) account = AdAccount(account_id) try: insights = account.get_insights( fields=["campaign_id", "campaign_name", "spend"], params={ "level": "campaign", "time_range": { "since": start.isoformat(), "until": end.isoformat(), }, "limit": 1000, }, ) except FacebookRequestError as e: print( f" failed: {e.api_error_message()}", file=sys.stderr, ) continue for row in insights: cid = row.get("campaign_id") if not cid: continue spend = float(row.get("spend", 0) or 0) existing = campaigns.get(cid) if existing: existing["spend_usd"] += spend else: campaigns[cid] = { "campaign_id": cid, "campaign_name": row.get("campaign_name", ""), "spend_usd": spend, } print( f" {len(campaigns)} campaigns with spend.", file=sys.stderr, flush=True, ) return campaigns def _extract_email(field_data: list[dict]) -> Optional[str]: """Lead form fields come back as [{name: 'email', values: ['x@y.com']}, ...].""" for f in field_data: name = (f.get("name") or "").lower() if name in ("email", "email_address"): values = f.get("values") or [] if values: return values[0] return None if __name__ == "__main__": import json import yaml from dotenv import load_dotenv load_dotenv() with open("config.yaml") as f: config = yaml.safe_load(f) leads = fetch_leads(config) print(f"\nFetched {len(leads)} Facebook leads.") if leads: print("Sample lead:") print(json.dumps(leads[0], indent=2, default=str)) campaigns = fetch_campaign_spend(config) print(f"\n{len(campaigns)} campaigns with spend.") if campaigns: sample_cid = next(iter(campaigns)) print("Sample campaign:") print(json.dumps(campaigns[sample_cid], indent=2, default=str))