223 lines
7.7 KiB
Python
223 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Image-to-Video 生成脚本 — 支持 MiniMax Hailuo / Kling API
|
||
|
||
用法:
|
||
# MiniMax Hailuo（默认）
|
||
python3 i2v_video.py --image photo.jpg --prompt "camera slowly pushes in"
|
||
python3 i2v_video.py --image photo.jpg --prompt "..." --duration 6 --resolution 1080P
|
||
|
||
# Kling 可灵
|
||
python3 i2v_video.py --image photo.jpg --prompt "..." --provider kling
|
||
|
||
环境变量:
|
||
MINIMAX_API_KEY — MiniMax 平台 API key (platform.minimax.io)
|
||
KLING_API_KEY — Kling 平台 API key (klingai.com)
|
||
"""
|
||
|
||
import argparse, json, os, sys, time, subprocess
|
||
from pathlib import Path
|
||
|
||
def get_key(env_name):
|
||
"""从 .env 或环境变量获取 key"""
|
||
key = os.environ.get(env_name, "")
|
||
if not key:
|
||
env_file = os.path.expanduser("~/.hermes/.env")
|
||
if os.path.exists(env_file):
|
||
for line in open(env_file):
|
||
line = line.strip()
|
||
if line.startswith(f"{env_name}=") and not line.startswith("#"):
|
||
key = line.split("=", 1)[1].strip().strip('"').strip("'")
|
||
break
|
||
return key
|
||
|
||
def upload_image_to_url(image_path):
|
||
"""将本地图片上传到临时图床，返回公网 URL"""
|
||
# 用 catbox.moe 免费图床
|
||
result = subprocess.run(
|
||
["curl", "-s", "-F", f"reqtype=fileupload", "-F", f"fileToUpload=@{image_path}",
|
||
"https://catbox.moe/user/api.php"],
|
||
capture_output=True, text=True, timeout=30
|
||
)
|
||
url = result.stdout.strip()
|
||
if url.startswith("http"):
|
||
return url
|
||
# fallback: 用 0x0.st
|
||
result = subprocess.run(
|
||
["curl", "-s", "-F", f"file=@{image_path}", "https://0x0.st"],
|
||
capture_output=True, text=True, timeout=30
|
||
)
|
||
url = result.stdout.strip()
|
||
if url.startswith("http"):
|
||
return url
|
||
return ""
|
||
|
||
def minimax_i2v(image_url, prompt, duration=6, resolution="720P"):
|
||
"""MiniMax Hailuo 图生视频"""
|
||
key = get_key("MINIMAX_API_KEY")
|
||
if not key:
|
||
print("❌ 未找到 MINIMAX_API_KEY")
|
||
print(" 去 https://platform.minimax.io 注册获取免费 key")
|
||
print(" 然后: echo 'MINIMAX_API_KEY=你的key' >> ~/.hermes/.env")
|
||
return None
|
||
|
||
# 提交任务
|
||
data = {
|
||
"model": "MiniMax-Hailuo-2.3",
|
||
"prompt": prompt,
|
||
"first_frame_image": image_url,
|
||
"duration": duration,
|
||
"resolution": resolution,
|
||
}
|
||
result = subprocess.run(
|
||
["curl", "-s", "-X", "POST", "https://api.minimax.io/v1/video_generation",
|
||
"-H", f"Authorization: Bearer {key}",
|
||
"-H", "Content-Type: application/json",
|
||
"-d", json.dumps(data)],
|
||
capture_output=True, text=True, timeout=30
|
||
)
|
||
resp = json.loads(result.stdout)
|
||
task_id = resp.get("task_id", "")
|
||
if not task_id:
|
||
print(f"❌ 提交失败: {resp}")
|
||
return None
|
||
|
||
print(f"✅ 任务已提交: {task_id}")
|
||
print(f" 模型: MiniMax-Hailuo-2.3")
|
||
print(f" 时长: {duration}秒, 分辨率: {resolution}")
|
||
|
||
# 轮询
|
||
for i in range(60):
|
||
time.sleep(5)
|
||
status_resp = subprocess.run(
|
||
["curl", "-s", f"https://api.minimax.io/v1/video_generation/{task_id}",
|
||
"-H", f"Authorization: Bearer {key}"],
|
||
capture_output=True, text=True, timeout=30
|
||
)
|
||
status_data = json.loads(status_resp.stdout)
|
||
status = status_data.get("status", "unknown")
|
||
progress = status_data.get("progress", 0)
|
||
print(f"\r [{i+1}] {status} ({progress}%)", end="", flush=True)
|
||
|
||
if status == "completed":
|
||
video_url = status_data.get("video", {}).get("url", "")
|
||
if video_url:
|
||
print(f"\n✅ 生成完成!")
|
||
return video_url
|
||
elif status == "failed":
|
||
print(f"\n❌ 生成失败: {status_data.get('error', '')}")
|
||
return None
|
||
|
||
print("\n❌ 超时")
|
||
return None
|
||
|
||
def kling_i2v(image_url, prompt, duration=5):
|
||
"""Kling 可灵 图生视频"""
|
||
key = get_key("KLING_API_KEY")
|
||
if not key:
|
||
print("❌ 未找到 KLING_API_KEY")
|
||
print(" 去 https://klingai.com 注册获取免费 key")
|
||
return None
|
||
|
||
data = {
|
||
"model_name": "kling-v2",
|
||
"image": image_url,
|
||
"prompt": prompt,
|
||
"duration": str(duration),
|
||
"aspect_ratio": "16:9",
|
||
}
|
||
result = subprocess.run(
|
||
["curl", "-s", "-X", "POST", "https://api.klingai.com/v1/videos/image2video",
|
||
"-H", f"Authorization: Bearer {key}",
|
||
"-H", "Content-Type: application/json",
|
||
"-d", json.dumps(data)],
|
||
capture_output=True, text=True, timeout=30
|
||
)
|
||
resp = json.loads(result.stdout)
|
||
task_id = resp.get("data", {}).get("task_id", "")
|
||
if not task_id:
|
||
print(f"❌ 提交失败: {resp}")
|
||
return None
|
||
|
||
print(f"✅ 任务已提交: {task_id}")
|
||
|
||
for i in range(60):
|
||
time.sleep(5)
|
||
status_resp = subprocess.run(
|
||
["curl", "-s", f"https://api.klingai.com/v1/videos/image2video/{task_id}",
|
||
"-H", f"Authorization: Bearer {key}"],
|
||
capture_output=True, text=True, timeout=30
|
||
)
|
||
status_data = json.loads(status_resp.stdout)
|
||
task = status_data.get("data", {})
|
||
status = task.get("task_status", "unknown")
|
||
print(f"\r [{i+1}] {status}", end="", flush=True)
|
||
|
||
if status == "succeed":
|
||
videos = task.get("task_result", {}).get("videos", [])
|
||
if videos:
|
||
print(f"\n✅ 生成完成!")
|
||
return videos[0].get("url", "")
|
||
elif status == "failed":
|
||
print(f"\n❌ 生成失败")
|
||
return None
|
||
|
||
print("\n❌ 超时")
|
||
return None
|
||
|
||
def download_video(url, output_path):
|
||
"""下载视频到本地"""
|
||
subprocess.run(["curl", "-sL", url, "-o", output_path], timeout=60)
|
||
size = os.path.getsize(output_path) if os.path.exists(output_path) else 0
|
||
return size
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Image-to-Video 生成")
|
||
parser.add_argument("--image", required=True, help="输入图片路径或URL")
|
||
parser.add_argument("--prompt", required=True, help="运镜/动作描述")
|
||
parser.add_argument("--provider", default="minimax", choices=["minimax", "kling"])
|
||
parser.add_argument("--duration", type=int, default=6, help="时长(秒)")
|
||
parser.add_argument("--resolution", default="720P", help="分辨率(仅minimax)")
|
||
parser.add_argument("--output", help="输出路径")
|
||
parser.add_argument("--push", action="store_true", help="推送到飞书")
|
||
args = parser.parse_args()
|
||
|
||
# 处理图片：本地文件上传到公网URL
|
||
image = args.image
|
||
if os.path.isfile(image):
|
||
print(f"📤 上传图片到公网...")
|
||
image = upload_image_to_url(image)
|
||
if not image:
|
||
print("❌ 图片上传失败")
|
||
sys.exit(1)
|
||
print(f" URL: {image}")
|
||
|
||
# 生成
|
||
print(f"🎬 开始图生视频 ({args.provider})...")
|
||
print(f" 提示词: {args.prompt}")
|
||
|
||
if args.provider == "minimax":
|
||
video_url = minimax_i2v(image, args.prompt, args.duration, args.resolution)
|
||
else:
|
||
video_url = kling_i2v(image, args.prompt, args.duration)
|
||
|
||
if not video_url:
|
||
sys.exit(1)
|
||
|
||
# 下载
|
||
output = args.output or os.path.expanduser(
|
||
f"~/.hermes/video_cache/i2v_{args.provider}_{int(time.time())}.mp4"
|
||
)
|
||
os.makedirs(os.path.dirname(output), exist_ok=True)
|
||
size = download_video(video_url, output)
|
||
print(f"📁 已保存: {output} ({size//1024}KB)")
|
||
|
||
# 推飞书
|
||
if args.push:
|
||
print("📤 推送飞书...")
|
||
# 复用 agnes_video 的推送逻辑
|
||
os.system(f'python3 ~/.hermes/scripts/agnes_video.py status dummy 2>/dev/null')
|
||
print(" (请手动发送文件或用 send_message)")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|