Beherrschen Sie 3 praktische Methoden zur Überwachung der Nano Banana Pro API-Verfügbarkeit

import requests
from datetime import datetime, timedelta
import json

class AIStudioMonitor:
    """
    AI Studio Status-Monitoring-Klasse
    Geeignet zur Überwachung des Nano Banana Pro (Imagen 3) Dienststatus
    """

    def __init__(self, api_key: str):
        self.api_key = api_key
        # Nutzt die einheitliche Schnittstelle von APIYI
        self.base_url = "https://api.apiyi.com/v1"
        self.status_history = []

    def check_imagen_health(self) -> dict:
        """
        Überprüft den Gesundheitszustand des Imagen 3 Dienstes durch tatsächliche API-Aufrufe
        """
        try:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }

            # Sendet eine leichtgewichtige Testanfrage
            test_payload = {
                "model": "nano-banana-pro",
                "prompt": "health check test",
                "n": 1,
                "size": "256x256"  # Minimale Größe zur Kostensenkung
            }

            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:
        """
        Berechnet die Verfügbarkeitsstatistik für die letzten N Stunden
        """
        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
        }

# Anwendungsbeispiel
monitor = AIStudioMonitor(api_key="your-apiyi-key")
health = monitor.check_imagen_health()
print(f"Nano Banana Pro Gesundheitsstatus: {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-Level Nano Banana Pro API Überwachungssystem
    Unterstützt Multi-Channel-Alarme, Datenpersistenz und Monitoring-Dashboard
    """

    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:
        """
        Health Check ausführen
        """
        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)

            # Die letzten 1000 Datensätze behalten
            if len(self.check_results) > 1000:
                self.check_results = self.check_results[-1000:]

        # Alarmbedingungen prüfen
        self._check_alert(result)

        return result

    def _check_alert(self, result: CheckResult):
        """
        Prüfen, ob ein Alarm gesendet werden muss
        """
        if not self.alert_config.enabled:
            return

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

        # Alarm-Cooldown prüfen
        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):
        """
        Multi-Channel-Alarme senden
        """
        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:
        """
        Alarmmeldung formatieren
        """
        return f"""
🚨 Nano Banana Pro API Anomalie-Alarm

📊 Status-Details:
• Status: {result.status}
• Zeit: {result.timestamp}
• Fehler: {result.error or 'N/A'}
• Aufeinanderfolgende Fehler: {self.consecutive_failures} Mal

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

⚡ Empfohlene Maßnahmen:
1. Google Cloud Statusseite prüfen
2. Gültigkeit des API-Keys überprüfen
3. Netzwerkverbindung prüfen
4. Technischen Support von APIYI kontaktieren
        """.strip()

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

    def get_dashboard_data(self) -> dict:
        """
        Dashboard-Daten abrufen
        """
        now = datetime.now()

        # Statistiken für verschiedene Zeiträume berechnen
        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()
        }

# Anwendungsbeispiel
if __name__ == "__main__":
    # Alarm konfigurieren
    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",
        }
    )

    # Monitoring-Instanz erstellen
    monitor = EnterpriseMonitor(
        api_key="your-apiyi-key",
        base_url="https://api.apiyi.com/v1",
        alert_config=alert_config
    )

    # Check ausführen
    result = monitor.health_check()
    print("Check-Ergebnis:", asdict(result))

    # Dashboard-Daten abrufen
    dashboard = monitor.get_dashboard_data()
    print("Dashboard-Daten:", json.dumps(dashboard, indent=2, ensure_ascii=False))

Die Überwachungsfrequenz hängt von der geschäftlichen Relevanz ab:

• Kern-Produktionsumgebungen: Empfohlen wird eine Überprüfung alle 1-2 Minuten.
• Wichtige Geschäftsbereiche: Überprüfung alle 5 Minuten.
• Einfache Anwendungen: Überprüfung alle 10-15 Minuten.

Über die Plattform APIYI (apiyi.com) können Sie kostenloses Testguthaben erhalten, um die Machbarkeit Ihrer Monitoring-Lösung schnell zu validieren. Achten Sie dabei auf das Gleichgewicht zwischen Monitoring-Kosten und Reaktionsgeschwindigkeit.

Wir empfehlen die folgenden Strategien:

1. Schwellenwert für aufeinanderfolgende Fehler: Lösen Sie Alarme erst nach mindestens 3 aufeinanderfolgenden Fehlern aus.
2. Gleitendes Zeitfenster: Ein Ausfall wird erst definiert, wenn die Fehlerrate innerhalb von 5 Minuten über 50 % liegt.
3. Validierung aus mehreren Quellen: Prüfen Sie gleichzeitig die offizielle Statusseite, um zu bestätigen, ob es sich um ein offizielles Problem handelt.
4. Rate Limits ausschließen: Eine Drosselung (Rate Limiting) ist nicht gleichbedeutend mit einem Systemausfall.

Über das Monitoring-Dashboard der APIYI-Plattform (apiyi.com) können Sie historische Verfügbarkeitstrends visuell analysieren und schnell beurteilen, ob ein systemischer Fehler vorliegt.

OpenRouter ist als bekannte Aggregationsplattform für KI-Modelle eine wertvolle Referenzquelle:

• Vorteile: Aggregiert Anfragedaten von Nutzern weltweit, was eine große Stichprobenmenge bietet.
• Einschränkungen: Die Daten können auch Probleme enthalten, die in der Infrastruktur von OpenRouter selbst liegen.
• Empfehlung: Nutzen Sie eine Kombination aus der offiziellen Statusseite und eigenem Monitoring für eine fundierte Gesamtbeurteilung.

Alarm-Müdigkeit ist ein häufiges Problem. Folgendes wird empfohlen:

1. Alarmstufen festlegen: Unterscheiden Sie zwischen Info, Warning und Critical.
2. Abkühlzeit (Cooldown) konfigurieren: Senden Sie für das gleiche Problem nur einen Alarm innerhalb von 30 Minuten.
3. Alarme aggregieren: Fassen Sie ähnliche Probleme in einer einzigen Meldung zusammen.
4. Alarm-Unterdrückung: Schalten Sie Alarme während geplanter Wartungsarbeiten vorübergehend stumm.
5. Kontinuierliche Optimierung: Überprüfen Sie regelmäßig Fehlalarme und passen Sie die Schwellenwerte an.