视频生成

import os
import time
import requests

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


# --- 步骤 1: 发起视频生成任务 ---
# MiniMax-H3 使用多模态 content[] 结构：每个元素通过 type（text / image_url / video_url / audio_url）区分，
# 并可用 role 标注用途。以下四个函数分别对应文生视频、图生视频、首尾帧、多模态参考四种模式，
# 都会发起一个异步任务并返回唯一的 task_id。

def invoke_text_to_video() -> str:
    """（模式一）纯文本生成视频（t2va）。t2va 场景 ratio 必填且不能为 adaptive。"""
    url = f"{BASE_URL}/v2/video_generation"
    payload = {
        "model": MODEL,
        "content": [
            # type=text 为必填项，用于描述视频的动态内容。
            {"type": "text", "text": "镜头拍摄一个女性坐在咖啡馆里，女人抬头看着窗外，镜头缓缓移动拍摄到窗外的街道，画面呈现暖色调，色彩浓郁，氛围轻松惬意。"},
        ],
        "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:
    """（模式二）首帧图 + 文本生成视频（i2va）。"""
    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 指定视频起始帧；图生视频场景下宽高比由输入图片决定，ratio 恒为 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:
    """（模式三）首帧图 + 尾帧图 + 文本生成视频。"""
    url = f"{BASE_URL}/v2/video_generation"
    payload = {
        "model": MODEL,
        "content": [
            {"type": "text", "text": "A little girl grows up."},
            # role=first_frame 指定起始画面
            {"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/fe9d04da-f60e-444d-a2e0-18ae743add33.jpeg"}, "role": "first_frame"},
            # role=last_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:
    """（模式四）多模态参考生视频（r2va）：可组合参考图 / 参考视频 / 参考音频。"""
    url = f"{BASE_URL}/v2/video_generation"
    payload = {
        "model": MODEL,
        "content": [
            {"type": "text", "text": "On an overcast day, in an ancient cobbled alleyway, the model walks and adjusts a vintage beret with a smile; natural lighting and cinematic colors."},
            # role=reference_image 提供人物/主体参考；也可加入 role=reference_video / reference_audio 作为参考。
            {"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/54be8fbe-5694-4422-9c95-99cf785eb90e.PNG"}, "role": "reference_image"},
        ],
        "duration": 5,
        "resolution": "2K",
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()["task_id"]


# --- 步骤 2: 轮询查询任务状态 ---
# 视频生成是一个耗时过程，因此 API 设计为异步模式。
# 提交任务后，需使用 task_id 通过此函数进行轮询。任务成功后直接返回成片下载地址（content.url），无需再换 file_id。
def query_task_status(task_id: str) -> str:
    """根据 task_id 轮询任务状态，成功后返回成片下载地址。"""
    url = f"{BASE_URL}/v2/query/video_generation/{task_id}"
    while True:
        # 推荐的轮询间隔为 10 秒，以避免对服务器造成不必要的压力。
        time.sleep(10)
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        task = response.json()["task"]
        status = task["status"]
        print(f"当前任务状态: {status}")
        # 成功时 task.content.url 即为成片下载地址。
        if status == "succeeded":
            return task["content"]["url"]
        # 终态失败：failed / cancelled / expired。
        if status in ("failed", "cancelled", "expired"):
            raise Exception(f"视频生成未成功: status={status}, error={task.get('error')}")


# --- 步骤 3: 下载并保存视频文件 ---
# 任务成功后直接得到成片下载地址，下载内容并保存到本地即可。
def fetch_video(download_url: str):
    """下载成片并保存到本地。"""
    with open("output.mp4", "wb") as f:
        video_response = requests.get(download_url)
        video_response.raise_for_status()
        f.write(video_response.content)
    print("视频已成功保存至 output.mp4")


# --- 主流程: 完整调用示例 ---
# 该部分演示了从发起任务到最终保存视频的完整调用链路。
if __name__ == "__main__":
    # 选择一种方式创建任务
    task_id = invoke_text_to_video()  # 方式一：文生视频
    # task_id = invoke_image_to_video() # 方式二：图生视频
    # task_id = invoke_start_end_to_video() # 方式三: 根据首尾帧生成视频
    # task_id = invoke_reference_to_video() # 方式四: 多模态参考生视频

    print(f"视频生成任务已提交，任务 ID: {task_id}")
    download_url = query_task_status(task_id)
    print(f"任务处理成功，成片地址: {download_url}")
    fetch_video(download_url)