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

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:

  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:

  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:

  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.

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.

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)