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
        # APIYI 통합 인터페이스 사용
        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)

            # 최근 1,000개 기록 유지
            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
• 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):
        """기업용 위챗 봇 알림"""
        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",
            # "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분마다 한 번씩 체크해도 충분합니다.

APIYI apiyi.com 플랫폼에서 무료 테스트 크레딧을 받아 모니터링 방안의 실현 가능성을 빠르게 검증해 볼 수 있습니다. 모니터링 비용과 응답 속도 사이의 균형을 잘 맞추는 것이 중요합니다.

다음과 같은 전략을 도입해 보세요:

1. 연속 실패 임계값: 연속 3회 이상 실패할 때만 경고 알림을 보내도록 설정하세요.
2. 슬라이딩 윈도우: 5분 이내 실패율이 50%를 넘을 때만 장애로 판단합니다.
3. 교차 검증: 공식 상태 페이지를 함께 확인하여 공식적인 장애인지 파악하세요.
4. Rate Limit 제외: 할당량 제한(Rate Limit)으로 인한 실패는 시스템 장애와 구분해야 합니다.

APIYI apiyi.com 플랫폼의 모니터링 대시보드를 활용하면 과거 가용성 트렌드를 직관적으로 확인하고, 시스템적인 장애 여부를 빠르게 판단할 수 있습니다.

OpenRouter는 인지도 높은 AI 모델 허브 플랫폼이므로, 해당 모니터링 데이터는 충분히 참고할 가치가 있습니다:

• 장점: 전 세계 사용자들의 요청 데이터를 취합하므로 표본이 방대합니다.
• 한계: OpenRouter 자체적인 네트워크 경로 문제가 포함되어 있을 수 있습니다.
• 권장: 공식 상태 페이지와 자체 구축한 모니터링 데이터를 결합하여 종합적으로 판단하는 것이 좋습니다.

알림 피로(Alert Fatigue)는 흔히 발생하는 문제예요. 이렇게 개선해 보세요:

1. 알림 레벨 설정: Info, Warning, Critical로 중요도를 나누세요.
2. 쿨다운 시간 설정: 동일한 문제에 대해서는 30분 이내에 한 번만 알림이 가도록 설정하세요.
3. 알림 그룹화: 유사한 문제들을 하나의 알림으로 묶어서 보내세요.
4. 알림 억제: 시스템 점검 중에는 일시적으로 알림을 꺼둡니다.
5. 지속적인 최적화: 정기적으로 오탐 사례를 검토하고 임계값을 조정하세요.