掌握 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. 持续优化: 定期回顾误报，调整阈值