Go to Qiita Advent Calendar Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@pan0x11

Google Veo 3.1 API を呼び出し EC UGC 動画をバッチ生成する方法

Posted at

機能概要

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

最終的な効果サンプル：

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

準備作業

環境要件

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

API キーの取得

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

依存関係のインストール

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
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?