# macOS / Linux
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": 1024
  }'

{
  "code": 0,
  "message": "Task queued successfully",
  "data": {
    "task_id": "task_seedance_20261022_ja_9c0d1e",
    "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_ja_9c0d1e"   -H "Authorization: Bearer $TOAPIS_API_KEY"

"""
Seedance 2.0 API 非同期動画生成クライアント
実行準備: pip install requests
"""

import os
import sys
import time
import requests

API_KEY = os.getenv("TOAPIS_API_KEY")
if not API_KEY:
    print("[エラー] TOAPIS_API_KEY が設定されていません。")
    print("実行例: export TOAPIS_API_KEY='sk-toapis-...'")
    sys.exit(1)

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

def submit_task(prompt: str, duration: int = 5, aspect_ratio: str = "16:9") -> str:
    """非同期タスクを投入し task_id を取得"""
    url = f"{BASE_URL}/videos/generations"
    payload = {
        "model": "doubao-seedance-2-0",
        "prompt": prompt,
        "aspect_ratio": aspect_ratio,
        "duration": duration,
        "seed": 42
    }

    print(f"[*] タスクを投入中: {url}")
    resp = requests.post(url, headers=HEADERS, json=payload, timeout=30)
    
    if resp.status_code not in (200, 201, 202):
        print(f"[!] 投入エラー: HTTP {resp.status_code} - {resp.text}")
        resp.raise_for_status()

    task_id = resp.json().get("data", {}).get("task_id")
    if not task_id:
        raise ValueError(f"task_id が見つかりません: {resp.text}")

    print(f"[✔] タスク登録完了！ Task ID: {task_id}")
    return task_id

def poll_status(task_id: str, interval: int = 3, timeout: int = 180) -> dict:
    """タスク完了までポーリングを実施"""
    url = f"{BASE_URL}/tasks/video-status"
    start_time = time.time()
    params = {"task_id": task_id}

    print(f"[*] ポーリング開始 (確認間隔={interval}秒, タイムアウト={timeout}秒)...")

    while True:
        elapsed = int(time.time() - start_time)
        if elapsed > timeout:
            raise TimeoutError(f"タスク {task_id} がタイムアウトしました（{timeout}秒経過）。")

        try:
            r = requests.get(url, headers=HEADERS, params=params, timeout=15)
            r.raise_for_status()
            data = r.json().get("data", {})
        except requests.exceptions.RequestException as e:
            print(f"[警告] 通信エラー ({e})。{interval}秒後に再試行します...")
            time.sleep(interval)
            continue

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

        if status in ("SUCCESS", "COMPLETED", "SUCCEEDED"):
            print(f"
[✔] 動画レンダリング完了！（所要時間: {elapsed}秒）")
            return data
        elif status in ("FAILED", "ERROR"):
            err = data.get("error_message", "不明なエラー")
            raise RuntimeError(f"レンダリング失敗: {err}")
        elif status in ("PROCESSING", "RUNNING", "QUEUED"):
            sys.stdout.write(f"
[-] ステータス: {status} | 進捗率: {progress}% | 経過: {elapsed}秒")
            sys.stdout.flush()

        time.sleep(interval)

if __name__ == "__main__":
    prompt = (
        "特写：秋のイチョウ並木を嬉しそうに駆け回るゴールデンレトリバーの子犬、"
        "舞い散る黄色の落ち葉、夕日の温かいシネマティックライティング、24fps、4k画質"
    )

    try:
        t_id = submit_task(prompt=prompt, duration=5, aspect_ratio="16:9")
        result = poll_status(t_id)

        video_url = result.get("video_url") or result.get("output", {}).get("video_url")
        print("
" + "=" * 60)
        print("🎬 Seedance 2.0 生成完了:")
        print(f"🔗 動画URL:   {video_url}")
        print(f"⏱ 再生時間: {result.get('duration', 5)} 秒")
        print(f"📐 解像度:   {result.get('resolution', '1280x720')}")
        print("=" * 60)
    except Exception as exc:
        print(f"
[致命的エラー] {exc}")
        sys.exit(1)

python3 test_seedance.py

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

import os
import requests

API_KEY = os.getenv("TOAPIS_API_KEY")

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

info = res.json().get("data", {})
print(f"[*] アカウントステータス: {info.get('status')}")
print(f"[*] 利用可能テスト残高:   ${info.get('balance', 0.0):.4f} {info.get('currency', 'USD')}")
print(f"[*] 最大同時実行数:       {info.get('max_concurrency', 5)} タスク")

import time
import random
import requests

def dispatch_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_sec = min(max_backoff, (base_backoff * (2 ** (attempt - 1)))) + jitter
                sleep_sec = max(1.0, sleep_sec)
                print(f"[!] 429レートリミット検知。{attempt}回目の再試行。{sleep_sec:.2f}秒待機します...")
                time.sleep(sleep_sec)
                continue

            resp.raise_for_status()
        except requests.exceptions.RequestException as err:
            print(f"[警告] 通信障害 ({err})。再試行中...")
            time.sleep(base_backoff)

    raise RuntimeError("最大再試行回数を超過しました。")

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

def resilient_call(prompt: str) -> dict:
    for model in BACKUP_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("全モデルクラスタが応答しません。")