> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mavera.io/llms.txt
> Use this file to discover all available pages before exploring further.
# Ad Video Analysis Pipeline
> Pull TikTok video ad URLs, upload to Mavera, and run Video Analysis to extract hook scoring, emotional arc, and cognitive load diagnostics.
### Scenario
Your TikTok ad account has dozens of video creatives, but you only see surface metrics — impressions, CTR, CPM. This job pulls video ad URLs, uploads them to Mavera Assets, and runs Video Analysis to extract behavioral intelligence that TikTok doesn't provide: hook scoring (first 3 seconds), emotional arc across the full duration, and cognitive load at each cut point. The result is a creative diagnostic that tells you *why* an ad works, not just *that* it works.
### Architecture
```mermaid theme={"dark"}
flowchart LR
A["TikTok GET /ad/get/"] --> B["Download video URLs"]
B --> C["Mavera POST /assets (upload)"]
C --> D["POST /video-analysis"]
D --> E["Hook score, emotional arc, cognitive load"]
```
### Code
```python Python theme={"dark"}
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")
```
```javascript JavaScript theme={"dark"}
const TT = process.env.TIKTOK_ACCESS_TOKEN;
const ADV = process.env.TIKTOK_ADVERTISER_ID;
const MV = process.env.MAVERA_API_KEY;
const TT_BASE = "https://business-api.tiktok.com/open_api/v1.3";
const MV_BASE = "https://app.mavera.io/api/v1";
const TT_H = { "Access-Token": TT };
const MV_H = { Authorization: `Bearer ${MV}`, "Content-Type": "application/json" };
// 1. Pull active video ads
const adsRes = await fetch(
`${TT_BASE}/ad/get/?advertiser_id=${ADV}` +
`&filtering=${encodeURIComponent('{"status":"AD_STATUS_DELIVERY_OK","creative_type":"VIDEO"}')}` +
`&fields=${encodeURIComponent('["ad_id","ad_name","video_id"]')}&page_size=20`,
{ headers: TT_H }
);
const adsData = await adsRes.json();
if (adsData.code !== 0) throw new Error(`TikTok: ${adsData.message}`);
const ads = adsData.data?.list || [];
// 2. Process each ad
const videoAnalyses = [];
for (const ad of ads.slice(0, 10)) {
if (!ad.video_id) continue;
const vidRes = await fetch(
`${TT_BASE}/file/video/ad/info/?advertiser_id=${ADV}&video_ids=${encodeURIComponent(`["${ad.video_id}"]`)}`,
{ headers: TT_H }
).then(r => r.json());
const videoUrl = vidRes.data?.list?.[0]?.video_url || vidRes.data?.list?.[0]?.preview_url;
if (!videoUrl) continue;
// 3. Download and upload to Mavera
const vidBytes = await fetch(videoUrl).then(r => r.arrayBuffer());
const formData = new FormData();
formData.append("file", new Blob([vidBytes], { type: "video/mp4" }), `${ad.ad_name}.mp4`);
const upload = await fetch(`${MV_BASE}/assets`, {
method: "POST",
headers: { Authorization: `Bearer ${MV}` },
body: formData,
}).then(r => r.json());
// 4. Run Video Analysis
const analysis = await fetch(`${MV_BASE}/video-analysis`, {
method: "POST", headers: MV_H,
body: JSON.stringify({
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 },
}),
}).then(r => r.json());
// 5. Poll
let status;
for (let i = 0; i < 30; i++) {
await new Promise(r => setTimeout(r, 3000));
status = await fetch(`${MV_BASE}/video-analysis/${analysis.id}`, { headers: MV_H }).then(r => r.json());
if (status.status === "completed") break;
}
videoAnalyses.push({
ad_id: ad.ad_id, ad_name: ad.ad_name,
hook_score: status.results?.hook_score || {},
emotional_arc: status.results?.emotional_arc || {},
cognitive_load: status.results?.cognitive_load || {},
});
await new Promise(r => setTimeout(r, 1000));
}
// 6. Scorecard
for (const va of videoAnalyses) {
console.log(`\n${"=".repeat(50)}`);
console.log(`AD: ${va.ad_name} (${va.ad_id})`);
console.log(` Hook Score: ${va.hook_score.score ?? "N/A"}/100 — ${va.hook_score.assessment || ""}`);
console.log(` Emotional Peak: ${va.emotional_arc.peak_emotion ?? "N/A"} at ${va.emotional_arc.peak_timestamp ?? "N/A"}s`);
console.log(` Avg Cognitive Load: ${va.cognitive_load.average ?? "N/A"}/10`);
}
```
### Example Output
```text theme={"dark"}
==================================================
AD: Summer Sale — UGC Creator (ad_12345)
Hook Score: 87/100 — Strong opening: text overlay + face close-up in first 0.8s
Emotional Peak: excitement at 4.2s
Avg Cognitive Load: 3.8/10
==================================================
AD: Product Demo — B-Roll (ad_12346)
Hook Score: 42/100 — Slow start: logo animation for first 2.1s loses attention
Emotional Peak: curiosity at 8.5s
Avg Cognitive Load: 6.2/10
==================================================
AD: Founder Story (ad_12347)
Hook Score: 71/100 — Direct eye contact + bold claim in first 1.2s
Emotional Peak: trust at 12.0s
Avg Cognitive Load: 4.1/10
```
### Error Handling
TikTok returns `code: 0` for success. Non-zero codes include `40001` (auth failure), `40100` (permission denied), and `40002` (invalid params). Always check the `code` field, not just HTTP status.
Some ad video URLs are CDN-signed and expire. If download fails, re-fetch the video info endpoint for a fresh URL.
TikTok ads over 60s or high-resolution may take 30s+ to upload. The poll loop allows 90s. Increase for longer creatives.
***