Seedance-2.5 选 15 秒还是 30 秒？成本测算、分镜拆解与任务状态机实战

CPUS = (总生成成功扣费金额 + 废片消耗) ÷ 最终进入剪辑轨道的优质视频条数
     = 单条单价 ÷ 质检通过率 (Pass Rate)

{
  "model": "seedance2.5",
  "content": [
    {
      "type": "text",
      "text": "特写镜头：一瓶磨砂质感的极简护肤精华液放置在平滑的大理石台面上。镜头平稳缓慢环绕右推，周围有细腻水雾喷洒并泛起柔和晨光丁达尔效应，大理石倒影清晰，4K 影视质感，商业广告级打光，15 秒无剪辑稳定运镜。"
    }
  ],
  "duration": 15
}

{
  "model": "seedance2.5",
  "content": [
    {
      "type": "text",
      "text": "长镜头分镜演进：\n[0-10秒] 远景航拍，阳光破开晨雾照亮雪山峡谷，一辆深灰色越野车沿盘山公路平稳前行，卷起微弱尘土；\n[11-20秒] 镜头无缝前推逼近越野车，切换至车身侧前方低角度追踪视角，车轮与悬挂在碎石路面上发生逼真机械颠簸；\n[21-30秒] 镜头再次升起并越过车顶，越野车冲上山顶开阔观景台刹停，车尾扬起金色夕阳余晖。整段镜头运镜平滑流畅，光影与车身材质绝对一致。"
    }
  ],
  "duration": 30
}

import os
import time
import uuid
import requests

API_KEY = os.getenv("GPT345_API_KEY", "sk-your-gpt345-key")
BASE_URL = "https://api.gpt345.com/v1"

class SeedanceVideoClient:
    def __init__(self, api_key: str, base_url: str = BASE_URL):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def submit_task(self, prompt: str, duration: int = 15, idempotency_key: str = None) -> str:
        """提交视频生成任务，返回 task_id。支持 15 秒或 30 秒"""
        if duration not in (15, 30):
            raise ValueError("Seedance-2.5 仅支持 15 秒或 30 秒时长参数")
        
        headers = dict(self.headers)
        if idempotency_key:
            headers["X-Idempotency-Key"] = idempotency_key

        payload = {
            "model": "seedance2.5",
            "content": [
                {"type": "text", "text": prompt}
            ],
            "duration": duration
        }

        resp = requests.post(f"{self.base_url}/videos/generations", json=payload, headers=headers, timeout=15)
        if resp.status_code == 401:
            raise PermissionError("401 Unauthorized: API Key 无效或未开通 Seedance 权限，请联系客服QQ 49098022。")
        elif resp.status_code != 200:
            raise RuntimeError(f"提交任务失败 [HTTP {resp.status_code}]: {resp.text}")

        data = resp.json()
        task_id = data.get("id")
        if not task_id:
            raise ValueError(f"响应缺少任务 ID: {data}")
        print(f"[+] 任务提交成功，Task ID: {task_id} (时长: {duration}s)")
        return task_id

    def poll_and_download(self, task_id: str, output_path: str, max_wait_sec: int = 300) -> str:
        """带退避状态机轮询任务状态，完成时下载视频"""
        poll_url = f"{self.base_url}/videos/generations/{task_id}"
        start_time = time.time()
        interval = 5.0  # 初始轮询间隔 5 秒

        while time.time() - start_time < max_wait_sec:
            try:
                resp = requests.get(poll_url, headers=self.headers, timeout=10)
                if resp.status_code == 200:
                    task_data = resp.json()
                    status = task_data.get("status")

                    if status == "succeeded":
                        video_url = task_data.get("content", {}).get("video_url")
                        if not video_url:
                            raise ValueError(f"任务已完成但缺少 video_url: {task_data}")
                        print(f"[✓] 任务生成成功！正在下载: {video_url}")
                        self._download_file(video_url, output_path)
                        return output_path
                    elif status in ("failed", "cancelled"):
                        raise RuntimeError(f"任务终态异常: status={status}, 详细信息: {task_data}")
                    else:
                        # queued 或 running 状态，等待重试
                        elapsed = int(time.time() - start_time)
                        print(f"[*] 任务渲染中 (status: {status}, 已耗时: {elapsed}s)...")
                elif resp.status_code == 429:
                    print("[!] 遭遇速率限制 429，等待 15 秒退避...")
                    time.sleep(15)
                    continue
                else:
                    print(f"[!] 查询返回异常码 HTTP {resp.status_code}，重试中...")
            except requests.exceptions.RequestException as e:
                print(f"[!] 网络抖动: {e}，将在下一轮重试...")

            time.sleep(interval)
            interval = min(interval * 1.15, 12.0)  # 指数退避，上限 12 秒

        raise TimeoutError(f"任务轮询超时（超过 {max_wait_sec} 秒），请后台人工核实 task_id: {task_id}")

    def _download_file(self, url: str, target_file: str):
        res = requests.get(url, stream=True, timeout=60)
        res.raise_for_status()
        with open(target_file, "wb") as f:
            for chunk in res.iter_content(chunk_size=65536):
                if chunk:
                    f.write(chunk)
        size_mb = os.path.getsize(target_file) / (1024 * 1024)
        print(f"[✓] 文件已保存至 {target_file} (文件大小: {size_mb:.2f} MB)")

if __name__ == "__main__":
    client = SeedanceVideoClient(api_key=API_KEY)
    
    # 场景 1：15 秒电商产品展示
    prompt_15s = "极简白色陶瓷咖啡杯，杯口升起缕缕热气，置于木质托盘上，晨光斜射，微距慢动作环绕运镜。"
    order_uuid = str(uuid.uuid4())
    
    try:
        task_id = client.submit_task(prompt=prompt_15s, duration=15, idempotency_key=order_uuid)
        client.poll_and_download(task_id, output_path="coffee_cup_15s.mp4", max_wait_sec=180)
    except PermissionError as pe:
        print(f"[权限拦截] {pe}")
    except Exception as e:
        print(f"[运行出错] {e}")