Veo 3 × Lyria 3 Pro API で構築する自動コンテンツ生成パイプライン — 動画×音楽の量産システムを個人で作る

# 必要なパッケージをインストール
pip install google-genai python-dotenv ffmpeg-python
 
# ffmpeg本体もインストールが必要（macOSの場合）
brew install ffmpeg
 
# Linuxの場合
# apt-get install -y ffmpeg

# config.py — 設定と認証の初期化
import os
from dotenv import load_dotenv
from google import genai
 
load_dotenv()
 
def get_client() -> genai.Client:
    """
    Google Gen AI クライアントを初期化して返す。
    API キーは環境変数 GOOGLE_AI_API_KEY から読み込む。
    """
    api_key = os.getenv("GOOGLE_AI_API_KEY")
    if not api_key:
        raise ValueError(
            "GOOGLE_AI_API_KEY が設定されていません。"
            ".env ファイルに GOOGLE_AI_API_KEY=your_key_here を追加してください。"
        )
    return genai.Client(api_key=api_key)
 
# モデル名定数
VEO3_MODEL = "veo-3.0-generate-preview"
LYRIA3_MODEL = "lyria-3.0-pro"
 
# 出力ディレクトリ
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 動画生成モジュール
import time
import logging
import httpx
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:
    """
    Veo 3 API を使ってテキストから動画を生成し、ファイルとして保存する。
 
    Args:
        prompt: 動画の内容を説明するテキスト（英語が精度良好）
        output_filename: 保存するファイル名（拡張子なし）
        aspect_ratio: "16:9" または "9:16"
        duration_seconds: 動画の長さ（最大8秒）
        max_retries: エラー時の最大リトライ回数
        poll_interval: ポーリング間隔（秒）
        timeout_seconds: 全体のタイムアウト（秒）
 
    Returns:
        生成された動画ファイルのパス。失敗した場合は None。
    """
    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 動画生成開始（試行 {attempt}/{max_retries}）: {prompt[:50]}...")
 
            # 動画生成リクエストを送信
            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,  # プロンプトの自動補完を有効化
                ),
            )
 
            # 完了を待つ（ポーリング）
            start_time = time.time()
            while not operation.done:
                elapsed = time.time() - start_time
                if elapsed > timeout_seconds:
                    raise TimeoutError(f"動画生成がタイムアウトしました（{timeout_seconds}秒超過）")
 
                logger.info(f"  生成中... {elapsed:.0f}秒経過")
                time.sleep(poll_interval)
                operation = client.operations.get(operation)
 
            # エラーチェック
            if operation.error:
                raise RuntimeError(f"動画生成エラー: {operation.error.message}")
 
            # 動画をダウンロードして保存
            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)
 
            logger.info(f"✅ 動画保存完了: {output_path} ({output_path.stat().st_size // 1024}KB)")
            return output_path
 
        except TimeoutError as e:
            logger.warning(f"⚠️ タイムアウト（試行 {attempt}）: {e}")
            if attempt == max_retries:
                logger.error("最大リトライ回数に達しました")
                return None
 
        except Exception as e:
            logger.error(f"❌ エラー（試行 {attempt}）: {type(e).__name__}: {e}")
            if attempt < max_retries:
                wait_time = 2 ** attempt * 5  # 指数バックオフ: 10秒 → 20秒 → 40秒
                logger.info(f"  {wait_time}秒後にリトライします...")
                time.sleep(wait_time)
            else:
                logger.error("最大リトライ回数に達しました。この動画の生成をスキップします。")
                return None
 
    return None

# music_generator.py — Lyria 3 Pro 音楽生成モジュール
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:
    """
    Lyria 3 Pro API を使ってテキストから音楽を生成し、ファイルとして保存する。
 
    Args:
        prompt: 音楽のスタイルを説明するテキスト
        output_filename: 保存するファイル名（拡張子なし）
        duration_seconds: 音楽の長さ（最大210秒）
        output_format: "mp3" または "wav"
        max_retries: エラー時の最大リトライ回数
 
    Returns:
        生成された音楽ファイルのパス。失敗した場合は None。
    """
    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 音楽生成開始（試行 {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 response.audio_data:
                with open(output_path, "wb") as f:
                    f.write(response.audio_data)
                logger.info(f"✅ 音楽保存完了: {output_path}")
                return output_path
            else:
                raise ValueError("音楽データが空です")
 
        except Exception as e:
            logger.error(f"❌ エラー（試行 {attempt}）: {type(e).__name__}: {e}")
            if attempt < max_retries:
                wait_time = 2 ** attempt * 3
                logger.info(f"  {wait_time}秒後にリトライします...")
                time.sleep(wait_time)
            else:
                logger.error("最大リトライ回数に達しました")
                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:
    """
    音楽生成プロンプトを構造化して組み立てるヘルパー関数。
    Lyria 3 Pro は細かい指定ほど品質が上がる。
    """
    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 — 統合パイプライン（非同期並列処理版）
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:
    """
    ffmpeg を使って動画と音楽を合成する。
 
    - 音楽が動画より長い場合は動画の長さに合わせてカット
    - フェードアウト処理を末尾に適用
    - 音量を video_volume パラメータで調整
 
    Args:
        audio_volume: 音楽の音量（0.0〜1.0、0.5が推奨）
    """
    output_path = Path(OUTPUT_DIR) / f"{output_filename}.mp4"
 
    try:
        # 動画の長さを取得
        probe = ffmpeg.probe(str(video_path))
        video_duration = float(probe["format"]["duration"])
 
        # 入力ストリームを定義
        video_input = ffmpeg.input(str(video_path))
        audio_input = ffmpeg.input(str(audio_path))
 
        # 音楽を動画の長さに合わせてカット + フェードアウト適用
        fade_start = max(0, video_duration - fade_out_seconds)
        audio_processed = (
            audio_input
            .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)  # フェードアウト
        )
 
        # 動画と処理済み音楽を合成
        output = ffmpeg.output(
            video_input.video,
            audio_processed,
            str(output_path),
            vcodec="copy",          # 動画は再エンコードなしでコピー（高速）
            acodec="aac",           # 音声はAAC形式に変換
            audio_bitrate="192k",
        )
 
        ffmpeg.run(output, overwrite_output=True, quiet=True)
        logger.info(f"✅ 合成完了: {output_path}")
        return output_path
 
    except ffmpeg.Error as e:
        logger.error(f"❌ ffmpeg エラー: {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 = 30,
) -> dict:
    """
    動画生成と音楽生成を並列実行して合成する統合パイプライン。
 
    Returns:
        実行結果を含む辞書（status, output_path, elapsed_seconds）
    """
    start_time = asyncio.get_event_loop().time()
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    base_name = f"{output_name}_{timestamp}"
 
    logger.info(f"🎬 パイプライン開始: {output_name}")
 
    # 動画生成と音楽生成を並列実行
    loop = asyncio.get_event_loop()
 
    video_task = loop.run_in_executor(
        None,
        generate_video,
        video_prompt, f"{base_name}_video", "16:9", video_duration
    )
    music_task = loop.run_in_executor(
        None,
        generate_music,
        music_prompt, f"{base_name}_music", music_duration
    )
 
    # 両方の完了を待つ
    video_path, music_path = await asyncio.gather(video_task, music_task)
 
    if not video_path:
        return {"status": "FAILED", "reason": "動画生成失敗"}
    if not music_path:
        return {"status": "FAILED", "reason": "音楽生成失敗"}
 
    # 合成処理
    output_path = merge_video_audio(video_path, music_path, base_name)
 
    elapsed = asyncio.get_event_loop().time() - start_time
 
    if output_path:
        # 一時ファイルを削除
        video_path.unlink(missing_ok=True)
        music_path.unlink(missing_ok=True)
        return {
            "status": "SUCCESS",
            "output_path": str(output_path),
            "elapsed_seconds": round(elapsed, 1),
        }
    else:
        return {"status": "FAILED", "reason": "合成処理失敗"}
 
# 実行例
if __name__ == "__main__":
    import logging
    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=15,
    ))
 
    print(result)
    # 期待する出力:
    # {'status': 'SUCCESS', 'output_path': 'output/japanese_garden_20260418_104500.mp4', 'elapsed_seconds': 87.3}