機能概要

EC 運営において、ユーザー生成コンテンツ（UGC）動画はコンバージョン率向上の重要なツールです。本記事では、Python と Google Veo AI 動画生成 API を活用し、静的な商品写真からアンboxing動画、商品デモ動画、ライフスタイル動画をバッチ生成する方法を解説します。

最終的な効果サンプル：

入力：商品の静止画 + テキストプロンプト
出力：高品質な商品紹介動画
バッチ処理：一度に数十商品を処理し、対応する動画を自動生成

準備作業

環境要件

Python 3.7+
requests ライブラリ
利用可能な Veo API アクセス権限

API キーの取得

DEFAPIプラットフォームでアカウント登録
コントロールパネルで API キーを取得
重要：セキュリティ注意：コードに直接キーを記述せず、環境変数で管理

依存関係のインストール

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: コールバック通知アドレス

タスク照会インターフェース

エンドポイント: 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:
        辞書：{商品ID: 動画URL}
    """
    
    # 動画タイプ対応プロンプトテンプレート
    prompt_templates = {
        'unboxing': "高品質な{product}のプロフェッショナルなアンboxing動画。パッケージ、開封プロセス、商品のお披露目を表示",
        '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