视频生成指南

import os
import time
import requests

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


# --- 步骤 1: 发起视频生成任务 ---
# API 支持两种视频生成模式：纯文本生成（文生视频）和图像加文本生成（图生视频）。
# 以下两个函数分别对应这两种模式。它们都会发起一个异步的生成任务，并返回一个唯一的 task_id。

def invoke_text_to_video() -> str:
    """（模式一）通过文本描述发起视频生成任务。"""
    url = "https://api.minimaxi.com/v1/video_generation"
    payload = {
        # 'prompt' 是核心参数，用于描述视频的动态内容。
        "prompt": "男子拿起一本书[上升]，然后阅读[固定]。",
        "model": "MiniMax-Hailuo-02",
        "duration": 6,
        "resolution": "1080P",
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    task_id = response.json()["task_id"]
    return task_id


def invoke_image_to_video() -> str:
    """（模式二）通过首帧图像和文本描述发起视频生成任务。"""
    url = "https://api.minimaxi.com/v1/video_generation"
    payload = {
        # 在图生视频模式下，'prompt' 用于描述基于首帧图像的动态变化。
        "prompt": "老鼠跑向镜头前，微笑着眨了眨眼睛。",
        # 'first_frame_image' 指定了视频的起始画面
        "first_frame_image": "https://cdn.hailuoai.com/prod/2024-09-18-16/user/multi_chat_file/9c0b5c14-ee88-4a5b-b503-4f626f018639.jpeg",
        "model": "MiniMax-Hailuo-02",
        "duration": 6,
        "resolution": "1080P",
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    task_id = response.json()["task_id"]
    return task_id


# --- 步骤 2: 轮询查询任务状态 ---
# 视频生成是一个耗时过程，因此 API 设计为异步模式。
# 提交任务后，需使用 task_id 通过此函数进行轮询，以获取任务的最终状态。
def query_task_status(task_id: str):
    """根据 task_id 轮询任务状态，直至任务成功或失败。"""
    url = "https://api.minimaxi.com/v1/query/video_generation"
    params = {"task_id": task_id}
    while True:
        # 推荐的轮询间隔为 10 秒，以避免对服务器造成不必要的压力。
        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"当前任务状态: {status}")
        # 任务成功时，API 会返回一个 'file_id'，用于下一步获取视频文件。
        if status == "Success":
            return response_json["file_id"]
        elif status == "Fail":
            raise Exception(f"视频生成失败: {response_json.get('error_message', '未知错误')}")


# --- 步骤 3: 获取并保存视频文件 ---
# 任务成功后，我们得到的是 file_id 而非直接的下载链接。
# 此函数首先使用 file_id 从文件服务获取下载 URL，然后下载视频内容并保存到本地。
def fetch_video(file_id: str):
    """根据 file_id 获取视频下载链接，并将其保存到本地。"""
    url = "https://api.minimaxi.com/v1/files/retrieve"
    params = {"file_id": file_id}
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    download_url = response.json()["file"]["download_url"]

    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() # 方式二：图生视频

    print(f"视频生成任务已提交，任务 ID: {task_id}")
    file_id = query_task_status(task_id)
    print(f"任务处理成功，文件 ID: {file_id}")
    fetch_video(file_id)