# Linux / macOS
export TOAPIS_API_KEY="sk-toapis-your-key-here"

# 验证环境变量配置
echo $TOAPIS_API_KEY

$env:TOAPIS_API_KEY="sk-toapis-your-key-here"

curl -s -X POST "https://api.toapis.com/v1/videos/generations"   -H "Authorization: Bearer $TOAPIS_API_KEY"   -H "Content-Type: application/json"   -d '{
    "model": "doubao-seedance-2-0",
    "prompt": "电影级运镜：赛博朋克霓虹街道上穿行的机械猎豹，地面水洼倒映粉蓝光芒，4k分辨率，24fps丝滑动态",
    "aspect_ratio": "16:9",
    "duration": 5,
    "seed": 2048
  }'

{
  "code": 0,
  "message": "Task queued successfully",
  "data": {
    "task_id": "task_seedance_20261022_zh_7a8b9c",
    "status": "PROCESSING",
    "model": "doubao-seedance-2-0",
    "created_at": 1729587600
  }
}

curl -s -X GET "https://api.toapis.com/v1/tasks/video-status?task_id=task_seedance_20261022_zh_7a8b9c"   -H "Authorization: Bearer $TOAPIS_API_KEY"

"""
Seedance 2.0 API 生产级异步视频生成与轮询客户端
依赖安装: pip install requests
"""

import os
import sys
import time
import requests

TOAPIS_API_KEY = os.getenv("TOAPIS_API_KEY")
if not TOAPIS_API_KEY:
    print("[错误] 未检测到 TOAPIS_API_KEY 环境变量，请先执行: export TOAPIS_API_KEY='sk-toapis-...'")
    sys.exit(1)

BASE_URL = "https://api.toapis.com/v1"
HEADERS = {
    "Authorization": f"Bearer {TOAPIS_API_KEY}",
    "Content-Type": "application/json"
}

def submit_video_task(prompt: str, duration: int = 5, aspect_ratio: str = "16:9") -> str:
    """提交异步视频生成任务并返回 task_id"""
    endpoint = f"{BASE_URL}/videos/generations"
    payload = {
        "model": "doubao-seedance-2-0",
        "prompt": prompt,
        "aspect_ratio": aspect_ratio,
        "duration": duration,
        "seed": 42
    }

    print(f"[*] 正在向网关提交生成任务: {endpoint}")
    response = requests.post(endpoint, headers=HEADERS, json=payload, timeout=30)
    
    if response.status_code not in (200, 201, 202):
        print(f"[!] 任务派发失败，HTTP状态码: {response.status_code}")
        print(f"[!] 错误详情: {response.text}")
        response.raise_for_status()

    resp_json = response.json()
    task_id = resp_json.get("data", {}).get("task_id")
    if not task_id:
        raise ValueError(f"响应体中未包含 task_id 字段: {resp_json}")

    print(f"[✔] 任务派发成功！任务识别码: {task_id}")
    return task_id

def poll_video_status(task_id: str, poll_interval: int = 3, max_timeout: int = 180) -> dict:
    """
    按指定间隔轮询任务状态直至 SUCCESS 或 FAILED
    推荐轮询间隔: 3 秒；默认超时阈值: 180 秒
    """
    status_url = f"{BASE_URL}/tasks/video-status"
    start_time = time.time()
    params = {"task_id": task_id}

    print(f"[*] 开始轮询任务状态 (间隔={poll_interval}s, 最大等待={max_timeout}s)...")

    while True:
        elapsed = int(time.time() - start_time)
        if elapsed > max_timeout:
            raise TimeoutError(f"任务 {task_id} 运行超时，已等待 {max_timeout} 秒。")

        try:
            resp = requests.get(status_url, headers=HEADERS, params=params, timeout=15)
            resp.raise_for_status()
            task_data = resp.json().get("data", {})
        except requests.exceptions.RequestException as exc:
            print(f"[警告] 轮询网络瞬时波动 ({exc})，将在 {poll_interval} 秒后重试...")
            time.sleep(poll_interval)
            continue

        status = task_data.get("status", "UNKNOWN").upper()
        progress = task_data.get("progress", 0)

        if status in ("SUCCESS", "COMPLETED", "SUCCEEDED"):
            print(f"
[✔] 视频渲染成功！耗时: {elapsed} 秒")
            return task_data
        elif status in ("FAILED", "ERROR"):
            err_msg = task_data.get("error_message", "未知的生成异常")
            raise RuntimeError(f"视频生成失败: {err_msg}")
        elif status in ("PROCESSING", "RUNNING", "QUEUED"):
            sys.stdout.write(f"
[-] 当前状态: {status} | 渲染进度: {progress}% | 已等待: {elapsed}s")
            sys.stdout.flush()
        else:
            print(f"
[?] 收到非预期状态响应: {status}")

        time.sleep(poll_interval)

if __name__ == "__main__":
    prompt_text = (
        "特写镜头：金毛幼犬在秋季枫树林中欢快奔跑，脚下落叶飞溅，"
        "柔和金色夕阳逆光照明，24fps 流畅慢动作，4k超清画质"
    )

    try:
        current_task_id = submit_video_task(prompt=prompt_text, duration=5, aspect_ratio="16:9")
        final_result = poll_video_status(current_task_id)

        video_download_url = final_result.get("video_url") or final_result.get("output", {}).get("video_url")
        print("
" + "=" * 62)
        print("🎬 Seedance 2.0 视频生成完成：")
        print(f"🔗 播放直链: {video_download_url}")
        print(f"⏱ 成片时长: {final_result.get('duration', 5)} 秒")
        print(f"📐 画幅规格: {final_result.get('resolution', '1280x720')}")
        print("=" * 62)
    except Exception as error:
        print(f"
[致命异常] 执行中断: {error}")
        sys.exit(1)

python3 seedance_test.py

{
  "code": 0,
  "message": "Task completed successfully",
  "data": {
    "task_id": "task_seedance_20261022_zh_7a8b9c",
    "status": "SUCCESS",
    "model": "doubao-seedance-2-0",
    "video_url": "https://cdn.toapis.com/outputs/seedance_20261022_zh_7a8b9c.mp4",
    "duration": 5,
    "resolution": "1280x720",
    "cost": 0.14,
    "metrics": {
      "queue_time_ms": 1380,
      "inference_time_ms": 25100,
      "total_time_ms": 26480
    }
  }
}

import os
import requests

API_KEY = os.getenv("TOAPIS_API_KEY")

response = requests.get(
    "https://api.toapis.com/v1/account/token-balance",
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=10
)
response.raise_for_status()

account_info = response.json().get("data", {})
print(f"[*] 账户运行状态: {account_info.get('status')}")
print(f"[*] 可用测试余额: ${account_info.get('balance', 0.0):.4f} {account_info.get('currency', 'USD')}")
print(f"[*] 最大并发配额: {account_info.get('max_concurrency', 5)} 个并发任务")

import time
import random
import requests

def submit_task_with_retry(payload: dict, max_retries: int = 4) -> dict:
    url = "https://api.toapis.com/v1/videos/generations"
    headers = {
        "Authorization": f"Bearer {os.getenv('TOAPIS_API_KEY')}",
        "Content-Type": "application/json"
    }

    base_backoff = 2.0  # 基础退避时间(秒)
    max_backoff = 30.0

    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.post(url, headers=headers, json=payload, timeout=30)
            if resp.status_code in (200, 202):
                return resp.json()

            if resp.status_code == 429:
                jitter = random.uniform(-0.5, 0.5)
                sleep_time = min(max_backoff, (base_backoff * (2 ** (attempt - 1)))) + jitter
                sleep_time = max(1.0, sleep_time)
                print(f"[!] 触发 429 限流，第 {attempt}/{max_retries} 次重试，退避等待 {sleep_time:.2f} 秒...")
                time.sleep(sleep_time)
                continue

            resp.raise_for_status()
        except requests.exceptions.RequestException as err:
            print(f"[警告] 网络通信异常 ({err})，尝试重试...")
            time.sleep(base_backoff)

    raise RuntimeError("重试次数耗尽，任务派发失败。")

FALLBACK_MODELS = [
    "doubao-seedance-2-0",
    "doubao-seedance-2-0-mini",
    "kling-3-omni"
]

def resilient_dispatch(prompt: str) -> dict:
    for model in FALLBACK_MODELS:
        payload = {"model": model, "prompt": prompt, "duration": 5, "aspect_ratio": "16:9"}
        try:
            res = requests.post(
                "https://api.toapis.com/v1/videos/generations",
                headers={"Authorization": f"Bearer {os.getenv('TOAPIS_API_KEY')}", "Content-Type": "application/json"},
                json=payload,
                timeout=25
            )
            if res.status_code in (200, 202):
                print(f"[✔] 成功在模型 {model} 建立生成任务！")
                return res.json()
        except Exception:
            continue
    raise SystemError("所有后备模型算力集群均不可用。")