💬

Google Veo 3.1 APIを使ったEC向けUGC動画の一括制作

に公開
AI
API
Google
tech

はじめに

EC事業において、ユーザー生成コンテンツ（UGC）動画はコンバージョン率向上の強力なツールです。本記事では、PythonとGoogleのVeo AI動画生成APIを使用して、商品の静止画から開封動画、製品デモ、ライフスタイル動画を一括生成する方法を解説します。

期待できる成果：

  • 入力：商品画像＋テキストプロンプト
  • 出力：高品質な商品紹介動画
  • 一括処理：数十点の商品をまとめて処理し、対応する動画を自動生成

準備作業

必要なもの

  • Python 3.7以上
  • requestsライブラリ
  • 有効なVeo APIアクセス権

APIキーの取得方法

  1. DEFAPIにアクセスしてアカウント登録
  2. コンソールでAPIキーを取得
  3. 重要セキュリティTips：キーは直接コードに書かず、環境変数を活用しましょう

依存関係のインストール

pip install requests python-dotenv

APIインターフェース詳細

動画生成エンドポイント

  • エンドポイント: POST https://api.defapi.org/api/google/veo/generate
  • 認証: Authorization: Bearer YOUR_API_KEY
  • 主要パラメータ:
    • model: 適切なVeoモデルの選択
    • prompt: 動画の説明テキスト
    • images: 商品画像URLの配列
    • aspect_ratio: 動画比率（16:9または9:16）
    • callback_url: コールバック通知URL

タスク問い合わせエンドポイント

  • エンドポイント: GET https://api.defapi.org/api/task/query
  • パラメータ: task_id - タスクID

実装コード

ステップ1：ライブラリのインポートと設定

import os
import requests
import time
import json
from typing import List, Dict, Optional
from dotenv import load_dotenv

# 環境変数の読み込み
load_dotenv()

class VeoVideoGenerator:
    def __init__(self):
        # 環境変数からAPIキーを読み込み（ハードコーディング回避）
        self.api_key = os.getenv('VEO_API_KEY')
        if not self.api_key:
            raise ValueError("VEO_API_KEY環境変数を設定してください")
        
        self.base_url = "https://api.defapi.org"
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

ステップ2：動画生成関数

def generate_video(self, 
                  product_images: List[str], 
                  prompt: str,
                  model: str = "google/veo3.1-components",
                  aspect_ratio: str = "9:16",
                  callback_url: Optional[str] = None) -> str:
    """
    商品動画を生成
    
    Args:
        product_images: 商品画像URLのリスト
        prompt: 動画説明プロンプト
        model: Veoモデル選択
        aspect_ratio: 動画アスペクト比
        callback_url: コールバックURL（オプション）
    
    Returns:
        task_id: 進捗確認用タスクID
    """
    
    # リクエストペイロードの構築
    payload = {
        "model": model,
        "prompt": prompt,
        "images": product_images,
        "aspect_ratio": aspect_ratio
    }
    
    # コールバックURLが提供された場合は追加
    if callback_url:
        payload["callback_url"] = callback_url
    
    try:
        # 動画生成リクエストの送信
        response = requests.post(
            f"{self.base_url}/api/google/veo/generate",
            headers=self.headers,
            json=payload,
            timeout=30
        )
        
        # レスポンスステータスの確認
        if response.status_code == 200:
            result = response.json()
            if result.get('code') == 0:
                task_id = result['data']['task_id']
                print(f"✅ 動画生成タスクを作成しました: {task_id}")
                return task_id
            else:
                raise Exception(f"APIエラー: {result.get('message')}")
        else:
            # エラーレスポンスの処理
            error_msg = response.json().get('message', '不明なエラー')
            raise Exception(f"リクエスト失敗 (HTTP {response.status_code}): {error_msg}")
            
    except requests.exceptions.RequestException as e:
        raise Exception(f"ネットワークリクエスト失敗: {str(e)}")

ステップ3：タスクステータス照会関数

def query_task_status(self, task_id: str) -> Dict:
    """
    タスクステータスを照会
    
    Args:
        task_id: タスクID
    
    Returns:
        タスクステータス情報
    """
    try:
        response = requests.get(
            f"{self.base_url}/api/task/query",
            headers=self.headers,
            params={"task_id": task_id},
            timeout=10
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"ステータス照会失敗: HTTP {response.status_code}")
            
    except requests.exceptions.RequestException as e:
        raise Exception(f"ステータス照会リクエスト失敗: {str(e)}")

def wait_for_task_completion(self, task_id: str, max_wait: int = 600) -> str:
    """
    タスク完了を待機し、動画URLを返す
    
    Args:
        task_id: タスクID
        max_wait: 最大待機時間（秒）
    
    Returns:
        動画URL
    """
    start_time = time.time()
    
    while time.time() - start_time < max_wait:
        # タスクステータスを照会
        status_result = self.query_task_status(task_id)
        
        if status_result.get('code') == 0:
            task_data = status_result['data']
            status = task_data['status']
            
            print(f"タスクステータス: {status}")
            
            if status == "success":
                # タスク成功、動画URLを返す
                video_url = task_data['result']['video_url']
                print(f"🎉 動画生成成功！ URL: {video_url}")
                return video_url
                
            elif status == "failed":
                # タスク失敗
                error_msg = task_data['status_reason'].get('message', '不明なエラー')
                raise Exception(f"動画生成失敗: {error_msg}")
                
            elif status in ["pending", "submitted", "in_progress"]:
                # タスク進行中、待機して再試行
                time.sleep(10)
            else:
                raise Exception(f"不明なタスクステータス: {status}")
        else:
            raise Exception(f"ステータス照会APIエラー: {status_result.get('message')}")
    
    raise Exception("タスク待機タイムアウト")

ステップ4：EC向けUGC動画の一括生成

def batch_generate_ugc_videos(self, products_data: List[Dict]) -> Dict[str, str]:
    """
    EC向けUGC動画を一括生成
    
    Args:
        products_data: 商品データのリスト、各要素に以下を含む:
            - product_id: 商品ID
            - images: 商品画像URLリスト
            - product_name: 商品名
            - video_type: 動画タイプ ('unboxing', 'demo', 'lifestyle')
    
    Returns:
        辞書: {product_id: video_url}
    """
    
    # 動画タイプ別プロンプトテンプレート
    prompt_templates = {
        'unboxing': "高品質な{product}の開封動画。パッケージ、開封プロセス、商品のお披露目をプロフェッショナルに表現",
        'demo': "現実的な設定での{product}の製品デモ動画。特徴、使用方法、利点をプロフェッショナルに紹介",
        'lifestyle': "自然な日常使用シーンでの{product}をフィーチャーしたライフスタイル動画。本格的で魅力的なコンテンツ"
    }
    
    results = {}
    
    for product in products_data:
        product_id = product['product_id']
        product_name = product['product_name']
        video_type = product['video_type']
        
        print(f"\n🚀 処理開始: {product_name} ({video_type})")
        
        try:
            # 動画プロンプトを生成
            prompt = prompt_templates[video_type].format(product=product_name)
            
            # 動画生成タスクを送信
            task_id = self.generate_video(
                product_images=product['images'],
                prompt=prompt,
                model="google/veo3.1-components",  # 複数画像対応モデルを使用
                aspect_ratio="9:16"  # モバイル向け縦比率
            )
            
            # タスク完了を待機
            video_url = self.wait_for_task_completion(task_id)
            results[product_id] = video_url
            
            print(f"✅ {product_name}の動画が完成しました")
            
        except Exception as e:
            print(f"❌ {product_name}の動画生成に失敗: {str(e)}")
            results[product_id] = None
    
    return results

ステップ5：コールバック処理（オプション、非同期処理用）

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/veo-callback', methods=['POST'])
def handle_veo_callback():
    """
    Veo APIコールバック通知を処理
    """
    callback_data = request.json
    
    task_id = callback_data.get('task_id')
    status = callback_data.get('status')
    
    print(f"コールバック受信: タスク {task_id} のステータスが {status} に変更")
    
    if status == "success":
        video_url = callback_data['result']['video_url']
        consumed = callback_data['consumed']
        
        print(f"🎉 動画生成成功！")
        print(f"動画URL: {video_url}")
        print(f"消費クレジット: {consumed}")
        
        # ここにビジネスロジックを追加：データベース更新、通知送信など
        
    elif status == "failed":
        error_msg = callback_data['status_reason'].get('message')
        print(f"❌ 動画生成失敗: {error_msg}")
    
    return jsonify({"status": "received"}), 200

使用例

# 環境変数の設定例
# VEO_API_KEY=your_actual_api_key_here

# 使用例
if __name__ == "__main__":
    generator = VeoVideoGenerator()
    
    # 商品データの準備
    products = [
        {
            "product_id": "prod_001",
            "product_name": "ワイヤレスイヤホン",
            "video_type": "unboxing",
            "images": [
                "https://example.com/images/earphone1.jpg",
                "https://example.com/images/earphone2.jpg"
            ]
        },
        {
            "product_id": "prod_002", 
            "product_name": "スマートウォッチ",
            "video_type": "demo",
            "images": [
                "https://example.com/images/watch1.jpg",
                "https://example.com/images/watch2.jpg"
            ]
        }
    ]
    
    # 一括動画生成の実行
    results = generator.batch_generate_ugc_videos(products)
    
    # 結果の表示
    for product_id, video_url in results.items():
        if video_url:
            print(f"商品 {product_id}: 動画生成成功 → {video_url}")
        else:
            print(f"商品 {product_id}: 動画生成失敗")

まとめ

Google Veo 3.1 APIを活用することで、ECサイト向けの高品質なUGC動画を効率的に一括生成できます。この手法により、手作業では時間がかかる動画制作を自動化し、マーケティング効果を大幅に向上させることが可能です。

特に、商品点数が多いEC事業者にとっては、このような自動化システムが業務効率化とコンバージョン率向上の両方に貢献するでしょう。

Discussion