#!/usr/bin/env python3 """ Provider Manager - مدیریت ارائه‌دهندگان API و استراتژی‌های Rotation """ import json import asyncio import aiohttp import time from typing import Dict, List, Optional, Any from dataclasses import dataclass, field from datetime import datetime from enum import Enum import random class ProviderStatus(Enum): """وضعیت ارائه‌دهنده""" ONLINE = "online" OFFLINE = "offline" DEGRADED = "degraded" RATE_LIMITED = "rate_limited" class RotationStrategy(Enum): """استراتژی‌های چرخش""" ROUND_ROBIN = "round_robin" PRIORITY = "priority" WEIGHTED = "weighted" LEAST_USED = "least_used" FASTEST_RESPONSE = "fastest_response" @dataclass(init=False) class RateLimitInfo: """اطلاعات محدودیت نرخ""" requests_per_second: Optional[int] = None requests_per_minute: Optional[int] = None requests_per_hour: Optional[int] = None requests_per_day: Optional[int] = None requests_per_week: Optional[int] = None requests_per_month: Optional[int] = None weight_per_minute: Optional[int] = None current_usage: int = 0 reset_time: Optional[float] = None extra_limits: Dict[str, Any] = field(default_factory=dict) def __init__( self, requests_per_second: Optional[int] = None, requests_per_minute: Optional[int] = None, requests_per_hour: Optional[int] = None, requests_per_day: Optional[int] = None, requests_per_week: Optional[int] = None, requests_per_month: Optional[int] = None, weight_per_minute: Optional[int] = None, current_usage: int = 0, reset_time: Optional[float] = None, **extra: Any, ): self.requests_per_second = requests_per_second self.requests_per_minute = requests_per_minute self.requests_per_hour = requests_per_hour self.requests_per_day = requests_per_day self.requests_per_week = requests_per_week self.requests_per_month = requests_per_month self.weight_per_minute = weight_per_minute self.current_usage = current_usage self.reset_time = reset_time self.extra_limits = extra @classmethod def from_dict(cls, data: Optional[Dict[str, Any]]) -> "RateLimitInfo": """ساخت نمونه از دیکشنری و مدیریت کلیدهای ناشناخته.""" if isinstance(data, cls): return data if not data: return cls() return cls(**data) def is_limited(self) -> bool: """بررسی محدودیت نرخ""" now = time.time() if self.reset_time and now < self.reset_time: if self.requests_per_second and self.current_usage >= self.requests_per_second: return True if self.requests_per_minute and self.current_usage >= self.requests_per_minute: return True if self.requests_per_hour and self.current_usage >= self.requests_per_hour: return True if self.requests_per_day and self.current_usage >= self.requests_per_day: return True return False def increment(self): """افزایش شمارنده استفاده""" self.current_usage += 1 @dataclass class Provider: """کلاس ارائه‌دهنده API""" provider_id: str name: str category: str base_url: str endpoints: Dict[str, str] rate_limit: RateLimitInfo requires_auth: bool = False priority: int = 5 weight: int = 50 status: ProviderStatus = ProviderStatus.ONLINE # آمار total_requests: int = 0 successful_requests: int = 0 failed_requests: int = 0 avg_response_time: float = 0.0 last_check: Optional[datetime] = None last_error: Optional[str] = None # Circuit Breaker consecutive_failures: int = 0 circuit_breaker_open: bool = False circuit_breaker_open_until: Optional[float] = None def __post_init__(self): """مقداردهی اولیه""" if isinstance(self.rate_limit, dict): self.rate_limit = RateLimitInfo.from_dict(self.rate_limit) elif not isinstance(self.rate_limit, RateLimitInfo): self.rate_limit = RateLimitInfo() @property def success_rate(self) -> float: """نرخ موفقیت""" if self.total_requests == 0: return 100.0 return (self.successful_requests / self.total_requests) * 100 @property def is_available(self) -> bool: """آیا ارائه‌دهنده در دسترس است؟""" # بررسی Circuit Breaker if self.circuit_breaker_open: if self.circuit_breaker_open_until and time.time() > self.circuit_breaker_open_until: self.circuit_breaker_open = False self.consecutive_failures = 0 else: return False # بررسی محدودیت نرخ if self.rate_limit and self.rate_limit.is_limited(): self.status = ProviderStatus.RATE_LIMITED return False # بررسی وضعیت return self.status in [ProviderStatus.ONLINE, ProviderStatus.DEGRADED] def record_success(self, response_time: float): """ثبت درخواست موفق""" self.total_requests += 1 self.successful_requests += 1 self.consecutive_failures = 0 # محاسبه میانگین متحرک زمان پاسخ if self.avg_response_time == 0: self.avg_response_time = response_time else: self.avg_response_time = (self.avg_response_time * 0.8) + (response_time * 0.2) self.status = ProviderStatus.ONLINE self.last_check = datetime.now() if self.rate_limit: self.rate_limit.increment() def record_failure(self, error: str, circuit_breaker_threshold: int = 5): """ثبت درخواست ناموفق""" self.total_requests += 1 self.failed_requests += 1 self.consecutive_failures += 1 self.last_error = error self.last_check = datetime.now() # فعال‌سازی Circuit Breaker if self.consecutive_failures >= circuit_breaker_threshold: self.circuit_breaker_open = True self.circuit_breaker_open_until = time.time() + 60 # ۶۰ ثانیه self.status = ProviderStatus.OFFLINE else: self.status = ProviderStatus.DEGRADED @dataclass class ProviderPool: """استخر ارائه‌دهندگان با استراتژی چرخش""" pool_id: str pool_name: str category: str rotation_strategy: RotationStrategy providers: List[Provider] = field(default_factory=list) current_index: int = 0 enabled: bool = True total_rotations: int = 0 def add_provider(self, provider: Provider): """افزودن ارائه‌دهنده به استخر""" if provider not in self.providers: self.providers.append(provider) # مرتب‌سازی بر اساس اولویت if self.rotation_strategy == RotationStrategy.PRIORITY: self.providers.sort(key=lambda p: p.priority, reverse=True) def remove_provider(self, provider_id: str): """حذف ارائه‌دهنده از استخر""" self.providers = [p for p in self.providers if p.provider_id != provider_id] def get_next_provider(self) -> Optional[Provider]: """دریافت ارائه‌دهنده بعدی بر اساس استراتژی""" if not self.providers or not self.enabled: return None # فیلتر ارائه‌دهندگان در دسترس available = [p for p in self.providers if p.is_available] if not available: return None provider = None if self.rotation_strategy == RotationStrategy.ROUND_ROBIN: provider = self._round_robin(available) elif self.rotation_strategy == RotationStrategy.PRIORITY: provider = self._priority_based(available) elif self.rotation_strategy == RotationStrategy.WEIGHTED: provider = self._weighted_random(available) elif self.rotation_strategy == RotationStrategy.LEAST_USED: provider = self._least_used(available) elif self.rotation_strategy == RotationStrategy.FASTEST_RESPONSE: provider = self._fastest_response(available) if provider: self.total_rotations += 1 return provider def _round_robin(self, available: List[Provider]) -> Provider: """چرخش Round Robin""" provider = available[self.current_index % len(available)] self.current_index += 1 return provider def _priority_based(self, available: List[Provider]) -> Provider: """بر اساس اولویت""" return max(available, key=lambda p: p.priority) def _weighted_random(self, available: List[Provider]) -> Provider: """انتخاب تصادفی وزن‌دار""" weights = [p.weight for p in available] return random.choices(available, weights=weights, k=1)[0] def _least_used(self, available: List[Provider]) -> Provider: """کمترین استفاده شده""" return min(available, key=lambda p: p.total_requests) def _fastest_response(self, available: List[Provider]) -> Provider: """سریع‌ترین پاسخ""" return min(available, key=lambda p: p.avg_response_time if p.avg_response_time > 0 else float('inf')) def get_stats(self) -> Dict[str, Any]: """آمار استخر""" total_providers = len(self.providers) available_providers = len([p for p in self.providers if p.is_available]) return { "pool_id": self.pool_id, "pool_name": self.pool_name, "category": self.category, "rotation_strategy": self.rotation_strategy.value, "total_providers": total_providers, "available_providers": available_providers, "total_rotations": self.total_rotations, "enabled": self.enabled, "providers": [ { "provider_id": p.provider_id, "name": p.name, "status": p.status.value, "success_rate": p.success_rate, "total_requests": p.total_requests, "avg_response_time": p.avg_response_time, "is_available": p.is_available } for p in self.providers ] } class ProviderManager: """مدیر ارائه‌دهندگان""" def __init__(self, config_path: str = "providers_config_extended.json"): self.config_path = config_path self.providers: Dict[str, Provider] = {} self.pools: Dict[str, ProviderPool] = {} self.session: Optional[aiohttp.ClientSession] = None self.load_config() def load_config(self): """بارگذاری پیکربندی از فایل JSON""" try: with open(self.config_path, 'r', encoding='utf-8') as f: config = json.load(f) # بارگذاری ارائه‌دهندگان for provider_id, provider_data in config.get('providers', {}).items(): rate_limit_data = provider_data.get('rate_limit', {}) rate_limit = RateLimitInfo.from_dict(rate_limit_data) provider = Provider( provider_id=provider_id, name=provider_data['name'], category=provider_data['category'], base_url=provider_data['base_url'], endpoints=provider_data.get('endpoints', {}), rate_limit=rate_limit, requires_auth=provider_data.get('requires_auth', False), priority=provider_data.get('priority', 5), weight=provider_data.get('weight', 50) ) self.providers[provider_id] = provider # بارگذاری Pool‌ها for pool_config in config.get('pool_configurations', []): pool_id = pool_config['pool_name'].lower().replace(' ', '_') pool = ProviderPool( pool_id=pool_id, pool_name=pool_config['pool_name'], category=pool_config['category'], rotation_strategy=RotationStrategy(pool_config['rotation_strategy']) ) # افزودن ارائه‌دهندگان به Pool for provider_id in pool_config.get('providers', []): if provider_id in self.providers: pool.add_provider(self.providers[provider_id]) self.pools[pool_id] = pool print(f"✅ بارگذاری موفق: {len(self.providers)} ارائه‌دهنده، {len(self.pools)} استخر") except FileNotFoundError: print(f"❌ خطا: فایل {self.config_path} یافت نشد") except Exception as e: print(f"❌ خطا در بارگذاری پیکربندی: {e}") async def init_session(self): """مقداردهی اولیه HTTP Session""" if not self.session: timeout = aiohttp.ClientTimeout(total=10) self.session = aiohttp.ClientSession(timeout=timeout) async def close_session(self): """بستن HTTP Session""" if self.session: await self.session.close() self.session = None async def health_check(self, provider: Provider) -> bool: """بررسی سلامت ارائه‌دهنده""" await self.init_session() # انتخاب اولین endpoint برای تست if not provider.endpoints: return False endpoint = list(provider.endpoints.values())[0] url = f"{provider.base_url}{endpoint}" start_time = time.time() try: async with self.session.get(url) as response: response_time = (time.time() - start_time) * 1000 # میلی‌ثانیه if response.status == 200: provider.record_success(response_time) return True else: provider.record_failure(f"HTTP {response.status}") return False except asyncio.TimeoutError: provider.record_failure("Timeout") return False except Exception as e: provider.record_failure(str(e)) return False async def health_check_all(self, silent: bool = False): """بررسی سلامت همه ارائه‌دهندگان""" tasks = [self.health_check(provider) for provider in self.providers.values()] results = await asyncio.gather(*tasks, return_exceptions=True) online = sum(1 for r in results if r is True) if not silent: print(f"✅ بررسی سلامت: {online}/{len(self.providers)} ارائه‌دهنده آنلاین") return online, len(self.providers) def get_provider(self, provider_id: str) -> Optional[Provider]: """دریافت ارائه‌دهنده با ID""" return self.providers.get(provider_id) def get_pool(self, pool_id: str) -> Optional[ProviderPool]: """دریافت Pool با ID""" return self.pools.get(pool_id) def get_next_from_pool(self, pool_id: str) -> Optional[Provider]: """دریافت ارائه‌دهنده بعدی از Pool""" pool = self.get_pool(pool_id) if pool: return pool.get_next_provider() return None def get_all_stats(self) -> Dict[str, Any]: """آمار کامل سیستم""" total_providers = len(self.providers) online_providers = len([p for p in self.providers.values() if p.status == ProviderStatus.ONLINE]) offline_providers = len([p for p in self.providers.values() if p.status == ProviderStatus.OFFLINE]) degraded_providers = len([p for p in self.providers.values() if p.status == ProviderStatus.DEGRADED]) total_requests = sum(p.total_requests for p in self.providers.values()) successful_requests = sum(p.successful_requests for p in self.providers.values()) return { "summary": { "total_providers": total_providers, "online": online_providers, "offline": offline_providers, "degraded": degraded_providers, "total_requests": total_requests, "successful_requests": successful_requests, "overall_success_rate": (successful_requests / total_requests * 100) if total_requests > 0 else 0 }, "providers": { provider_id: { "name": p.name, "category": p.category, "status": p.status.value, "success_rate": p.success_rate, "total_requests": p.total_requests, "avg_response_time": p.avg_response_time, "is_available": p.is_available, "priority": p.priority, "weight": p.weight } for provider_id, p in self.providers.items() }, "pools": { pool_id: pool.get_stats() for pool_id, pool in self.pools.items() } } def export_stats(self, filepath: str = "provider_stats.json"): """صادرکردن آمار به فایل JSON""" stats = self.get_all_stats() with open(filepath, 'w', encoding='utf-8') as f: json.dump(stats, f, indent=2, ensure_ascii=False) print(f"✅ آمار در {filepath} ذخیره شد") # تست و نمونه استفاده async def main(): """تابع اصلی برای تست""" manager = ProviderManager() print("\n📊 بررسی سلامت ارائه‌دهندگان...") await manager.health_check_all() print("\n🔄 تست Pool چرخشی...") pool = manager.get_pool("primary_market_data_pool") if pool: for i in range(5): provider = pool.get_next_provider() if provider: print(f" Round {i+1}: {provider.name}") print("\n📈 آمار کلی:") stats = manager.get_all_stats() summary = stats['summary'] print(f" کل: {summary['total_providers']}") print(f" آنلاین: {summary['online']}") print(f" آفلاین: {summary['offline']}") print(f" نرخ موفقیت: {summary['overall_success_rate']:.2f}%") # صادرکردن آمار manager.export_stats() await manager.close_session() print("\n✅ اتمام") if __name__ == "__main__": asyncio.run(main())