Nano Banana Pro API 可用性モニタリングをマスターする3つの実用的な方法

import requests
from datetime import datetime, timedelta
import json

class AIStudioMonitor:
    """
    AI Studio 監視クラス
    Nano Banana Pro (Imagen 3) のサービスステータス監視に適しています
    """

    def __init__(self, api_key: str):
        self.api_key = api_key
        # APIYI の統合インターフェースを使用
        self.base_url = "https://api.apiyi.com/v1"
        self.status_history = []

    def check_imagen_health(self) -> dict:
        """
        実際の API 呼び出しを通じて Imagen 3 サービスの健全性を検出します
        """
        try:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }

            # 軽量なテストリクエストを送信
            test_payload = {
                "model": "nano-banana-pro",
                "prompt": "health check test",
                "n": 1,
                "size": "256x256"  # コスト削減のため最小サイズ
            }

            start_time = datetime.now()
            response = requests.post(
                f"{self.base_url}/images/generations",
                headers=headers,
                json=test_payload,
                timeout=60
            )
            latency = (datetime.now() - start_time).total_seconds()

            if response.status_code == 200:
                return {
                    'status': 'healthy',
                    'latency': latency,
                    'timestamp': datetime.now().isoformat()
                }
            elif response.status_code == 429:
                return {
                    'status': 'rate_limited',
                    'latency': latency,
                    'timestamp': datetime.now().isoformat()
                }
            else:
                return {
                    'status': 'degraded',
                    'error_code': response.status_code,
                    'timestamp': datetime.now().isoformat()
                }

        except requests.Timeout:
            return {
                'status': 'timeout',
                'timestamp': datetime.now().isoformat()
            }
        except Exception as e:
            return {
                'status': 'error',
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }

    def get_uptime_stats(self, hours: int = 24) -> dict:
        """
        過去 N 時間の稼働率統計を計算します
        """
        cutoff = datetime.now() - timedelta(hours=hours)
        recent_checks = [
            s for s in self.status_history
            if datetime.fromisoformat(s['timestamp']) > cutoff
        ]

        if not recent_checks:
            return {'uptime': 'N/A', 'sample_count': 0}

        healthy_count = sum(1 for s in recent_checks if s['status'] == 'healthy')
        uptime_pct = (healthy_count / len(recent_checks)) * 100

        return {
            'uptime': f"{uptime_pct:.2f}%",
            'sample_count': len(recent_checks),
            'healthy_count': healthy_count,
            'period_hours': hours
        }

# 使用例
monitor = AIStudioMonitor(api_key="your-apiyi-key")
health = monitor.check_imagen_health()
print(f"Nano Banana Pro ヘルスステータス: {health}")

import openai
import requests
import time
import json
import schedule
import threading
from datetime import datetime, timedelta
from typing import Optional, List, Dict
from dataclasses import dataclass, asdict
from enum import Enum

class AlertChannel(Enum):
    FEISHU = "feishu"
    WECHAT = "wechat"
    DINGTALK = "dingtalk"
    SLACK = "slack"
    EMAIL = "email"

@dataclass
class CheckResult:
    timestamp: str
    status: str
    latency: Optional[float] = None
    error: Optional[str] = None
    model: str = "nano-banana-pro"

@dataclass
class AlertConfig:
    enabled: bool = True
    channels: List[AlertChannel] = None
    threshold: int = 3
    cooldown_minutes: int = 30
    webhook_urls: Dict[str, str] = None

class EnterpriseMonitor:
    """
    エンタープライズ級 Nano Banana Pro API 監視システム
    マルチチャネルアラート、データの永続化、監視ダッシュボードをサポート
    """

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.apiyi.com/v1",
        alert_config: Optional[AlertConfig] = None
    ):
        self.client = openai.OpenAI(
            api_key=api_key,
            base_url=base_url
        )
        self.alert_config = alert_config or AlertConfig()
        self.check_results: List[CheckResult] = []
        self.consecutive_failures = 0
        self.last_alert_time: Optional[datetime] = None
        self._lock = threading.Lock()

    def health_check(self) -> CheckResult:
        """
        ヘルスチェックを実行
        """
        start_time = time.time()

        try:
            response = self.client.images.generate(
                model="nano-banana-pro",
                prompt="health check: simple geometric shape",
                n=1,
                size="256x256"
            )

            latency = time.time() - start_time
            result = CheckResult(
                timestamp=datetime.now().isoformat(),
                status="success",
                latency=round(latency, 2)
            )

            with self._lock:
                self.consecutive_failures = 0

        except openai.RateLimitError:
            result = CheckResult(
                timestamp=datetime.now().isoformat(),
                status="rate_limited",
                latency=round(time.time() - start_time, 2)
            )

        except openai.APIError as e:
            result = CheckResult(
                timestamp=datetime.now().isoformat(),
                status="api_error",
                error=str(e)
            )
            with self._lock:
                self.consecutive_failures += 1

        except Exception as e:
            result = CheckResult(
                timestamp=datetime.now().isoformat(),
                status="error",
                error=str(e)
            )
            with self._lock:
                self.consecutive_failures += 1

        with self._lock:
            self.check_results.append(result)

            # 直近 1000 件のレコードを保持
            if len(self.check_results) > 1000:
                self.check_results = self.check_results[-1000:]

        # アラート条件の確認
        self._check_alert(result)

        return result

    def _check_alert(self, result: CheckResult):
        """
        アラート送信が必要か確認
        """
        if not self.alert_config.enabled:
            return

        if self.consecutive_failures < self.alert_config.threshold:
            return

        # アラートのクールダウン（抑制期間）を確認
        if self.last_alert_time:
            cooldown = timedelta(minutes=self.alert_config.cooldown_minutes)
            if datetime.now() - self.last_alert_time < cooldown:
                return

        self._send_alerts(result)
        self.last_alert_time = datetime.now()

    def _send_alerts(self, result: CheckResult):
        """
        マルチチャネルでアラートを送信
        """
        message = self._format_alert_message(result)

        if not self.alert_config.webhook_urls:
            print(f"[ALERT] {message}")
            return

        for channel, webhook_url in self.alert_config.webhook_urls.items():
            try:
                if channel == "feishu":
                    self._send_feishu_alert(webhook_url, message)
                elif channel == "wechat":
                    self._send_wechat_alert(webhook_url, message)
                elif channel == "dingtalk":
                    self._send_dingtalk_alert(webhook_url, message)
                elif channel == "slack":
                    self._send_slack_alert(webhook_url, message)
            except Exception as e:
                print(f"Failed to send {channel} alert: {e}")

    def _format_alert_message(self, result: CheckResult) -> str:
        """
        アラートメッセージのフォーマット
        """
        return f"""
🚨 Nano Banana Pro API 異常アラート

📊 ステータス詳細:
• ステータス: {result.status}
• 時刻: {result.timestamp}
• エラー: {result.error or 'N/A'}
• 連続失敗: {self.consecutive_failures} 回

🔗 監視ソース:
• Google Cloud Status: status.cloud.google.com
• OpenRouter Uptime: openrouter.ai/google/gemini-3-pro-image-preview/uptime
• APIYI 監視: apiyi.com

⚡ 推奨アクション:
1. Google Cloud ステータスページを確認
2. APIキーが有効か確認
3. ネットワーク接続状態を確認
4. APIYI テクニカルサポートに連絡
        """.strip()

    def _send_feishu_alert(self, webhook_url: str, message: str):
        """Lark（飛書）ロボットアラート"""
        payload = {
            "msg_type": "text",
            "content": {"text": message}
        }
        requests.post(webhook_url, json=payload, timeout=10)

    def _send_wechat_alert(self, webhook_url: str, message: str):
        """WeCom（企業微信）ロボットアラート"""
        payload = {
            "msgtype": "text",
            "text": {"content": message}
        }
        requests.post(webhook_url, json=payload, timeout=10)

    def _send_dingtalk_alert(self, webhook_url: str, message: str):
        """DingTalk（釘釘）ロボットアラート"""
        payload = {
            "msgtype": "text",
            "text": {"content": message}
        }
        requests.post(webhook_url, json=payload, timeout=10)

    def _send_slack_alert(self, webhook_url: str, message: str):
        """Slack ロボットアラート"""
        payload = {"text": message}
        requests.post(webhook_url, json=payload, timeout=10)

    def get_dashboard_data(self) -> dict:
        """
        ダッシュボード用のデータを取得
        """
        now = datetime.now()

        # 異なる期間の統計を計算
        periods = {
            '1h': timedelta(hours=1),
            '6h': timedelta(hours=6),
            '24h': timedelta(hours=24),
            '7d': timedelta(days=7)
        }

        stats = {}
        for period_name, delta in periods.items():
            cutoff = now - delta
            recent = [
                r for r in self.check_results
                if datetime.fromisoformat(r.timestamp) > cutoff
            ]

            if recent:
                success = sum(1 for r in recent if r.status == 'success')
                latencies = [r.latency for r in recent if r.latency]

                stats[period_name] = {
                    'total': len(recent),
                    'success': success,
                    'uptime': round((success / len(recent)) * 100, 2),
                    'avg_latency': round(sum(latencies) / len(latencies), 2) if latencies else None
                }
            else:
                stats[period_name] = None

        return {
            'model': 'nano-banana-pro',
            'current_status': self.check_results[-1].status if self.check_results else 'unknown',
            'consecutive_failures': self.consecutive_failures,
            'statistics': stats,
            'last_update': now.isoformat()
        }

# 使用例
if __name__ == "__main__":
    # アラート設定
    alert_config = AlertConfig(
        enabled=True,
        threshold=3,
        cooldown_minutes=30,
        webhook_urls={
            "feishu": "https://open.feishu.cn/open-apis/bot/v2/hook/xxx",
            # "wechat": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx",
        }
    )

    # 監視インスタンスの作成
    monitor = EnterpriseMonitor(
        api_key="your-apiyi-key",
        base_url="https://api.apiyi.com/v1",
        alert_config=alert_config
    )

    # チェック実行
    result = monitor.health_check()
    print("チェック結果:", asdict(result))

    # ダッシュボードデータの取得
    dashboard = monitor.get_dashboard_data()
    print("ダッシュボードデータ:", json.dumps(dashboard, indent=2, ensure_ascii=False))

監視頻度は、ビジネスの重要性によって異なります：

• コアとなる本番環境: 1〜2分おきにチェックすることをお勧めします
• 重要な業務: 5分おきにチェック
• 一般的な業務: 10〜15分おきにチェック

APIYI（apiyi.com）プラットフォームでは、無料のテストクレジットを取得して、監視スキームの実現可能性を素早く検証できます。監視コストとレスポンス速度のバランスに注意してください。

以下の戦略を採用することをお勧めします：

1. 連続失敗のしきい値: 3回以上連続で失敗した場合にのみアラートをトリガーするように設定する
2. スライディングウィンドウ: 5分以内の失敗率が50%を超えた場合に故障とみなす
3. 複数ソースによる検証: 公式ステータスページを同時にチェックし、公式側の障害かどうかを確認する
4. レート制限（Rate Limit）の除外: レート制限は故障ではありません

APIYI（apiyi.com）プラットフォームの監視パネルを使用すると、過去の可用性の傾向を直感的に確認でき、システム的な故障かどうかを迅速に判断できます。

OpenRouter は有名な AI モデルアグリゲータープラットフォームであり、その監視データは参考価値があります：

• メリット: 世界中のユーザーのリクエストデータを集約しているため、サンプル数が多い
• 限界: OpenRouter 自身のネットワーク経路の問題が含まれている可能性がある
• アドバイス: 公式ステータスページと自前の監視データを組み合わせて総合的に判断する

アラート疲れはよくある問題です。以下の対策を推奨します：

1. アラートレベルの設定: Info、Warning、Critical を区別する
2. クールダウン時間の設定: 同一の問題については、30分以内に1回だけ通知する
3. アラートの集約: 類似した問題を1つのアラートにまとめる
4. アラートの抑制: メンテナンス期間中は一時的にアラートをオフにする
5. 継続的な最適化: 定期的に誤報を振り返り、しきい値を調整する