异步调用 API

Content-Type: application/json
Authorization: sk-APIKEY

{
  "id": "video_abc123",
  "object": "video",
  "created": 1762181811,
  "status": "queued",
  "model": "veo-3.1"
}

{
  "id": "video_abc123",
  "object": "video",
  "created": 1762181811,
  "status": "completed",
  "model": "veo-3.1",
  "prompt": "画小猫",
  "url": "https://veo-video.gptkey.asia/assets/flow/xxx.mp4",
  "duration": 8,
  "resolution": "720x1280"
}

import requests
import time

class VEOClient:
    """VEO 3.1 异步调用客户端"""

    def __init__(self, api_key: str, base_url: str = "https://api.apiyi.com"):
        self.base_url = base_url
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": api_key
        }

    def create_video(self, prompt: str, model: str = "veo-3.1") -> str:
        """创建视频任务，返回 video_id"""
        response = requests.post(
            f"{self.base_url}/v1/videos",
            headers=self.headers,
            json={"prompt": prompt, "model": model}
        )
        response.raise_for_status()
        return response.json()["id"]

    def get_status(self, video_id: str) -> dict:
        """查询任务状态"""
        response = requests.get(
            f"{self.base_url}/v1/videos/{video_id}",
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def get_content(self, video_id: str) -> dict:
        """获取视频内容"""
        response = requests.get(
            f"{self.base_url}/v1/videos/{video_id}/content",
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def wait_for_completion(self, video_id: str, timeout: int = 600, interval: int = 5) -> dict:
        """等待任务完成"""
        start_time = time.time()
        while time.time() - start_time < timeout:
            status_data = self.get_status(video_id)
            status = status_data.get("status")

            if status == "completed":
                return self.get_content(video_id)
            elif status == "failed":
                raise Exception(f"视频生成失败: {status_data}")

            print(f"任务进行中... 状态: {status}")
            time.sleep(interval)

        raise TimeoutError("等待超时")


# 使用示例
if __name__ == "__main__":
    client = VEOClient("sk-your-api-key")

    try:
        # 1. 创建任务
        video_id = client.create_video(
            prompt="一只猫咪在阳光下的花园里漫步",
            model="veo-3.1-fast"
        )
        print(f"任务已创建，ID: {video_id}")

        # 2. 等待完成并获取结果
        result = client.wait_for_completion(video_id)
        print(f"视频生成成功！")
        print(f"视频URL: {result['url']}")
        print(f"分辨率: {result['resolution']}")

    except Exception as e:
        print(f"错误: {e}")

{
  "id": "video_xyz789",
  "object": "video",
  "created": 1762181811,
  "status": "queued",
  "model": "veo-3.1-landscape-fl"
}

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