import aiohttp import json from datetime import datetime, timedelta from . import BaseAdPlatform class TikTokAdsPlatform(BaseAdPlatform): """TikTok Marketing API integration for paid ads.""" BASE_URL = "https://business-api.tiktok.com/open_api/v1.3" def _headers(self, account): return { "Access-Token": account.access_token, "Content-Type": "application/json", } async def upload_video(self, video_url: str, account) -> str: """Upload video to TikTok Ad Account via URL.""" url = f"{self.BASE_URL}/file/video/ad/upload/" payload = { "advertiser_id": account.account_id, "upload_type": "UPLOAD_BY_URL", "video_url": video_url, } async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, headers=self._headers(account)) as resp: result = await resp.json() if result.get("code") == 0: return result["data"]["video_id"] raise Exception(f"TikTok video upload failed: {result.get('message', result)}") async def create_campaign(self, ad_campaign, account) -> dict: """Create full TikTok ad: Campaign → Ad Group → Ad.""" headers = self._headers(account) targeting = ad_campaign.targeting_json or {} advertiser_id = account.account_id async with aiohttp.ClientSession() as session: # 1. Create Campaign camp_url = f"{self.BASE_URL}/campaign/create/" camp_payload = { "advertiser_id": advertiser_id, "campaign_name": ad_campaign.name, "objective_type": "VIDEO_VIEWS", "budget_mode": "BUDGET_MODE_TOTAL", "budget": ad_campaign.budget_amount, } async with session.post(camp_url, json=camp_payload, headers=headers) as resp: camp_result = await resp.json() if camp_result.get("code") != 0: return {"status": "error", "error": f"Campaign failed: {camp_result.get('message')}"} campaign_id = camp_result["data"]["campaign_id"] # 2. Upload video video_id = None if ad_campaign.campaign and ad_campaign.campaign.video_url: try: video_id = await self.upload_video(ad_campaign.campaign.video_url, account) except Exception as e: return {"status": "error", "error": f"Video upload failed: {e}"} # 3. Create Ad Group (targeting + budget) adgroup_url = f"{self.BASE_URL}/adgroup/create/" 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)) location_ids = targeting.get("tiktok_location_ids", []) if not location_ids: location_ids = ["6252001"] # Default: Saudi Arabia adgroup_payload = { "advertiser_id": advertiser_id, "campaign_id": campaign_id, "adgroup_name": f"{ad_campaign.name} - Ad Group", "placement_type": "PLACEMENT_TYPE_AUTOMATIC", "budget_mode": "BUDGET_MODE_DAY", "budget": ad_campaign.daily_budget, "schedule_type": "SCHEDULE_START_END", "schedule_start_time": start_date.strftime("%Y-%m-%d %H:%M:%S"), "schedule_end_time": end_date.strftime("%Y-%m-%d %H:%M:%S"), "optimization_goal": "VIDEO_VIEW", "billing_event": "CPC", "location_ids": location_ids, } if targeting.get("age_min") or targeting.get("age_max"): age_groups = self._map_age_to_tiktok(targeting.get("age_min", 18), targeting.get("age_max", 65)) if age_groups: adgroup_payload["age_groups"] = age_groups if targeting.get("gender") and targeting["gender"] != "all": adgroup_payload["gender"] = "GENDER_MALE" if targeting["gender"] == "male" else "GENDER_FEMALE" if targeting.get("interests"): adgroup_payload["interest_category_ids"] = targeting["interests"] async with session.post(adgroup_url, json=adgroup_payload, headers=headers) as resp: adgroup_result = await resp.json() if adgroup_result.get("code") != 0: return {"status": "error", "error": f"Ad Group failed: {adgroup_result.get('message')}"} adgroup_id = adgroup_result["data"]["adgroup_id"] # 4. Create Ad ad_url = f"{self.BASE_URL}/ad/create/" ad_payload = { "advertiser_id": advertiser_id, "adgroup_id": adgroup_id, "creatives": [{ "ad_name": f"{ad_campaign.name} - Ad", "ad_text": ad_campaign.ad_caption or ad_campaign.name, "call_to_action": ad_campaign.ad_cta or "LEARN_MORE", }], } if video_id: ad_payload["creatives"][0]["video_id"] = video_id async with session.post(ad_url, json=ad_payload, headers=headers) as resp: ad_result = await resp.json() if ad_result.get("code") != 0: return {"status": "error", "error": f"Ad creation failed: {ad_result.get('message')}"} ad_ids = ad_result["data"].get("ad_ids", []) return { "status": "active", "campaign_id": campaign_id, "adgroup_id": adgroup_id, "ad_ids": ad_ids, "video_id": video_id, } async def get_metrics(self, platform_campaign_id: str, account, date_from=None, date_to=None) -> list: """Fetch daily metrics from TikTok Reporting API.""" url = f"{self.BASE_URL}/report/integrated/get/" 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") payload = { "advertiser_id": account.account_id, "report_type": "BASIC", "dimensions": ["campaign_id", "stat_time_day"], "data_level": "AUCTION_CAMPAIGN", "start_date": date_from, "end_date": date_to, "metrics": [ "spend", "impressions", "clicks", "ctr", "reach", "video_views_p25", "video_views_p50", "video_views_p75", "video_views_p100", "likes", "comments", "shares", "frequency", ], "filters": [{"field_name": "campaign_id", "filter_type": "IN", "filter_value": json.dumps([platform_campaign_id])}], "page_size": 100, } metrics_list = [] async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, headers=self._headers(account)) as resp: result = await resp.json() for row in result.get("data", {}).get("list", []): dims = row.get("dimensions", {}) m = row.get("metrics", {}) views = int(m.get("video_views_p25", 0)) impressions = int(m.get("impressions", 0)) clicks = int(m.get("clicks", 0)) spend = float(m.get("spend", 0)) metrics_list.append({ "date": dims.get("stat_time_day", ""), "impressions": impressions, "views": views, "clicks": clicks, "ctr": float(m.get("ctr", 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)), "comments": int(m.get("comments", 0)), "shares": int(m.get("shares", 0)), "reach": int(m.get("reach", 0)), "frequency": float(m.get("frequency", 0)), }) return metrics_list async def pause_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/campaign/status/update/" payload = { "advertiser_id": account.account_id, "campaign_ids": [platform_campaign_id], "opt_status": "DISABLE", } async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, headers=self._headers(account)) as resp: result = await resp.json() return {"status": "paused"} if result.get("code") == 0 else {"status": "error", "error": result.get("message")} async def resume_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/campaign/status/update/" payload = { "advertiser_id": account.account_id, "campaign_ids": [platform_campaign_id], "opt_status": "ENABLE", } async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, headers=self._headers(account)) as resp: result = await resp.json() return {"status": "active"} if result.get("code") == 0 else {"status": "error", "error": result.get("message")} async def delete_campaign(self, platform_campaign_id: str, account) -> dict: url = f"{self.BASE_URL}/campaign/status/update/" payload = { "advertiser_id": account.account_id, "campaign_ids": [platform_campaign_id], "opt_status": "DELETE", } async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, headers=self._headers(account)) as resp: result = await resp.json() return {"status": "deleted"} if result.get("code") == 0 else {"status": "error", "error": result.get("message")} @staticmethod def _map_age_to_tiktok(age_min, age_max): """Map age range to TikTok age group codes.""" groups = [] tiktok_ages = [ ("AGE_13_17", 13, 17), ("AGE_18_24", 18, 24), ("AGE_25_34", 25, 34), ("AGE_35_44", 35, 44), ("AGE_45_54", 45, 54), ("AGE_55_100", 55, 100), ] for code, lo, hi in tiktok_ages: if age_min <= hi and age_max >= lo: groups.append(code) return groups