Kuasai 3 Metode Praktis untuk Memantau Ketersediaan API Nano Banana Pro

import requests
from datetime import datetime, timedelta
import json

class AIStudioMonitor:
    """
    Kelas pemantauan status AI Studio
    Cocok untuk memantau status layanan Nano Banana Pro (Imagen 3)
    """

    def __init__(self, api_key: str):
        self.api_key = api_key
        # Menggunakan antarmuka terpadu APIYI
        self.base_url = "https://api.apiyi.com/v1"
        self.status_history = []

    def check_imagen_health(self) -> dict:
        """
        Mendeteksi status kesehatan layanan Imagen 3 melalui pemanggilan API nyata
        """
        try:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }

            # Mengirim permintaan uji coba ringan
            test_payload = {
                "model": "nano-banana-pro",
                "prompt": "health check test",
                "n": 1,
                "size": "256x256"  # Ukuran minimum untuk mengurangi biaya
            }

            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:
        """
        Menghitung statistik ketersediaan selama N jam terakhir
        """
        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
        }

# Contoh penggunaan
monitor = AIStudioMonitor(api_key="your-apiyi-key")
health = monitor.check_imagen_health()
print(f"Status kesehatan 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:
    """
    Sistem pemantauan API Nano Banana Pro tingkat perusahaan
    Mendukung peringatan multi-saluran, persistensi data, dan dasbor pemantauan
    """

    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:
        """
        Menjalankan pemeriksaan kesehatan
        """
        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)

            # Menyimpan 1000 catatan terakhir
            if len(self.check_results) > 1000:
                self.check_results = self.check_results[-1000:]

        # Memeriksa kondisi peringatan
        self._check_alert(result)

        return result

    def _check_alert(self, result: CheckResult):
        """
        Memeriksa apakah perlu mengirim peringatan
        """
        if not self.alert_config.enabled:
            return

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

        # Memeriksa pendinginan (cooldown) peringatan
        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):
        """
        Mengirim peringatan multi-saluran
        """
        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"Gagal mengirim peringatan {channel}: {e}")

    def _format_alert_message(self, result: CheckResult) -> str:
        """
        Memformat pesan peringatan
        """
        return f"""
🚨 Peringatan Anomali API Nano Banana Pro

📊 Detail Status:
• Status: {result.status}
• Waktu: {result.timestamp}
• Kesalahan: {result.error or 'N/A'}
• Kegagalan Berturut-turut: {self.consecutive_failures} kali

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

⚡ Saran Tindakan:
1. Periksa halaman status Google Cloud
2. Verifikasi apakah kunci API valid
3. Periksa status koneksi jaringan
4. Hubungi dukungan teknis APIYI
        """.strip()

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

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

    def get_dashboard_data(self) -> dict:
        """
        Mendapatkan data dasbor pemantauan
        """
        now = datetime.now()

        # Menghitung statistik untuk periode waktu yang berbeda
        periods = {
            '1j': timedelta(hours=1),
            '6j': timedelta(hours=6),
            '24j': timedelta(hours=24),
            '7h': 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()
        }

# Contoh penggunaan
if __name__ == "__main__":
    # Konfigurasi peringatan
    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",
        }
    )

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

    # Jalankan pemeriksaan
    result = monitor.health_check()
    print("Hasil Pemeriksaan:", asdict(result))

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

Frekuensi pemantauan bergantung pada seberapa krusial layanan tersebut bagi bisnis Anda:

• Lingkungan Produksi Inti: Disarankan untuk memeriksa setiap 1-2 menit.
• Bisnis Penting: Periksa setiap 5 menit.
• Bisnis Biasa: Periksa setiap 10-15 menit.

Melalui platform APIYI (apiyi.com), Anda bisa mendapatkan kuota uji coba gratis untuk memvalidasi kelayakan skema pemantauan Anda dengan cepat. Pastikan untuk menyeimbangkan antara biaya pemantauan dan kecepatan respons.

Kami menyarankan strategi berikut:

1. Ambang Batas Kegagalan Beruntun: Atur agar peringatan (alert) hanya muncul setelah terjadi minimal 3 kali kegagalan berturut-turut.
2. Jendela Geser (Sliding Window): Anggap sebagai gangguan jika tingkat kegagalan melebihi 50% dalam waktu 5 menit.
3. Validasi Multi-Sumber: Periksa juga halaman status resmi secara bersamaan untuk mengonfirmasi apakah gangguan tersebut berasal dari pihak penyedia.
4. Kecualikan Rate Limit: Batasan frekuensi (rate limit) tidak sama dengan gangguan sistem.

Dengan panel pemantauan di platform APIYI (apiyi.com), Anda dapat melihat tren ketersediaan historis secara visual untuk menentukan dengan cepat apakah masalah tersebut merupakan gangguan sistemik.

Sebagai platform agregator model AI yang ternama, data pemantauan OpenRouter memiliki nilai referensi yang baik:

• Keunggulan: Mengagregasi data permintaan dari pengguna global dengan jumlah sampel yang besar.
• Keterbatasan: Mungkin mencakup masalah koneksi dari sisi OpenRouter itu sendiri.
• Saran: Gunakan sebagai referensi tambahan yang dikombinasikan dengan halaman status resmi dan pemantauan mandiri Anda.

Kelelahan akibat peringatan adalah masalah umum. Berikut sarannya:

1. Tetapkan Tingkat Peringatan: Bedakan antara Info, Warning, dan Critical.
2. Konfigurasi Waktu Pendinginan (Cooldown): Untuk masalah yang sama, kirimkan peringatan hanya sekali dalam 30 menit.
3. Agregasi Peringatan: Gabungkan masalah-masalah yang serupa menjadi satu notifikasi.
4. Penekanan Peringatan (Suppression): Matikan peringatan sementara selama masa pemeliharaan (maintenance).
5. Optimasi Berkelanjutan: Tinjau laporan positif palsu (false alarm) secara berkala dan sesuaikan ambang batasnya.