utils/openai_api.py

"""OpenAI API処理モジュール"""
import os
import json
import base64
from typing import List, Dict, Optional
from openai import OpenAI
import asyncio
from PIL import Image
import io
import httpx  # httpx が既にインストールされている前提

# === パッチ: httpx.Client で 'proxies' 引数を無視 ===
_original_httpx_client_init = httpx.Client.__init__


def _patched_httpx_client_init(self, *args, **kwargs):
    # 'proxies' キーワードが存在しても無視して初期化
    if 'proxies' in kwargs:
        kwargs.pop('proxies')
    return _original_httpx_client_init(self, *args, **kwargs)


# 一度だけパッチを適用
if not getattr(httpx.Client, '_proxies_patch_applied', False):
    httpx.Client.__init__ = _patched_httpx_client_init
    httpx.Client._proxies_patch_applied = True

class OpenAIPromptSplitter:
    """OpenAI APIを使用したプロンプト分割クラス"""
    
    def __init__(self):
        api_key = os.getenv('OPENAI_API_KEY')
        if not api_key:
            print("警告: OpenAI APIキーが設定されていません。AI自動プロンプト分割は使用できません。")
            self.client = None
        else:
            try:
                # プロキシ問題はhttpxパッチで解決済み
                self.client = OpenAI()
                print(f"✅ OpenAI APIクライアントの初期化に成功しました")
            except Exception as e:
                print(f"❌ OpenAI APIクライアントの初期化に失敗しました: {str(e)}")
                self.client = None
                raise e  # エラーを上位に伝播
    
    async def split_prompt(
        self,
        base_prompt: str,
        num_clips: int,
        start_description: str,
        end_description: str
    ) -> List[Dict[str, str]]:
        """単一プロンプトを複数のクリップ用プロンプトに分割"""
        
        system_prompt = """あなたは映像制作の専門家です。
与えられた基本プロンプトと開始・終了の画像説明を元に、
自然で流れのあるストーリーを持つ複数の動画クリップ用のプロンプトを生成してください。

重要な要件：
1. 入力プロンプトが日本語の場合は、必ず英語に翻訳してから各クリップのプロンプトを作成してください
2. 各クリップのプロンプトは必ず英語で出力してください
3. 英語への翻訳は、映像生成に適した詳細で具体的な表現を使用してください

各プロンプトは以下の要素を含めてください：
1. 具体的な動作や動き（specific actions and movements）
2. カメラワーク（zoom, pan, rotation など）
3. 照明や雰囲気の変化（lighting and atmosphere changes）
4. 背景要素の動き（background element movements）

JSON形式で返してください。"""
        
        user_prompt = f"""
基本プロンプト: {base_prompt}
開始画像の説明: {start_description}
終了画像の説明: {end_description}
クリップ数: {num_clips}

{num_clips}個のクリップ用プロンプトを生成してください。
最初のクリップは開始画像から自然に始まり、
最後のクリップは終了画像に向かって自然に終わるようにしてください。

重要：すべてのプロンプトは英語で出力してください。入力が日本語の場合は英語に翻訳してください。

返答形式:
{{
  "clips": [
    {{"clip_number": 1, "prompt": "Detailed English prompt 1"}},
    {{"clip_number": 2, "prompt": "Detailed English prompt 2"}},
    ...
  ]
}}
"""
        
        if not self.client:
            # OpenAI APIが利用できない場合はフォールバック
            return self._fallback_split(base_prompt, num_clips)
        
        # 試行するモデルの順序を定義
        # 2025年6月時点で利用可能なモデル
        models_to_try = [
            ("gpt-4.1-2025-04-14", "GPT-4.1モデル（メイン）"),
            ("o4-mini-2025-04-16", "o4-miniモデル（フォールバック）"),
            ("gpt-4o-2024-11-20", "GPT-4oモデル（バックアップ）"),
            ("gpt-4-turbo", "GPT-4 Turboモデル"),
            ("gpt-4", "GPT-4モデル"),
            ("gpt-3.5-turbo", "GPT-3.5 Turboモデル")
        ]
        
        response = None
        
        for model, model_name in models_to_try:
            print(f"{model_name}を試行中...")
            response = await self._try_model(model, system_prompt, user_prompt, timeout=180)
            
            if response:
                print(f"{model_name}での生成に成功しました。")
                break
            else:
                print(f"{model_name}での生成に失敗しました。")
        
        if response:
            # JSONをパース
            try:
                result = json.loads(response)
                return result.get("clips", [])
            except json.JSONDecodeError:
                print("JSON解析エラー: フォールバック処理を使用します。")
                return self._fallback_split(base_prompt, num_clips)
        else:
            print("すべてのモデルでの生成に失敗しました。フォールバック処理を使用します。")
            return self._fallback_split(base_prompt, num_clips)
    
    async def _try_model(self, model: str, system_prompt: str, user_prompt: str, timeout: int) -> Optional[str]:
        """指定されたモデルでプロンプト生成を試行"""
        try:
            # 非同期実行のためのラッパー
            loop = asyncio.get_event_loop()
            
            def _make_request():
                # o3/o4モデルの特殊要件に対応
                if "o3" in model or "o4" in model:
                    return self.client.chat.completions.create(
                        model=model,
                        messages=[
                            {"role": "system", "content": system_prompt},
                            {"role": "user", "content": user_prompt}
                        ],
                        max_completion_tokens=2000  # max_tokensの代わり、temperatureなし
                    )
                else:
                    return self.client.chat.completions.create(
                        model=model,
                        messages=[
                            {"role": "system", "content": system_prompt},
                            {"role": "user", "content": user_prompt}
                        ],
                        temperature=0.8,
                        max_tokens=2000,
                        response_format={"type": "json_object"}
                    )
            
            # タイムアウト付きで実行
            response = await asyncio.wait_for(
                loop.run_in_executor(None, _make_request),
                timeout=timeout
            )
            
            return response.choices[0].message.content
            
        except asyncio.TimeoutError:
            print(f"  - タイムアウト（{timeout}秒）")
            return None
        except Exception as e:
            error_message = str(e)
            error_message_lower = error_message.lower()
            
            # より詳細なエラー情報を表示
            if "model_not_found" in error_message_lower or "does not exist" in error_message_lower:
                print(f"  - モデルが存在しません: {model}")
            elif "invalid_request_error" in error_message_lower:
                print(f"  - 無効なリクエスト: {error_message}")
            elif "rate_limit_exceeded" in error_message_lower:
                print(f"  - レート制限に達しました")
            elif "insufficient_quota" in error_message_lower:
                print(f"  - APIクレジットが不足しています")
            elif "invalid_api_key" in error_message_lower:
                print(f"  - APIキーが無効です")
            else:
                print(f"  - エラー: {error_message}")
            return None
    
    def _fallback_split(self, base_prompt: str, num_clips: int) -> List[Dict[str, str]]:
        """フォールバック用のシンプルなプロンプト分割"""
        clips = []
        
        # 日本語を検出して簡単な英訳を行う
        # 日本語文字（ひらがな、カタカナ、漢字）を含むかチェック
        import re
        has_japanese = bool(re.search(r'[\u3040-\u309F\u30A0-\u30FF\u4E00-\u9FAF]', base_prompt))
        
        if has_japanese:
            # 簡単な英訳マッピング（実際の翻訳はAIに任せるべきですが、フォールバック用）
            base_prompt_en = base_prompt  # 本来はここで翻訳すべきだが、フォールバックなので元のまま使用
            print("警告: 日本語プロンプトが検出されましたが、翻訳APIが利用できません。")
        else:
            base_prompt_en = base_prompt
        
        # 基本的な動きのバリエーション
        movements = [
            "slowly zooming in, gentle movement",
            "panning from left to right, smooth transition",
            "subtle rotation, dynamic perspective",
            "pulling back slowly, revealing more context",
            "focusing on details, intimate view"
        ]
        
        # 照明のバリエーション
        lighting = [
            "soft morning light",
            "warm afternoon glow",
            "dramatic sunset lighting",
            "cool evening atmosphere",
            "mystical twilight ambiance"
        ]
        
        for i in range(num_clips):
            movement = movements[i % len(movements)]
            light = lighting[i % len(lighting)]
            
            if i == 0:
                # 最初のクリップ
                prompt = f"{base_prompt_en}, starting scene, {movement}, {light}"
            elif i == num_clips - 1:
                # 最後のクリップ
                prompt = f"{base_prompt_en}, concluding scene, {movement}, {light}, transitioning to end"
            else:
                # 中間のクリップ
                prompt = f"{base_prompt_en}, {movement}, {light}, continuous flow"
            
            clips.append({
                "clip_number": i + 1,
                "prompt": prompt
            })
        
        return clips
    
    def _encode_image(self, image_path: str) -> str:
        """画像をbase64エンコード"""
        with Image.open(image_path) as img:
            # JPEGに変換してサイズを最適化
            if img.mode != 'RGB':
                img = img.convert('RGB')
            
            # 画像が大きすぎる場合はリサイズ（最大1024px）
            max_size = 1024
            if img.width > max_size or img.height > max_size:
                img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
            
            # バイトストリームに保存
            buffer = io.BytesIO()
            img.save(buffer, format='JPEG', quality=85)
            
            # base64エンコード
            image_bytes = buffer.getvalue()
            return base64.b64encode(image_bytes).decode('utf-8')
    
    async def generate_image_description(self, image_path: str) -> str:
        """GPT-4 Visionを使用して画像から英語の説明を生成"""
        
        if not self.client:
            return "A dynamic scene with smooth camera movement and natural transitions. The camera is fixed. Camera does not move."
        
        # 画像をbase64エンコード
        base64_image = self._encode_image(image_path)
        
        system_prompt = """You are an expert at analyzing images and creating detailed descriptions for video generation.
Analyze the provided image and create a comprehensive English description that captures:
1. The main subject and its appearance
2. The background and environment
3. Lighting conditions and atmosphere
4. Colors and visual style
5. Any notable objects or elements

Your description should be suitable for video generation AI to recreate the scene.
Be specific and detailed, using vivid descriptive language."""
        
        user_prompt = "Analyze this image and provide a detailed English description suitable for video generation. Focus on visual elements, composition, and atmosphere."
        
        # Vision対応モデルのリスト
        vision_models = [
            ("gpt-4o", "GPT-4o Vision"),
            ("gpt-4-turbo", "GPT-4 Turbo Vision"),
            ("gpt-4-vision-preview", "GPT-4 Vision Preview")
        ]
        
        for model, model_name in vision_models:
            print(f"{model_name}で画像を分析中...")
            
            try:
                response = self.client.chat.completions.create(
                    model=model,
                    messages=[
                        {
                            "role": "system",
                            "content": system_prompt
                        },
                        {
                            "role": "user",
                            "content": [
                                {
                                    "type": "text",
                                    "text": user_prompt
                                },
                                {
                                    "type": "image_url",
                                    "image_url": {
                                        "url": f"data:image/jpeg;base64,{base64_image}"
                                    }
                                }
                            ]
                        }
                    ],
                    max_tokens=500,
                    temperature=0.7
                )
                
                description = response.choices[0].message.content
                print(f"{model_name}での画像分析に成功しました。")
                
                # カメラ固定の指示を追加
                return f"{description} The camera is fixed. Camera does not move."
                
            except Exception as e:
                print(f"{model_name}でエラー: {str(e)}")
                continue
        
        # すべてのモデルで失敗した場合のフォールバック
        print("画像分析に失敗しました。デフォルトの説明を使用します。")
        return "A dynamic scene with natural elements and smooth transitions. The camera is fixed. Camera does not move."

    async def translate_prompt(self, prompt: str) -> str:
        """日本語プロンプトを英語に翻訳"""
        
        if not self.client:
            return prompt
        
        # 日本語が含まれているかチェック
        import re
        has_japanese = bool(re.search(r'[\u3040-\u309F\u30A0-\u30FF\u4E00-\u9FAF]', prompt))
        
        if not has_japanese:
            # 日本語が含まれていない場合はそのまま返す
            return prompt
        
        system_prompt = """You are a professional translator specializing in image generation prompts.
Translate the given Japanese prompt into detailed, descriptive English suitable for AI image generation.
Focus on visual elements, artistic style, and atmosphere.
Make the translation more descriptive and specific than the original.
Only output the translated prompt, nothing else."""
        
        user_prompt = f"Translate this Japanese prompt to English for image generation:\n{prompt}"
        
        # 試行するモデルの順序を定義
        models_to_try = [
            ("gpt-4o", "GPT-4o"),
            ("gpt-4-turbo", "GPT-4 Turbo"),
            ("gpt-3.5-turbo", "GPT-3.5 Turbo")
        ]
        
        for model, model_name in models_to_try:
            try:
                # 非同期実行のためのラッパー
                loop = asyncio.get_event_loop()
                
                def _make_request():
                    return self.client.chat.completions.create(
                        model=model,
                        messages=[
                            {"role": "system", "content": system_prompt},
                            {"role": "user", "content": user_prompt}
                        ],
                        temperature=0.3,  # 翻訳なので低めの温度
                        max_tokens=500
                    )
                
                # タイムアウト付きで実行
                response = await asyncio.wait_for(
                    loop.run_in_executor(None, _make_request),
                    timeout=30
                )
                
                translated = response.choices[0].message.content.strip()
                print(f"{model_name}での翻訳に成功しました。")
                return translated
                
            except Exception as e:
                print(f"{model_name}でエラー: {str(e)}")
                continue
        
        # すべてのモデルで失敗した場合は元のプロンプトを返す
        print("翻訳に失敗しました。元のプロンプトを使用します。")
        return prompt