import os, requests, time, tempfile
TT = os.environ["TIKTOK_ACCESS_TOKEN"]
ADV = os.environ["TIKTOK_ADVERTISER_ID"]
MV = os.environ["MAVERA_API_KEY"]
TT_BASE = "https://business-api.tiktok.com/open_api/v1.3"
MV_BASE = "https://app.mavera.io/api/v1"
TT_H = {"Access-Token": TT}
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
# 1. Pull active video ads
r = requests.get(f"{TT_BASE}/ad/get/",
headers=TT_H,
params={
"advertiser_id": ADV,
"filtering": '{"status": "AD_STATUS_DELIVERY_OK", "creative_type": "VIDEO"}',
"fields": '["ad_id", "ad_name", "video_id", "image_ids", "landing_page_url"]',
"page_size": 20,
})
r.raise_for_status()
data = r.json()
if data.get("code") != 0:
raise SystemExit(f"TikTok API error: {data.get('message')}")
ads = data.get("data", {}).get("list", [])
# 2. Get video URLs for each ad
video_analyses = []
for ad in ads[:10]:
video_id = ad.get("video_id")
if not video_id:
continue
vid_r = requests.get(f"{TT_BASE}/file/video/ad/info/",
headers=TT_H,
params={"advertiser_id": ADV, "video_ids": f'["{video_id}"]'})
vid_r.raise_for_status()
vid_data = vid_r.json().get("data", {}).get("list", [])
if not vid_data:
continue
video_url = vid_data[0].get("video_url") or vid_data[0].get("preview_url", "")
if not video_url:
continue
# 3. Download and upload to Mavera
vid_bytes = requests.get(video_url).content
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
tmp.write(vid_bytes)
tmp_path = tmp.name
upload = requests.post(f"{MV_BASE}/assets",
headers={"Authorization": f"Bearer {MV}"},
files={"file": (f"{ad['ad_name']}.mp4", open(tmp_path, "rb"), "video/mp4")},
).json()
# 4. Run Video Analysis
analysis = requests.post(f"{MV_BASE}/video-analysis", headers=MV_H, json={
"asset_id": upload["id"],
"analysis_types": ["hook_score", "emotional_arc", "cognitive_load", "visual_complexity", "pacing"],
"metadata": {"source": "tiktok", "ad_id": ad["ad_id"], "ad_name": ad["ad_name"]},
}).json()
# 5. Poll for completion
for _ in range(30):
time.sleep(3)
status = requests.get(f"{MV_BASE}/video-analysis/{analysis['id']}", headers=MV_H).json()
if status.get("status") == "completed":
break
video_analyses.append({
"ad_id": ad["ad_id"],
"ad_name": ad["ad_name"],
"hook_score": status.get("results", {}).get("hook_score", {}),
"emotional_arc": status.get("results", {}).get("emotional_arc", {}),
"cognitive_load": status.get("results", {}).get("cognitive_load", {}),
})
os.unlink(tmp_path)
time.sleep(1)
# 6. Output scorecard
for va in video_analyses:
hook = va["hook_score"]
print(f"\n{'='*50}")
print(f"AD: {va['ad_name']} ({va['ad_id']})")
print(f" Hook Score: {hook.get('score', 'N/A')}/100 — {hook.get('assessment', '')}")
print(f" Emotional Peak: {va['emotional_arc'].get('peak_emotion', 'N/A')} at {va['emotional_arc'].get('peak_timestamp', 'N/A')}s")
print(f" Avg Cognitive Load: {va['cognitive_load'].get('average', 'N/A')}/10")
