services/gemini_client.py

"""Gemini API クライアント"""

import asyncio
import time
from typing import Optional
from dataclasses import dataclass
import numpy as np
from PIL import Image

from config.settings import settings
from utils.image_utils import frame_to_pil


@dataclass
class GeminiResponse:
    """Gemini APIレスポンス"""

    text: str
    success: bool
    error: Optional[str] = None


class GeminiClient:
    """
    Gemini API クライアント
    - 10 RPM 制限を遵守
    - 指数バックオフでリトライ
    """

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or settings.gemini_api_key
        self._client = None
        self._model = None
        self._last_request_time: float = 0
        self._min_interval: float = 6.0  # 10 RPM = 6秒間隔
        self._initialized = False

    def _init_client(self) -> bool:
        """クライアントの遅延初期化"""
        if self._initialized:
            return self._client is not None

        if not self.api_key:
            print("Warning: GEMINI_API_KEY is not set")
            self._initialized = True
            return False

        try:
            import google.generativeai as genai

            genai.configure(api_key=self.api_key)
            self._client = genai
            self._model = genai.GenerativeModel(settings.gemini_model)
            self._initialized = True
            return True
        except ImportError:
            print("Warning: google-generativeai not installed")
            self._initialized = True
            return False
        except Exception as e:
            print(f"Gemini initialization error: {e}")
            self._initialized = True
            return False

    def _wait_for_rate_limit(self) -> None:
        """レート制限のための待機"""
        elapsed = time.time() - self._last_request_time
        if elapsed < self._min_interval:
            time.sleep(self._min_interval - elapsed)

    async def _async_wait_for_rate_limit(self) -> None:
        """非同期レート制限待機"""
        elapsed = time.time() - self._last_request_time
        if elapsed < self._min_interval:
            await asyncio.sleep(self._min_interval - elapsed)

    def analyze_image(
        self, image: Image.Image, prompt: str, max_retries: int = 3
    ) -> GeminiResponse:
        """
        画像を分析

        Args:
            image: PIL Image
            prompt: 分析プロンプト
            max_retries: 最大リトライ回数

        Returns:
            GeminiResponse
        """
        if not self._init_client():
            return GeminiResponse(
                text="",
                success=False,
                error="Gemini client not initialized",
            )

        for attempt in range(max_retries):
            try:
                self._wait_for_rate_limit()
                self._last_request_time = time.time()

                response = self._model.generate_content([prompt, image])
                return GeminiResponse(text=response.text, success=True)

            except Exception as e:
                error_msg = str(e)
                if "429" in error_msg or "quota" in error_msg.lower():
                    # レート制限エラー: 指数バックオフ
                    wait_time = (2**attempt) * 10
                    print(f"Rate limited, waiting {wait_time}s...")
                    time.sleep(wait_time)
                elif attempt < max_retries - 1:
                    time.sleep(2**attempt)
                else:
                    return GeminiResponse(
                        text="",
                        success=False,
                        error=error_msg,
                    )

        return GeminiResponse(
            text="",
            success=False,
            error="Max retries exceeded",
        )

    def analyze_frame(
        self, frame: np.ndarray, prompt: str
    ) -> GeminiResponse:
        """
        NumPyフレームを分析

        Args:
            frame: NumPy配列（BGR形式）
            prompt: 分析プロンプト

        Returns:
            GeminiResponse
        """
        image = frame_to_pil(frame)
        return self.analyze_image(image, prompt)

    async def analyze_image_async(
        self, image: Image.Image, prompt: str, max_retries: int = 3
    ) -> GeminiResponse:
        """非同期で画像を分析"""
        if not self._init_client():
            return GeminiResponse(
                text="",
                success=False,
                error="Gemini client not initialized",
            )

        for attempt in range(max_retries):
            try:
                await self._async_wait_for_rate_limit()
                self._last_request_time = time.time()

                response = await asyncio.to_thread(
                    self._model.generate_content, [prompt, image]
                )
                return GeminiResponse(text=response.text, success=True)

            except Exception as e:
                error_msg = str(e)
                if "429" in error_msg or "quota" in error_msg.lower():
                    wait_time = (2**attempt) * 10
                    await asyncio.sleep(wait_time)
                elif attempt < max_retries - 1:
                    await asyncio.sleep(2**attempt)
                else:
                    return GeminiResponse(
                        text="",
                        success=False,
                        error=error_msg,
                    )

        return GeminiResponse(
            text="",
            success=False,
            error="Max retries exceeded",
        )

    @property
    def is_available(self) -> bool:
        """クライアントが利用可能かどうか"""
        return self._init_client()