84 lines
3.4 KiB
Python
84 lines
3.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""镜01 · Seedance视频生成 · 高空云海下推"""
|
||
|
|
import os, sys, json, time, subprocess
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
ROOT = Path('D:/WorkBuddy/guanghulab/video-ai-system')
|
||
|
|
env = {}
|
||
|
|
for line in open(ROOT/".env"):
|
||
|
|
line = line.strip()
|
||
|
|
if not line or line.startswith("#"): continue
|
||
|
|
if "=" in line:
|
||
|
|
k, v = line.split("=", 1)
|
||
|
|
env[k.strip()] = v.strip()
|
||
|
|
|
||
|
|
AK = env["JIMENG_API_KEY"]
|
||
|
|
URL = env["JIMENG_BASE_URL"]
|
||
|
|
OUT = ROOT / "outputs/shots"
|
||
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
P = """中国修仙世界云海之上，极致大远景。洁白云层铺展如无边大海，灵气蒸腾飘渺。数名修仙者御剑飞行穿梭云间，浑身散发淡淡灵光拖出光尾。一位仙人驾仙鹤从云海掠过化作流光。远处仙山浮于云海之上若隐若现。镜头从高空云层缓慢向下推镜，云层渐渐变薄，下方隐约可见巨型修仙广场。竖屏。"""
|
||
|
|
|
||
|
|
def gen():
|
||
|
|
payload = {
|
||
|
|
"model": "doubao-seedance-2-0-260128",
|
||
|
|
"content": [{"type": "text", "text": P}],
|
||
|
|
"duration": 5,
|
||
|
|
"resolution": "720p"
|
||
|
|
}
|
||
|
|
body = json.dumps(payload)
|
||
|
|
r = subprocess.run(["curl", "-s", "-m", "60", "--noproxy", "*", "-X", "POST",
|
||
|
|
f"{URL}/contents/generations/tasks",
|
||
|
|
"-H", f"Authorization: Bearer {AK}", "-H", "Content-Type: application/json",
|
||
|
|
"-d", body], capture_output=True, text=True, timeout=70)
|
||
|
|
print(f"POST status: curl exit={r.returncode}")
|
||
|
|
try:
|
||
|
|
return json.loads(r.stdout)
|
||
|
|
except:
|
||
|
|
print(f"Raw: {r.stdout[:500]}")
|
||
|
|
return {}
|
||
|
|
|
||
|
|
def poll(task_id, max_attempts=60):
|
||
|
|
for i in range(max_attempts):
|
||
|
|
time.sleep(10)
|
||
|
|
r = subprocess.run(["curl", "-s", "-m", "30", "--noproxy", "*",
|
||
|
|
f"{URL}/contents/generations/tasks/{task_id}?Authorization=Bearer {AK}"],
|
||
|
|
capture_output=True, text=True, timeout=40)
|
||
|
|
try:
|
||
|
|
data = json.loads(r.stdout)
|
||
|
|
except:
|
||
|
|
continue
|
||
|
|
status = (data.get("status") or data.get("data",{}).get("status") or "").lower()
|
||
|
|
if i % 3 == 0:
|
||
|
|
print(f" poll {i+1}: {status}")
|
||
|
|
if status in ["completed", "succeeded", "done", "success"]:
|
||
|
|
video_url = (data.get("content",{}).get("video_url") or
|
||
|
|
data.get("output",{}).get("video_url") or
|
||
|
|
data.get("data",{}).get("output",{}).get("video_url") or
|
||
|
|
data.get("content",[{}])[0].get("video_url"))
|
||
|
|
return video_url, data
|
||
|
|
if status in ["failed", "error"]:
|
||
|
|
return None, data
|
||
|
|
return None, {}
|
||
|
|
|
||
|
|
print(f"镜01 · 高空云海下推 · 5秒")
|
||
|
|
resp = gen()
|
||
|
|
task_id = resp.get("id") or resp.get("task_id") or resp.get("data",{}).get("id") or resp.get("data",{}).get("task_id")
|
||
|
|
|
||
|
|
if task_id:
|
||
|
|
print(f"任务: {task_id}")
|
||
|
|
video_url, final = poll(task_id)
|
||
|
|
if video_url:
|
||
|
|
ts = time.strftime("%Y%m%d-%H%M%S")
|
||
|
|
dst = OUT / f"ep01-shot01-{ts}.mp4"
|
||
|
|
res = subprocess.run(["curl", "-s", "-m", "300", "-L", "--noproxy", "*", "-o", str(dst), video_url],
|
||
|
|
capture_output=True, timeout=310)
|
||
|
|
if dst.exists() and dst.stat().st_size > 1024:
|
||
|
|
print(f"✅ {dst.name} ({dst.stat().st_size//1024}KB)")
|
||
|
|
else:
|
||
|
|
print("❌ 下载失败")
|
||
|
|
else:
|
||
|
|
print(f"❌ 无视频: {json.dumps(final,ensure_ascii=False)[:500]}")
|
||
|
|
else:
|
||
|
|
print(f"❌ 无任务ID: {json.dumps(resp,ensure_ascii=False)[:500]}")
|