Building a Narration Channel with Gemini TTS and Veo 3.1 — A

import anthropic
import base64
from pathlib import Path
from typing import Optional
 
def generate_flash_narration(
    text: str,
    voice_style: str = "neutral",
    speed_factor: float = 1.0,
    output_path: str = "output.mp3"
) -> dict:
    """
    Generate narration using Gemini Flash TTS.
 
    Flash TTS excels at:
    - Fast turnaround (ideal for automated pipelines)
    - Clear, crisp delivery
    - Minimal latency for real-time applications
 
    Args:
        text: Japanese narration script
        voice_style: "neutral", "warm", "authoritative"
        speed_factor: 0.8 (slow) to 1.3 (fast)
        output_path: Where to save MP3
 
    Returns:
        dict with file path, duration, and metadata
    """
    client = anthropic.Anthropic(api_key="YOUR_GEMINI_API_KEY")
 
    # Style-specific instructions
    style_prompts = {
        "neutral": "Professional, news-anchor style delivery.",
        "warm": "Friendly, conversational tone. Like talking to a friend.",
        "authoritative": "Expert, commanding presence. Like a documentary narrator."
    }
 
    prompt = f"""Generate Japanese narration in {voice_style} style.
 
Specifications:
- Pronunciation: Standard Tokyo dialect
- Pacing: {speed_factor}x normal speed
- Output: High-quality MP3, 128kbps minimum
 
Style: {style_prompts.get(voice_style, style_prompts['neutral'])}
 
Text to narrate:
{text}
"""
 
    response = client.messages.create(
        model="gemini-2.0-flash",
        max_tokens=2048,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )
 
    # Decode base64 audio
    audio_bytes = base64.b64decode(response.content[0].text)
    output_file = Path(output_path)
    output_file.parent.mkdir(parents=True, exist_ok=True)
    output_file.write_bytes(audio_bytes)
 
    # Rough duration estimate (Japanese: ~80 chars/min at 1.0x speed)
    estimated_duration_seconds = (len(text) / 80) * 60 / speed_factor
 
    return {
        "file": str(output_file.absolute()),
        "size_mb": len(audio_bytes) / 1024 / 1024,
        "estimated_duration": estimated_duration_seconds,
        "model": "gemini-2.0-flash",
        "style": voice_style
    }

def generate_pro_narration(
    text: str,
    emphasis_markers: Optional[list] = None,
    output_path: str = "output_pro.mp3"
) -> dict:
    """
    Generate high-fidelity narration using Gemini Pro TTS.
 
    Pro TTS is worth the extra cost for:
    - Emotional depth and nuance
    - Proper handling of technical terms
    - Cinematic-quality narration
    - Longer-form content (5+ minutes)
 
    Args:
        text: Full narration script
        emphasis_markers: List of phrases to emphasize
        output_path: Output file location
 
    Returns:
        Metadata dict
    """
    client = anthropic.Anthropic(api_key="YOUR_GEMINI_API_KEY")
 
    emphasis_instructions = ""
    if emphasis_markers:
        emphasis_instructions = "\n\nEmphasize these phrases with slight vocal lift:\n"
        emphasis_instructions += "\n".join(f"- {phrase}" for phrase in emphasis_markers)
 
    response = client.messages.create(
        model="gemini-2.0-pro",
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": f"""Generate studio-quality Japanese narration.
 
Target: YouTube educational video (7–10 minutes)
Quality: Near-professional voice acting level
Tone: Warm, credible, engaging
 
{emphasis_instructions}
 
Script:
{text}
"""
            }
        ]
    )
 
    audio_bytes = base64.b64decode(response.content[0].text)
    output_file = Path(output_path)
    output_file.write_bytes(audio_bytes)
 
    return {
        "file": str(output_file.absolute()),
        "size_mb": len(audio_bytes) / 1024 / 1024,
        "model": "gemini-2.0-pro",
        "quality_tier": "studio"
    }
 
# Example usage
if __name__ == "__main__":
    sample_script = """
    今日は、AI動画生成の最前線についてお話しします。
    過去5年間で、動画制作の民主化が急速に進みました。
    かつては大企業の専有技術だった映像制作が、
    今では個人でも数クリックで実現できる時代です。
    """
 
    # Quick version (30 seconds)
    flash_result = generate_flash_narration(
        sample_script,
        voice_style="warm"
    )
    print(f"Flash TTS: {flash_result}")
 
    # High-quality version
    pro_result = generate_pro_narration(
        sample_script,
        emphasis_markers=["AI動画生成", "動画制作の民主化"]
    )
    print(f"Pro TTS: {pro_result}")

import json
import anthropic
from typing import List, Dict
 
def extract_visual_scenes(
    narration_script: str,
    num_scenes: int = 4,
    duration_per_scene: int = 8
) -> List[Dict]:
    """
    Analyze narration script and generate Veo 3.1 prompts for each scene.
 
    This function:
    1. Identifies narrative beats in the script
    2. Creates cinematic visual descriptions for each beat
    3. Returns Veo 3.1-ready prompts with shot composition details
 
    Args:
        narration_script: Full text of narration
        num_scenes: Number of background videos to generate
        duration_per_scene: Seconds per video (Veo max is 8)
 
    Returns:
        List of scene dicts with prompts and metadata
    """
    client = anthropic.Anthropic(api_key="YOUR_GEMINI_API_KEY")
 
    extraction_prompt = f"""Analyze this YouTube narration script and create {num_scenes} visual scenes.
 
For EACH scene, write a cinematic Veo 3.1 prompt that:
- Specifies shot type (wide, close-up, tracking, etc.)
- Includes lighting/mood (cool blue, warm amber, high-key, etc.)
- Describes motion (slow pan, subtle parallax, static, etc.)
- Suggests color palette
- Is 1–2 sentences max (Veo prompts work best when concise)
 
Output strict JSON:
{{
    "scenes": [
        {{
            "scene_number": 1,
            "timestamp": "0:00-0:08",
            "script_excerpt": "First 10 seconds of narration...",
            "veo_prompt": "Wide establishing shot of a minimalist tech office with soft blue lighting...",
            "duration": 8,
            "motion_intensity": "low"
        }},
        ...
    ]
}}
 
Narration script:
{narration_script}
"""
 
    response = client.messages.create(
        model="gemini-2.0-flash",
        max_tokens=2048,
        messages=[
            {"role": "user", "content": extraction_prompt}
        ]
    )
 
    return json.loads(response.content[0].text)["scenes"]
 
def generate_veo_backgrounds(
    scenes: List[Dict],
    resolution: str = "720p"
) -> List[Dict]:
    """
    Call Veo 3.1 API to generate background videos for each scene.
 
    Args:
        scenes: Output from extract_visual_scenes()
        resolution: "720p", "1080p", or "4k"
 
    Returns:
        List of generated video URLs + metadata
    """
    import requests
 
    veo_base_url = "https://generativelanguage.googleapis.com/v1/veo:generateVideo"
 
    generated_videos = []
 
    for scene in scenes:
        print(f"Generating Veo video: Scene {scene['scene_number']}...")
 
        payload = {
            "prompt": scene["veo_prompt"],
            "duration_seconds": scene["duration"],
            "resolution": resolution,
            "output_format": "mp4"
        }
 
        try:
            response = requests.post(
                veo_base_url,
                json=payload,
                headers={
                    "Authorization": f"Bearer {YOUR_GEMINI_API_KEY}",
                    "Content-Type": "application/json"
                },
                timeout=180
            )
            response.raise_for_status()
 
            result = response.json()
            video_url = result.get("videoUrl")
 
            generated_videos.append({
                "scene_number": scene["scene_number"],
                "video_url": video_url,
                "resolution": resolution,
                "status": "success"
            })
            print(f"  ✓ Scene {scene['scene_number']}: {video_url}")
 
        except requests.exceptions.RequestException as e:
            print(f"  ✗ Scene {scene['scene_number']}: {str(e)}")
            generated_videos.append({
                "scene_number": scene["scene_number"],
                "status": "failed",
                "error": str(e)
            })
 
    return generated_videos
 
# Example
if __name__ == "__main__":
    script = """
    AI video generation represents a fundamental shift in creative production.
    What once required a crew of cinematographers can now be done by one person.
    This democratization is reshaping how we tell stories.
    """
 
    scenes = extract_visual_scenes(script, num_scenes=3)
    for scene in scenes:
        print(f"Scene {scene['scene_number']}: {scene['veo_prompt']}")
 
    # Uncomment to actually generate (requires valid API key)
    # videos = generate_veo_backgrounds(scenes)

import os
import json
from datetime import datetime
from pathlib import Path
 
class VideoProductionFactory:
    """
    Orchestrates the entire blog-to-YouTube conversion pipeline.
 
    Workflow:
    1. Script optimization (Gemini Flash)
    2. Narration generation (Gemini TTS)
    3. Scene extraction (Gemini Flash)
    4. Background video generation (Veo 3.1)
    5. Metadata generation (Gemini Flash)
    6. Final package ready for Vrew
    """
 
    def __init__(self, api_key: str, output_dir: str = "production_output"):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
 
    def process_article(
        self,
        article_text: str,
        article_title: str,
        use_pro_tts: bool = False
    ) -> dict:
        """
        Process a blog article into a complete video production.
 
        Args:
            article_text: Full blog article
            article_title: Article title
            use_pro_tts: Use Pro TTS for higher quality (costs more)
 
        Returns:
            Complete production package ready for Vrew
        """
        print(f"\n🎬 Starting production: {article_title}")
        job_id = datetime.now().strftime("%Y%m%d_%H%M%S")
 
        # Step 1: Optimize script
        print("  [1/5] Optimizing script...")
        script = self._create_script(article_text, article_title)
 
        # Step 2: Generate narration
        print("  [2/5] Generating narration...")
        audio_file = self._generate_narration(
            script,
            article_title,
            use_pro=use_pro_tts
        )
 
        # Step 3: Extract scenes
        print("  [3/5] Planning visual scenes...")
        scenes = extract_visual_scenes(script, num_scenes=4)
 
        # Step 4: Generate backgrounds
        print("  [4/5] Generating background videos...")
        backgrounds = generate_veo_backgrounds(scenes, resolution="1080p")
 
        # Step 5: Generate metadata
        print("  [5/5] Creating YouTube metadata...")
        metadata = self._generate_metadata(article_title, script)
 
        # Package results
        production_package = {
            "job_id": job_id,
            "article_title": article_title,
            "script": script,
            "assets": {
                "narration_audio": audio_file,
                "background_videos": [bg["video_url"] for bg in backgrounds],
                "scene_metadata": scenes
            },
            "youtube": metadata,
            "status": "ready_for_vrew",
            "created_at": datetime.now().isoformat()
        }
 
        # Save production log
        log_file = self.output_dir / f"{job_id}_production.json"
        with open(log_file, "w", encoding="utf-8") as f:
            json.dump(production_package, f, ensure_ascii=False, indent=2)
 
        print(f"✓ Production complete: {log_file}")
        return production_package
 
    def _create_script(self, article: str, title: str) -> str:
        """Convert blog article to narration script."""
        response = self.client.messages.create(
            model="gemini-2.0-flash",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""Convert this blog article into a YouTube narration script.
 
Requirements:
- 400–600 words
- Conversational, spoken-word tone
- Short sentences (under 12 words each)
- Include [PAUSE X] markers for natural breathing room
- Define technical terms on first mention
 
Article title: {title}
 
Article text:
{article}
"""
            }]
        )
        return response.content[0].text
 
    def _generate_narration(
        self,
        script: str,
        title: str,
        use_pro: bool = False
    ) -> str:
        """Generate narration audio."""
        output_file = self.output_dir / f"{title.replace(' ', '_')}.mp3"
 
        if use_pro:
            result = generate_pro_narration(script, output_path=str(output_file))
        else:
            result = generate_flash_narration(script, output_path=str(output_file))
 
        return result["file"]
 
    def _generate_metadata(self, title: str, script: str) -> dict:
        """Generate YouTube title, description, tags."""
        response = self.client.messages.create(
            model="gemini-2.0-flash",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"""Generate YouTube metadata for this narration video.
 
Article title: {title}
Script excerpt: {script[:300]}...
 
Output JSON with:
- youtube_title (under 60 chars, SEO-optimized)
- youtube_description (under 160 chars)
- tags (array of 5–8 relevant tags)
- thumbnail_description (visual direction for thumbnail)
 
JSON format:
{{
    "youtube_title": "...",
    "youtube_description": "...",
    "tags": [...],
    "thumbnail_description": "..."
}}
"""
            }]
        )
 
        return json.loads(response.content[0].text)
 
# Usage
if __name__ == "__main__":
    factory = VideoProductionFactory(api_key="YOUR_KEY")
 
    blog_content = """
    Video production has entered a new era.
    AI tools are enabling individual creators to compete with studios.
    This isn't science fiction—it's happening right now.
    [... full article ...]
    """
 
    package = factory.process_article(
        article_text=blog_content,
        article_title="The AI Video Revolution",
        use_pro_tts=True  # Upgrade to Pro TTS for premium content
    )
 
    print("\n📦 Production Package Ready:")
    print(f"  Narration: {package['assets']['narration_audio']}")
    print(f"  Background Videos: {len(package['assets']['background_videos'])} clips")
    print(f"  YouTube Title: {package['youtube']['youtube_title']}")

max_retries = 3
for attempt in range(max_retries):
    try:
        video = generate_veo_backgrounds(scenes)
        break
    except Exception as e:
        wait_time = 2 ** attempt  # 2s, 4s, 8s
        time.sleep(wait_time)

import queue
import time
 
request_queue = queue.Queue()
for scene in scenes:
    request_queue.put(scene)
    time.sleep(0.6)  # Spread requests over time