Seedance 2.5 API 中转站：价格表、火山方舟协议、高并发对账与状态机实战

POST https://api.gpt345.com/v1/videos/generations
Authorization: Bearer sk-your-gpt345-key
Content-Type: application/json
X-Idempotency-Key: your-unique-order-id-20260916

{
  "model": "seedance2.5",
  "content": [
    {
      "type": "text",
      "text": "4K 商业产品摄影：一瓶高档透明玻璃香水瓶置于清澈波纹水面，阳光折射出彩虹光晕，缓慢环绕特写，15 秒无剪辑运镜"
    },
    {
      "type": "image_url",
      "image_url": {
        "url": "https://your-domain.com/sample-product.png"
      }
    }
  ],
  "duration": 15
}

{
  "id": "cgt-seedance-88942b08-6f11-4a3e-9081",
  "model": "seedance2.5",
  "status": "queued",
  "created_at": 1789540000
}

{
  "id": "cgt-seedance-88942b08-6f11-4a3e-9081",
  "model": "seedance2.5",
  "status": "succeeded",
  "content": {
    "video_url": "https://cdn.gpt345.com/outputs/seedance/2026/09/cgt-seedance-88942.mp4?expires=1789626400&sig=abcdef"
  },
  "usage": {
    "completion_tokens": 0,
    "total_cost_cny": 2.50
  }
}

import os
import time
import uuid
import requests
from typing import Dict, Any, Optional

class SeedanceEnterpriseGateway:
    PRICING_MAP = {
        "seedance2.0fast": 1.50,
        "seedance2.5-15": 2.50,
        "seedance2.5-30": 3.50,
    }

    def __init__(self, api_key: str, base_url: str = "https://api.gpt345.com/v1"):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def create_generation_job(self, prompt: str, duration: int = 15, is_fast: bool = False, image_url: Optional[str] = None, order_id: Optional[str] = None) -> Dict[str, Any]:
        """创建生成任务，返回任务元数据并记录预期扣费"""
        if is_fast:
            model_id = "seedance2.0fast"
            expected_cost = self.PRICING_MAP["seedance2.0fast"]
            actual_duration = 10
        else:
            model_id = "seedance2.5"
            if duration not in (15, 30):
                raise ValueError("Seedance 2.5 仅支持 15 秒或 30 秒")
            actual_duration = duration
            expected_cost = self.PRICING_MAP[f"seedance2.5-{duration}"]

        content_payload = [{"type": "text", "text": prompt}]
        if image_url:
            content_payload.append({"type": "image_url", "image_url": {"url": image_url}})

        body = {
            "model": model_id,
            "content": content_payload,
            "duration": actual_duration
        }

        req_headers = dict(self.headers)
        order_key = order_id or f"SEEDANCE_{int(time.time())}_{uuid.uuid4().hex[:8]}"
        req_headers["X-Idempotency-Key"] = order_key

        url = f"{self.base_url}/videos/generations"
        resp = requests.post(url, json=body, headers=req_headers, timeout=20)
        
        if resp.status_code == 401:
            raise PermissionError("HTTP 401 权限不足：请联系微信 GPT774 开通 Seedance 白名单")
        elif resp.status_code != 200:
            raise RuntimeError(f"创建失败 [{resp.status_code}]: {resp.text}")

        res_json = resp.json()
        task_id = res_json.get("id")
        return {
            "order_id": order_key,
            "task_id": task_id,
            "model": model_id,
            "duration": actual_duration,
            "expected_cost": expected_cost,
            "submitted_at": time.time()
        }

    def track_until_completed(self, task_id: str, timeout_sec: int = 360) -> Dict[str, Any]:
        """指数退避跟踪任务，输出最终对账信息"""
        poll_url = f"{self.base_url}/videos/generations/{task_id}"
        start_t = time.time()
        sleep_interval = 4.0

        while time.time() - start_t < timeout_sec:
            resp = requests.get(poll_url, headers=self.headers, timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                status = data.get("status")
                
                if status == "succeeded":
                    video_url = data.get("content", {}).get("video_url")
                    return {
                        "status": "succeeded",
                        "video_url": video_url,
                        "duration_used": int(time.time() - start_t),
                        "billed": True
                    }
                elif status in ("failed", "cancelled"):
                    return {
                        "status": status,
                        "error": data.get("error", "Generation aborted"),
                        "billed": False  # 明确标记未扣费
                    }
                # queued 或 running
                time.sleep(sleep_interval)
                sleep_interval = min(sleep_interval * 1.2, 10.0)
            elif resp.status_code == 429:
                time.sleep(12)
            else:
                time.sleep(5)

        raise TimeoutError(f"Task {task_id} 轮询超过 {timeout_sec} 秒")

# 生产执行验证
if __name__ == "__main__":
    gateway = SeedanceEnterpriseGateway(api_key=os.getenv("GPT345_API_KEY", "sk-placeholder"))
    try:
        # 发起 15 秒高清商品展示生成
        job = gateway.create_generation_job(
            prompt="极简磨砂保温杯，微距慢动作环绕运镜，蒸汽升腾，4K 影视打光",
            duration=15
        )
        print(f"[*] 任务已录入数据库，Order: {job['order_id']}, Task: {job['task_id']}")
        
        # 轮询终态
        result = gateway.track_until_completed(job['task_id'], timeout_sec=240)
        print(f"[✓] 任务结算终态: {result['status']}, 扣费状态: {result['billed']}")
        if result['status'] == 'succeeded':
            print(f"[✓] 成片下载地址: {result['video_url']}")
    except PermissionError as pe:
        print(f"[拦截提示] {pe}")
    except Exception as e:
        print(f"[异常] {e}")