""" Client for Meta Marketing API. Docs: https://developers.facebook.com/docs/marketing-api SDK: facebook-business """ from facebook_business.api import FacebookAdsApi from facebook_business.adobjects.adaccount import AdAccount from facebook_business.adobjects.campaign import Campaign from facebook_business.adobjects.adset import AdSet from facebook_business.adobjects.ad import Ad from facebook_business.adobjects.adcreative import AdCreative import config from datetime import datetime, timedelta class MetaAdsClient: def __init__(self): FacebookAdsApi.init( app_id=config.META_APP_ID, app_secret=config.META_APP_SECRET, access_token=config.META_ACCESS_TOKEN, ) self.account = AdAccount(config.META_AD_ACCOUNT_ID) @staticmethod def _count_conversions(actions: list) -> float: """Prioritize 'lead' to avoid double-counting with lead_grouped; include click-to-call.""" by_type = {a["action_type"]: float(a["value"]) for a in actions} if "lead" in by_type: return by_type["lead"] if "onsite_conversion.lead_grouped" in by_type: return by_type["onsite_conversion.lead_grouped"] # Click-to-call campaigns (Llamadas): click_to_call_call_confirm o call_confirm_grouped if "click_to_call_call_confirm" in by_type: return by_type["click_to_call_call_confirm"] if "call_confirm_grouped" in by_type: return by_type["call_confirm_grouped"] return by_type.get("call_confirm", by_type.get("contact", 0.0)) def _parse_insights_row(self, row: dict) -> dict: spend = float(row.get("spend", 0)) impressions = int(row.get("impressions", 0)) clicks = int(row.get("clicks", 0)) ctr = float(row.get("ctr", 0)) cpm = float(row.get("cpm", 0)) leads = self._count_conversions(row.get("actions", [])) cpl = round(spend / leads, 2) if leads > 0 else 0.0 return { "campaign_id": row.get("campaign_id", ""), "name": row.get("campaign_name", ""), "status": "ACTIVE", "spend": round(spend, 2), "impressions": impressions, "clicks": clicks, "ctr": round(ctr, 4), "cpm": round(cpm, 2), "leads": int(leads), "cpl": cpl, } def get_campaign_metrics(self, date_from: str, date_to: str) -> dict: """ Campaign-level metrics aggregated over a date range. Returns {campaign_id: metrics}, spend > 0 only. """ prefix = config.META_CAMPAIGN_PREFIX.upper() insights = self.account.get_insights( fields=["campaign_id", "campaign_name", "spend", "impressions", "clicks", "ctr", "cpm", "actions"], params={ "level": "campaign", "time_range": {"since": date_from, "until": date_to}, } ) result = {} for row in insights: if not row.get("campaign_name", "").upper().startswith(prefix): continue m = self._parse_insights_row(row) if m["spend"] == 0: continue result[m["campaign_id"]] = m return result def get_daily_campaign_rows(self, date_from: str, date_to: str) -> list: """ Per-campaign per-day rows for a date range. Returns [{date, campaign_name, spend, leads}] sorted by date. """ if date_from > date_to: return [] prefix = config.META_CAMPAIGN_PREFIX.upper() insights = self.account.get_insights( fields=["date_start", "campaign_name", "spend", "actions"], params={ "level": "campaign", "time_range": {"since": date_from, "until": date_to}, "time_increment": 1, } ) result = [] for row in insights: if not row.get("campaign_name", "").upper().startswith(prefix): continue spend = float(row.get("spend", 0)) leads = self._count_conversions(row.get("actions", [])) result.append({ "date": row["date_start"], "campaign_name": row.get("campaign_name", ""), "spend": round(spend, 2), "leads": int(leads), }) return sorted(result, key=lambda x: x["date"]) def _get_sub_insights_range(self, campaign_id: str, level: str, date_from: str, date_to: str) -> list: """Ad set or ad level insights for a date range, spend > 0, sorted by spend desc.""" id_field = f"{level}_id" name_field = f"{level}_name" try: insights = Campaign(campaign_id).get_insights( fields=[id_field, name_field, "spend", "impressions", "clicks", "ctr", "cpm", "actions"], params={ "level": level, "time_range": {"since": date_from, "until": date_to}, } ) except Exception: return [] result = [] for row in insights: spend = float(row.get("spend", 0)) if spend == 0: continue leads = self._count_conversions(row.get("actions", [])) cpl = round(spend / leads, 2) if leads > 0 else 0.0 result.append({ "id": row.get(id_field, ""), "name": row.get(name_field, ""), "spend": round(spend, 2), "impressions": int(row.get("impressions", 0)), "clicks": int(row.get("clicks", 0)), "ctr": round(float(row.get("ctr", 0)), 4), "cpm": round(float(row.get("cpm", 0)), 2), "leads": int(leads), "cpl": cpl, }) # For ads, exclude currently paused ads (they may have historic spend in the window) if level == "ad": try: active_ads = Campaign(campaign_id).get_ads( fields=["id"], params={"effective_status": ["ACTIVE"]}, ) active_ids = {a["id"] for a in active_ads} result = [r for r in result if r["id"] in active_ids] except Exception: pass return sorted(result, key=lambda x: -x["spend"]) def get_yesterday_metrics(self) -> dict: yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") return self.get_campaign_metrics(yesterday, yesterday) def get_monthly_daily_totals(self) -> list: """Per-campaign daily rows for the current month (used by run.py).""" now = datetime.now() date_start = f"{now.year}-{now.month:02d}-01" yesterday = (now - timedelta(days=1)).strftime("%Y-%m-%d") return self.get_daily_campaign_rows(date_start, yesterday) def get_adset_metrics(self, campaign_id: str, date_from: str, date_to: str) -> list: return self._get_sub_insights_range(campaign_id, "adset", date_from, date_to) def get_ad_metrics(self, campaign_id: str, date_from: str, date_to: str) -> list: return self._get_sub_insights_range(campaign_id, "ad", date_from, date_to) def get_yesterday_adset_metrics(self, campaign_id: str) -> list: yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") return self.get_adset_metrics(campaign_id, yesterday, yesterday) def get_yesterday_ad_metrics(self, campaign_id: str) -> list: yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") return self.get_ad_metrics(campaign_id, yesterday, yesterday) def get_period_campaign_metrics(self, days: int) -> dict: today = datetime.now() date_to = (today - timedelta(days=1)).strftime("%Y-%m-%d") date_from = (today - timedelta(days=days)).strftime("%Y-%m-%d") return self.get_campaign_metrics(date_from, date_to) def get_period_adset_metrics(self, campaign_id: str, days: int) -> list: today = datetime.now() date_to = (today - timedelta(days=1)).strftime("%Y-%m-%d") date_from = (today - timedelta(days=days)).strftime("%Y-%m-%d") return self.get_adset_metrics(campaign_id, date_from, date_to) def get_period_ad_metrics(self, campaign_id: str, days: int) -> list: today = datetime.now() date_to = (today - timedelta(days=1)).strftime("%Y-%m-%d") date_from = (today - timedelta(days=days)).strftime("%Y-%m-%d") return self.get_ad_metrics(campaign_id, date_from, date_to) def get_ads_with_creatives(self, campaign_id: str) -> list: """ Returns active ads for a campaign with their thumbnail URLs for creative analysis. """ campaign = Campaign(campaign_id) ads = campaign.get_ads( fields=[Ad.Field.id, Ad.Field.name, Ad.Field.status, Ad.Field.creative], params={"effective_status": ["ACTIVE"]}, ) result = [] for ad in ads: creative_ref = ad.get("creative", {}) creative_id = creative_ref.get("id") if creative_ref else None thumbnail = "" if creative_id: try: creative = AdCreative(creative_id).api_get( fields=["thumbnail_url", "image_url"] ) thumbnail = creative.get("thumbnail_url") or creative.get("image_url", "") except Exception: pass result.append({ "ad_id": ad["id"], "ad_name": ad["name"], "campaign_id": campaign_id, "thumbnail_url": thumbnail, }) return result def get_campaign_bid_config(self, campaign_id: str) -> dict: """Fetch bid strategy and daily/lifetime budget at campaign level.""" try: data = Campaign(campaign_id).api_get( fields=["bid_strategy", "daily_budget", "lifetime_budget"] ) daily = float(data.get("daily_budget", 0) or 0) lifetime = float(data.get("lifetime_budget", 0) or 0) return { "bid_strategy": data.get("bid_strategy", ""), "daily_budget_eur": round(daily / 100, 2) if daily else None, "lifetime_budget_eur": round(lifetime / 100, 2) if lifetime else None, } except Exception: return {} def get_adset_bid_configs(self, campaign_id: str) -> dict: """Returns {adset_id: {bid_strategy, cost_cap_eur, daily_budget_eur}}.""" try: adsets = Campaign(campaign_id).get_ad_sets( fields=[AdSet.Field.id, AdSet.Field.bid_strategy, AdSet.Field.bid_amount, AdSet.Field.daily_budget] ) result = {} for as_obj in adsets: bid_amount = float(as_obj.get("bid_amount", 0) or 0) daily = float(as_obj.get("daily_budget", 0) or 0) result[as_obj["id"]] = { "bid_strategy": as_obj.get("bid_strategy", ""), "cost_cap_eur": round(bid_amount / 100, 2) if bid_amount else None, "daily_budget_eur": round(daily / 100, 2) if daily else None, } return result except Exception: return {} def set_campaign_budget(self, campaign_id: str, daily_budget_cents: int): """Set campaign daily budget (amount in cents).""" campaign = Campaign(campaign_id) campaign.api_update(params={"daily_budget": daily_budget_cents}) def pause_ad(self, ad_id: str): ad = Ad(fbid=ad_id) ad.api_update(params={"status": Ad.Status.paused}) def pause_campaign(self, campaign_id: str): """Pause a campaign.""" campaign = Campaign(campaign_id) campaign.api_update(params={"status": Campaign.Status.paused})