"""Facebook Ads connector via OAuth 2.0.
Syncs ad accounts, campaigns, ad sets, ads, and performance metrics
from the Facebook Marketing API (Graph API). Uses OAuth 2.0 login
so the user just clicks "Connect with Facebook" — no manual tokens.
Meta OAuth quirk: no refresh_token. Long-lived tokens last ~60 days
and are re-exchanged by posting the same token back to the token endpoint.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, date, timezone, timedelta
from typing import Any, Dict, List, Optional
from urllib.parse import urlencode
import requests
from . import OAuthConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata -------------------------------------------------
register_connector(
"facebook_ads",
{
"service": "facebook_ads",
"name": "Facebook Ads",
"category": "advertising",
"description": "Ad spend, campaign performance, and creative metrics from Meta Ads Manager.",
"auth_type": "oauth2",
"auth_fields": ["access_token", "ad_account_id"],
"capabilities": ["accounts", "campaigns", "ad_sets", "ads", "insights"],
"rate_limit": "200 requests/minute per account; call-timeout 60s",
"docs_url": "https://developers.facebook.com/docs/marketing-api/",
},
)
# -- Constants ---------------------------------------------------------------
_BASE_URL = "https://graph.facebook.com/v20.0"
_DEFAULT_FIELDS = "id,name,status,budget,daily_budget,lifetime_budget,start_time,end_time,created_time"
# -- Internal API helpers ---------------------------------------------------
# Meta enforces strict rate limits. A new app in dev mode can be blocked
# after a burst of requests. Enforce minimum spacing to stay under the radar.
_MIN_CALL_INTERVAL = 1.0 # seconds between Graph API calls
_last_call_time: float = 0
def _enforce_rate_limit() -> None:
"""Block until minimum interval has passed since the last call."""
global _last_call_time
now = time.monotonic()
elapsed = now - _last_call_time
if elapsed < _MIN_CALL_INTERVAL:
time.sleep(_MIN_CALL_INTERVAL - elapsed)
_last_call_time = time.monotonic()
def _is_api_blocked_error(exc: Exception) -> bool:
"""Check if an exception is Meta's 'API access blocked' (non-retryable)."""
if isinstance(exc, requests.HTTPError):
resp = getattr(exc, "response", None)
if resp and hasattr(resp, "status_code"):
try:
body = resp.json()
if body.get("error", {}).get("message", "").startswith("API access blocked"):
return True
except Exception:
pass
# Also check ValueError that wraps JSON errors from our connector code
msg = str(exc)
if "API access blocked" in msg:
return True
return False
def _graph_get(
*,
path: str,
access_token: str,
params: Optional[Dict[str, str]] = None,
base_url: str = _BASE_URL,
) -> Dict[str, Any]:
"""Make a GET request to the Facebook Graph API with rate limiting."""
_enforce_rate_limit()
url = f"{base_url}/{path.lstrip('/')}"
qs = dict(params or {})
qs["access_token"] = access_token
resp = requests.get(url, params=qs, timeout=30)
# Detect "API access blocked" early — don't retry, tell the user directly
if resp.status_code == 400:
try:
body = resp.json()
err_msg = body.get("error", {}).get("message", "")
if err_msg.startswith("API access blocked"):
raise ValueError(
"Meta has blocked API access for this app. "
"Go to developers.facebook.com → your app → check for "
"a block notification and follow the resolution steps. "
"This is not a code issue — the app needs to be unblocked in Meta's console."
)
except ValueError:
raise
except Exception:
pass
resp.raise_for_status()
return resp.json()
def _paginate_endpoint(
*,
path: str,
access_token: str,
params: Optional[Dict[str, str]] = None,
base_url: str = _BASE_URL,
) -> List[Dict[str, Any]]:
"""Fetch all pages from a paginated Graph API endpoint with rate limiting."""
all_data: List[Dict[str, Any]] = []
qs = dict(params or {})
qs["access_token"] = access_token
url = f"{base_url}/{path.lstrip('/')}"
while url:
_enforce_rate_limit()
if qs:
resp = requests.get(url, params=qs, timeout=30)
qs = None # pagination URLs are self-contained
else:
resp = requests.get(url, params=None, timeout=30)
# Same block detection for paginated calls
if resp.status_code == 400:
try:
body = resp.json()
err_msg = body.get("error", {}).get("message", "")
if err_msg.startswith("API access blocked"):
raise ValueError(
"Meta has blocked API access for this app. "
"Go to developers.facebook.com → your app → check for "
"a block notification and follow the resolution steps."
)
except ValueError:
raise
except Exception:
pass
resp.raise_for_status()
body = resp.json()
data = body.get("data", [])
if not data:
break
all_data.extend(data)
logger.info(
"Facebook Ads: fetched %d items (total %d)",
len(data), len(all_data),
)
paging = body.get("paging", {})
url = paging.get("next")
# Respect Meta's rate limits — 1s between pages
time.sleep(_MIN_CALL_INTERVAL)
return all_data
def _parse_fb_datetime(val: Optional[str]) -> Optional[datetime]:
"""Parse Facebook ISO-8601 timestamp into aware datetime."""
if not val:
return None
try:
return datetime.fromisoformat(val.replace("Z", "+00:00"))
except (ValueError, TypeError):
try:
return datetime.strptime(val, "%Y-%m-%d").replace(tzinfo=timezone.utc)
except (ValueError, TypeError):
return None
def _safe_float(val: Any) -> Optional[float]:
"""Convert a value to float safely."""
if val is None:
return None
try:
return float(val)
except (ValueError, TypeError):
return None
# -- Connector class --------------------------------------------------------
class FacebookAdsConnector(OAuthConnector):
"""Facebook Ads integration — OAuth 2.0 login with Graph API sync."""
_SERVICE = "facebook_ads"
_BASE_URL = _BASE_URL
# -- OAuth 2.0 ------------------------------------------------------------
OAUTH_AUTHORIZE_URL = "https://www.facebook.com/v20.0/dialog/oauth"
OAUTH_TOKEN_URL = "https://graph.facebook.com/v20.0/oauth/access_token"
OAUTH_SCOPES = [
"ads_management",
"ads_read",
"pages_read_engagement",
"business_management",
]
@property
def _oauth_client_id(self) -> str:
return self._get_client_id()
@property
def _oauth_client_secret(self) -> str:
return self._get_client_secret()
def oauth_authorize_url(self, state: str) -> str:
"""Build the Meta OAuth authorization URL."""
params = {
"client_id": self._oauth_client_id,
"redirect_uri": self._get_redirect_uri(),
"scope": ",".join(self.OAUTH_SCOPES),
"state": state,
"response_type": "code",
}
return f"{self.OAUTH_AUTHORIZE_URL}?{urlencode(params)}"
def exchange_code_for_tokens(self, code: str, **kwargs) -> Dict[str, Any]:
"""Exchange authorization code for a long-lived token + ad account info.
Meta flow:
1. Exchange code → short-lived token (~2 hours)
2. Swap short-lived → long-lived token (~60 days)
3. Fetch /me/adaccounts to get the first ad account ID
4. Return everything the callback needs
"""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError(
"Facebook Ads OAuth credentials not configured. "
"Set OAUTH_FACEBOOK_ADS_CLIENT_ID and OAUTH_FACEBOOK_ADS_CLIENT_SECRET."
)
redirect_uri = self._get_redirect_uri()
# Step 1: Exchange code for short-lived token
short_resp = requests.get(
self.OAUTH_TOKEN_URL,
params={
"client_id": client_id,
"client_secret": client_secret,
"grant_type": "authorization_code",
"code": code,
"redirect_uri": redirect_uri,
},
timeout=30,
)
if short_resp.status_code != 200:
try:
err_body = short_resp.json()
except Exception:
err_body = short_resp.text
raise ValueError(
f"Meta short-lived token exchange failed ({short_resp.status_code}): "
f"{err_body}"
)
short_data = short_resp.json()
short_token = short_data.get("access_token", "")
if not short_token:
raise ValueError("Meta did not return an access token: " + str(short_data))
# Step 2: Swap to long-lived token (~60 days)
long_resp = requests.get(
self.OAUTH_TOKEN_URL,
params={
"grant_type": "fb_exchange_token",
"client_id": client_id,
"client_secret": client_secret,
"fb_exchange_token": short_token,
},
timeout=30,
)
if long_resp.status_code != 200:
try:
err_body = long_resp.json()
except Exception:
err_body = long_resp.text
raise ValueError(
f"Meta long-lived token exchange failed ({long_resp.status_code}): "
f"{err_body}"
)
long_data = long_resp.json()
long_token = long_data.get("access_token", "")
if not long_token:
raise ValueError("Failed to get long-lived token: " + str(long_data))
# Step 3: Fetch ad accounts
ad_account_id = ""
ad_account_name = ""
try:
accounts_data = _graph_get(
path="me/adaccounts",
access_token=long_token,
params={"fields": "id,name,status"},
)
accounts = accounts_data.get("data", [])
if accounts:
# Prefer the first ACTIVE account
active = [a for a in accounts if a.get("status") == "ACTIVE"]
chosen = active[0] if active else accounts[0]
ad_account_id = chosen.get("id", "")
ad_account_name = chosen.get("name", "")
logger.info(
"Facebook Ads OAuth: found %d ad account(s), selected %s (%s)",
len(accounts), ad_account_id, ad_account_name,
)
else:
logger.warning("Facebook Ads OAuth: no ad accounts found for user")
except Exception as exc:
logger.warning("Facebook Ads OAuth: could not fetch ad accounts: %s", exc)
return {
"access_token": long_token,
"token_type": long_data.get("token_type", "Bearer"),
"expires_in": long_data.get("expires_in", ""),
"scope": long_data.get("scope", ""),
"ad_account_id": ad_account_id,
"ad_account_name": ad_account_name,
# Meta has no refresh_token — we re-exchange the long-lived token
"refresh_token": "",
}
def refresh_access_token(self, refresh_token: str) -> Dict[str, str]:
"""Re-exchange the long-lived token for a fresh 60-day token.
Meta doesn't use refresh_token — you re-exchange the same long-lived
token back to the token endpoint to get a new 60-day token.
"""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError("Facebook Ads OAuth credentials not configured.")
# Use the current access_token as the exchange token
current_token = self.config.get("access_token", "")
if not current_token:
raise ValueError("No access_token to re-exchange.")
resp = requests.get(
self.OAUTH_TOKEN_URL,
params={
"grant_type": "fb_exchange_token",
"client_id": client_id,
"client_secret": client_secret,
"fb_exchange_token": current_token,
},
timeout=30,
)
if resp.status_code != 200:
try:
err_body = resp.json()
except Exception:
err_body = resp.text
raise ValueError(
f"Meta token refresh failed ({resp.status_code}): "
f"{err_body}"
)
data = resp.json()
return {
"access_token": data.get("access_token", ""),
"token_type": data.get("token_type", "Bearer"),
"expires_in": data.get("expires_in", ""),
"refresh_token": "",
}
# -- connect / disconnect -------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate access token by querying /me endpoint."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
access_token = self.config.get("access_token", "")
if not access_token:
raise ValueError("Facebook Ads access_token is required")
response = self._retry(
_graph_get,
path="me",
access_token=access_token,
params={"fields": "name"},
)
self._connected = True
account_name = response.get("name", "Unknown")
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={
"account_name": account_name,
"ad_account_id": self.config.get("ad_account_id"),
},
)
return {
"status": "connected",
"service": "facebook_ads",
"account_name": account_name,
"ad_account_id": self.config.get("ad_account_id"),
"duration_ms": duration_ms,
}
except Exception as exc:
self._log(event_type="connect_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
def disconnect(self) -> Dict[str, Any]:
"""Clear credentials and reset connection state."""
self._connected = False
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "facebook_ads"}
# -- sync ----------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Sync ad accounts, campaigns, ad sets, ads, and insights from Facebook."""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total_records = 0
try:
results: Dict[str, Any] = {}
campaigns = self._sync_campaigns()
total_records += campaigns
results["campaigns"] = campaigns
ad_sets = self._sync_ad_sets()
total_records += ad_sets
results["ad_sets"] = ad_sets
ads = self._sync_ads()
total_records += ads
results["ads"] = ads
metrics = self._sync_insights()
total_records += metrics
results["metrics"] = metrics
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=total_records,
duration_ms=duration_ms,
details=results,
)
self.config["last_sync_at"] = datetime.now(timezone.utc).isoformat()
return {
"status": "success",
"record_count": total_records,
"duration_ms": duration_ms,
"details": results,
"last_sync_at": self.config.get("last_sync_at"),
}
except Exception as exc:
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_error",
status="error",
record_count=total_records,
duration_ms=duration_ms,
error_message=str(exc),
)
return {
"status": "error",
"error": str(exc),
"record_count": total_records,
}
# -- Campaign sync ------------------------------------------------------
def _sync_campaigns(self) -> int:
"""Pull campaigns and upsert to AdCampaign."""
from app.models import db, AdCampaign
account_id = self.config.get("ad_account_id", "")
access_token = self.config.get("access_token", "")
if not account_id or not access_token:
logger.warning("Facebook Ads sync: missing credentials")
return 0
acct = account_id if account_id.startswith("act_") else f"act_{account_id}"
campaigns = self._retry(
_paginate_endpoint,
path=f"{acct}/campaigns",
access_token=access_token,
params={"fields": _DEFAULT_FIELDS},
)
count = 0
for camp in campaigns:
campaign_id = str(camp.get("id", ""))
if not campaign_id:
continue
budget = _safe_float(camp.get("lifetime_budget")) or _safe_float(camp.get("budget")) or _safe_float(camp.get("daily_budget"))
budget_type = "lifetime" if camp.get("lifetime_budget") else "daily"
count += self._merge(
AdCampaign,
self.company_id,
campaign_id,
{
"source_service": "facebook_ads",
"name": camp.get("name", ""),
"status": camp.get("status", ""),
"budget": budget,
"budget_type": budget_type,
"start_date": _parse_fb_datetime(camp.get("start_time")),
"end_date": _parse_fb_datetime(camp.get("end_time")),
"channel_type": "facebook_ads",
"metadata_json": camp,
},
)
db.session.commit()
logger.info("Facebook Ads campaigns synced: %d records merged", count)
return count
# -- Ad set sync ---------------------------------------------------------
def _sync_ad_sets(self) -> int:
"""Pull ad sets and upsert to AdCampaign (as children of campaigns)."""
from app.models import db, AdCampaign
account_id = self.config.get("ad_account_id", "")
access_token = self.config.get("access_token", "")
if not account_id or not access_token:
logger.warning("Facebook Ads sync: missing credentials")
return 0
acct = account_id if account_id.startswith("act_") else f"act_{account_id}"
ad_sets = self._retry(
_paginate_endpoint,
path=f"{acct}/adsets",
access_token=access_token,
params={"fields": _DEFAULT_FIELDS},
)
count = 0
for adset in ad_sets:
adset_id = str(adset.get("id", ""))
if not adset_id:
continue
budget = _safe_float(adset.get("lifetime_budget")) or _safe_float(adset.get("daily_budget"))
count += self._merge(
AdCampaign,
self.company_id,
adset_id,
{
"source_service": "facebook_ads",
"name": adset.get("name", ""),
"status": adset.get("status", ""),
"budget": budget,
"budget_type": "lifetime" if adset.get("lifetime_budget") else "daily",
"start_date": _parse_fb_datetime(adset.get("start_time")),
"end_date": _parse_fb_datetime(adset.get("end_time")),
"channel_type": "facebook_ads",
"parent_campaign_id": adset.get("campaign_id"),
"metadata_json": adset,
},
)
db.session.commit()
logger.info("Facebook Ads ad sets synced: %d records merged", count)
return count
# -- Ad sync -------------------------------------------------------------
def _sync_ads(self) -> int:
"""Pull individual ads and upsert to AdCampaign."""
from app.models import db, AdCampaign
account_id = self.config.get("ad_account_id", "")
access_token = self.config.get("access_token", "")
if not account_id or not access_token:
logger.warning("Facebook Ads sync: missing credentials")
return 0
acct = account_id if account_id.startswith("act_") else f"act_{account_id}"
ads = self._retry(
_paginate_endpoint,
path=f"{acct}/ads",
access_token=access_token,
params={
"fields": "id,name,status,creative,adset_id,campaign_id,"
"start_time,stop_time,created_time,promoted_object"
},
)
count = 0
for ad in ads:
ad_id = str(ad.get("id", ""))
if not ad_id:
continue
count += self._merge(
AdCampaign,
self.company_id,
ad_id,
{
"source_service": "facebook_ads",
"name": ad.get("name", ""),
"status": ad.get("status", ""),
"channel_type": "facebook_ads",
"parent_campaign_id": ad.get("adset_id"),
"start_date": _parse_fb_datetime(ad.get("start_time")),
"end_date": _parse_fb_datetime(ad.get("stop_time")),
"metadata_json": ad,
},
)
db.session.commit()
logger.info("Facebook Ads ads synced: %d records merged", count)
return count
# -- Insights (metrics) sync --------------------------------------------
def _sync_insights(self) -> int:
"""Pull account-level insights and upsert to AdMetric."""
from app.models import db, AdMetric
account_id = self.config.get("ad_account_id", "")
access_token = self.config.get("access_token", "")
if not account_id or not access_token:
logger.warning("Facebook Ads sync: missing credentials")
return 0
acct = account_id if account_id.startswith("act_") else f"act_{account_id}"
end_date = date.today()
insights = self._retry(
_paginate_endpoint,
path=f"{acct}/insights",
access_token=access_token,
params={
"fields": "campaign_id,impressions,clicks,spend,actions,"
"cpc,cpm,ctr,unique_clicks,reach,results",
"level": "campaign",
},
)
count = 0
for insight in insights:
campaign_id = str(insight.get("campaign_id", ""))
if not campaign_id:
continue
impressions = int(insight.get("impressions", 0) or 0)
clicks = int(insight.get("clicks", 0) or 0)
spend = _safe_float(insight.get("spend")) or 0.0
conversions = _safe_float(insight.get("results")) or 0.0
cpc = _safe_float(insight.get("cpc"))
cpm = _safe_float(insight.get("cpm"))
ctr = _safe_float(insight.get("ctr"))
# Compute derived metrics when not provided
if ctr is None and impressions > 0:
ctr = (clicks / impressions * 100)
if cpc is None and clicks > 0:
cpc = spend / clicks
ext_id = f"{campaign_id}_{end_date.isoformat()}"
count += self._merge(
AdMetric,
self.company_id,
ext_id,
{
"external_campaign_id": campaign_id,
"source_service": "facebook_ads",
"metric_date": end_date,
"spend": spend,
"impressions": impressions,
"clicks": clicks,
"conversions": conversions,
"ctr": ctr,
"cpc": cpc,
"cpv": cpm,
"roas": None,
"metadata_json": insight,
},
)
db.session.commit()
logger.info("Facebook Ads insights synced: %d records merged", count)
return count
# -- Merge (upsert) helper -----------------------------------------------
@staticmethod
def _merge(model, company_id: str, external_id: str, defaults: Dict[str, Any]) -> int:
"""Find or create a record by (company_id, external_id) and update fields."""
from app.models import db
record = model.query.filter_by(
company_id=company_id, external_id=external_id
).first()
if record:
for key, value in defaults.items():
if hasattr(record, key):
setattr(record, key, value)
else:
record = model(
company_id=company_id,
external_id=external_id,
**{k: v for k, v in defaults.items() if hasattr(model, k)}
)
db.session.add(record)
return 1
# -- status --------------------------------------------------------------
def status(self) -> Dict[str, Any]:
"""Check connection health by hitting the /me endpoint."""
if not self._connected:
return {
"service": "facebook_ads",
"connected": False,
"config_present": bool(self.config.get("access_token")),
}
try:
response = self._retry(
_graph_get,
path="me",
access_token=self.config["access_token"],
params={"fields": "name"},
)
return {
"service": "facebook_ads",
"connected": True,
"account_name": response.get("name"),
"ad_account_id": self.config.get("ad_account_id"),
"last_sync_at": self.config.get("last_sync_at"),
}
except Exception as exc:
return {
"service": "facebook_ads",
"connected": False,
"error": str(exc),
}
# -- Write operations (Lead Gen Engine) ---------------------------------
def create_campaign(self, name: str, budget: float, objective: str = "CONVERSIONS", status: str = "ACTIVE", daily_budget: bool = True) -> Dict[str, Any]:
"""Create a campaign on the connected ad account.
Args:
name: Campaign name
budget: Daily budget in dollars
objective: Campaign objective (CONVERSIONS, LEAD_GENERATION, etc.)
status: ACTIVE or PAUSED
daily_budget: True for daily budget, False for lifetime
Returns:
Dict with campaign_id, status, etc.
"""
ad_account_id = self.config.get("ad_account_id", "")
access_token = self.config.get("access_token", "")
if not ad_account_id or not access_token:
return {"error": "Missing ad_account_id or access_token"}
budget_amount = int(budget * 100) # Micros (1/10000 of a dollar)
data = {
"name": name,
"objective": objective,
"status": status,
"special_ad_categories": [],
}
if daily_budget:
data["daily_budget"] = budget_amount
else:
data["lifetime_budget"] = budget_amount
# Allow billing events
data["billing_event"] = "IMPRESSIONS"
_enforce_rate_limit()
url = f"{_BASE_URL}/{ad_account_id}/campaigns"
qs = {"access_token": access_token}
resp = requests.post(url, data=data, params=qs, timeout=30)
resp.raise_for_status()
result = resp.json()
return {
"campaign_id": result.get("id", ""),
"name": name,
"status": status,
"objective": objective,
}
def create_ad_set(self, campaign_id: str, name: str, budget: float, start_time: Optional[str] = None, end_time: Optional[str] = None, targeting: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Create an ad set within a campaign."""
access_token = self.config.get("access_token", "")
if not access_token:
return {"error": "Missing access_token"}
budget_amount = int(budget * 100)
data = {
"name": name,
"campaign_id": campaign_id,
"daily_budget": budget_amount,
"optimization_goal": "CONVERSIONS",
"billing_event": "IMPRESSIONS",
"status": "ACTIVE",
"targeting": targeting or {
"geo_locations": {
"countries": ["US"],
},
"age_min": 25,
"age_max": 65,
},
}
if start_time:
data["start_time"] = start_time
if end_time:
data["end_time"] = end_time
import json
data["targeting"] = json.dumps(data["targeting"])
_enforce_rate_limit()
url = f"{_BASE_URL}/{campaign_id}/adsets"
qs = {"access_token": access_token}
resp = requests.post(url, data=data, params=qs, timeout=30)
resp.raise_for_status()
result = resp.json()
return {
"ad_set_id": result.get("id", ""),
"name": name,
"status": "ACTIVE",
}
def create_ad(self, ad_set_id: str, name: str, headline: str, body: str, image_hash: Optional[str] = None, call_to_action: Optional[Dict[str, str]] = None, url: str = "") -> Dict[str, Any]:
"""Create an ad within an ad set."""
access_token = self.config.get("access_token", "")
if not access_token:
return {"error": "Missing access_token"}
# Create ad creative first (if using image)
creative_id = ""
if image_hash:
creative_data = {
"name": name,
"object_story_spec": {
"page_id": self.config.get("facebook_page_id", ""),
"link_data": {
"image_hash": image_hash,
"link": url,
"message": body,
"call_to_action": call_to_action or {"type": "LEARN_MORE"},
},
},
"call_to_action": call_to_action or {"type": "LEARN_MORE"},
}
import json
creative_data["object_story_spec"] = json.dumps(creative_data["object_story_spec"])
_enforce_rate_limit()
url = f"{_BASE_URL}/act_{self.config.get('ad_account_id', '')}/adcreatives"
qs = {"access_token": access_token}
creative_resp = requests.post(url, data=creative_data, params=qs, timeout=30)
creative_resp.raise_for_status()
creative_id = creative_resp.json().get("id", "")
# Create the ad
ad_data = {
"name": name,
"adset_id": ad_set_id,
"status": "ACTIVE",
}
if creative_id:
ad_data["creative"] = creative_id
else:
# Simple text ad
ad_data["creative"] = json.dumps({
"name": name,
"object_story_spec": {
"page_id": self.config.get("facebook_page_id", ""),
"link_data": {
"link": url,
"message": body,
"call_to_action": call_to_action or {"type": "LEARN_MORE"},
},
},
"call_to_action": call_to_action or {"type": "LEARN_MORE"},
})
import json as json_mod
if isinstance(ad_data.get("creative"), dict):
ad_data["creative"] = json_mod.dumps(ad_data["creative"])
_enforce_rate_limit()
url = f"{_BASE_URL}/act_{self.config.get('ad_account_id', '')}/ads"
qs = {"access_token": access_token}
resp = requests.post(url, data=ad_data, params=qs, timeout=30)
resp.raise_for_status()
result = resp.json()
return {
"ad_id": result.get("id", ""),
"name": name,
"status": "ACTIVE",
}
def pause_campaign(self, campaign_id: str) -> Dict[str, Any]:
"""Pause a campaign."""
access_token = self.config.get("access_token", "")
_enforce_rate_limit()
url = f"{_BASE_URL}/{campaign_id}"
qs = {"access_token": access_token}
resp = requests.post(url, data={"status": "PAUSED"}, params=qs, timeout=30)
resp.raise_for_status()
result = resp.json()
return {
"status": "paused" if result else "unknown",
"campaign_id": campaign_id,
}
def update_campaign_budget(self, campaign_id: str, new_budget: float, daily: bool = True) -> Dict[str, Any]:
"""Update campaign budget."""
access_token = self.config.get("access_token", "")
budget_amount = int(new_budget * 100)
data: Dict[str, Any] = {}
if daily:
data["daily_budget"] = budget_amount
else:
data["lifetime_budget"] = budget_amount
_enforce_rate_limit()
url = f"{_BASE_URL}/{campaign_id}"
qs = {"access_token": access_token}
resp = requests.post(url, data=data, params=qs, timeout=30)
resp.raise_for_status()
return {
"status": "updated",
"campaign_id": campaign_id,
"new_budget": new_budget,
"budget_type": "daily" if daily else "lifetime",
}
# -- End write operations -----------------------------------------------
# -- Register in the framework registry -------------------------------------
_REGISTRY["facebook_ads"] = FacebookAdsConnector