import requests
import time
import os
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.yelinai.com/v1"
# Step 1: Create video task
def create_video_task(prompt, model="veo-3.1", image_paths=None):
"""Create video generation task
Args:
prompt: Video generation prompt
model: Model name
image_paths: Image path(s), can be:
- None: Text-to-video
- str: Single image path
- list: Multiple image paths (first/last frame mode, max 2)
"""
url = f"{BASE_URL}/videos"
headers = {"Authorization": f"Bearer {API_KEY}"}
if image_paths:
# Convert to list for uniform handling
if isinstance(image_paths, str):
image_paths = [image_paths]
# Image-to-Video: Use multipart/form-data to upload images
files = []
for path in image_paths:
if not os.path.exists(path):
raise FileNotFoundError(f"Image file not found: {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)
# Close file handles
for _, (_, f, _) in files:
f.close()
else:
# Text-to-Video: Use JSON format
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()
# Step 2: Poll for status
def wait_for_video(video_id, poll_interval=5, timeout=600):
"""Wait for video generation to complete"""
url = f"{BASE_URL}/videos/{video_id}"
headers = {"Authorization": f"Bearer {API_KEY}"}
start_time = time.time()
while True:
# Check timeout
if time.time() - start_time > timeout:
raise TimeoutError(f"Video generation timeout ({timeout}s)")
# Query status
response = requests.get(url, headers=headers)
response.raise_for_status()
task = response.json()
status = task["status"]
print(f"Status: {status}")
if status == "completed":
return task
elif status == "failed":
raise Exception("Generation failed")
# Wait and retry
time.sleep(poll_interval)
# Step 3: Get video content
def get_video_content(video_id):
"""Get video content and 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()
# Step 4: Download video
def download_video(video_url, save_path="video.mp4"):
"""Download video file"""
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"Video saved to: {save_path}")
# Complete workflow
def generate_video_async(prompt, model="veo-3.1"):
"""Complete async video generation workflow"""
print("1. Creating video task...")
task = create_video_task(prompt, model)
video_id = task["id"]
print(f" Task ID: {video_id}")
print("\n2. Waiting for video generation...")
completed_task = wait_for_video(video_id)
print(" Generation complete!")
print("\n3. Getting video content...")
content = get_video_content(video_id)
video_url = content["url"]
print(f" Video URL: {video_url}")
print("\n4. Downloading video...")
download_video(video_url)
print("\n✅ Done!")
# Usage example
if __name__ == "__main__":
# Text-to-video - Portrait standard
# Text-to-video - Portrait standard
generate_video_async(
prompt="A cute cat playing with a ball in a sunny garden",
model="veo-3.1"
)
# Image-to-Video - Single image mode
# task = create_video_task(
# prompt="Make this cat blink slowly",
# model="veo-3.1-fl",
# image_paths="test.jpg"
# )
# Image-to-Video - First/Last frame mode (2 images)
# First image = start frame, Second image = end frame
# task = create_video_task(
# prompt="Smoothly transition from start to end frame with dynamic effects",
# model="veo-3.1-landscape-fl",
# image_paths=["start_frame.jpg", "end_frame.jpg"]
# )
