José Manuel Gómez
9239e2f67f
Ports meta-optimizer's Meta Ads execution/approval/creative-analysis layer (agent.py, meta_ads_client.py, baserow_client.py, slack_notifier.py, approval_server.py) and replaces the per-vertical CPL model with the PPL + monthly-capping-per-course model already used by leads-optimizer, via a new airtable_client.py that shares Cursos/Familias/CentroCurso/ CursoMes/Leads Lake with that project and adds Meta Ads Campaigns / MetaCampaignMes alongside its Google Ads Campaigns / GACampaignMes.
324 lines
14 KiB
Python
324 lines
14 KiB
Python
"""
|
|
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
|
|
from facebook_business.adobjects.advideo import AdVideo
|
|
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_all_campaigns(self) -> list:
|
|
"""
|
|
All campaigns matching the prefix, regardless of spend/status
|
|
(for catalog sync — insights-based methods only return campaigns
|
|
with spend > 0 in the queried window).
|
|
"""
|
|
prefix = config.META_CAMPAIGN_PREFIX.upper()
|
|
campaigns = self.account.get_campaigns(
|
|
fields=["id", "name", "effective_status"],
|
|
params={"limit": 500},
|
|
)
|
|
result = []
|
|
for c in campaigns:
|
|
name = c.get("name", "")
|
|
if not name.upper().startswith(prefix):
|
|
continue
|
|
result.append({"id": c["id"], "name": name, "status": c.get("effective_status", "PAUSED")})
|
|
return result
|
|
|
|
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, "adset_id"],
|
|
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_url = ""
|
|
image_url = ""
|
|
video_thumbnail_url = ""
|
|
|
|
if creative_id:
|
|
try:
|
|
creative = AdCreative(creative_id).api_get(
|
|
fields=["thumbnail_url", "image_url", "video_id"]
|
|
)
|
|
thumbnail_url = creative.get("thumbnail_url", "")
|
|
image_url = creative.get("image_url", "")
|
|
video_id = creative.get("video_id", "")
|
|
|
|
# For video creatives: fetch a permanent thumbnail via AdVideo.picture
|
|
if video_id and not image_url:
|
|
try:
|
|
vdata = AdVideo(video_id).api_get(fields=["picture"])
|
|
video_thumbnail_url = vdata.get("picture", "")
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
result.append({
|
|
"ad_id": ad["id"],
|
|
"ad_name": ad["name"],
|
|
"campaign_id": campaign_id,
|
|
"adset_id": ad.get("adset_id", ""),
|
|
"thumbnail_url": thumbnail_url,
|
|
"image_url": image_url,
|
|
"video_thumbnail_url": video_thumbnail_url,
|
|
})
|
|
|
|
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})
|