archive/monitor.py

"""
API Health Monitoring Engine
Async health checks with retry logic, caching, and metrics tracking
"""

import asyncio
import aiohttp
import time
import logging
from typing import Dict, List, Tuple, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
from enum import Enum

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class HealthStatus(Enum):
    """Health status enumeration"""
    ONLINE = "online"
    DEGRADED = "degraded"
    OFFLINE = "offline"
    UNKNOWN = "unknown"


@dataclass
class HealthCheckResult:
    """Result of a health check"""
    provider_name: str
    category: str
    status: HealthStatus
    response_time: float  # in milliseconds
    status_code: Optional[int] = None
    error_message: Optional[str] = None
    timestamp: float = None
    endpoint_tested: str = ""

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = time.time()

    def to_dict(self) -> Dict:
        """Convert to dictionary"""
        d = asdict(self)
        d['status'] = self.status.value
        d['timestamp_human'] = datetime.fromtimestamp(self.timestamp).strftime('%Y-%m-%d %H:%M:%S')
        return d

    def get_badge(self) -> str:
        """Get emoji badge for status"""
        badges = {
            HealthStatus.ONLINE: "🟢",
            HealthStatus.DEGRADED: "🟡",
            HealthStatus.OFFLINE: "🔴",
            HealthStatus.UNKNOWN: "⚪"
        }
        return badges.get(self.status, "⚪")


class APIMonitor:
    """Asynchronous API health monitor"""

    def __init__(self, config, timeout: int = 10, max_concurrent: int = 10):
        self.config = config
        self.timeout = timeout
        self.max_concurrent = max_concurrent
        self.cache = {}  # Simple in-memory cache
        self.cache_ttl = 60  # 1 minute cache
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results_history = []  # Store recent results

    async def check_endpoint(
        self,
        resource: Dict,
        use_proxy: bool = False,
        proxy_index: int = 0
    ) -> HealthCheckResult:
        """Check a single endpoint health"""
        provider_name = resource.get('name', 'Unknown')
        category = resource.get('category', 'Other')

        # Check cache first
        cache_key = f"{provider_name}:{category}"
        if cache_key in self.cache:
            cached_result, cache_time = self.cache[cache_key]
            if time.time() - cache_time < self.cache_ttl:
                logger.debug(f"Cache hit for {provider_name}")
                return cached_result

        # Construct URL
        url = resource.get('url', '')
        endpoint = resource.get('endpoint', '')
        test_url = f"{url}{endpoint}" if endpoint else url

        # Add API key if available
        api_key = resource.get('key', '')
        if not api_key:
            # Try to get from config
            key_name = provider_name.lower().replace(' ', '').replace('(', '').replace(')', '')
            api_key = self.config.get_api_key(key_name)

        # Apply proxy if needed
        if use_proxy:
            proxy_url = self.config.get_cors_proxy(proxy_index)
            if 'allorigins' in proxy_url:
                test_url = f"{proxy_url}{test_url}"
            else:
                test_url = f"{proxy_url}{test_url}"

        start_time = time.time()

        try:
            async with self.semaphore:
                async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=self.timeout)) as session:
                    headers = {}

                    # Add API key to headers if available
                    if api_key:
                        if 'coinmarketcap' in provider_name.lower():
                            headers['X-CMC_PRO_API_KEY'] = api_key
                        elif 'etherscan' in provider_name.lower() or 'bscscan' in provider_name.lower():
                            # Add as query parameter instead
                            separator = '&' if '?' in test_url else '?'
                            test_url = f"{test_url}{separator}apikey={api_key}"

                    async with session.get(test_url, headers=headers, ssl=False) as response:
                        response_time = (time.time() - start_time) * 1000  # Convert to ms
                        status_code = response.status

                        # Determine health status
                        if status_code == 200:
                            # Try to parse JSON to ensure valid response
                            try:
                                data = await response.json()
                                if data:
                                    status = HealthStatus.ONLINE
                                else:
                                    status = HealthStatus.DEGRADED
                            except:
                                status = HealthStatus.DEGRADED
                        elif 200 < status_code < 300:
                            status = HealthStatus.ONLINE
                        elif 400 <= status_code < 500:
                            status = HealthStatus.DEGRADED
                        else:
                            status = HealthStatus.OFFLINE

                        result = HealthCheckResult(
                            provider_name=provider_name,
                            category=category,
                            status=status,
                            response_time=response_time,
                            status_code=status_code,
                            endpoint_tested=test_url[:100]  # Truncate long URLs
                        )

        except asyncio.TimeoutError:
            response_time = (time.time() - start_time) * 1000
            result = HealthCheckResult(
                provider_name=provider_name,
                category=category,
                status=HealthStatus.OFFLINE,
                response_time=response_time,
                error_message="Timeout",
                endpoint_tested=test_url[:100]
            )

        except Exception as e:
            response_time = (time.time() - start_time) * 1000
            result = HealthCheckResult(
                provider_name=provider_name,
                category=category,
                status=HealthStatus.OFFLINE,
                response_time=response_time,
                error_message=str(e)[:200],  # Truncate long errors
                endpoint_tested=test_url[:100]
            )
            logger.error(f"Error checking {provider_name}: {e}")

        # Cache the result
        self.cache[cache_key] = (result, time.time())

        # Add to history
        self.results_history.append(result)
        # Keep only last 1000 results
        if len(self.results_history) > 1000:
            self.results_history = self.results_history[-1000:]

        return result

    async def check_all(
        self,
        resources: Optional[List[Dict]] = None,
        use_proxy: bool = False
    ) -> List[HealthCheckResult]:
        """Check all endpoints"""
        if resources is None:
            resources = self.config.get_all_resources()

        logger.info(f"Checking {len(resources)} endpoints...")

        # Create tasks with stagger to avoid overwhelming APIs
        tasks = []
        for i, resource in enumerate(resources):
            # Stagger requests by 0.1 seconds each
            await asyncio.sleep(0.1)
            task = asyncio.create_task(self.check_endpoint(resource, use_proxy))
            tasks.append(task)

        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Filter out exceptions
        valid_results = []
        for result in results:
            if isinstance(result, HealthCheckResult):
                valid_results.append(result)
            elif isinstance(result, Exception):
                logger.error(f"Task failed with exception: {result}")

        logger.info(f"Completed {len(valid_results)} checks")
        return valid_results

    async def check_by_category(
        self,
        category: str,
        use_proxy: bool = False
    ) -> List[HealthCheckResult]:
        """Check all endpoints in a category"""
        resources = self.config.get_by_category(category)
        return await self.check_all(resources, use_proxy)

    async def check_single(
        self,
        provider_name: str,
        use_proxy: bool = False
    ) -> Optional[HealthCheckResult]:
        """Check a single provider by name"""
        resources = self.config.get_all_resources()
        resource = next((r for r in resources if r.get('name') == provider_name), None)

        if resource:
            return await self.check_endpoint(resource, use_proxy)
        return None

    def get_summary_stats(self, results: List[HealthCheckResult]) -> Dict:
        """Calculate summary statistics from results"""
        if not results:
            return {
                'total': 0,
                'online': 0,
                'degraded': 0,
                'offline': 0,
                'unknown': 0,
                'online_percentage': 0,
                'avg_response_time': 0,
                'critical_issues': 0
            }

        online = sum(1 for r in results if r.status == HealthStatus.ONLINE)
        degraded = sum(1 for r in results if r.status == HealthStatus.DEGRADED)
        offline = sum(1 for r in results if r.status == HealthStatus.OFFLINE)
        unknown = sum(1 for r in results if r.status == HealthStatus.UNKNOWN)

        response_times = [r.response_time for r in results if r.response_time]
        avg_response_time = sum(response_times) / len(response_times) if response_times else 0

        # Critical issues: Tier 1 APIs that are offline
        critical_issues = sum(
            1 for r in results
            if r.status == HealthStatus.OFFLINE and self._is_tier1(r.provider_name)
        )

        return {
            'total': len(results),
            'online': online,
            'degraded': degraded,
            'offline': offline,
            'unknown': unknown,
            'online_percentage': round((online / len(results)) * 100, 2) if results else 0,
            'avg_response_time': round(avg_response_time, 2),
            'critical_issues': critical_issues
        }

    def _is_tier1(self, provider_name: str) -> bool:
        """Check if provider is Tier 1"""
        resources = self.config.get_all_resources()
        resource = next((r for r in resources if r.get('name') == provider_name), None)
        return resource.get('tier', 3) == 1 if resource else False

    def get_category_stats(self, results: List[HealthCheckResult]) -> Dict[str, Dict]:
        """Get statistics grouped by category"""
        category_results = {}

        for result in results:
            category = result.category
            if category not in category_results:
                category_results[category] = []
            category_results[category].append(result)

        return {
            category: self.get_summary_stats(cat_results)
            for category, cat_results in category_results.items()
        }

    def get_recent_history(self, hours: int = 24) -> List[HealthCheckResult]:
        """Get recent history within specified hours"""
        cutoff_time = time.time() - (hours * 3600)
        return [r for r in self.results_history if r.timestamp >= cutoff_time]

    def clear_cache(self):
        """Clear the results cache"""
        self.cache.clear()
        logger.info("Cache cleared")

    def get_uptime_percentage(
        self,
        provider_name: str,
        hours: int = 24
    ) -> float:
        """Calculate uptime percentage for a provider"""
        recent = self.get_recent_history(hours)
        provider_results = [r for r in recent if r.provider_name == provider_name]

        if not provider_results:
            return 0.0

        online_count = sum(1 for r in provider_results if r.status == HealthStatus.ONLINE)
        return round((online_count / len(provider_results)) * 100, 2)


# Convenience function for synchronous usage
def check_all_sync(config, use_proxy: bool = False) -> List[HealthCheckResult]:
    """Synchronous wrapper for checking all endpoints"""
    monitor = APIMonitor(config)
    return asyncio.run(monitor.check_all(use_proxy=use_proxy))