إتقان 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 عبر استدعاء واجهة برمجة تطبيقات فعلي
        """
        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"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. 检查 network 连接状态
4. 联系 APIYI 技术支持
        """.strip()

    def _send_feishu_alert(self, webhook_url: str, message: str):
        """飞书机器人告警"""
        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):
        """钉钉机器人告警"""
        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))

يعتمد معدل تكرار المراقبة على مدى أهمية أعمالك:

• بيئة الإنتاج الأساسية: نوصي بالفحص كل دقيقة إلى دقيقتين.
• الأعمال الهامة: الفحص كل 5 دقائق.
• الأعمال العادية: الفحص كل 10 إلى 15 دقيقة.

يمكنك الحصول على حصة اختبار مجانية من خلال منصة APIYI apiyi.com للتحقق بسرعة من جدوى خطة المراقبة الخاصة بك. احرص على موازنة تكاليف المراقبة مع سرعة الاستجابة المطلوبة.

نوصي باتباع الاستراتيجيات التالية:

1. عتبة الفشل المتتالي: لا تقم بتفعيل التنبيه إلا بعد حدوث 3 إخفاقات متتالية على الأقل.
2. النافذة المنزلقة: اعتبار العطل حقيقياً إذا تجاوز معدل الفشل 50% خلال 5 دقائق.
3. التحقق متعدد المصادر: تحقق من صفحة الحالة الرسمية في نفس الوقت للتأكد مما إذا كان العطل من المصدر نفسه.
4. استبعاد حدود المعدل (Rate Limit): تجاوز حدود الاستخدام لا يعني بالضرورة وجود عطل في النظام.

من خلال لوحة مراقبة منصة APIYI apiyi.com، يمكنك عرض اتجاهات التوفر التاريخية بشكل مرئي للحكم بسرعة على ما إذا كان العطل نظامياً أم لا.

باعتبار OpenRouter منصة تجميع معروفة لنماذج الذكاء الاصطناعي، فإن بيانات المراقبة الخاصة بها لها قيمة مرجعية:

• المزايا: تجمع بيانات طلبات المستخدمين من جميع أنحاء العالم، مما يوفر عينة كبيرة.
• القيود: قد تتأثر البيانات بمشاكل في الربط الخاص بمنصة OpenRouter نفسها.
• النصيحة: اجمع بين صفحة الحالة الرسمية وبين المراقبة التي قمت ببنائها بنفسك للوصول إلى حكم شامل.

إجهاد التنبيهات مشكلة شائعة، وننصح بما يلي:

1. تحديد مستويات التنبيه: التمييز بين (معلومات Info، تحذير Warning، حرج Critical).
2. إعداد فترة التهدئة: إرسال التنبيه مرة واحدة فقط لنفس المشكلة خلال 30 دقيقة.
3. تجميع التنبيهات: دمج المشاكل المتشابهة في تنبيه واحد.
4. كتم التنبيهات: إيقاف التنبيهات مؤقتاً أثناء فترات الصيانة.
5. التحسين المستمر: مراجعة التنبيهات الكاذبة بانتظام وتعديل العتبات.