Async API (Temporarily Unavailable)

import requests
import time
import os

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api2.laozhang.ai/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"]
    # )

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

const API_KEY = 'YOUR_API_KEY';
const BASE_URL = 'https://api2.laozhang.ai/v1';

// Create video task
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;
}

// Poll for status
async function waitForVideo(videoId, pollInterval = 5000, timeout = 600000) {
  const startTime = Date.now();

  while (true) {
    // Check timeout
    if (Date.now() - startTime > timeout) {
      throw new Error(`Video generation timeout (${timeout/1000}s)`);
    }

    // Query status
    const response = await axios.get(`${BASE_URL}/videos/${videoId}`, {
      headers: { 'Authorization': `Bearer ${API_KEY}` }
    });

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

    console.log(`Status: ${status}`);

    if (status === 'completed') {
      return task;
    } else if (status === 'failed') {
      throw new Error('Generation failed');
    }

    // Wait and retry
    await new Promise(resolve => setTimeout(resolve, pollInterval));
  }
}

// Get video content
async function getVideoContent(videoId) {
  const response = await axios.get(`${BASE_URL}/videos/${videoId}/content`, {
    headers: { 'Authorization': `Bearer ${API_KEY}` }
  });

  return response.data;
}

// Download video
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(`Video saved to: ${savePath}`);
      resolve();
    });
    writer.on('error', reject);
  });
}

// Complete workflow
async function generateVideoAsync(prompt, model = 'veo-3.1') {
  try {
    console.log('1. Creating video task...');
    const task = await createVideoTask(prompt, model);
    const videoId = task.id;
    console.log(`   Task ID: ${videoId}`);

    console.log('\n2. Waiting for video generation...');
    await waitForVideo(videoId);
    console.log('   Generation complete!');

    console.log('\n3. Getting video content...');
    const content = await getVideoContent(videoId);
    const videoUrl = content.url;
    console.log(`   Video URL: ${videoUrl}`);

    console.log('\n4. Downloading video...');
    await downloadVideo(videoUrl);
    console.log('\n✅ Done!');

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

// Usage example
generateVideoAsync(
  'A cute cat playing with a ball in a sunny garden',
  'veo-3.1'
);

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