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-2.3",
"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": "Contemporary dance,the people in the picture are performing contemporary dance.",
# 'first_frame_image' 指定了视频的起始画面
"first_frame_image": "https://filecdn.minimax.chat/public/85c96368-6ead-4eae-af9c-116be878eac3.png",
"model": "MiniMax-Hailuo-2.3",
"duration": 6,
"resolution": "1080P",
}
def invoke_start_end_to_video() -> str:
"""(模式三) 使用首帧图像、尾帧图像和文本描述发起视频生成任务。"""
url = "https://api.minimaxi.com/v1/video_generation"
payload = {
"prompt": "A little girl grow up.",
# 'first_frame_image' 指定了视频的起始画面
"first_frame_image": "https://filecdn.minimax.chat/public/fe9d04da-f60e-444d-a2e0-18ae743add33.jpeg",
# 'last_frame_image' 指定了视频的结束画面
"last_frame_image": "https://filecdn.minimax.chat/public/97b7cd08-764e-4b8b-a7bf-87a0bd898575.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
def invoke_subject_reference() -> str:
"""(模式四) 使用人物主体图片和文本描述发起视频生成任务"""
url = "https://api.minimaxi.com/v1/video_generation"
payload = {
"prompt": "On an overcast day, in an ancient cobbled alleyway, the model is dressed in a brown corduroy jacket paired with beige trousers and ankle boots, topped with a vintage beret. The shot starts from over the model's shoulder, following his steps as it captures his swaying figure. Then, the camera moves slightly sideways to the front, showcasing his natural gesture of adjusting the beret with a smile. Next, the shot slightly tilts down, capturing the model's graceful stance as he leans against the wall at a corner. The video concludes with an upward shot, showing the model smiling at the camera. The lighting and colors are natural, giving the footage a cinematic quality.",
"subject_reference": [
{
"type": "character",
"image": [
"https://filecdn.minimax.chat/public/54be8fbe-5694-4422-9c95-99cf785eb90e.PNG"
],
}
],
"model": "S2V-01",
"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() # 方式二:图生视频
# task_id = invoke_start_end_to_video() # 方式三: 根据首尾帧生成视频
# task_id = invoke_subject_reference() # 方式四: 主体参考生成视频
print(f"视频生成任务已提交,任务 ID: {task_id}")
file_id = query_task_status(task_id)
print(f"任务处理成功,文件 ID: {file_id}")
fetch_video(file_id)