Video Generation with Templates Guide

import os
import time
import requests

api_key = os.environ["MINIMAX_API_KEY"]
headers = {"Authorization": f"Bearer {api_key}"}

# --- Step 1: Submit a video generation task ---
# This function calls the API to start an asynchronous template-based video generation task.
# Upon success, the API immediately returns a task_id for querying task status.
def invoke_template_task() -> str:
    """Submit a template-based video generation task and return the task ID"""
    url = "https://api.minimax.io/v1/video_template_generation"
    payload = {
        # 'template_id' specifies the base video template.
        "template_id": "393769180141805569",  # Example: Run for Life style
        # Media assets such as images or videos to fill in the template
        "media_inputs": [
            {
                "value": "https://cdn.hailuoai.com/prod/2024-09-18-16/user/multi_chat_file/9c0b5c14-ee88-4a5b-b503-4f626f018639.jpeg"
            }
        ],
        # Text inputs for filling text placeholders in the template
        "text_inputs": [{"value": "Lion"}],
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    task_id = response.json()["task_id"]
    return task_id

# --- Step 2: Poll task status ---
# Since video generation is asynchronous, you need to check status periodically using task_id.
# Once status becomes "Success", the function returns the video URL; if failed, an exception is raised.
def query_task_status(task_id: str):
    url = "https://api.minimax.io/v1/query/video_template_generation"
    params = {"task_id": task_id}
    while True:
        # A reasonable polling interval is recommended to avoid excessive requests.
        time.sleep(10)
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        response_json = response.json()
        status = response_json["status"]
        print(f"Current task status: {status}")
        if status == "Success":
            return response_json["video_url"]
        elif status == "Fail":
            raise Exception(f"Video generation failed: {response_json}")

# --- Step 3: Save the video file ---
# This helper function downloads the generated video from the provided URL and saves it locally.
def save_video_from_url(video_url: str):
    print(f"Downloading video from {video_url}...")
    response = requests.get(video_url)
    response.raise_for_status()
    with open("output.mp4", "wb") as f:
        f.write(response.content)
    print("Video successfully saved as output.mp4")

# --- Main process: Full workflow ---
# Execute the entire flow in the order: submit -> poll -> save.
if __name__ == "__main__":
    task_id = invoke_template_task()
    print(f"Video generation task submitted successfully, task_id: {task_id}")
    final_video_url = query_task_status(task_id)
    print(f"Task completed successfully, video URL: {final_video_url}")
    save_video_from_url(final_video_url)