掌握 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
        # 使用 API易 統一接口
        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
• API易 監控: apiyi.com

⚡ 建議操作:
1. 檢查 Google Cloud 狀態頁
2. 驗證 API 密鑰是否有效
3. 檢查網絡連接狀態
4. 聯繫 API易 技術支持
        """.strip（）

    def _send_feishu_alert（self， webhook_url: str， message: str）:
        """飛書機器人告警"""
        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）:
        """企業微信機器人告警"""
        payload = {
            "msgtype": "text"，
            "text": {"content": message}
        }
        requests.post（webhook_url， json=payload， timeout=10）

    def _send_dingtalk_alert（self， webhook_url: str， message: str）:
        """釘釘機器人告警"""
        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 分鐘檢查一次

通過 API易 apiyi.com 平臺可以獲取免費測試額度，快速驗證監控方案可行性。注意平衡監控成本和響應速度。

建議採用以下策略:

1. 連續失敗閾值: 設置連續 3 次以上失敗才觸發告警
2. 滑動窗口: 5 分鐘內失敗率超過 50% 才認定故障
3. 多源驗證: 同時檢查官方狀態頁確認是否官方故障
4. 排除 Rate Limit: 限流不等於故障

通過 API易 apiyi.com 平臺的監控面板，可以直觀查看歷史可用性趨勢，快速判斷是否爲系統性故障。

OpenRouter 作爲知名的 AI 模型聚合平臺，其監控數據具有參考價值:

• 優勢: 聚合全球用戶請求數據，樣本量大
• 侷限: 可能包含 OpenRouter 自身鏈路問題
• 建議: 結合官方狀態頁和自建監控綜合判斷

告警疲勞是常見問題，建議:

1. 設置告警級別: 區分 Info、Warning、Critical
2. 配置冷卻時間: 同一問題 30 分鐘內只告警一次
3. 聚合告警: 相似問題合併爲一條告警
4. 告警抑制: 維護期間臨時關閉告警
5. 持續優化: 定期回顧誤報，調整閾值