3 практических способа мониторинга доступности Nano Banana Pro API

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:
        """
        Проверка работоспособности Imagen 3 через фактический вызов API
        """
        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"Не удалось отправить алерт в {channel}: {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):
        """Алерт для Feishu бота"""
        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 бота"""
        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",
            # "slack": "https://hooks.slack.com/services/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: если вы уперлись в лимиты запросов — это не поломка API.

В панели управления APIYI на apiyi.com можно наглядно посмотреть графики доступности и быстро понять, носит ли сбой системный характер.

OpenRouter — это популярный агрегатор ИИ-моделей, и его данные вполне можно использовать как ориентир:

• Плюсы: огромный объем данных на основе реальных запросов пользователей со всего мира.
• Минусы: в статистику могут подмешиваться проблемы самого OpenRouter или их сетевых маршрутов.
• Совет: используйте эти данные как дополнение к официальным статус-страницам и собственным проверкам.

Усталость от алертов (alert fatigue) — частая проблема. Вот как с ней бороться:

1. Разделите уровни важности: используйте градации Info, Warning и Critical.
2. Настройте период охлаждения (cooldown): не шлите уведомления чаще одного раза в 30 минут по одной и той же проблеме.
3. Агрегируйте алерты: объединяйте похожие ошибки в одно сообщение.
4. Режим обслуживания: временно отключайте уведомления на время плановых работ.
5. Постоянная оптимизация: регулярно анализируйте ложные срабатывания и корректируйте пороги чувствительности.