异步调用 API

curl -X POST "https://api.yelinai.com/v1/videos" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "veo-3.1",
    "prompt": "一只可爱的猫咪在阳光明媚的花园里玩球"
  }'

# 响应示例
{
  "id": "video_abc123",
  "object": "video",
  "created": 1762181811,
  "status": "queued",
  "model": "veo-3.1"
}

import requests
import time
import os

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.yelinai.com/v1"

# 步骤1: 创建视频任务
def create_video_task(prompt, model="veo-3.1", image_paths=None):
    """创建视频生成任务

    Args:
        prompt: 视频生成提示词
        model: 模型名称
        image_paths: 图片路径，可以是：
            - None: 文生视频
            - str: 单张图片路径
            - list: 多张图片路径（首尾帧模式，最多2张）
    """
    url = f"{BASE_URL}/videos"
    headers = {"Authorization": f"Bearer {API_KEY}"}

    if image_paths:
        # 统一转为列表处理
        if isinstance(image_paths, str):
            image_paths = [image_paths]

        # 图生视频：使用 multipart/form-data 上传图片
        files = []
        for path in image_paths:
            if not os.path.exists(path):
                raise FileNotFoundError(f"图片文件不存在: {path}")
            files.append(("input_reference", (os.path.basename(path), open(path, 'rb'), "image/jpeg")))

        data = {"model": model, "prompt": prompt}
        response = requests.post(url, headers=headers, files=files, data=data)

        # 关闭文件句柄
        for _, (_, f, _) in files:
            f.close()
    else:
        # 文生视频：使用 JSON 格式
        headers["Content-Type"] = "application/json"
        data = {"model": model, "prompt": prompt}
        response = requests.post(url, headers=headers, json=data)

    response.raise_for_status()
    return response.json()

# 步骤2: 轮询查询状态
def wait_for_video(video_id, poll_interval=5, timeout=600):
    """等待视频生成完成"""
    url = f"{BASE_URL}/videos/{video_id}"
    headers = {"Authorization": f"Bearer {API_KEY}"}
    start_time = time.time()

    while True:
        # 检查超时
        if time.time() - start_time > timeout:
            raise TimeoutError(f"视频生成超时（{timeout}秒）")

        # 查询状态
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        task = response.json()

        status = task["status"]
        print(f"状态: {status}")

        if status == "completed":
            return task
        elif status == "failed":
            raise Exception(f"生成失败")

        # 等待后重试
        time.sleep(poll_interval)

# 步骤3: 获取视频内容
def get_video_content(video_id):
    """获取视频内容和URL"""
    url = f"{BASE_URL}/videos/{video_id}/content"
    headers = {"Authorization": f"Bearer {API_KEY}"}

    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# 步骤4: 下载视频
def download_video(video_url, save_path="video.mp4"):
    """下载视频文件"""
    response = requests.get(video_url, stream=True)
    response.raise_for_status()

    with open(save_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)

    print(f"视频已保存到: {save_path}")

# 完整流程
def generate_video_async(prompt, model="veo-3.1"):
    """异步生成视频的完整流程"""
    print("1. 创建视频任务...")
    task = create_video_task(prompt, model)
    video_id = task["id"]
    print(f"   任务ID: {video_id}")

    print("\n2. 等待视频生成...")
    completed_task = wait_for_video(video_id)
    print("   生成完成！")

    print("\n3. 获取视频内容...")
    content = get_video_content(video_id)
    video_url = content["url"]
    print(f"   视频URL: {video_url}")

    print("\n4. 下载视频...")
    download_video(video_url)
    print("\n✅ 完成！")

# 使用示例
if __name__ == "__main__":
    # 文生视频 - 竖屏标准版
    # 文生视频 - 竖屏标准版
    generate_video_async(
        prompt="一只可爱的猫咪在阳光明媚的花园里玩球",
        model="veo-3.1"
    )

    # 图生视频 - 单图模式
    # task = create_video_task(
    #     prompt="让这只猫咪慢慢眨眼睛",
    #     model="veo-3.1-fl",
    #     image_paths="test.jpg"
    # )

    # 图生视频 - 首尾帧模式（传入2张图片）
    # 第一张图作为开始帧，第二张图作为结束帧
    # task = create_video_task(
    #     prompt="让画面从开始帧平滑过渡到结束帧，加入动态效果",
    #     model="veo-3.1-landscape-fl",
    #     image_paths=["start_frame.jpg", "end_frame.jpg"]
    # )

    # 文生视频 - 横屏快速版
    # generate_video_async(
    #     prompt="日落时分，海浪轻轻拍打沙滩",
    #     model="veo-3.1-landscape-fast"
    # )

const axios = require('axios');
const fs = require('fs');

const API_KEY = 'YOUR_API_KEY';
const BASE_URL = 'https://api.yelinai.com/v1';

// 创建视频任务
async function createVideoTask(prompt, model = 'veo-3.1') {
  const response = await axios.post(`\`$\{BASE_URL}/videos`, {
    model,
    prompt
  }, {
    headers: { 'Authorization': `Bearer \`$\{API_KEY}` }
  });

  return response.data;
}

// 轮询查询状态
async function waitForVideo(videoId, pollInterval = 5000, timeout = 600000) {
  const startTime = Date.now();

  while (true) {
    // 检查超时
    if (Date.now() - startTime > timeout) {
      throw new Error(`视频生成超时（\`$\{timeout/1000}秒）`);
    }

    // 查询状态
    const response = await axios.get(`\`$\{BASE_URL}/videos/\`$\{videoId}`, {
      headers: { 'Authorization': `Bearer \`$\{API_KEY}` }
    });

    const task = response.data;
    const { status } = task;

    console.log(`状态: \`$\{status}`);

    if (status === 'completed') {
      return task;
    } else if (status === 'failed') {
      throw new Error('生成失败');
    }

    // 等待后重试
    await new Promise(resolve => setTimeout(resolve, pollInterval));
  }
}

// 获取视频内容
async function getVideoContent(videoId) {
  const response = await axios.get(`\`$\{BASE_URL}/videos/\`$\{videoId}/content`, {
    headers: { 'Authorization': `Bearer \`$\{API_KEY}` }
  });

  return response.data;
}

// 下载视频
async function downloadVideo(videoUrl, savePath = 'video.mp4') {
  const response = await axios.get(videoUrl, { responseType: 'stream' });

  const writer = fs.createWriteStream(savePath);
  response.data.pipe(writer);

  return new Promise((resolve, reject) => {
    writer.on('finish', () => {
      console.log(`视频已保存到: \`$\{savePath}`);
      resolve();
    });
    writer.on('error', reject);
  });
}

// 完整流程
async function generateVideoAsync(prompt, model = 'veo-3.1') {
  try {
    console.log('1. 创建视频任务...');
    const task = await createVideoTask(prompt, model);
    const videoId = task.id;
    console.log(`   任务ID: \`$\{videoId}`);

    console.log('\n2. 等待视频生成...');
    await waitForVideo(videoId);
    console.log('   生成完成！');

    console.log('\n3. 获取视频内容...');
    const content = await getVideoContent(videoId);
    const videoUrl = content.url;
    console.log(`   视频URL: \`$\{videoUrl}`);

    console.log('\n4. 下载视频...');
    await downloadVideo(videoUrl);
    console.log('\n✅ 完成！');

  } catch (error) {
    console.error('错误:', error.message);
  }
}

// 使用示例
generateVideoAsync(
  '一只可爱的猫咪在阳光明媚的花园里玩球',
  'veo-3.1'
);

# 推荐
poll_interval = 5  # 每5秒查询一次

# 不推荐
poll_interval = 1  # 太频繁，浪费请求
poll_interval = 30 # 太慢，用户体验差

def wait_for_video(video_id, timeout=600):
    start_time = time.time()

    while True:
        if time.time() - start_time > timeout:
            # 超时后的处理
            print(f"任务 {video_id} 超时，稍后可继续查询")
            break

        # 查询逻辑...

def create_video_with_retry(prompt, model, max_retries=3):
    for i in range(max_retries):
        try:
            return create_video_task(prompt, model)
        except Exception as e:
            if i < max_retries - 1:
                print(f"创建失败，{5}秒后重试... ({i+1}/{max_retries})")
                time.sleep(5)
            else:
                raise

import concurrent.futures

def batch_generate_videos(prompts, model="veo-3.1", max_workers=5):
    """批量生成视频，控制并发数"""
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 创建所有任务
        future_to_prompt = {
            executor.submit(create_video_task, prompt, model): prompt
            for prompt in prompts
        }

        video_ids = []
        for future in concurrent.futures.as_completed(future_to_prompt):
            task = future.result()
            video_ids.append(task['id'])

        # 并发等待所有任务完成
        futures = [executor.submit(wait_for_video, vid) for vid in video_ids]
        results = [f.result() for f in concurrent.futures.as_completed(futures)]

        return results

try:
    response = requests.get(f"{BASE_URL}/videos/{video_id}", headers=headers)
    task = response.json()
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 404:
        print("任务不存在或已过期")
    else:
        raise

{
  "error": {
    "code": "invalid_api_key",
    "message": "Invalid API key provided",
    "type": "authentication_error"
  }
}