""" AnyAPI Rotation - Multi-account management Rotação automática entre múltiplas contas AnyAPI (https://anyapi.io) Contas disponíveis: 1. gitakira (GITAKIRAANY_API_KEY) 2. netflix (NETFLIXANY_API_KEY) """ import os import logging from typing import Optional, Dict from datetime import datetime, timedelta logger = logging.getLogger(__name__) class AnyAPIRotation: """Gerencia rotação de múltiplas contas AnyAPI""" def __init__(self): self.accounts = { 'gitakira': os.getenv('GITAKIRAANY_API_KEY', ''), 'netflix': os.getenv('NETFLIXANY_API_KEY', ''), } self.accounts = {k: v for k, v in self.accounts.items() if v} self.current_index = 0 self.account_names = list(self.accounts.keys()) self.rate_limit_cache: Dict[str, dict] = {} if self.account_names: logger.info(f"✅ AnyAPI Rotation inicializado com {len(self.account_names)} conta(s)") for i, name in enumerate(self.account_names, 1): logger.info(f" [{i}] {name.upper():12} ✅ ATIVA") else: logger.warning("⚠️ Nenhuma conta AnyAPI encontrada nas variáveis de ambiente") def get_current_account_name(self) -> str: if not self.account_names: return "nenhuma" return self.account_names[self.current_index] def get_current_api_key(self) -> Optional[str]: if not self.account_names: return None current_name = self.account_names[self.current_index] return self.accounts.get(current_name) def rotate_to_next(self): if not self.account_names: return old_index = self.current_index self.current_index = (self.current_index + 1) % len(self.account_names) old_name = self.account_names[old_index] new_name = self.account_names[self.current_index] logger.info(f"🔄 Rotacionando AnyAPI: '{old_name}' → '{new_name}'") def handle_rate_limit_error(self) -> bool: """ Trata erro 429 (rate limit) Retorna True se conseguiu rotacionar, False se todas contas limitadas """ if len(self.account_names) <= 1: logger.error("❌ [429] Apenas 1 conta AnyAPI disponível e limitada") return False current_name = self.account_names[self.current_index] logger.warning(f"⚠️ [429 RATE LIMIT] Conta AnyAPI '{current_name}' esgotada") self.rate_limit_cache[current_name] = { 'limited': True, 'until': datetime.now() + timedelta(minutes=10) } self.rotate_to_next() new_name = self.account_names[self.current_index] logger.info(f"✅ [429 RECOVERY] Mudando AnyAPI para '{new_name}'") return True def get_all_api_keys(self) -> Dict[str, str]: return self.accounts.copy() def is_account_limited(self, account_name: str) -> bool: if account_name not in self.rate_limit_cache: return False cache = self.rate_limit_cache[account_name] if datetime.now() > cache.get('until', datetime.now()): del self.rate_limit_cache[account_name] return False return cache.get('limited', False) _anyapi_rotation_instance: Optional[AnyAPIRotation] = None def get_anyapi_rotation() -> AnyAPIRotation: global _anyapi_rotation_instance if _anyapi_rotation_instance is None: _anyapi_rotation_instance = AnyAPIRotation() return _anyapi_rotation_instance def reset_anyapi_rotation(): global _anyapi_rotation_instance _anyapi_rotation_instance = None