ICE-GL-ZY001 · 镜01高空云海下推·5秒·Seedance生成 · D169

video-ai-system/tools/shot01_gen.py

@@ -0,0 +1,83 @@
+#!/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]}")