Generate B-Roll Footage with OpenAI Sora API in Python

pip install openai httpx python-dotenv

# .env
OPENAI_API_KEY=sk-...

# sora_client.py
import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

# The client reads OPENAI_API_KEY from env automatically
client = OpenAI()

import time
import httpx
from pathlib import Path

def generate_broll(prompt: str, output_path: str, duration: int = 5) -> str:
    """
    Submit a Sora generation job and return the job ID.
    duration: seconds of video (5 or 10 supported)
    """
    response = client.videos.generate(
        model="sora",
        prompt=prompt,
        duration=duration,          # 5 or 10 seconds
        resolution="1080p",         # 720p | 1080p
        n=1,                        # number of variants to generate
    )

    # Response contains a job ID, not the video itself
    job_id = response.id
    print(f"Job submitted: {job_id}")
    return job_id

def poll_until_done(job_id: str, max_wait: int = 300) -> dict:
    """
    Poll the job status endpoint until done or timeout.
    Uses exponential backoff to avoid rate limiting.
    """
    interval = 5      # Start polling every 5 seconds
    elapsed = 0

    while elapsed < max_wait:
        job = client.videos.retrieve(job_id)

        if job.status == "succeeded":
            return job  # Has download_url at this point

        if job.status == "failed":
            raise RuntimeError(f"Generation failed: {job.error}")

        # Still running — wait before next poll
        time.sleep(interval)
        elapsed += interval
        interval = min(interval * 1.5, 30)  # Cap backoff at 30s

    raise TimeoutError(f"Job {job_id} did not complete within {max_wait}s")

def download_clip(job: dict, output_path: str) -> Path:
    """
    Download the finished video to disk.
    The signed URL expires in ~1 hour — download immediately.
    """
    url = job.download_url  # Only present when status == "succeeded"
    output = Path(output_path)
    output.parent.mkdir(parents=True, exist_ok=True)

    # Stream download to avoid loading entire file into memory
    with httpx.stream("GET", url) as r:
        r.raise_for_status()
        with open(output, "wb") as f:
            for chunk in r.iter_bytes(chunk_size=8192):
                f.write(chunk)

    print(f"Saved: {output} ({output.stat().st_size / 1024:.1f} KB)")
    return output

def create_broll(prompt: str, filename: str, duration: int = 5) -> Path:
    """End-to-end: prompt in, MP4 file out."""
    job_id = generate_broll(prompt, filename, duration)
    job = poll_until_done(job_id)
    return download_clip(job, f"./broll/{filename}.mp4")


if __name__ == "__main__":
    clips = [
        ("Aerial drone shot of a city skyline at golden hour", "city-skyline"),
        ("Close-up of hands typing on a mechanical keyboard", "keyboard-typing"),
        ("Coffee being poured into a white ceramic mug, slow motion", "coffee-pour"),
    ]

    for prompt, name in clips:
        path = create_broll(prompt, name)
        print(f"Done: {path}")

python sora_client.py

ls -lh ./broll/
# Should show 3 non-zero .mp4 files

import concurrent.futures

def batch_generate(clips: list[tuple[str, str]]) -> list[Path]:
    # Submit all jobs immediately
    jobs = [(name, generate_broll(prompt, name)) for prompt, name in clips]

    # Poll all in parallel
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {
            executor.submit(poll_until_done, job_id): (name, job_id)
            for name, job_id in jobs
        }
        for future in concurrent.futures.as_completed(futures):
            name, _ = futures[future]
            job = future.result()
            results.append(download_clip(job, f"./broll/{name}.mp4"))

    return results