import os, time, requests
BASE = os.environ["ANYINT_API_BASE"]
API_KEY = os.environ["ANYINT_API_KEY"]
prompt = "摇摄的广角镜头:一只三花猫在阳光下睡觉"
# 步骤 1 — 文本→图像(例如 gemini-2.5-flash-image)
img_resp = requests.post(
f"{BASE}/images/generations",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
},
json={
"model": "gemini-2.5-flash-image",
"prompt": prompt,
"n": 1,
"size": "1024x1024",
# 如果需要,您也可以请求 base64 而不是 URL:
# "response_format": "b64_json"
},
timeout=60,
)
img_resp.raise_for_status()
img_data = img_resp.json()
# 假设第一个图像 URL 是我们想要的
image_url = img_data["data"][0]["url"]
print("生成的图像 URL:", image_url)
# 步骤 2 — 图像 + 文本→视频(Veo 3.1)
create_resp = requests.post(
f"{BASE}/videos/generations",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
},
json={
"model": "veo-3.1-generate-preview",
"prompt": prompt,
"image_url": image_url, # 将引导图像传递给视频模型
# 可选:
# "duration_seconds": 6,
# "aspect_ratio": "16:9"
},
timeout=60,
)
create_resp.raise_for_status()
operation = create_resp.json()
operation_id = operation["id"]
print("已创建视频生成任务:", operation_id)
# 步骤 3 — 轮询视频生成任务
while True:
status_resp = requests.get(
f"{BASE}/videos/operations/{operation_id}",
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=60,
)
status_resp.raise_for_status()
op = status_resp.json()
status = op.get("status")
print("当前状态:", status)
if status == "succeeded":
video_url = op["result"]["videos"][0]["url"]
break
if status == "failed":
raise RuntimeError(f"视频生成失败:{op.get('error')}")
time.sleep(10)
print("视频已就绪,地址:", video_url)
# 步骤 4 — 下载最终视频
video_resp = requests.get(video_url, timeout=300)
video_resp.raise_for_status()
with open("veo3_with_image_input.mp4", "wb") as f:
f.write(video_resp.content)
print("生成的视频已保存为 veo3_with_image_input.mp4")