# Hybrid Recap Video Production

> Produce high-fidelity, premium marketing recap videos blending real video clips and static images animated by Veo 3.1, featuring custom Lyria AI soundtracks and ultra-realistic TTS voiceovers.

- Skill: `izzyfresh/hybrid-recap-video-production` (Agent Skill)
- Install (CLI): `npx skillmds@latest add izzyfresh/hybrid-recap-video-production`
- Raw SKILL.md: https://api.skillmd.com/api/skills/izzyfresh/hybrid-recap-video-production/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Marketing & Growth
- Author: IzzyFresh (https://skillmd.com/u/izzyfresh)
- Updated: 2026-09-21
- Page: https://skillmd.com/skills/izzyfresh/hybrid-recap-video-production

---


# 🎬 Hybrid AI-Video Recap Production Skill

This skill documents the end-to-end engineering pipeline for producing premium, professional-grade marketing recap videos. It implements a **hybrid workflow** that seamlessly blends **real video footage** (to preserve 100% authenticity and save API credits) and **static images brought to life by Veo 3.1**, layered with **custom AI soundtracks (Vertex AI Lyria)** and **ultra-realistic voiceovers (Google Cloud TTS)**.

---

## 🛠️ Pipeline Architecture & Flow

```mermaid
graph TD
    A[Raw Assets: HEIC/JPG/MOV/MP4] --> B[Asset Normalization: HEIC to PNG & H.264 Re-encoding]
    B --> C[Veo 3.1 Video Generation: Image-to-Video & Text-to-Video]
    C --> D[Hybrid Video Stitching: Concat AI segments & Real video clips]
    D --> E[Audio Synthesis: Google Cloud TTS es-US-Neural2-A & Vertex AI Lyria-002]
    E --> F[Post-Production Mixing: Layer audio on video & Render final H.264 MP4]
```

---

## ⚠️ Critical Gotchas & Technical Solutions

When building multi-modal video pipelines combining GenAI models and traditional Python video libraries, you will encounter three major technical barriers. This skill provides the exact solutions to bypass them:

### 1. 🛡️ MoviePy iPhone Video Metadata Crash (`TypeError`)
*   **The Problem**: Modern iPhones (e.g., iPhone 15/16/17 Pro) embed proprietary stream-level **Side Data** inside the H.264 video stream packets (such as Apple's `Ambient Viewing Environment` float metrics: `ambient_illuminance=314.000000`). When MoviePy's ffmpeg metadata parser reads the file, it tries to concatenate these float values as strings, crashing instantly with:  
    `TypeError: unsupported operand type(s) for +: 'float' and 'str'`  
    Standard metadata stripping (`-map_metadata -1`) with stream copying (`-c:v copy`) **does not remove Side Data** because it is embedded inside the stream packets.
*   **The Solution**: You must **re-encode and normalize the video stream** on the fly. Re-encoding completely regenerates the H.264 video packets, stripping out all proprietary Apple Side Data. Crucially, you must also apply a **universal aspect-ratio scaling and padding filter** to normalize all mismatched clip dimensions (e.g., horizontal 16:9 vs vertical 9:16) to exactly **1920x1080**. This prevents pixel stride misalignment, tiling, and horizontal scanline corruption when MoviePy concatenates the clips.
*   **Ffmpeg Command**:
    ```bash
    ffmpeg -y -i input.mov -map_metadata -1 -vf "scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2:black" -c:v libx264 -c:a aac -pix_fmt yuv420p -crf 23 output.mp4
    ```

### 2. 🎵 Vertex AI Lyria Recitation Check Blocks (`400 Audio generation failed`)
*   **The Problem**: Google's Lyria music generation model (`lyria-002`) has strict safety filters to prevent generating AI music that too closely resembles copyrighted commercial songs. If your prompt describes a specific song arrangement (e.g., *"starts with a 10-second intro then drops into a beat"*) or uses highly commercial club-music genres (e.g., *"tech-house track"*, *"four-on-the-floor beat"*), the safety filters will block the request:  
    `generic::invalid_argument: Audio generation failed: All responses were blocked by recitation checks.`
*   **The Solution**: Use **highly abstract, textured, and atmospheric prompt engineering**. Avoid commercial genre tags, structural timelines, or BPM specifications. Focus on describing instruments, general mood, clean rhythms, and atmospheric textures.
*   **Safe Prompt Blueprint**:
    > *"A modern, upbeat electronic ambient soundtrack with a driving, steady synthesizer pulse. Features uplifting, positive acoustic piano chords, warm synth pads, organic light percussion, and a sophisticated, inspiring corporate tech atmosphere. The rhythm is clean and energetic."*

### 3. ⏳ Veo 3.1 Video Duration Constraints
*   **The Problem**: The Veo 3.1 API (`veo-3.1-generate-001`) has strict validation on video durations. If you request a duration not supported by the model feature (such as 3 or 5 seconds), it will fail with:  
    `Unsupported output video duration X seconds, supported durations are [8,4,6] for feature text_to_video.`
*   **The Solution**: You must strictly enforce and constrain all AI-generated scene durations to exactly **4, 6, or 8 seconds** in your playbook. Real video clips can have any arbitrary duration.

### 4. 🖼️ HEIC Image Formats
*   **The Problem**: Google Cloud Storage and the Veo 3.1 API do not natively support iPhone's proprietary HEIC image format.
*   **The Solution**: Auto-detect HEIC extensions and use ImageMagick (`convert`) to convert them to high-quality PNGs locally before uploading to GCS.
*   **Command**: `convert input.heic output.png`

### 5. 🎙️ Speech-to-Text (Chirp 3) vs. Text-to-Speech (Gemini TTS / Gacrux)
*   **The Distinction**: When building speech pipelines on Google Cloud, it is critical to distinguish between transcription (STT) and synthesis (TTS) models:
    1.  **Chirp 3 (Speech-to-Text / Transcription)**: Google Cloud's massive multilingual model designed to transcribe spoken audio files or streams into written text. Accessed via the `google.cloud.speech_v2` library with `model="chirp_3"`.
    2.  **Gemini TTS (Text-to-Speech / Synthesis)**: Google's next-generation generative voice model (like the warm, mature female voice **Gacrux**) designed to synthesize written text into incredibly realistic, expressive human-like speech. Accessed via the Gemini Multimodal API (`gemini-2.0-flash-001`) with `response_modalities=["AUDIO"]`.
*   **Chirp 3 (STT) Code Blueprint**:
    ```python
    from google.cloud import speech_v2 as speech
    from google.api_core.client_options import ClientOptions

    def transcribe_audio_chirp3(project_id, audio_uri):
        client = speech.SpeechClient(client_options=ClientOptions(api_endpoint="us-speech.googleapis.com"))
        config = speech.RecognitionConfig(
            auto_decoding_config=speech.AutoDetectDecodingConfig(),
            language_codes=["en-US"],
            model="chirp_3",
            features=speech.RecognitionFeatures(diarization_config=speech.SpeakerDiarizationConfig())
        )
        request = speech.BatchRecognizeRequest(
            recognizer=f"projects/{project_id}/locations/us/recognizers/_",
            config=config,
            files=[speech.BatchRecognizeFileMetadata(uri=audio_uri)],
            recognition_output_config=speech.RecognitionOutputConfig(inline_response_config=speech.InlineOutputConfig())
        )
        operation = client.batch_recognize(request=request)
        response = operation.result(timeout=120)
        return response.results[audio_uri].transcript
    ```
*   **Gemini Gacrux (TTS) Code Blueprint**:
    ```python
    from google import genai
    from google.genai import types

    def synthesize_speech_gacrux(text):
        client = genai.Client(vertexai=True)
        config = types.GenerateContentConfig(
            response_modalities=["AUDIO"],
            speech_config=types.SpeechConfig(
                voice_config=types.VoiceConfig(
                    prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Gacrux")
                )
            )
        )
        response = client.models.generate_content(
            model="gemini-2.0-flash-001",
            contents=text,
            config=config
        )
        for part in response.candidates[0].content.parts:
            if part.inline_data:
                return part.inline_data.data # Returns raw audio bytes
    ```

---

## 📋 Storyboard Playbook Blueprint

Define your video timeline in a structured list of dictionaries (a "Playbook"). This allows your script to dynamically decide whether to copy a real video, convert a HEIC, or call Veo 3.1.

```python
ROADSHOW_PLAYBOOK = [
    {
        "name": "1. Guatemala City Aerial",
        "type": "text-to-video",
        "prompt": "Cinematic, slow drone shot flying over Guatemala City at sunrise...",
        "duration": 4 # Must be 4, 6, or 8
    },
    {
        "name": "2. Arrival Hallway",
        "type": "real-video",
        "local_path": "/path/to/raw_clip.mov",
        "duration": 10 # Real clips can be any duration
    },
    {
        "name": "3. Costa Rica Classroom",
        "type": "image-to-video",
        "local_image": "/path/to/selfie.heic", # Will be auto-converted to PNG
        "prompt": "Animate the people smiling and waving...",
        "duration": 6 # Must be 4, 6, or 8
    }
]
```

---

## 🐍 Complete Python Pipeline Blueprint

Below is the complete, production-ready Python script utilizing the new `google-genai` SDK, `moviepy>=2.0.0`, and Google Cloud APIs to execute the entire hybrid recap production pipeline.

```python
import os
import sys
import time
import uuid
import json
import logging
import subprocess
import tempfile
from dotenv import load_dotenv
from google import genai
from google.genai import types
from google.cloud import storage
import google.cloud.texttospeech as texttospeech
from google.cloud import aiplatform
from moviepy import VideoFileClip, AudioFileClip, CompositeAudioClip, concatenate_videoclips, concatenate_audioclips

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()
PROJECT_ID = os.getenv("PROJECT_ID")
VIDEO_BUCKET = os.getenv("VIDEO_BUCKET")

# Initialize clients
storage_client = storage.Client(project=PROJECT_ID)
genai_client = genai.Client(
    vertexai=True,
    project=PROJECT_ID,
    location="us-central1",
)

STATE_FILE = "video_play_state.json"

def strip_video_metadata(input_path, output_path):
    """Strips all metadata and normalizes the video to a perfect 1920x1080 H.264 MP4 using scale and pad filters."""
    logger.info(f"🧹 Normalizing and re-encoding video: {os.path.basename(input_path)} -> {os.path.basename(output_path)}...")
    subprocess.run([
        "ffmpeg", "-y", "-i", input_path, 
        "-map_metadata", "-1", 
        "-vf", "scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2:black",
        "-c:v", "libx264", "-c:a", "aac", 
        "-pix_fmt", "yuv420p", "-crf", "23",
        output_path
    ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

def convert_heic_to_png(heic_path):
    """Converts local HEIC to PNG using ImageMagick."""
    png_path = heic_path.rsplit('.', 1)[0] + ".png"
    logger.info(f"Converting HEIC to PNG: {heic_path} -> {png_path}...")
    subprocess.run(["convert", heic_path, png_path], check=True)
    return png_path

def upload_file_to_gcs(local_path, destination_blob_name, mime_type, bucket_name):
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    blob.upload_from_filename(local_path, content_type=mime_type)
    return f"gs://{bucket_name}/{destination_blob_name}"

def download_from_gcs(gcs_uri):
    path = gcs_uri[5:]
    bucket_name, blob_name = path.split("/", 1)
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    return blob.download_as_bytes()

def generate_lyria_music(prompt, project_id):
    """Generates background music using Vertex AI Lyria-002."""
    aiplatform.init(project=project_id, location="us-central1")
    LYRIA_ENDPOINT = f"projects/{project_id}/locations/global/publishers/google/models/lyria-002"
    client_options = {"api_endpoint": "us-central1-aiplatform.googleapis.com"}
    client = aiplatform.gapic.PredictionServiceClient(client_options=client_options)
    
    response = client.predict(
        endpoint=LYRIA_ENDPOINT,
        instances=[{"prompt": prompt}],
        parameters={"sampleCount": 1},
    )
    
    import base64
    encoded_bytes = response.predictions[0]["bytesBase64Encoded"]
    return base64.b64decode(encoded_bytes)

def generate_tts_voiceover(text):
    """Synthesizes high-fidelity speech using es-US-Neural2-A Journey voice."""
    client = texttospeech.TextToSpeechClient()
    voice = texttospeech.VoiceSelectionParams(language_code="es-US", name="es-US-Neural2-A")
    audio_config = texttospeech.AudioConfig(audio_encoding=texttospeech.AudioEncoding.LINEAR16)
    response = client.synthesize_speech(input=texttospeech.SynthesisInput(text=text), voice=voice, audio_config=audio_config)
    return response.audio_content

def run_video_generation(playbook, bucket_name):
    state = load_state()
    segment_uris = state.get("segment_uris", [])
    start_index = len(segment_uris)
    
    for i in range(start_index, len(playbook)):
        scene = playbook[i]
        
        # Scenario A: Real Video
        if scene["type"] == "real-video":
            filename = os.path.basename(scene["local_path"])
            gcs_uri = upload_file_to_gcs(scene["local_path"], f"raw_clips/{filename}", "video/mp4", bucket_name)
            segment_uris.append(gcs_uri)
            save_state(segment_uris)
            continue
            
        # Scenario B: AI Generated
        image_input = None
        if scene["type"] == "image-to-video":
            local_img = scene["local_image"]
            if local_img.lower().endswith((".heic", ".heif")):
                local_img = convert_heic_to_png(local_img)
            filename = os.path.basename(local_img)
            gcs_img_uri = upload_file_to_gcs(local_img, f"inputs/{filename}", "image/png", bucket_name)
            image_input = types.Image(gcs_uri=gcs_img_uri, mime_type="image/png")
            
        # Call Veo 3.1
        gen_config = types.GenerateVideosConfig(
            aspect_ratio="16:9",
            duration_seconds=scene["duration"], # Constrained to 4, 6, or 8
            enhance_prompt=True,
            output_gcs_uri=f"gs://{VIDEO_BUCKET}",
            resolution="1080p",
            person_generation="allow_all"
        )
        operation = genai_client.models.generate_videos(
            model="veo-3.1-generate-001",
            prompt=scene["prompt"],
            config=gen_config,
            image=image_input
        )
        while not operation.done:
            time.sleep(10)
            operation = genai_client.operations.get(operation)
            
        segment_uri = operation.result.generated_videos[0].video.uri
        segment_uris.append(segment_uri)
        save_state(segment_uris)
        
    return segment_uris

def run_post_production(segment_uris, voiceover_script, music_prompt):
    bucket_name = VIDEO_BUCKET.split("/", 1)[0]
    with tempfile.TemporaryDirectory() as tmpdir:
        clips = []
        for idx, uri in enumerate(segment_uris):
            video_bytes = download_from_gcs(uri)
            ext = ".mp4" if "mp4" in uri.lower() else ".mov"
            local_raw_path = os.path.join(tmpdir, f"segment_{idx}_raw{ext}")
            with open(local_raw_path, "wb") as f:
                f.write(video_bytes)
                
            clean_mp4_path = os.path.join(tmpdir, f"segment_{idx}.mp4")
            strip_video_metadata(local_raw_path, clean_mp4_path)
            clips.append(VideoFileClip(clean_mp4_path))
            
        final_video_clip = concatenate_videoclips(clips)
        video_duration = final_video_clip.duration
        
        # Audio Production
        music_path = os.path.join(tmpdir, "music.wav")
        with open(music_path, "wb") as f:
            f.write(generate_lyria_music(music_prompt, PROJECT_ID))
            
        voice_path = os.path.join(tmpdir, "voice.wav")
        with open(voice_path, "wb") as f:
            f.write(generate_tts_voiceover(voiceover_script))
            
        voice_clip = AudioFileClip(voice_path)
        music_clip = AudioFileClip(music_path).with_volume_scaled(0.14)
        
        if music_clip.duration < video_duration:
            repeats = int(video_duration / music_clip.duration) + 1
            music_clip = concatenate_audioclips([music_clip] * repeats)
        music_clip = music_clip.subclipped(0, video_duration)
        
        final_video_clip.audio = CompositeAudioClip([voice_clip, music_clip])
        
        output_filename = f"masterpiece_{uuid.uuid4().hex[:8]}.mp4"
        local_output_path = os.path.join(tmpdir, output_filename)
        
        final_video_clip.write_videofile(local_output_path, codec="libx264", audio_codec="aac")
        return upload_file_to_gcs(local_output_path, f"showcases/{output_filename}", "video/mp4", bucket_name)
```