import requests import json import time from datetime import datetime, timedelta from typing import List, Optional, Dict, Any, Tuple from requests.exceptions import RequestException # --- Constantes de API (HARDCODED) --- AUTH_URL = 'https://api.oplab.com.br/v3/domain/users/authenticate' TARGET_URL_BASE = 'https://api.oplab.com.br/v3/market/instruments' # Credenciais (ATENÇÃO: Armazenar credenciais diretamente no código é INSEGURO!) # Estas credenciais são baseadas no seu exemplo Delphi, mas devem ser carregadas de forma segura. AUTH_PAYLOAD = { "email": "jrussi@gmail.com", "password": "jkiller77" } # --- Classes de Suporte (Conversão de Delphi/Pascal para Python) --- class TApiResponse: """Simula o record de resposta da API.""" def __init__(self): self.Success: bool = False self.ErrorMessage: str = "" self.Content: str = "" self.NeedsReAuthentication: bool = False class TOplabStockData: """Simula o record de dados de cotação (simplificado para o uso em Estrutura.py).""" def __init__(self): self.Symbol: str = "" self.Strike: float = 0.0 self.Vencimento: int = 0 self.Tipo: str = "ativo" # 'call', 'put', 'ativo' self.AtivoBase: str = "" self.ValorAtivo: float = 0.0 # Preço do ativo-objeto (spot_price ou Ask) self.Sigma: float = 0.0 # Volatilidade (stdv_1y) self.Bid: float = 0.0 self.Ask: float = 0.0 self.Close: float = 0.0 self.DueDate: str = "" def __repr__(self): return f"TOplabStockData(Symbol='{self.Symbol}', Strike={self.Strike}, Vencimento={self.Vencimento}, Tipo='{self.Tipo}', Sigma={self.Sigma:.2f})" # --- Classe TOpLab (Versão Python Simplificada) --- class TOpLab: def __init__(self): self.Token: str = "" self.Initialized: bool = False self.FMaxRetries: int = 1 # Reduzido para 1 para simplicidade/velocidade (o Delphi usava 3) self.FRetryDelayMs: int = 1000 self.FLastAuthAttempt: Optional[datetime] = None # Mantido, mas a lógica será removida # Headers padrão. O Python lida com compressão automaticamente. # self.headers: Dict[str, str] = { # 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', # Usando User-Agent de navegador para ser menos detectável # 'Accept': 'application/json', # 'Content-Type': 'application/json' # CRUCIAL para POST JSON # } def Create(self) -> 'TOpLab': # Em Python, o construtor __init__ é o que cria. return self def Destroy(self): # Em Python, a destruição é gerenciada pelo GC. pass def _ProtectedRequest(self, url: str, method: str, data: Optional[Dict[str, Any]], response_obj: TApiResponse, authenticated: bool = True) -> bool: """Executa a requisição HTTP com headers e lógica de retries.""" # Headers padrão. Recriado localmente para segurança. headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'application/json', 'Content-Type': 'application/json' } # --- CORREÇÃO CRÍTICA AQUI --- # A API Oplab usa o cabeçalho 'Access-Token' em vez de 'Authorization: Bearer' if authenticated and self.Token: headers['Access-Token'] = self.Token # <--- CORRIGIDO # Lógica simplificada de Retries (apenas para falhas de conexão/timeout) for attempt in range(self.FMaxRetries): try: if method == 'GET': response = requests.get(url, headers=headers, timeout=30) elif method == 'POST': response = requests.post(url, headers=headers, json=data, timeout=30) else: response_obj.ErrorMessage = f"Método {method} não suportado." return False response_obj.Content = response.text response_obj.Success = response.status_code == 200 response_obj.NeedsReAuthentication = response.status_code == 401 if response_obj.Success: return True if response.status_code == 401: response_obj.ErrorMessage = "Token expirado ou inválido (401)." return False # Se não for sucesso e não for 401, tentar novamente em caso de erro transiente response_obj.ErrorMessage = f"Erro HTTP {response.status_code}: {response.reason}" if attempt < self.FMaxRetries - 1 and response.status_code not in (400, 403, 404): time.sleep(self.FRetryDelayMs / 1000) except RequestException as e: response_obj.ErrorMessage = f"Erro de Conexão: {e}" if attempt < self.FMaxRetries - 1: time.sleep(self.FRetryDelayMs / 1000) return False def DoAuthenticate(self, auth_response: TApiResponse): """Implementação mais direta da autenticação (replicando o fluxo do Delphi).""" # REMOVENDO A LÓGICA DE TIMING PARA SIMPLIFICAR E FORÇAR A TENTATIVA self.FLastAuthAttempt = datetime.now() # Apenas registra a tentativa payload = AUTH_PAYLOAD # Sem autenticação no header para esta chamada (authenticated=False) if self._ProtectedRequest(AUTH_URL, 'POST', payload, auth_response, authenticated=False): try: data = json.loads(auth_response.Content) token = data.get('access-token') if token: self.Token = token self.Initialized = True auth_response.Success = True else: self.Token = '' self.Initialized = False auth_response.Success = False # Se falhar e retornar 200, é porque a API retornou um erro em JSON error_msg = data.get('message', 'Token de acesso não encontrado.') auth_response.ErrorMessage = error_msg except json.JSONDecodeError: auth_response.ErrorMessage = 'Falha ao parsear JSON da resposta de autenticação.' else: self.Token = '' self.Initialized = False # O erro já está em auth_response.ErrorMessage do _ProtectedRequest print('DoAuthenticate: ' + ('Sucesso' if auth_response.Success else f'Falha - {auth_response.ErrorMessage}')) def Authenticate(self): """Método público de autenticação.""" response = TApiResponse() self.DoAuthenticate(response) return response def CotarSymbolsplus(self, symbols: str) -> List[TOplabStockData]: """ Simula o método CotarSymbolsPlus, lidando com autenticação e parsing JSON. Retorna uma lista de TOplabStockData. """ stock_data: List[TOplabStockData] = [] auth_attempted = False target_url = f'{TARGET_URL_BASE}?tickers={symbols}' while True: # 1. Lógica de Autenticação/Inicialização if not self.Initialized and not auth_attempted: auth_response = TApiResponse() self.DoAuthenticate(auth_response) auth_attempted = True if not auth_response.Success: print(f'CotarSymbolsPlus: Falha autenticação: {auth_response.ErrorMessage}') return [] elif not self.Initialized and auth_attempted: print('CotarSymbolsPlus: Autenticação falhou.') return [] # 2. Requisição Protegida api_response = TApiResponse() if self._ProtectedRequest(target_url, 'GET', None, api_response, authenticated=True): # Requisição bem-sucedida try: json_array: List[Dict[str, Any]] = json.loads(api_response.Content) #print(json_array) for item in json_array: stock_item = TOplabStockData() # Mapeamento e Preenchimento dos Dados stock_item.Symbol = item.get('symbol', '') stock_item.Close = item.get('close', 0.0) stock_item.Strike = item.get('strike', 0.0) or 0.0 # Trata null como 0.0 stock_item.Vencimento = item.get('days_to_maturity', 0) stock_item.Bid = item.get('bid', 0.0) stock_item.Ask = item.get('ask', 0.0) stock_item.DueDate = item.get('due_date', '') # Tipo e Ativo Base category = item.get('category', '').lower().replace('vista', 'ativo') stock_item.Tipo = category item_type = item.get('type', '') if item_type == 'OPTION': stock_item.AtivoBase = item.get('parent_symbol', '') stock_item.ValorAtivo = item.get('spot_price', 0.0) elif item_type == 'STOCK': stock_item.AtivoBase = stock_item.Symbol stock_item.ValorAtivo = stock_item.Ask stock_item.Sigma = item.get('stdv_1y', 0.0) # Volatilidade para Ativos stock_data.append(stock_item) #print('\n\nStockdata:',stock_data) #print(f"DEBUG: {stock_data[1].Symbol} - Bid: {stock_data[1].Bid}, Ask: {stock_data[1].Ask}, Due: {stock_data[1].DueDate}") return stock_data except json.JSONDecodeError: print(f'CotarSymbolsPlus: Erro parse JSON. JSON: {api_response.Content[:100]}...') return [] # 3. Lógica de Reautenticação else: if api_response.NeedsReAuthentication and not auth_attempted: print('CotarSymbolsPlus: Requer reautenticação. Tentando novamente.') self.Initialized = False auth_attempted = True else: print(f'CotarSymbolsPlus: Falha para {symbols}. Erro: {api_response.ErrorMessage}') return [] # --- Exemplo de Uso (para testar o módulo) --- if __name__ == '__main__': oplab = TOpLab() # 1. Tenta autenticar auth_result = oplab.Authenticate() print(f"\nAutenticação: {'Sucesso' if auth_result.Success else 'Falha'}. Token: {'SIM' if oplab.Token else 'NÃO'}") if oplab.Initialized: # 2. Tenta cotar símbolos symbols_to_quote = "PETR4,BOVAM130,BOVA11" print(f"\nCotando símbolos: {symbols_to_quote}") quoted_data = oplab.CotarSymbolsplus(symbols_to_quote) print("\n--- Dados de Cotação Recebidos ---") if quoted_data: for item in quoted_data: print(f"Symbol: {item.Symbol}, Strike: {item.Strike:.2f}, Venc: {item.Vencimento} dias, AtivoBase: {item.AtivoBase}, Sigma: {item.Sigma:.4f}") else: print("Nenhum dado recebido.")