import aiohttp from datetime import datetime, timedelta from . import BaseAdPlatform class MetaAdsPlatform(BaseAdPlatform): """Meta Marketing API integration for Facebook + Instagram ads.""" API_VERSION = "v21.0" BASE_URL = f"https://graph.facebook.com/{API_VERSION}" async def upload_video(self, video_url: str, account) -> str: """Upload video to Meta Ad Account. Returns video ID.""" url = f"{self.BASE_URL}/act_{account.account_id}/advideos" async with aiohttp.ClientSession() as session: data = aiohttp.FormData() data.add_field("file_url", video_url) data.add_field("access_token", account.access_token) async with session.post(url, data=data) as resp: result = await resp.json() if "id" in result: return result["id"] raise Exception(f"Meta video upload failed: {result}") async def create_campaign(self, ad_campaign, account) -> dict: """Create full Meta ad campaign: Campaign → Ad Set → Ad Creative → Ad.""" headers = {"Authorization": f"Bearer {account.access_token}"} act_id = f"act_{account.account_id}" targeting = ad_campaign.targeting_json or {} async with aiohttp.ClientSession() as session: # 1. Create Campaign camp_url = f"{self.BASE_URL}/{act_id}/campaigns" camp_data = { "name": ad_campaign.name, "objective": "OUTCOME_AWARENESS", "status": "PAUSED", "special_ad_categories": "[]", "access_token": account.access_token, } async with session.post(camp_url, data=camp_data) as resp: camp_result = await resp.json() if "id" not in camp_result: return {"status": "error", "error": f"Campaign creation failed: {camp_result}"} campaign_id = camp_result["id"] # 2. Upload video creative video_url = None if ad_campaign.campaign and ad_campaign.campaign.video_url: video_url = ad_campaign.campaign.video_url video_id = None if video_url: try: video_id = await self.upload_video(video_url, account) except Exception as e: return {"status": "error", "error": f"Video upload failed: {e}"} # 3. Create Ad Set (targeting + budget) adset_url = f"{self.BASE_URL}/{act_id}/adsets" geo_locations = {} locations = targeting.get("locations", []) if locations: geo_locations["countries"] = locations else: geo_locations["countries"] = ["SA"] # Default: Saudi Arabia targeting_spec = { "geo_locations": geo_locations, } if targeting.get("age_min"): targeting_spec["age_min"] = targeting["age_min"] if targeting.get("age_max"): targeting_spec["age_max"] = targeting["age_max"] if targeting.get("gender"): gender_map = {"male": 1, "female": 2, "all": 0} g = gender_map.get(targeting["gender"], 0) if g: targeting_spec["genders"] = [g] if targeting.get("interests"): targeting_spec["flexible_spec"] = [{"interests": [{"name": i} for i in targeting["interests"]]}] import json start_date = ad_campaign.start_date or datetime.utcnow() end_date = ad_campaign.end_date or (start_date + timedelta(days=ad_campaign.duration_days or 7)) adset_data = { "name": f"{ad_campaign.name} - Ad Set", "campaign_id": campaign_id, "daily_budget": int(ad_campaign.daily_budget * 100), # in cents "billing_event": "IMPRESSIONS", "optimization_goal": "REACH", "targeting": json.dumps(targeting_spec), "start_time": start_date.strftime("%Y-%m-%dT%H:%M:%S+0000"), "end_time": end_date.strftime("%Y-%m-%dT%H:%M:%S+0000"), "status": "PAUSED", "access_token": account.access_token, } async with session.post(adset_url, data=adset_data) as resp: adset_result = await resp.json() if "id" not in adset_result: return {"status": "error", "error": f"Ad Set creation failed: {adset_result}"} adset_id = adset_result["id"] # 4. Create Ad Creative creative_url = f"{self.BASE_URL}/{act_id}/adcreatives" creative_data = { "name": f"{ad_campaign.name} - Creative", "access_token": account.access_token, } if video_id: creative_data["object_story_spec"] = json.dumps({ "page_id": targeting.get("page_id", account.account_id), "video_data": { "video_id": video_id, "message": ad_campaign.ad_caption or ad_campaign.name, "call_to_action": { "type": ad_campaign.ad_cta or "LEARN_MORE", } } }) async with session.post(creative_url, data=creative_data) as resp: creative_result = await resp.json() if "id" not in creative_result: return {"status": "error", "error": f"Creative creation failed: {creative_result}"} creative_id = creative_result["id"] # 5. Create Ad ad_url = f"{self.BASE_URL}/{act_id}/ads" ad_data = { "name": f"{ad_campaign.name} - Ad", "adset_id": adset_id, "creative": json.dumps({"creative_id": creative_id}), "status": "PAUSED", "access_token": account.access_token, } async with session.post(ad_url, data=ad_data) as resp: ad_result = await resp.json() if "id" not in ad_result: return {"status": "error", "error": f"Ad creation failed: {ad_result}"} ad_id = ad_result["id"] # 6. Activate campaign activate_url = f"{self.BASE_URL}/{campaign_id}" await session.post(activate_url, data={"status": "ACTIVE", "access_token": account.access_token}) activate_url2 = f"{self.BASE_URL}/{adset_id}" await session.post(activate_url2, data={"status": "ACTIVE", "access_token": account.access_token}) activate_url3 = f"{self.BASE_URL}/{ad_id}" await session.post(activate_url3, data={"status": "ACTIVE", "access_token": account.access_token}) return { "status": "active", "campaign_id": campaign_id, "adset_id": adset_id, "creative_id": creative_id, "ad_id": ad_id, "video_id": video_id, } async def get_metrics(self, platform_campaign_id: str, account, date_from=None, date_to=None) -> list: """Fetch daily insights from Meta.""" url = f"{self.BASE_URL}/{platform_campaign_id}/insights" params = { "fields": "impressions,reach,clicks,ctr,spend,video_views,actions,frequency,cost_per_action_type", "time_increment": 1, # daily breakdown "access_token": account.access_token, } if date_from: params["time_range"] = f'{{"since":"{date_from}","until":"{date_to or date_from}"}}' metrics_list = [] async with aiohttp.ClientSession() as session: async with session.get(url, params=params) as resp: data = await resp.json() for row in data.get("data", []): views = int(row.get("video_views", 0)) impressions = int(row.get("impressions", 0)) clicks = int(row.get("clicks", 0)) spend = float(row.get("spend", 0)) reach = int(row.get("reach", 0)) frequency = float(row.get("frequency", 0)) ctr = float(row.get("ctr", 0)) likes = 0 comments = 0 shares = 0 for action in row.get("actions", []): if action.get("action_type") == "like": likes = int(action.get("value", 0)) elif action.get("action_type") == "comment": comments = int(action.get("value", 0)) elif action.get("action_type") == "post": shares = int(action.get("value", 0)) metrics_list.append({ "date": row.get("date_start"), "impressions": impressions, "views": views, "clicks": clicks, "ctr": ctr, "spend": spend, "cost_per_view": round(spend / views, 4) if views > 0 else 0, "cost_per_click": round(spend / clicks, 4) if clicks > 0 else 0, "likes": likes, "comments": comments, "shares": shares, "reach": reach, "frequency": frequency, }) return metrics_list async def pause_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/{platform_campaign_id}" async with aiohttp.ClientSession() as session: async with session.post(url, data={"status": "PAUSED", "access_token": account.access_token}) as resp: result = await resp.json() return {"status": "paused"} if result.get("success") else {"status": "error", "error": str(result)} async def resume_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/{platform_campaign_id}" async with aiohttp.ClientSession() as session: async with session.post(url, data={"status": "ACTIVE", "access_token": account.access_token}) as resp: result = await resp.json() return {"status": "active"} if result.get("success") else {"status": "error", "error": str(result)} async def delete_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/{platform_campaign_id}" async with aiohttp.ClientSession() as session: async with session.post(url, data={"status": "DELETED", "access_token": account.access_token}) as resp: result = await resp.json() return {"status": "deleted"} if result.get("success") else {"status": "error", "error": str(result)}