Sora 2 视频生成（异步调用）

curl -X POST "https://api.apiyi.com/v1/videos" \
  -H "Authorization: YOUR_API_KEY" \
  -H "Accept: application/json" \
  -F "prompt=一只可爱的金毛犬在公园草地上追逐飞盘，阳光明媚，背景是绿树和蓝天" \
  -F "model=sora-2" \
  -F "size=1280x720" \
  -F "seconds=10"

curl -X POST "https://api.apiyi.com/v1/videos" \
  -H "Authorization: YOUR_API_KEY" \
  -H "Accept: application/json" \
  -F "prompt=参考配图，使得动物们活跃起来，在场地里绕场一周的追逐" \
  -F "model=sora-2" \
  -F "size=1280x720" \
  -F "seconds=10" \
  -F "input_reference=@/path/to/image.png"

curl -X GET "https://api.apiyi.com/v1/videos/video_abc123def456" \
  -H "Authorization: YOUR_API_KEY"

curl -X GET "https://api.apiyi.com/v1/videos/video_abc123def456/content" \
  -H "Authorization: YOUR_API_KEY" \
  -o video.mp4

import requests
import time
import os
from datetime import datetime

# API 配置
BASE_URL = "https://api.apiyi.com/v1/videos"
API_KEY = "YOUR_API_KEY"

# 参数配置
IMAGE_PATH = './dog-and-cat.png'  # 参考图片路径，设为 None 则为文生视频
PROMPT = '参考配图，使得动物们活跃起来，在场地里绕场一周的追逐'
MODEL = 'sora-2'
SIZE = '1280x720'
SECONDS = '10'

# 轮询配置
POLL_INTERVAL = 30
MAX_WAIT_TIME = 600


def step1_submit_request():
    """第一步：提交视频生成请求"""
    print("=" * 60)
    print("第一步：提交视频生成请求")
    print("=" * 60)

    files = {
        'prompt': (None, PROMPT),
        'model': (None, MODEL),
        'size': (None, SIZE),
        'seconds': (None, SECONDS)
    }

    headers = {
        'Accept': 'application/json',
        'Authorization': API_KEY
    }

    # 如果有图片，添加到请求中
    if IMAGE_PATH and os.path.exists(IMAGE_PATH):
        print(f"模式：图片生成视频")
        print(f"图片: {IMAGE_PATH}")
        with open(IMAGE_PATH, 'rb') as f:
            files['input_reference'] = (
                os.path.basename(IMAGE_PATH),
                f,
                'image/png'
            )
            response = requests.post(BASE_URL, headers=headers, files=files)
    else:
        print(f"模式：文字生成视频")
        response = requests.post(BASE_URL, headers=headers, files=files)

    if response.status_code == 200:
        result = response.json()
        video_id = result.get('id')
        print(f"✅ 请求提交成功!")
        print(f"   视频ID: {video_id}")
        print(f"   状态: {result.get('status')}")
        print(f"   预计生成时间: 3-5 分钟")
        return video_id
    else:
        print(f"❌ 请求失败: {response.status_code}")
        print(f"   响应: {response.text}")
        return None


def step2_poll_status(video_id):
    """第二步：轮询查询视频生成状态"""
    print("\n" + "=" * 60)
    print("第二步：查询视频生成状态")
    print("=" * 60)

    headers = {'Authorization': API_KEY}
    status_url = f"{BASE_URL}/{video_id}"
    start_time = time.time()

    while True:
        elapsed_time = time.time() - start_time

        if elapsed_time > MAX_WAIT_TIME:
            print(f"⏱️ 超时：已等待 {MAX_WAIT_TIME} 秒")
            return False

        response = requests.get(status_url, headers=headers)

        if response.status_code == 200:
            result = response.json()
            status = result.get('status')
            progress = result.get('progress', 0)

            print(f"\n🔍 查询状态... (已等待 {int(elapsed_time)} 秒)")
            print(f"   状态: {status}")
            print(f"   进度: {progress}%")

            if status == 'completed':
                print(f"✅ 视频生成完成!")
                print(f"   视频URL: {result.get('url')}")
                return True
            elif status == 'failed':
                print(f"❌ 视频生成失败")
                return False
            else:
                print(f"⏳ 等待 {POLL_INTERVAL} 秒后继续查询...")
                time.sleep(POLL_INTERVAL)
        else:
            print(f"❌ 状态查询失败: {response.status_code}")
            time.sleep(POLL_INTERVAL)


def step3_download_video(video_id, output_path=None):
    """第三步：下载生成的视频"""
    print("\n" + "=" * 60)
    print("第三步：下载视频")
    print("=" * 60)

    if output_path is None:
        output_path = f"sora_video_{video_id.split('_')[-1]}.mp4"

    headers = {'Authorization': API_KEY}
    content_url = f"{BASE_URL}/{video_id}/content"

    print(f"📥 开始下载视频...")
    response = requests.get(content_url, headers=headers, stream=True)

    if response.status_code == 200:
        total_size = int(response.headers.get('content-length', 0))

        with open(output_path, 'wb') as f:
            if total_size == 0:
                f.write(response.content)
            else:
                downloaded = 0
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
                        downloaded += len(chunk)
                        percent = (downloaded / total_size) * 100
                        print(f"\r   下载进度: {percent:.1f}%", end='')

        print(f"\n✅ 视频下载成功!")
        print(f"   保存路径: {os.path.abspath(output_path)}")
        print(f"   文件大小: {os.path.getsize(output_path)} 字节")
        return True
    else:
        print(f"❌ 下载失败: {response.status_code}")
        return False


def main():
    """完整的异步流程"""
    print("\n🎬 开始 Sora-2 视频生成完整流程")
    print(f"⏰ 开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")

    # 第一步：提交请求
    video_id = step1_submit_request()
    if not video_id:
        print("\n❌ 流程终止：视频请求提交失败")
        return

    # 第二步：轮询状态
    success = step2_poll_status(video_id)
    if not success:
        print("\n❌ 流程终止：视频生成未完成")
        return

    # 第三步：下载视频
    success = step3_download_video(video_id)

    if success:
        print("\n" + "=" * 60)
        print("🎉 完整流程执行成功!")
        print("=" * 60)
        print(f"⏰ 结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")


if __name__ == "__main__":
    main()