Mastering 3 Practical Methods for Nano Banana Pro API Availability Monitoring

import requests
from datetime import datetime, timedelta
import json

class AIStudioMonitor:
    """
    AI Studio status monitoring class.
    Suitable for monitoring Nano Banana Pro (Imagen 3) service status.
    """

    def __init__(self, api_key: str):
        self.api_key = api_key
        # Use the unified APIYI interface
        self.base_url = "https://api.apiyi.com/v1"
        self.status_history = []

    def check_imagen_health(self) -> dict:
        """
        Detect Imagen 3 service health via actual API calls.
        """
        try:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }

            # Send a lightweight test request
            test_payload = {
                "model": "nano-banana-pro",
                "prompt": "health check test",
                "n": 1,
                "size": "256x256"  # Minimum size to keep costs low
            }

            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:
        """
        Calculate availability statistics for the past N hours.
        """
        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
        }

# Usage Example
monitor = AIStudioMonitor(api_key="your-apiyi-key")
health = monitor.check_imagen_health()
print(f"Nano Banana Pro Health Status: {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:
    """
    Enterprise-grade Nano Banana Pro API monitoring system.
    Supports multi-channel alerts, data persistence, and monitoring dashboards.
    """

    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:
        """
        Perform a health check
        """
        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)

            # Keep the last 1,000 records
            if len(self.check_results) > 1000:
                self.check_results = self.check_results[-1000:]

        # Check alert conditions
        self._check_alert(result)

        return result

    def _check_alert(self, result: CheckResult):
        """
        Check if an alert needs to be sent
        """
        if not self.alert_config.enabled:
            return

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

        # Check alert cooldown
        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):
        """
        Send multi-channel alerts
        """
        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:
        """
        Format alert message
        """
        return f"""
🚨 Nano Banana Pro API Exception Alert

📊 Status Details:
• Status: {result.status}
• Time: {result.timestamp}
• Error: {result.error or 'N/A'}
• Consecutive Failures: {self.consecutive_failures}

🔗 Monitoring Sources:
• Google Cloud Status: status.cloud.google.com
• OpenRouter Uptime: openrouter.ai/google/gemini-3-pro-image-preview/uptime
• APIYI Monitoring: apiyi.com

⚡ Recommended Actions:
1. Check the Google Cloud status page
2. Verify if the API key is valid
3. Check network connection status
4. Contact APIYI technical support
        """.strip()

    def _send_feishu_alert(self, webhook_url: str, message: str):
        """Feishu bot alert"""
        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 bot alert"""
        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 bot alert"""
        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 bot alert"""
        payload = {"text": message}
        requests.post(webhook_url, json=payload, timeout=10)

    def get_dashboard_data(self) -> dict:
        """
        Get monitoring dashboard data
        """
        now = datetime.now()

        # Calculate statistics for different time periods
        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()
        }

# Usage example
if __name__ == "__main__":
    # Configure alerts
    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",
        }
    )

    # Create monitoring instance
    monitor = EnterpriseMonitor(
        api_key="your-apiyi-key",
        base_url="https://api.apiyi.com/v1",
        alert_config=alert_config
    )

    # Perform check
    result = monitor.health_check()
    print("Check Result:", asdict(result))

    # Get dashboard data
    dashboard = monitor.get_dashboard_data()
    print("Dashboard Data:", json.dumps(dashboard, indent=2, ensure_ascii=False))

The right frequency depends on how critical the service is to your business:

• Core Production Environments: We recommend checking every 1-2 minutes.
• Important Business Logic: Once every 5 minutes is usually enough.
• General Internal Tasks: Every 10-15 minutes should suffice.

You can grab some free test credits on the APIYI (apiyi.com) platform to quickly validate your monitoring setup. Just remember to find the right balance between monitoring costs and response speed.

It's best to use a combination of these strategies:

1. Consecutive Failure Threshold: Only trigger an alert after 3 or more consecutive failures.
2. Sliding Window: Define an outage only if the failure rate exceeds 50% within a 5-minute window.
3. Multi-source Verification: Check the official status page at the same time to see if it's a global issue.
4. Exclude Rate Limits: Remember, hitting a rate limit isn't the same as a system failure.

With the monitoring dashboard on the APIYI (apiyi.com) platform, you can visually track historical availability trends to quickly determine if you're dealing with a systemic issue.

As a well-known AI model aggregator, OpenRouter's monitoring data is a great reference point:

• Pros: It aggregates request data from users worldwide, providing a massive sample size.
• Limitations: The data might include latency or issues stemming from OpenRouter's own infrastructure.
• Advice: Use it as a secondary data point alongside official status pages and your own custom monitoring.

Alert fatigue is a common headache. Here’s how to manage it:

1. Set Alert Severity: Differentiate between Info, Warning, and Critical levels.
2. Configure Cooling-off Periods: Ensure the same issue only triggers one alert every 30 minutes.
3. Aggregate Alerts: Group similar issues into a single notification.
4. Alert Suppression: Temporarily disable alerts during scheduled maintenance.
5. Continuous Optimization: Regularly review false positives and adjust your thresholds accordingly.