Video Generation

import os
import time
import requests

api_key = os.environ["MINIMAX_API_KEY"]
headers = {"Authorization": f"Bearer {api_key}"}
BASE_URL = "https://api.minimax.io"
MODEL = "MiniMax-H3"


# --- Step 1: Create a video generation task ---
# MiniMax-H3 uses a multimodal content[] structure: each element is distinguished by type
# (text / image_url / video_url / audio_url) and can be labeled with a role. Each function below
# corresponds to one mode (text-to-video, image-to-video, first-and-last-frame, reference-to-video),
# starts an asynchronous task, and returns a unique task_id.

def invoke_text_to_video() -> str:
    """(Mode 1) Text-to-video (t2va). For t2va, ratio is required and cannot be 'adaptive'."""
    url = f"{BASE_URL}/v2/video_generation"
    payload = {
        "model": MODEL,
        "content": [
            # A type=text item is required and defines the video's content and motion.
            {"type": "text", "text": "A tiktok dancer is dancing on a drone, doing flips and tricks."},
        ],
        "duration": 5,
        "resolution": "2K",
        "ratio": "16:9",
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()["task_id"]


def invoke_image_to_video() -> str:
    """(Mode 2) Image-to-video (i2va) using a first-frame image and text."""
    url = f"{BASE_URL}/v2/video_generation"
    payload = {
        "model": MODEL,
        "content": [
            {"type": "text", "text": "Contemporary dance, the people in the picture are performing contemporary dance."},
            # role=first_frame specifies the opening frame; for image-to-video the aspect ratio is
            # determined by the input image and ratio is always 'adaptive'.
            {"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/85c96368-6ead-4eae-af9c-116be878eac3.png"}, "role": "first_frame"},
        ],
        "duration": 5,
        "resolution": "2K",
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()["task_id"]


def invoke_start_end_to_video() -> str:
    """(Mode 3) First-frame + last-frame image + text."""
    url = f"{BASE_URL}/v2/video_generation"
    payload = {
        "model": MODEL,
        "content": [
            {"type": "text", "text": "A little girl grows up."},
            # role=first_frame specifies the opening frame
            {"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/fe9d04da-f60e-444d-a2e0-18ae743add33.jpeg"}, "role": "first_frame"},
            # role=last_frame specifies the ending frame
            {"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/97b7cd08-764e-4b8b-a7bf-87a0bd898575.jpeg"}, "role": "last_frame"},
        ],
        "duration": 5,
        "resolution": "2K",
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()["task_id"]


def invoke_reference_to_video() -> str:
    """(Mode 4) Reference-to-video (r2va): combine reference images / videos / audio."""
    url = f"{BASE_URL}/v2/video_generation"
    payload = {
        "model": MODEL,
        "content": [
            {"type": "text", "text": "The character performs street dance, following the motion in reference video 1; the character's appearance follows reference images 1 and 2."},
            {"type": "image_url", "image_url": {"url": "https://cdn.hailuoai.com/prod/hailuo_demo/testsets/h3_promo_eval_ref2va/gallery/sr_v2p26_trio_seed42_20260724/inputs/9d5c7a33fa6e_01_%E5%9B%BE1_MHGgbVga3o_gpt4o-image-1780651118146.png"}, "role": "reference_image"},
            {"type": "image_url", "image_url": {"url": "https://cdn.hailuoai.com/prod/hailuo_demo/testsets/h3_promo_eval_ref2va/gallery/sr_v2p26_trio_seed42_20260724/inputs/7af326902315_00_%E5%9B%BE2_YqtKbY1jpo_u4391985813_Young_male_wearing_cream_hoodie_and_dark_brown_sh_45d56ed0-d626-4c37-9a5b-77f51f374982_1.png"}, "role": "reference_image"},
            {"type": "video_url", "video_url": {"url": "https://cdn.hailuoai.com/prod/hailuo_demo/testsets/h3_promo_eval_ref2va/gallery/sr_v2p26_trio_seed42_20260724/inputs/d9060f5cb9ab_02_%E8%A7%86%E9%A2%911_HDvmbpQrEo_%E8%A1%97%E8%88%9E3.mp4"}, "role": "reference_video"},
        ],
        "duration": 5,
        "resolution": "2K",
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()["task_id"]


# --- Step 2: Poll task status ---
# Since video generation is time-consuming, the API works asynchronously.
# After submitting a task, poll its status using the task_id. On success the response directly
# returns the video download URL (content.url) — no file_id exchange is needed.
def query_task_status(task_id: str) -> str:
    """Poll task status by task_id and return the video download URL on success."""
    url = f"{BASE_URL}/v2/query/video_generation/{task_id}"
    while True:
        # A recommended polling interval is 10 seconds to avoid unnecessary server load.
        time.sleep(10)
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        task = response.json()["task"]
        status = task["status"]
        print(f"Current task status: {status}")
        # On success, task.content.url is the video download URL.
        if status == "succeeded":
            return task["content"]["url"]
        # Terminal failure states: failed / cancelled.
        if status in ("failed", "cancelled"):
            raise Exception(f"Video generation did not succeed: status={status}, error={task.get('error')}")


# --- Step 3: Download and save the video file ---
# On success you get the download URL directly; download the content and save it locally.
def fetch_video(download_url: str):
    """Download the video and save it locally."""
    with open("output.mp4", "wb") as f:
        video_response = requests.get(download_url)
        video_response.raise_for_status()
        f.write(video_response.content)
    print("Video successfully saved as output.mp4")


# --- Main process: end-to-end example ---
# Demonstrates the full workflow from task creation to video retrieval.
if __name__ == "__main__":
    # Choose a task creation mode
    task_id = invoke_text_to_video()  # Mode 1: Text-to-Video
    # task_id = invoke_image_to_video() # Mode 2: Image-to-Video
    # task_id = invoke_start_end_to_video() # Mode 3: First-and-Last-Frame Video
    # task_id = invoke_reference_to_video() # Mode 4: Reference-to-Video
    print(f"Video generation task submitted, Task ID: {task_id}")
    download_url = query_task_status(task_id)
    print(f"Task succeeded, video URL: {download_url}")
    fetch_video(download_url)