File size: 5,844 Bytes
bb5a7c6 2b09552 bb5a7c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """
AnyAPIClient - Cliente centralizado para a API AnyAPI (anyapi.io)
Gerencia autenticação, rotação de contas e chamadas HTTP.
"""
import requests
import logging
from typing import Any, Dict, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
_BASE_URL = "https://api.anyapi.io/v1"
class AnyAPIClient:
"""Cliente HTTP para AnyAPI com rotação automática de chaves."""
_instance: Optional["AnyAPIClient"] = None
def __new__(cls) -> "AnyAPIClient":
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self) -> None:
if self._initialized:
return
self._initialized = True
self._rotation = None
self._session = requests.Session()
self._session.headers.update({
"Content-Type": "application/json",
"User-Agent": "AKIRA-SOFTEDGE/21"
})
logger.info("AnyAPIClient inicializado")
@property
def rotation(self):
if self._rotation is None:
from modules.anyapi_rotation import get_anyapi_rotation
self._rotation = get_anyapi_rotation()
return self._rotation
def get_raw(self, endpoint: str, params: Optional[Dict[str, Any]] = None, timeout: int = 10) -> Dict[str, Any]:
"""
Realiza GET na AnyAPI e retorna conteúdo binário (imagem, PDF, etc).
Args:
endpoint: caminho relativo (ex: '/v1/screenshot')
params: query params
timeout: timeout em segundos
Returns:
dict com 'sucesso', 'dados' (bytes do conteúdo binário) e metadados
"""
api_key = self.rotation.get_current_api_key()
if not api_key:
return self._error_response("Nenhuma chave AnyAPI disponível")
url = f"{_BASE_URL}{endpoint}"
query = params or {}
query["apikey"] = api_key
try:
resp = self._session.get(url, params=query, timeout=timeout)
if resp.status_code == 429:
rotated = self.rotation.handle_rate_limit_error()
if rotated:
return self.get_raw(endpoint, params, timeout)
return self._error_response("Rate limit em todas as contas AnyAPI")
resp.raise_for_status()
return {
"sucesso": True,
"dados": resp.content,
"content_type": resp.headers.get("Content-Type", "application/octet-stream"),
"provider": "anyapi",
"account": self.rotation.get_current_account_name(),
"timestamp": datetime.now().isoformat()
}
except requests.exceptions.Timeout:
logger.warning(f"AnyAPI timeout: {endpoint}")
return self._error_response(f"Timeout ao chamar {endpoint}")
except requests.exceptions.ConnectionError:
logger.error(f"AnyAPI connection error: {endpoint}")
return self._error_response("Erro de conexão com AnyAPI")
except requests.exceptions.HTTPError as e:
logger.error(f"AnyAPI HTTP error: {e}")
return self._error_response(f"Erro HTTP: {e.response.status_code}")
except Exception as e:
logger.error(f"AnyAPI erro inesperado: {type(e).__name__}: {e}")
return self._error_response(str(e))
def get(self, endpoint: str, params: Optional[Dict[str, Any]] = None, timeout: int = 10) -> Dict[str, Any]:
"""
Realiza GET na AnyAPI.
Args:
endpoint: caminho relativo (ex: '/v1/email')
params: query params
timeout: timeout em segundos
Returns:
dict com 'sucesso', 'dados' e metadados
"""
api_key = self.rotation.get_current_api_key()
if not api_key:
return self._error_response("Nenhuma chave AnyAPI disponível")
url = f"{_BASE_URL}{endpoint}"
query = params or {}
query["apikey"] = api_key
try:
resp = self._session.get(url, params=query, timeout=timeout)
if resp.status_code == 429:
rotated = self.rotation.handle_rate_limit_error()
if rotated:
return self.get(endpoint, params, timeout)
return self._error_response("Rate limit em todas as contas AnyAPI")
resp.raise_for_status()
data = resp.json()
return {
"sucesso": True,
"dados": data,
"provider": "anyapi",
"account": self.rotation.get_current_account_name(),
"timestamp": datetime.now().isoformat()
}
except requests.exceptions.Timeout:
logger.warning(f"AnyAPI timeout: {endpoint}")
return self._error_response(f"Timeout ao chamar {endpoint}")
except requests.exceptions.ConnectionError:
logger.error(f"AnyAPI connection error: {endpoint}")
return self._error_response("Erro de conexão com AnyAPI")
except requests.exceptions.HTTPError as e:
logger.error(f"AnyAPI HTTP error: {e}")
return self._error_response(f"Erro HTTP: {e.response.status_code}")
except Exception as e:
logger.error(f"AnyAPI erro inesperado: {type(e).__name__}: {e}")
return self._error_response(str(e))
def _error_response(self, message: str) -> Dict[str, Any]:
return {
"sucesso": False,
"erro": message,
"provider": "anyapi",
"account": self.rotation.get_current_account_name() if self.rotation else "nenhuma",
"timestamp": datetime.now().isoformat()
}
|