Cheapest Veo 3.1 API for YouTube Automation & Faceless Channels

import os, time, uuid, requests

BASE = "https://avatarity.dev"
HEADERS = {
    "Authorization": f"Bearer {os.environ['AVATARITY_API_KEY']}",
    "Content-Type": "application/json",
}

def render_veo_scene(prompt: str, aspect_ratio: str = "16:9", start_image_url: str | None = None) -> dict:
    """Start an 8-second 720p Veo 3.1 Lite render and poll until completion."""
    payload = {
        "model": "veo-3.1-lite",
        "prompt": prompt,
        "aspect_ratio": aspect_ratio,
        "resolution": "720p",
    }
    if start_image_url:
        payload["image"] = start_image_url

    resp = requests.post(
        f"{BASE}/v1/video/generations",
        headers={**HEADERS, "Idempotency-Key": str(uuid.uuid4())},
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    task_id = resp.json()["task_id"]

    deadline = time.time() + 600
    while time.time() < deadline:
        time.sleep(5)
        task = requests.get(f"{BASE}/v1/tasks/{task_id}", headers=HEADERS, timeout=30).json()
        if task["status"] == "succeeded":
            return task
        if task["status"] in ("failed", "cancelled"):
            raise RuntimeError((task.get("error") or {}).get("code", "RENDER_FAILED"))
    raise TimeoutError(f"Task {task_id} timed out")

def upscale_approved_cut_to_1080p(source_task_id: str) -> str:
    """Upscale a finished 720p Veo 3.1 Lite render to Full HD 1080p (costs 1 additional video unit)."""
    resp = requests.post(
        f"{BASE}/v1/video/generations",
        headers={**HEADERS, "Idempotency-Key": str(uuid.uuid4())},
        json={"model": "veo-3.1-lite", "prompt": "Upscale", "source_task_id": source_task_id},
        timeout=30,
    )
    resp.raise_for_status()
    upscale_id = resp.json()["task_id"]

    deadline = time.time() + 600
    while time.time() < deadline:
        time.sleep(5)
        task = requests.get(f"{BASE}/v1/tasks/{upscale_id}", headers=HEADERS, timeout=30).json()
        if task["status"] == "succeeded":
            return task["video_url"]
        if task["status"] in ("failed", "cancelled"):
            raise RuntimeError((task.get("error") or {}).get("code", "UPSCALE_FAILED"))
    raise TimeoutError(f"Upscale {upscale_id} timed out")

// Server-side Node.js 18+ / TypeScript helper for YouTube automation queues
const BASE = 'https://avatarity.dev';
const API_KEY = process.env.AVATARITY_API_KEY!;

export async function enqueueYoutubeScene(params: {
  prompt: string;
  aspectRatio: '16:9' | '9:16';
  startFrameUrl?: string;
  webhookUrl: string;
  sceneId: string;
}): Promise<{ taskId: string; remainingVeo: number | null }> {
  const res = await fetch(`${BASE}/v1/video/generations`, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${API_KEY}`,
      'Content-Type': 'application/json',
      'Idempotency-Key': `yt-scene-${params.sceneId}`,
    },
    body: JSON.stringify({
      model: 'veo-3.1-lite',
      prompt: params.prompt,
      aspect_ratio: params.aspectRatio,
      image: params.startFrameUrl,
      webhook_url: params.webhookUrl,
    }),
  });

  const body = await res.json();
  if (!res.ok) {
    throw new Error(`Avatarity error ${body?.error?.code ?? res.status}: ${body?.error?.message ?? ''}`);
  }

  const remainingHeader = res.headers.get('x-avatarity-remaining-veo');
  return {
    taskId: body.task_id,
    remainingVeo: remainingHeader ? Number(remainingHeader) : null,
  };
}