import requests
import os
from moviepy.editor import VideoFileClip # For extracting video metadata
def get_video_metadata(file_path: str) -> dict:
"""Extract metadata from video file."""
try:
with VideoFileClip(file_path) as clip:
return {
"width": clip.w,
"height": clip.h,
"duration_ms": int(clip.duration * 1000),
"file_size": os.path.getsize(file_path)
}
except Exception as e:
raise Exception(f"Failed to extract video metadata: {e}")
def upload_video_to_mosaic(api_key: str, file_path: str) -> str:
"""Upload a video file to Mosaic using the 3-step process with metadata validation."""
headers = {"Authorization": f"Bearer {api_key}"}
filename = os.path.basename(file_path)
# Step 0: Extract video metadata locally
print("📊 Extracting video metadata...")
try:
metadata = get_video_metadata(file_path)
print(f" ✅ Metadata: {metadata['width']}x{metadata['height']}, "
f"{metadata['duration_ms']/1000:.1f}s, {metadata['file_size']/(1024**2):.1f}MB")
except Exception as e:
print(f" ❌ Failed to extract metadata: {e}")
return None
# Step 1: Get upload URL with metadata validation
print("📤 Step 1: Getting upload URL with validation...")
# Determine content type from file extension
content_type_map = {
'.mp4': 'video/mp4',
'.mov': 'video/quicktime',
'.avi': 'video/x-msvideo',
'.webm': 'video/webm',
'.mkv': 'video/x-matroska',
'.m4v': 'video/x-m4v'
}
file_ext = os.path.splitext(filename)[1].lower()
content_type = content_type_map.get(file_ext, 'video/mp4')
response = requests.post(
"https://api.mosaic.so/videos/get_upload_url",
headers={**headers, "Content-Type": "application/json"},
json={
"filename": filename,
"content_type": content_type,
"file_size": metadata["file_size"],
"width": metadata["width"],
"height": metadata["height"],
"duration_ms": metadata["duration_ms"]
}
)
# Handle validation errors
if response.status_code == 413:
error_data = response.json()
print(f" ❌ File exceeds limits: {error_data.get('detail', 'File too large or too long')}")
return None
elif response.status_code == 400:
error_data = response.json()
print(f" ❌ Invalid metadata: {error_data.get('detail', 'Bad request')}")
return None
elif not response.ok:
print(f" ❌ Failed to get upload URL: {response.status_code}")
return None
upload_data = response.json()
video_id = upload_data["video_id"]
upload_url = upload_data["upload_url"]
method = upload_data["method"]
print(f" ✅ Got video_id: {video_id}")
print(f" ✅ Upload method: {method}")
# Step 2: Upload video using resumable upload
print("⬆️ Step 2: Uploading video...")
with open(file_path, "rb") as f:
if method == "POST":
# Resumable upload with proper headers
upload_response = requests.post(
upload_url,
headers={
"x-goog-resumable": "start",
"Content-Type": content_type,
"Content-Length": str(metadata["file_size"])
},
data=f
)
else:
# Fallback PUT method
upload_response = requests.put(
upload_url,
headers={"Content-Type": content_type},
data=f
)
if upload_response.status_code in [200, 201, 204]:
print(" ✅ Upload successful")
else:
print(f" ❌ Upload failed with status {upload_response.status_code}")
return None
# Step 3: Finalize upload (now faster without Lambda)
print("✅ Step 3: Finalizing upload...")
finalize_response = requests.post(
"https://api.mosaic.so/videos/finalize_upload",
headers={**headers, "Content-Type": "application/json"},
json={"video_id": video_id}
)
if finalize_response.status_code == 200:
print(" ✅ Finalization complete")
print(f"🎉 Upload complete! Video ID: {video_id}")
return video_id
else:
error_data = finalize_response.json()
print(f" ❌ Finalization failed: {error_data.get('detail', 'Unknown error')}")
return None
# Usage example
if __name__ == "__main__":
# Install required dependency: pip install moviepy
api_key = "mk_your_api_key_here"
video_file = "path/to/your/video.mp4"
try:
video_id = upload_video_to_mosaic(api_key, video_file)
if video_id:
print(f"✅ Success! Use video_id '{video_id}' in your agent runs")
else:
print("❌ Upload failed")
except Exception as e:
print(f"❌ Error: {e}")
