Building an Automated Content Pipeline with Veo 3 & Lyria 3 Pro API — Mass-Producing Video + Music

# Python packages
pip install google-genai python-dotenv ffmpeg-python
 
# ffmpeg binary (macOS)
brew install ffmpeg
 
# ffmpeg binary (Ubuntu/Debian)
# apt-get install -y ffmpeg

# config.py — project configuration and client initialization
import os
from dotenv import load_dotenv
from google import genai
 
load_dotenv()
 
def get_client() -> genai.Client:
    """
    Initialize and return the Google Gen AI client.
    API key is read from the GOOGLE_AI_API_KEY environment variable.
    """
    api_key = os.getenv("GOOGLE_AI_API_KEY")
    if not api_key:
        raise ValueError(
            "GOOGLE_AI_API_KEY is not set. "
            "Add GOOGLE_AI_API_KEY=your_key_here to your .env file."
        )
    return genai.Client(api_key=api_key)
 
# Model name constants — update these when new versions release
VEO3_MODEL = "veo-3.0-generate-preview"
LYRIA3_MODEL = "lyria-3.0-pro"
 
# Output directories
OUTPUT_DIR = "output"
TEMP_DIR = "temp"
 
os.makedirs(OUTPUT_DIR, exist_ok=True)
os.makedirs(TEMP_DIR, exist_ok=True)

# video_generator.py — Veo 3 video generation
import time
import logging
from pathlib import Path
from google import genai
from google.genai import types
from config import get_client, VEO3_MODEL, TEMP_DIR
 
logger = logging.getLogger(__name__)
 
def generate_video(
    prompt: str,
    output_filename: str,
    aspect_ratio: str = "16:9",
    duration_seconds: int = 8,
    max_retries: int = 3,
    poll_interval: int = 10,
    timeout_seconds: int = 300,
) -> Path | None:
    """
    Generate a video from a text prompt using Veo 3 and save it to disk.
 
    Args:
        prompt: Text describing the video content (English yields best results)
        output_filename: Output file name without extension
        aspect_ratio: "16:9" for landscape, "9:16" for portrait/Shorts
        duration_seconds: Video length in seconds (max 8)
        max_retries: Number of retry attempts on failure
        poll_interval: Seconds between polling requests
        timeout_seconds: Total timeout before giving up
 
    Returns:
        Path to the saved video file, or None on failure.
    """
    client = get_client()
    output_path = Path(TEMP_DIR) / f"{output_filename}.mp4"
 
    for attempt in range(1, max_retries + 1):
        try:
            logger.info(f"Veo 3 generation started (attempt {attempt}/{max_retries}): {prompt[:60]}...")
 
            # Submit the generation request — returns an Operation, not the video
            operation = client.models.generate_video(
                model=VEO3_MODEL,
                prompt=prompt,
                config=types.GenerateVideoConfig(
                    aspect_ratio=aspect_ratio,
                    duration_seconds=duration_seconds,
                    number_of_videos=1,
                    enhance_prompt=True,  # Let the model improve the prompt automatically
                ),
            )
 
            # Poll until done or timeout
            start_time = time.time()
            while not operation.done:
                elapsed = time.time() - start_time
                if elapsed > timeout_seconds:
                    raise TimeoutError(
                        f"Video generation timed out after {timeout_seconds}s. "
                        "This often means the prompt triggered a safety filter without "
                        "raising an explicit error. Try rephrasing the prompt."
                    )
                logger.info(f"  Generating... {elapsed:.0f}s elapsed")
                time.sleep(poll_interval)
                operation = client.operations.get(operation)
 
            # Check for explicit errors
            if operation.error:
                raise RuntimeError(f"Generation error: {operation.error.message}")
 
            # Download and save the video
            video = operation.result.generated_videos[0]
            video_bytes = client.files.download(file=video.video)
 
            with open(output_path, "wb") as f:
                f.write(video_bytes)
 
            size_kb = output_path.stat().st_size // 1024
            logger.info(f"✅ Video saved: {output_path} ({size_kb} KB)")
            return output_path
 
        except TimeoutError as e:
            logger.warning(f"⚠️ Timeout (attempt {attempt}): {e}")
            if attempt == max_retries:
                logger.error("Max retries reached — skipping this video")
                return None
 
        except Exception as e:
            logger.error(f"❌ Error (attempt {attempt}): {type(e).__name__}: {e}")
            if attempt < max_retries:
                wait = 2 ** attempt * 5  # 10s → 20s → 40s
                logger.info(f"  Retrying in {wait}s...")
                time.sleep(wait)
            else:
                logger.error("Max retries reached — skipping this video")
                return None
 
    return None

# music_generator.py — Lyria 3 Pro music generation
import time
import logging
from pathlib import Path
from google import genai
from google.genai import types
from config import get_client, LYRIA3_MODEL, TEMP_DIR
 
logger = logging.getLogger(__name__)
 
def generate_music(
    prompt: str,
    output_filename: str,
    duration_seconds: int = 30,
    output_format: str = "mp3",
    max_retries: int = 3,
) -> Path | None:
    """
    Generate music from a text prompt using Lyria 3 Pro and save it to disk.
 
    Args:
        prompt: Text describing the musical style, mood, and instrumentation
        output_filename: Output file name without extension
        duration_seconds: Track length in seconds (max 210)
        output_format: "mp3" or "wav"
        max_retries: Number of retry attempts on failure
 
    Returns:
        Path to the saved audio file, or None on failure.
    """
    client = get_client()
    output_path = Path(TEMP_DIR) / f"{output_filename}.{output_format}"
 
    for attempt in range(1, max_retries + 1):
        try:
            logger.info(f"Lyria 3 Pro generation started (attempt {attempt}/{max_retries})")
 
            response = client.models.generate_music(
                model=LYRIA3_MODEL,
                prompt=prompt,
                config=types.GenerateMusicConfig(
                    duration_seconds=duration_seconds,
                    output_format=output_format,
                    seamless_loop=False,
                ),
            )
 
            if not response.audio_data:
                raise ValueError("Response returned empty audio data")
 
            with open(output_path, "wb") as f:
                f.write(response.audio_data)
 
            logger.info(f"✅ Music saved: {output_path}")
            return output_path
 
        except Exception as e:
            logger.error(f"❌ Error (attempt {attempt}): {type(e).__name__}: {e}")
            if attempt < max_retries:
                wait = 2 ** attempt * 3  # 6s → 12s → 24s
                logger.info(f"  Retrying in {wait}s...")
                time.sleep(wait)
            else:
                logger.error("Max retries reached — skipping this track")
                return None
 
    return None
 
def build_music_prompt(
    genre: str,
    mood: str,
    tempo: str = "moderate",
    instruments: list[str] | None = None,
    no_lyrics: bool = True,
) -> str:
    """
    Build a structured music prompt that Lyria 3 Pro responds well to.
 
    Lyria performs best with explicit, comma-separated style descriptors
    rather than natural language sentences. This function assembles them
    in the right order.
 
    Example:
        build_music_prompt("ambient", "peaceful", "slow", ["koto", "piano"])
        → "ambient music, peaceful mood, slow tempo, featuring koto, piano,
           instrumental, no vocals, no lyrics, high quality, studio quality"
    """
    parts = [f"{genre} music", f"{mood} mood", f"{tempo} tempo"]
    if instruments:
        parts.append(f"featuring {', '.join(instruments)}")
    if no_lyrics:
        parts.append("instrumental, no vocals, no lyrics")
    parts.append("high quality, professional recording, studio quality")
    return ", ".join(parts)

# pipeline.py — the full automated pipeline
import asyncio
import logging
import ffmpeg
from pathlib import Path
from datetime import datetime
from config import OUTPUT_DIR, TEMP_DIR
from video_generator import generate_video
from music_generator import generate_music, build_music_prompt
 
logger = logging.getLogger(__name__)
 
def merge_video_audio(
    video_path: Path,
    audio_path: Path,
    output_filename: str,
    audio_volume: float = 0.5,
    fade_out_seconds: float = 1.0,
) -> Path | None:
    """
    Merge a video file and an audio file using ffmpeg.
 
    - Audio is trimmed to match the video duration
    - A fade-out is applied to the last `fade_out_seconds` of audio
    - Volume is controlled by `audio_volume` (0.0–1.0)
    - Video stream is copied without re-encoding (fast)
    - Audio is encoded as AAC 192k
    """
    output_path = Path(OUTPUT_DIR) / f"{output_filename}.mp4"
 
    try:
        probe = ffmpeg.probe(str(video_path))
        video_duration = float(probe["format"]["duration"])
 
        video_in = ffmpeg.input(str(video_path))
        audio_in = ffmpeg.input(str(audio_path))
 
        fade_start = max(0.0, video_duration - fade_out_seconds)
        audio_processed = (
            audio_in.audio
            .filter("atrim", duration=video_duration)
            .filter("asetpts", "PTS-STARTPTS")
            .filter("volume", audio_volume)
            .filter("afade", type="out", start_time=fade_start, duration=fade_out_seconds)
        )
 
        out = ffmpeg.output(
            video_in.video,
            audio_processed,
            str(output_path),
            vcodec="copy",
            acodec="aac",
            audio_bitrate="192k",
        )
        ffmpeg.run(out, overwrite_output=True, quiet=True)
 
        logger.info(f"✅ Merge complete: {output_path}")
        return output_path
 
    except ffmpeg.Error as e:
        logger.error(f"❌ ffmpeg error: {e.stderr.decode()}")
        return None
 
async def run_pipeline(
    video_prompt: str,
    music_prompt: str,
    output_name: str,
    video_duration: int = 8,
    music_duration: int = 10,
) -> dict:
    """
    Run video and music generation in parallel, then merge the results.
 
    Returns a result dict with keys: status, output_path, elapsed_seconds
    """
    start_time = asyncio.get_event_loop().time()
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    base = f"{output_name}_{timestamp}"
 
    logger.info(f"🎬 Pipeline started: {output_name}")
 
    loop = asyncio.get_event_loop()
 
    # Run both generations concurrently
    video_task = loop.run_in_executor(
        None, generate_video, video_prompt, f"{base}_video", "16:9", video_duration
    )
    music_task = loop.run_in_executor(
        None, generate_music, music_prompt, f"{base}_music", music_duration
    )
 
    video_path, music_path = await asyncio.gather(video_task, music_task)
 
    if not video_path:
        return {"status": "FAILED", "reason": "video generation failed"}
    if not music_path:
        return {"status": "FAILED", "reason": "music generation failed"}
 
    output_path = merge_video_audio(video_path, music_path, base)
 
    elapsed = round(asyncio.get_event_loop().time() - start_time, 1)
 
    # Clean up temp files
    video_path.unlink(missing_ok=True)
    music_path.unlink(missing_ok=True)
 
    if output_path:
        return {"status": "SUCCESS", "output_path": str(output_path), "elapsed_seconds": elapsed}
    return {"status": "FAILED", "reason": "merge failed"}
 
# ── Example usage ──────────────────────────────────────────────────────────
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
 
    result = asyncio.run(run_pipeline(
        video_prompt=(
            "A serene Japanese garden in autumn, koi pond with red and orange maple leaves "
            "falling gently onto the water surface, soft morning light filtering through bamboo, "
            "cinematic 4K quality, slow motion"
        ),
        music_prompt=build_music_prompt(
            genre="ambient",
            mood="peaceful and contemplative",
            tempo="slow",
            instruments=["koto", "shakuhachi flute", "gentle piano"],
        ),
        output_name="japanese_garden",
        video_duration=8,
        music_duration=10,
    ))
 
    print(result)
    # Expected output (times will vary):
    # {'status': 'SUCCESS', 'output_path': 'output/japanese_garden_20260418_104500.mp4', 'elapsed_seconds': 87.3}

# scheduler.py — batch pipeline runner with queue and logging
import asyncio
import json
import logging
import time
from datetime import datetime
from pathlib import Path
from pipeline import run_pipeline, build_music_prompt  # type: ignore
 
logger = logging.getLogger(__name__)
 
LOG_DIR = Path("logs")
LOG_DIR.mkdir(exist_ok=True)
 
# How many seconds to wait between jobs to avoid rate limit 429s
# Adjust based on your quota tier
INTER_JOB_DELAY_SECONDS = 90
 
def load_job_queue(queue_file: str) -> list[dict]:
    """
    Load the content queue from a JSON file.
 
    Expected format:
    [
      {
        "name": "autumn_garden",
        "video_prompt": "...",
        "music_genre": "ambient",
        "music_mood": "calm",
        "music_tempo": "slow",
        "music_instruments": ["piano", "strings"]
      },
      ...
    ]
    """
    with open(queue_file, "r", encoding="utf-8") as f:
        jobs = json.load(f)
    logger.info(f"Loaded {len(jobs)} jobs from {queue_file}")
    return jobs
 
async def process_queue(queue_file: str = "content_queue.json") -> list[dict]:
    """
    Process all jobs in the queue file, one at a time with rate-limit spacing.
 
    Returns a list of result dicts for downstream logging or alerting.
    """
    jobs = load_job_queue(queue_file)
    results = []
    log_file = LOG_DIR / f"batch_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jsonl"
 
    for i, job in enumerate(jobs, start=1):
        logger.info(f"\n{'='*60}")
        logger.info(f"Job {i}/{len(jobs)}: {job['name']}")
 
        music_prompt = build_music_prompt(
            genre=job.get("music_genre", "ambient"),
            mood=job.get("music_mood", "calm"),
            tempo=job.get("music_tempo", "moderate"),
            instruments=job.get("music_instruments"),
        )
 
        result = await run_pipeline(
            video_prompt=job["video_prompt"],
            music_prompt=music_prompt,
            output_name=job["name"],
        )
 
        result["job_name"] = job["name"]
        result["timestamp"] = datetime.utcnow().isoformat()
        results.append(result)
 
        # Append to JSONL log for easy parsing
        with open(log_file, "a", encoding="utf-8") as lf:
            lf.write(json.dumps(result, ensure_ascii=False) + "\n")
 
        status_icon = "✅" if result["status"] == "SUCCESS" else "❌"
        logger.info(f"{status_icon} {job['name']}: {result['status']}")
 
        # Rate-limit spacing — skip delay after the last job
        if i < len(jobs):
            logger.info(f"Waiting {INTER_JOB_DELAY_SECONDS}s before next job...")
            await asyncio.sleep(INTER_JOB_DELAY_SECONDS)
 
    successes = sum(1 for r in results if r["status"] == "SUCCESS")
    logger.info(f"\n🏁 Batch complete: {successes}/{len(jobs)} succeeded")
    logger.info(f"Log written to: {log_file}")
    return results
 
if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        handlers=[
            logging.StreamHandler(),
            logging.FileHandler(LOG_DIR / "scheduler.log"),
        ],
    )
    asyncio.run(process_queue("content_queue.json"))

[
  {
    "name": "morning_forest",
    "video_prompt": "Dense forest at dawn, sunlight breaking through tall cedar trees, mist rising from the ground, birds visible in the canopy, cinematic wide shot, 4K quality",
    "music_genre": "acoustic",
    "music_mood": "uplifting and serene",
    "music_tempo": "slow",
    "music_instruments": ["acoustic guitar", "light percussion", "flute"]
  },
  {
    "name": "ocean_timelapse",
    "video_prompt": "Ocean waves crashing against dark volcanic rocks at golden hour, seafoam swirling in slow motion, warm orange and purple sky, ultra high definition",
    "music_genre": "cinematic orchestral",
    "music_mood": "epic and expansive",
    "music_tempo": "moderate",
    "music_instruments": ["strings", "piano", "french horn"]
  }
]

# quality_filter.py — basic output validation
import logging
from pathlib import Path
 
logger = logging.getLogger(__name__)
 
# Expected size range for 8-second 1080p Veo 3 output (bytes)
MIN_VIDEO_SIZE_BYTES = 5 * 1024 * 1024   # 5 MB
MAX_VIDEO_SIZE_BYTES = 150 * 1024 * 1024 # 150 MB
 
def validate_video_output(video_path: Path, duration_seconds: int = 8) -> bool:
    """
    Run basic validation checks on a generated video file.
 
    Checks:
    1. File exists and is non-empty
    2. File size is within the expected range for the given duration
    3. File extension is .mp4
 
    Args:
        video_path: Path to the video file to validate
        duration_seconds: Expected video duration for size range scaling
 
    Returns:
        True if the file passes all checks, False otherwise.
    """
    if not video_path.exists():
        logger.warning(f"Validation FAIL: file does not exist: {video_path}")
        return False
 
    size = video_path.stat().st_size
    scaled_min = MIN_VIDEO_SIZE_BYTES * (duration_seconds / 8)
    scaled_max = MAX_VIDEO_SIZE_BYTES * (duration_seconds / 8)
 
    if size < scaled_min:
        logger.warning(
            f"Validation FAIL: file too small ({size // 1024}KB < {scaled_min // 1024:.0f}KB min). "
            "Likely a glitched or empty output."
        )
        return False
 
    if size > scaled_max:
        logger.warning(
            f"Validation FAIL: file unexpectedly large ({size // (1024*1024)}MB > {scaled_max // (1024*1024):.0f}MB max)."
        )
        return False
 
    if video_path.suffix.lower() \!= ".mp4":
        logger.warning(f"Validation FAIL: unexpected extension: {video_path.suffix}")
        return False
 
    logger.info(f"Validation PASS: {video_path.name} ({size // (1024*1024)}MB)")
    return True

from quality_filter import validate_video_output
 
# At the end of merge_video_audio(), before returning output_path:
if output_path and not validate_video_output(output_path, duration_seconds=video_duration):
    logger.warning(f"Output failed quality validation: {output_path}")
    output_path.unlink(missing_ok=True)  # Remove the bad file
    return {"status": "FAILED", "reason": "quality validation failed"}