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 integrated report — creative-level performance
report = requests.post(f"{TT_BASE}/reports/integrated/get/",
headers=TT_H,
json={
"advertiser_id": ADV,
"report_type": "BASIC",
"data_level": "AUCTION_AD",
"dimensions": ["ad_id"],
"metrics": ["ad_name", "impressions", "clicks", "ctr", "conversion", "cost_per_conversion",
"spend", "video_play_actions", "video_watched_2s", "video_watched_6s",
"average_video_play_per_user"],
"start_date": "2025-01-01",
"end_date": "2025-12-31",
"page_size": 50,
"page": 1,
}).json()
if report.get("code") != 0:
raise SystemExit(f"TikTok report error: {report.get('message')}")
rows = report.get("data", {}).get("list", [])
# 2. For each creative, fetch video and run analysis
leaderboard = []
for row in rows[:15]:
metrics = row.get("metrics", {})
dims = row.get("dimensions", {})
ad_id = dims.get("ad_id", "")
impressions = int(metrics.get("impressions", 0))
if impressions < 500:
continue
# Fetch ad details for video URL
ad_detail = requests.get(f"{TT_BASE}/ad/get/",
headers=TT_H,
params={"advertiser_id": ADV, "filtering": f'{{"ad_ids": ["{ad_id}"]}}',
"fields": '["video_id", "ad_name"]'}).json()
ad_info = (ad_detail.get("data", {}).get("list") or [{}])[0]
video_id = ad_info.get("video_id")
if not video_id:
continue
vid_info = requests.get(f"{TT_BASE}/file/video/ad/info/",
headers=TT_H,
params={"advertiser_id": ADV, "video_ids": f'["{video_id}"]'}).json()
video_url = (vid_info.get("data", {}).get("list") or [{}])[0].get("video_url", "")
if not video_url:
continue
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_id}.mp4", open(tmp_path, "rb"), "video/mp4")}).json()
analysis = requests.post(f"{MV_BASE}/video-analysis", headers=MV_H, json={
"asset_id": upload["id"],
"analysis_types": ["hook_score", "emotional_arc", "pacing", "cognitive_load"],
}).json()
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
results = status.get("results", {})
hook = results.get("hook_score", {}).get("score", 0)
emotion = results.get("emotional_arc", {}).get("intensity_avg", 0)
ctr = float(metrics.get("ctr", 0))
behavioral_score = (hook * 0.4 + emotion * 6 * 0.3 + (10 - results.get("cognitive_load", {}).get("average", 5)) * 10 * 0.3)
leaderboard.append({
"ad_id": ad_id,
"ad_name": metrics.get("ad_name", ad_info.get("ad_name", "")),
"impressions": impressions,
"ctr": ctr, "spend": float(metrics.get("spend", 0)),
"hook_score": hook, "emotion_avg": emotion,
"behavioral_score": round(behavioral_score, 1),
"verdict": "SCALE" if behavioral_score > 60 and ctr > 1.5 else
"TEST" if behavioral_score > 60 else
"KILL" if behavioral_score < 40 and ctr < 1.0 else "HOLD",
})
os.unlink(tmp_path)
time.sleep(1)
# 3. Print leaderboard
leaderboard.sort(key=lambda x: -x["behavioral_score"])
print(f"{'Rank':<5} {'Ad Name':<30} {'Hook':<6} {'Behavioral':<12} {'CTR':<8} {'Spend':<10} {'Verdict'}")
print("-" * 85)
for i, lb in enumerate(leaderboard, 1):
print(f"{i:<5} {lb['ad_name'][:28]:<30} {lb['hook_score']:<6} {lb['behavioral_score']:<12} {lb['ctr']:.2f}%{'':<4} ${lb['spend']:>8,.0f} {lb['verdict']}")