import aiohttp import json from datetime import datetime, timedelta from . import BaseAdPlatform class TwitterAdsPlatform(BaseAdPlatform): """X/Twitter Ads API integration for paid ads.""" BASE_URL = "https://ads-api.x.com/12" def _headers(self, account): return { "Authorization": f"Bearer {account.access_token}", "Content-Type": "application/json", } async def upload_video(self, video_url: str, account) -> str: """Twitter media upload for ads.""" # Twitter requires chunked media upload - simplified version async with aiohttp.ClientSession() as session: async with session.get(video_url) as vresp: video_data = await vresp.read() # INIT init_url = "https://upload.twitter.com/1.1/media/upload.json" init_data = { "command": "INIT", "total_bytes": len(video_data), "media_type": "video/mp4", "media_category": "amplify_video", } async with session.post(init_url, data=init_data, headers={"Authorization": f"Bearer {account.access_token}"}) as resp: result = await resp.json() media_id = result.get("media_id_string") if not media_id: raise Exception(f"Twitter media init failed: {result}") # APPEND form = aiohttp.FormData() form.add_field("command", "APPEND") form.add_field("media_id", media_id) form.add_field("segment_index", "0") form.add_field("media_data", video_data) async with session.post(init_url, data=form, headers={"Authorization": f"Bearer {account.access_token}"}) as resp: if resp.status not in (200, 201, 204): raise Exception(f"Twitter media append failed: {await resp.text()}") # FINALIZE async with session.post(init_url, data={"command": "FINALIZE", "media_id": media_id}, headers={"Authorization": f"Bearer {account.access_token}"}) as resp: result = await resp.json() return result.get("media_id_string", media_id) async def create_campaign(self, ad_campaign, account) -> dict: """Create Twitter ad campaign.""" headers = self._headers(account) ad_account_id = account.account_id targeting = ad_campaign.targeting_json or {} async with aiohttp.ClientSession() as session: # 1. Create Funding Instrument (use existing) fi_url = f"{self.BASE_URL}/accounts/{ad_account_id}/funding_instruments" async with session.get(fi_url, headers=headers) as resp: fi_result = await resp.json() fi_list = fi_result.get("data", []) if not fi_list: return {"status": "error", "error": "No funding instrument found"} funding_instrument_id = fi_list[0]["id"] # 2. Create Campaign camp_url = f"{self.BASE_URL}/accounts/{ad_account_id}/campaigns" 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)) camp_data = { "name": ad_campaign.name, "funding_instrument_id": funding_instrument_id, "daily_budget_amount_local_micro": int(ad_campaign.daily_budget * 1_000_000), "start_time": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), "end_time": end_date.strftime("%Y-%m-%dT%H:%M:%SZ"), "entity_status": "PAUSED", } async with session.post(camp_url, json=camp_data, headers=headers) as resp: camp_result = await resp.json() campaign_data = camp_result.get("data", {}) if not campaign_data.get("id"): return {"status": "error", "error": f"Campaign creation failed: {camp_result}"} campaign_id = campaign_data["id"] # 3. Create Line Item li_url = f"{self.BASE_URL}/accounts/{ad_account_id}/line_items" li_data = { "campaign_id": campaign_id, "name": f"{ad_campaign.name} - Line Item", "product_type": "PROMOTED_TWEETS", "placements": ["ALL_ON_TWITTER"], "objective": "VIDEO_VIEWS", "bid_amount_local_micro": 500000, "entity_status": "ACTIVE", } async with session.post(li_url, json=li_data, headers=headers) as resp: li_result = await resp.json() line_item_id = li_result.get("data", {}).get("id") # 4. Set targeting if line_item_id: tc_url = f"{self.BASE_URL}/accounts/{ad_account_id}/targeting_criteria" # Location targeting for loc in targeting.get("locations", ["SA"]): tc_data = { "line_item_id": line_item_id, "targeting_type": "LOCATION", "targeting_value": loc, } await session.post(tc_url, json=tc_data, headers=headers) # Age targeting if targeting.get("age_min"): tc_data = { "line_item_id": line_item_id, "targeting_type": "AGE", "targeting_value": f"AGE_{targeting.get('age_min', 18)}_TO_{targeting.get('age_max', 54)}", } await session.post(tc_url, json=tc_data, headers=headers) # 5. Activate activate_url = f"{self.BASE_URL}/accounts/{ad_account_id}/campaigns/{campaign_id}" await session.put(activate_url, json={"entity_status": "ACTIVE"}, headers=headers) return { "status": "active", "campaign_id": campaign_id, "line_item_id": line_item_id, } async def get_metrics(self, platform_campaign_id: str, account, date_from=None, date_to=None) -> list: """Fetch campaign analytics from Twitter.""" ad_account_id = account.account_id if not date_from: date_from = (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d") if not date_to: date_to = datetime.utcnow().strftime("%Y-%m-%d") url = f"{self.BASE_URL}/stats/accounts/{ad_account_id}" params = { "entity": "CAMPAIGN", "entity_ids": platform_campaign_id, "start_time": f"{date_from}T00:00:00Z", "end_time": f"{date_to}T23:59:59Z", "granularity": "DAY", "metric_groups": "ENGAGEMENT,VIDEO,BILLING", } metrics_list = [] async with aiohttp.ClientSession() as session: async with session.get(url, params=params, headers=self._headers(account)) as resp: data = await resp.json() for entry in data.get("data", []): id_data = entry.get("id_data", []) for daily in id_data: m = daily.get("metrics", {}) views = int(m.get("video_total_views", [0])[0]) if m.get("video_total_views") else 0 clicks = int(m.get("clicks", [0])[0]) if m.get("clicks") else 0 impressions = int(m.get("impressions", [0])[0]) if m.get("impressions") else 0 spend_micro = int(m.get("billed_charge_local_micro", [0])[0]) if m.get("billed_charge_local_micro") else 0 spend = spend_micro / 1_000_000 metrics_list.append({ "date": daily.get("segment", {}).get("start_time", "")[:10], "impressions": impressions, "views": views, "clicks": clicks, "ctr": round((clicks / impressions) * 100, 2) if impressions > 0 else 0, "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": int(m.get("likes", [0])[0]) if m.get("likes") else 0, "comments": int(m.get("replies", [0])[0]) if m.get("replies") else 0, "shares": int(m.get("retweets", [0])[0]) if m.get("retweets") else 0, "reach": 0, "frequency": 0, }) return metrics_list async def pause_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/accounts/{account.account_id}/campaigns/{platform_campaign_id}" async with aiohttp.ClientSession() as session: async with session.put(url, json={"entity_status": "PAUSED"}, headers=self._headers(account)) as resp: return {"status": "paused"} async def resume_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/accounts/{account.account_id}/campaigns/{platform_campaign_id}" async with aiohttp.ClientSession() as session: async with session.put(url, json={"entity_status": "ACTIVE"}, headers=self._headers(account)) as resp: return {"status": "active"} async def delete_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/accounts/{account.account_id}/campaigns/{platform_campaign_id}" async with aiohttp.ClientSession() as session: async with session.delete(url, headers=self._headers(account)) as resp: return {"status": "deleted"}