José Manuel Gómez
8fb0b69896
- Migrate from Airtable to Baserow: BaserowClient with snapshots, actions, creatives, logs - Claude agents (Haiku for decisions/units, Sonnet for creatives) with cost_cap_eur vs CPL comparison - Slack bot with colored action emojis, effect text before approval buttons, 500-char justifications - Streamlit dashboard with date-range navigation, campaign drill-down (adsets/ads), Histórico tab - Approval server (FastAPI + ngrok) for Slack button callbacks - backfill.py for historical snapshot regeneration with Claude re-analysis - Margin fix: 0-lead campaigns contribute -spend (not 0) to margin - CTR fix: Meta returns CTR as percentage already, removed *100 - Parameter fix: pass decision parameter to Slack action for correct budget effect display Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
243 lines
10 KiB
Python
243 lines
10 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
|
|
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)
|
|
|
|
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 = sum(float(a["value"]) for a in row.get("actions", [])
|
|
if a["action_type"] in ("lead", "onsite_conversion.lead_grouped"))
|
|
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 = sum(float(a["value"]) for a in row.get("actions", [])
|
|
if a["action_type"] in ("lead", "onsite_conversion.lead_grouped"))
|
|
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 = sum(float(a["value"]) for a in row.get("actions", [])
|
|
if a["action_type"] in ("lead", "onsite_conversion.lead_grouped"))
|
|
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,
|
|
})
|
|
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_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_campaign(self, campaign_id: str):
|
|
"""Pause a campaign."""
|
|
campaign = Campaign(campaign_id)
|
|
campaign.api_update(params={"status": Campaign.Status.paused})
|