Seedance — 完全版APIチュートリアル

# Install required packages
!pip install byteplus-python-sdk-v2 python-dotenv requests pillow -q

import os
import time
import json
import requests
from pathlib import Path
from dotenv import load_dotenv
from IPython.display import Video, Image, display, HTML
from byteplussdkarkruntime import Ark

# Load credentials from .env file
load_dotenv()

ARK_API_KEY = os.getenv("ARK_API_KEY")
ARK_BASE_URL = os.getenv("ARK_BASE_URL", "https://ark.ap-southeast.bytepluses.com/api/v3")

if not ARK_API_KEY:
    raise EnvironmentError(
        "ARK_API_KEY not found. Create a .env file with:\n"
        "ARK_API_KEY=your_key_here\n"
        "Get yours at: https://console.byteplus.com/ark/region:ark+ap-southeast-1/apikey"
    )

# Initialize the Ark client
client = Ark(
    base_url=ARK_BASE_URL,
    api_key=ARK_API_KEY,
)

print(f"✅ Client initialized | Base URL: {ARK_BASE_URL}")

# Output directory
OUTPUT_DIR = Path("seedance_outputs")
OUTPUT_DIR.mkdir(exist_ok=True)

def estimate_cost(resolution="720p", duration=5, audio=True):
    """Estimate video generation cost."""
    dims = {"480p": (854, 480), "720p": (1280, 720), "1080p": (1920, 1080)}
    w, h = dims.get(resolution, (1280, 720))
    fps = 24
    tokens = (h * w * fps * duration) / 1024
    rate = 2.4 if audio else 1.2
    cost = tokens * rate / 1_000_000
    print(f"Resolution: {resolution} ({w}x{h}) | Duration: {duration}s | Audio: {audio}")
    print(f"Tokens: {tokens:,.0f} | Estimated cost: ${cost:.4f}")
    return cost

estimate_cost("720p", 5, audio=True)

def poll_task(task_id, interval=10, max_wait=600, verbose=True):
    """
    Poll a video generation task until it's complete.
    
    Args:
        task_id: Task ID from create_task response
        interval: Seconds between polls
        max_wait: Maximum wait time in seconds
        verbose: Print status updates
    
    Returns:
        Task result dict or None on timeout
    """
    elapsed = 0
    while elapsed < max_wait:
        result = client.content_generation.tasks.get(id=task_id)
        status = result.status

        if verbose:
            print(f"  [{elapsed:>4}s] Status: {status}")

        if status == "succeeded":
            return result
        elif status in ("failed", "cancelled"):
            print(f"❌ Task {status}: {getattr(result, 'error', 'unknown error')}")
            return None

        time.sleep(interval)
        elapsed += interval

    print(f"⏱️ Timed out after {max_wait}s")
    return None


def download_video(url, filename, output_dir=OUTPUT_DIR):
    """Download a video from a URL and save locally."""
    path = output_dir / filename
    response = requests.get(url, stream=True, timeout=60)
    response.raise_for_status()
    with open(path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    size_mb = path.stat().st_size / 1_048_576
    print(f"✅ Saved: {path} ({size_mb:.1f} MB)")
    return path


def generate_video(model, content, filename="output.mp4", verbose=True):
    """
    Full pipeline: create task, poll, download.
    
    Args:
        model: Model ID string
        content: List of content dicts (text/image_url)
        filename: Output filename
        verbose: Print progress
    
    Returns:
        Local path to downloaded video
    """
    print(f"🎬 Creating task | Model: {model}")
    task = client.content_generation.tasks.create(model=model, content=content)
    task_id = task.id
    print(f"   Task ID: {task_id}")

    result = poll_task(task_id, verbose=verbose)
    if result is None:
        return None

    video_url = result.content.video_url
    print(f"🔗 Video URL: {video_url}")
    return download_video(video_url, filename)

print("✅ Helper functions defined")

# --- Basic Text-to-Video ---
T2V_MODEL = "seedance-1-5-pro-251215"

prompt_t2v = (
    "Cinematic close-up of a single white daisy in a sunlit meadow. "
    "Dewdrops glisten on the petals as a gentle breeze sways the flower. "
    "Slow zoom out, soft golden hour lighting, film grain texture. "
    "--resolution 720p --duration 5 --camerafixed false"
)

content_t2v = [{"type": "text", "text": prompt_t2v}]

# Uncomment to run (costs ~$0.26)
# video_path = generate_video(T2V_MODEL, content_t2v, filename="t2v_basic.mp4")
# if video_path:
#     display(Video(str(video_path), embed=True, width=640))

print("Prompt ready. Uncomment the generate_video call to run.")
print(f"Model: {T2V_MODEL}")
print(f"Prompt: {prompt_t2v}")

# --- Prompt Variations Showcase ---
prompt_examples = {
    "nature": (
        "Aerial view of a dense forest at dawn, morning mist weaving through treetops. "
        "Slow drone descent through the canopy, dappled light filtering down. "
        "--resolution 720p --duration 8 --camerafixed false"
    ),
    "character": (
        "A young woman in a red coat walks across a rain-slicked Tokyo street at night. "
        "Neon lights reflect in puddles. Medium tracking shot, handheld feel. "
        "She pauses, looks at camera, slight smile. "
        "--resolution 720p --duration 6 --camerafixed false"
    ),
    "product": (
        "Luxury perfume bottle rotating on a reflective obsidian surface. "
        "Deep purple and gold lighting. Slow 360 orbit shot. "
        "Particles of light drift around the bottle. Studio setup. "
        "--resolution 720p --duration 5 --camerafixed false"
    ),
    "multi_shot": (
        "Wide shot: A chef seasons vegetables in a professional kitchen, flames rising in a wok. "
        "Cut to close-up of vegetables sizzling with vibrant steam. "
        "Cut to the chef plating the dish with focused expression. "
        "Natural lighting, documentary style. "
        "--resolution 720p --duration 10 --camerafixed false"
    ),
}

for name, prompt in prompt_examples.items():
    print(f"\n📋 [{name.upper()}]")
    print(f"   {prompt[:120]}..." if len(prompt) > 120 else f"   {prompt}")

I2V_MODEL = "seedance-1-5-pro-251215"  # Same model does both T2V and I2V

# Public test image (replace with your own)
IMAGE_URL = "https://ark-doc.tos-ap-southeast-1.bytepluses.com/see_i2v.jpeg"

# For I2V: describe movement and camera — the image defines the scene
prompt_i2v = (
    "Traffic flows along the bridge toward Big Ben at dusk. "
    "Cars move steadily with glowing headlights and taillights. "
    "Clouds drift slowly across the dramatic sky. "
    "Slow zoom in, cinematic urban atmosphere. "
    "--resolution 720p --duration 5 --camerafixed false"
)

content_i2v = [
    {"type": "text", "text": prompt_i2v},
    {"type": "image_url", "image_url": {"url": IMAGE_URL}},
]

# Display the input image
try:
    display(Image(url=IMAGE_URL, width=400))
    print(f"Reference image: {IMAGE_URL}")
except Exception:
    print(f"Reference image URL: {IMAGE_URL}")

# Uncomment to run
# video_path = generate_video(I2V_MODEL, content_i2v, filename="i2v_output.mp4")
# if video_path:
#     display(Video(str(video_path), embed=True, width=640))

print("\nI2V content structure ready.")

# First + Last frame I2V (supported by seedance-1-0-pro-250528)
START_FRAME_URL = "https://ark-doc.tos-ap-southeast-1.bytepluses.com/see_i2v.jpeg"
END_FRAME_URL   = "https://ark-doc.tos-ap-southeast-1.bytepluses.com/see_i2v_last.jpeg"

content_first_last = [
    {
        "type": "text",
        "text": "Flowers blooming, petals opening slowly. Gentle sunlight. --resolution 720p --duration 5"
    },
    {"type": "image_url", "image_url": {"url": START_FRAME_URL}},  # first frame
    {"type": "image_url", "image_url": {"url": END_FRAME_URL}},    # last frame
]

print("First+Last frame content structure:")
print(json.dumps(content_first_last, indent=2))

# Retrieve a specific task by ID
def get_task_status(task_id):
    """Retrieve and display task status."""
    try:
        result = client.content_generation.tasks.get(id=task_id)
        print(f"Task ID  : {result.id}")
        print(f"Status   : {result.status}")
        print(f"Model    : {result.model}")
        if result.status == "succeeded":
            print(f"Video URL: {result.content.video_url}")
        elif result.status == "failed":
            print(f"Error    : {getattr(result, 'error', 'N/A')}")
        return result
    except Exception as e:
        print(f"Error fetching task: {e}")
        return None

# List recent tasks
def list_tasks(model=None, limit=5):
    """List recent video generation tasks."""
    try:
        params = {"page_size": limit}
        if model:
            params["model"] = model
        tasks = client.content_generation.tasks.list(**params)
        print(f"Recent tasks (limit {limit}):")
        for t in tasks.items:
            print(f"  {t.id} | {t.status:12} | {t.model}")
        return tasks
    except Exception as e:
        print(f"Error listing tasks: {e}")
        return None

# Example usage
# get_task_status("your-task-id-here")
# list_tasks(limit=5)

print("Task management functions defined.")

API_BASE = ARK_BASE_URL
HEADERS = {
    "Authorization": f"Bearer {ARK_API_KEY}",
    "Content-Type": "application/json",
}

def create_task_rest(model, content):
    """Create a video generation task via REST API."""
    payload = {"model": model, "content": content}
    resp = requests.post(
        f"{API_BASE}/content_generation/tasks",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


def get_task_rest(task_id):
    """Retrieve a video generation task via REST API."""
    resp = requests.get(
        f"{API_BASE}/content_generation/tasks/{task_id}",
        headers=HEADERS,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


def poll_task_rest(task_id, interval=10, max_wait=600):
    """Poll REST API until task completes."""
    elapsed = 0
    while elapsed < max_wait:
        data = get_task_rest(task_id)
        status = data.get("status")
        print(f"  [{elapsed:>4}s] {status}")
        if status == "succeeded":
            return data
        elif status in ("failed", "cancelled"):
            return data
        time.sleep(interval)
        elapsed += interval
    return None


# Example REST call (uncomment to run)
# task = create_task_rest(
#     model="seedance-1-5-pro-t2v-250612",
#     content=[{"type": "text", "text": "A cat playing piano. --resolution 480p --duration 4"}]
# )
# print(json.dumps(task, indent=2))
# result = poll_task_rest(task["id"])

print("REST API functions defined.")

# Prompt builder utility
def build_prompt(
    subject,
    subject_action,
    background,
    camera="",
    audio_hint="",
    resolution="720p",
    duration=5,
    camera_fixed=False,
):
    """
    Build a structured Seedance prompt.

    Args:
        subject: Who/what is in the scene
        subject_action: What the subject is doing
        

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)


def generate_with_retry(model, content, filename="output.mp4", max_retries=3, backoff=2):
    """
    Generate a video with exponential backoff retry.

    Retries on:
    - Rate limit errors (429)
    - Transient server errors (5xx)

    Does NOT retry on:
    - Invalid parameters (400)
    - Authentication errors (401)
    """
    for attempt in range(max_retries):
        try:
            logger.info(f"Attempt {attempt + 1}/{max_retries}")
            task = client.content_generation.tasks.create(
                model=model, content=content
            )
            logger.info(f"Task created: {task.id}")
            result = poll_task(task.id, verbose=True)
            if result:
                return download_video(result.content.video_url, filename)
            return None

        except Exception as e:
            error_str = str(e).lower()
            if "401" in error_str or "invalid" in error_str:
                logger.error(f"Non-retryable error: {e}")
                raise

            wait = backoff ** attempt
            logger.warning(f"Error (attempt {attempt+1}): {e}. Retrying in {wait}s...")
            if attempt < max_retries - 1:
                time.sleep(wait)
            else:
                logger.error("Max retries exceeded.")
                raise

    return None


# Status codes and their meanings
STATUS_GUIDE = {
    "queued": "Task waiting to be processed",
    "running": "Video generation in progress",
    "succeeded": "Video ready for download",
    "failed": "Generation failed — check error message",
    "cancelled": "Task was cancelled by user",
}

print("Status codes:")
for status, desc in STATUS_GUIDE.items():
    print(f"  {status:12} → {desc}")

def batch_generate(prompts, model=T2V_MODEL, resolution="720p", duration=5):
    """
    Submit multiple video tasks and collect results.
    
    Submits all tasks first, then polls — maximizes parallelism
    within the concurrency limit (10 concurrent tasks per account).
    """
    tasks = []

    # Submit all tasks
    for i, prompt in enumerate(prompts):
        full_prompt = f"{prompt} --resolution {resolution} --duration {duration}"
        try:
            task = client.content_generation.tasks.create(
                model=model,
                content=[{"type": "text", "text": full_prompt}],
            )
            tasks.append({"id": task.id, "prompt": prompt[:50], "status": "queued"})
            print(f"✅ Task {i+1}/{len(prompts)} submitted: {task.id}")
        except Exception as e:
            print(f"❌ Task {i+1} failed to submit: {e}")
            tasks.append({"id": None, "prompt": prompt[:50], "status": "error"})

    # Poll all tasks
    results = []
    for task_info in tasks:
        if task_info["id"] is None:
            results.append(None)
            continue
        print(f"\nPolling {task_info['id']}...")
        result = poll_task(task_info["id"], verbose=False)
        results.append(result)
        task_info["status"] = result.status if result else "timeout"

    print("\n📊 Batch Summary:")
    for t in tasks:
        print(f"  {t['status']:12} | {t['prompt']}")

    return results, tasks


# Example batch (3 prompts)
batch_prompts = [
    "Ocean waves crashing on rocky cliffs at sunset, slow motion aerial",
    "City timelapse at night, car light trails on a busy intersection",
    "Coffee being poured into a cup, macro shot, steam rising",
]

estimate_cost("720p", 5) 
print(f"\nEstimated cost for {len(batch_prompts)} videos:")
total = estimate_cost("720p", 5) * len(batch_prompts)
print(f"Total: ~${total:.4f}")

# Uncomment to run the batch
# results, summary = batch_generate(batch_prompts)

def cancel_task(task_id):
    """Cancel a running or queued task."""
    try:
        result = client.content_generation.tasks.cancel(id=task_id)
        print(f"Cancelled task {task_id}: {result.status}")
        return result
    except Exception as e:
        print(f"Failed to cancel: {e}")
        return None

# cancel_task("your-task-id")
print("cancel_task() defined.")

def run_full_example():
    """Complete end-to-end example with Seedance 1.5 Pro."""
    # 1. Build a structured prompt
    prompt = build_prompt(
        subject="A red fox",
        subject_action="trots through a snowy pine forest",
        background="Snowflakes fall softly, pine branches bend under the weight of snow",
        camera="Wide tracking shot follows the fox at low angle",
        audio_hint="Soft crunch of snow underfoot, distant wind",
        resolution="720p",
        duration=7,
        camera_fixed=False,
    )
    print("📝 Prompt:")
    print(f"   {prompt}\n")

    # 2. Estimate cost
    cost = estimate_cost("720p", 7, audio=True)
    print()

    # 3. Generate (with retry)
    video_path = generate_with_retry(
        model="seedance-1-5-pro-t2v-250612",
        content=[{"type": "text", "text": prompt}],
        filename="fox_in_snow.mp4",
        max_retries=2,
    )

    # 4. Display
    if video_path:
        print(f"\n🎬 Video saved to: {video_path}")
        display(Video(str(video_path), embed=True, width=640))
    else:
        print("Generation failed or timed out.")


# Uncomment to run the full example
# run_full_example()

print("run_full_example() defined — uncomment to execute.")