135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
import requests
|
|
import json
|
|
import pyodbc
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
class RGBTokenClient:
|
|
"""Cliente para obter token JWT da API do Grupo Boticário (execução única)"""
|
|
|
|
def __init__(self):
|
|
self.base_url = "https://api.grupoboticario.com.br/global/v2/jwt-token/token"
|
|
self.client_id = "88ymKwAUNfu06sD85i0RiokCxWGSkFBkx9ytgI5y1ZKxX3OQ"
|
|
self.client_secret = "YDFz43qAzL6ApNIKVCxu3dAmS9GWOqJbcc2aPnFDkmEaBXexSpsHGfcItg56i2dE"
|
|
|
|
# Configurações do banco de dados
|
|
self.driver = self._get_available_sql_server_driver()
|
|
self.connection_string = (
|
|
f'DRIVER={self.driver};'
|
|
'SERVER=10.77.77.10;'
|
|
'DATABASE=GINSENG;'
|
|
'UID=supginseng;'
|
|
'PWD=Ginseng@;'
|
|
'PORT=1433;'
|
|
'TrustServerCertificate=yes;'
|
|
'Encrypt=yes'
|
|
)
|
|
|
|
def _get_available_sql_server_driver(self) -> str:
|
|
"""Detecta automaticamente o driver SQL Server disponível"""
|
|
drivers_to_try = [
|
|
'{ODBC Driver 18 for SQL Server}',
|
|
'{ODBC Driver 17 for SQL Server}',
|
|
'{ODBC Driver 13 for SQL Server}',
|
|
'{ODBC Driver 11 for SQL Server}',
|
|
'{SQL Server Native Client 11.0}',
|
|
'{SQL Server Native Client 10.0}',
|
|
'{SQL Server}'
|
|
]
|
|
|
|
available_drivers = pyodbc.drivers()
|
|
print("Drivers ODBC disponíveis:")
|
|
for d in available_drivers:
|
|
print(f" - {d}")
|
|
|
|
for preferred in drivers_to_try:
|
|
if preferred.strip("{}") in available_drivers:
|
|
print(f"✅ Usando driver: {preferred}")
|
|
return preferred
|
|
|
|
if available_drivers:
|
|
fallback = f"{{{available_drivers[0]}}}"
|
|
print(f"⚠️ Nenhum driver padrão encontrado. Usando: {fallback}")
|
|
return fallback
|
|
|
|
raise Exception("Nenhum driver ODBC encontrado no sistema.")
|
|
|
|
def get_token(self):
|
|
"""Obtém o token JWT da API"""
|
|
try:
|
|
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Solicitando novo token...")
|
|
|
|
response = requests.post(
|
|
self.base_url,
|
|
params={"grant_type": "client_credentials"},
|
|
data={
|
|
"client_id": self.client_id,
|
|
"client_secret": self.client_secret
|
|
},
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"Accept": "application/json"
|
|
},
|
|
timeout=30
|
|
)
|
|
|
|
print(f"Status Code: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
token = data.get("access_token")
|
|
expires_in = data.get("expires_in", 0)
|
|
expira = datetime.now() + timedelta(seconds=expires_in)
|
|
|
|
print(f"✅ Token obtido com sucesso! Expira em: {expira}")
|
|
return token
|
|
else:
|
|
print(f"❌ Erro na requisição: {response.status_code}")
|
|
print(f"Resposta: {response.text}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
print(f"❌ Erro ao obter token: {e}")
|
|
return None
|
|
|
|
def save_token(self, token: str) -> bool:
|
|
"""Atualiza o token no banco de dados"""
|
|
try:
|
|
print("Conectando ao banco de dados...")
|
|
conn = pyodbc.connect(self.connection_string)
|
|
cursor = conn.cursor()
|
|
|
|
query = "UPDATE dbo.rgb_token SET token = ?, updatedAt = GETDATE() WHERE id = 1"
|
|
cursor.execute(query, token)
|
|
conn.commit()
|
|
|
|
rows = cursor.rowcount
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
if rows > 0:
|
|
print(f"✅ Token atualizado com sucesso no banco ({rows} registro(s)).")
|
|
return True
|
|
else:
|
|
print("⚠️ Nenhum registro atualizado (verifique o ID = 1).")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Erro ao salvar token no banco: {e}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
client = RGBTokenClient()
|
|
token = client.get_token()
|
|
|
|
if token:
|
|
client.save_token(token)
|
|
print("🎯 Execução finalizada com sucesso.")
|
|
else:
|
|
print("❌ Falha ao obter ou salvar o token.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|