Quickstart: Video Analysis

import requests

API_KEY = "mvra_live_your_key_here"
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
WORKSPACE_ID = "ws_your_workspace_id"
BASE = "https://app.mavera.io/api/v1"

# Path to your video file
VIDEO_PATH = "path/to/your/clip.mp4"

with open(VIDEO_PATH, "rb") as f:
    content = f.read()

file_size = len(content)
file_name = VIDEO_PATH.split("/")[-1]
file_type = "video/mp4"  # or video/quicktime for MOV

upload_resp = requests.post(
    f"{BASE}/files/upload-url",
    headers=HEADERS,
    json={
        "file_name": file_name,
        "file_type": file_type,
        "file_size": file_size,
        "workspace_id": WORKSPACE_ID,
    },
).json()

if "error" in upload_resp:
    raise Exception(upload_resp["error"]["message"])

upload_url = upload_resp["upload_url"]
public_url = upload_resp["public_url"]
print(f"Upload URL expires in {upload_resp.get('expires_in', '?')}s")

put_resp = requests.put(
    upload_url,
    data=content,
    headers={"Content-Type": file_type},
)
put_resp.raise_for_status()

file_resp = requests.post(
    f"{BASE}/files",
    headers=HEADERS,
    json={
        "name": file_name,
        "type": file_type,
        "url": public_url,
        "workspace_id": WORKSPACE_ID,
        "file_size": file_size,
    },
).json()

if "error" in file_resp:
    raise Exception(file_resp["error"]["message"])

asset_id = file_resp["id"]
print(f"File created; asset_id for video analysis: {asset_id}")

analysis_payload = {
    "title": "Quickstart: Ad Clip Analysis",
    "asset_id": asset_id,
    "goal": "Analyze viewer engagement and emotional response",
    "brand": "Your Brand",
    "product": "Product Name",
    "primary_intent": "Drive product awareness",
    "chunk_duration": 5,
    "frames_per_chunk": 3,
    "workspace_id": WORKSPACE_ID,
}

create_resp = requests.post(
    f"{BASE}/video-analyses",
    headers=HEADERS,
    json=analysis_payload,
).json()

if "error" in create_resp:
    raise Exception(create_resp["error"]["message"])

analysis_id = create_resp["id"]
status = create_resp["status"]
print(f"Analysis created: {analysis_id}")
print(f"Status: {status}")

import time

def get_analysis(aid):
    r = requests.get(f"{BASE}/video-analyses/{aid}", headers=HEADERS)
    return r.json()

for _ in range(40):
    analysis = get_analysis(analysis_id)
    if "error" in analysis:
        raise Exception(analysis["error"]["message"])
    if analysis.get("status") == "COMPLETED":
        break
    print(f"Status: {analysis.get('status')}; waiting 15s...")
    time.sleep(15)
else:
    raise Exception("Analysis did not complete in time")

metrics = analysis["results"]["full_video_metrics"]
print("Overall score:", metrics.get("overall_score"))
print("Emotional impact:", metrics.get("emotional_impact"))
print("Attention score:", metrics.get("attention_score"))
print("CTA effectiveness:", metrics.get("cta_effectiveness"))
print("Brand recall likelihood:", metrics.get("brand_recall_likelihood"))

print("\nChunk breakdown:")
for chunk in analysis["results"].get("chunks", []):
    print(f"  {chunk['start_time']}s–{chunk['end_time']}s: engagement={chunk.get('engagement_score')}, valence={chunk.get('emotional_valence')}")

print("\nRecommendations:", analysis["results"].get("recommendations", []))
print("Credits used:", analysis.get("usage", {}).get("credits_used"))

chat_resp = requests.post(
    f"{BASE}/video-analyses/{analysis_id}/chat",
    headers=HEADERS,
    json={
        "message": "What are the weakest moments in this video and how can I improve them?"
    },
).json()

if "error" not in chat_resp:
    print(chat_resp.get("content", ""))