import requests
import jwt
import urllib
import json
import threading
import webbrowser
try:
# Tkinter pode não estar disponível em alguns ambientes headless
import tkinter as tk
from tkinter import ttk, messagebox
except Exception: # pragma: no cover - fallback se não houver tkinter
tk = None
ttk = None
from pyarialib.ariaresponse import AriaResponse
from pyarialib.persona import Persona;
from pyarialib.sei import Sei;
from pyarialib.logos import ChatLogos
from .siafi import Siafi
from .otp import Otp
# Exportações de conveniência
__all__ = [
'Aria',
'Otp',
'Persona',
'Sei',
'ChatLogos',
'Siafi'
]
[documentos]
class Aria:
_tokens = {}
def __set_query_param(self, url: str, key: str, value: str) -> str:
partes_url = url.split('?')
base_url = partes_url[0]
query_string = partes_url[0] if len(partes_url) >= 2 else None
query_params = urllib.parse.parse_qs(query_string)
if key in query_params:
query_params[key] = value
else:
query_params[key] = [value]
encoded_query_params = urllib.parse.urlencode(query_params, doseq=True)
return f"{base_url}?{encoded_query_params}"
def __autentica(self) -> None:
token_antigo = self._tokens.get(self.username)
if token_antigo != None:
self.token = token_antigo
else:
data = {'login': self.username, 'password': self.password}
headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
url_token = self.base_url + ("/" if self.base_url[-1] != "/" else "") + "token"
resultado = requests.post(url_token, json=data, headers=headers)
resultado_json = resultado.json()
if resultado_json.get("status") == "ok":
self.token = resultado_json.get("token")
self._tokens[self.username] = self.token
else:
if self._tokens.get(self.username) != None:
self._tokens.pop(self.username)
raise Exception("Erro na autenticação.")
def __init__(self, username = None, password = None, base_url="https://apiapex.tesouro.gov.br/aria/", token = None):
if token != None:
self.token = token
self.revalida_token = False
elif username != None and password != None:
self.base_url = base_url
self.username = username
self.password = password
self.__autentica()
self.revalida_token = True
else:
self.token = None
self.revalida_token = False
# Sem estado OTP global: cada request receberá opcionalmente um objeto Otp
def __fetch(self, method, body, json_body, headers, url, data_append = None, auto_pagination = False) -> AriaResponse:
if method.lower() == 'get':
resultado = requests.request(method=method, url=url, headers=headers)
elif len(body) > 0:
resultado = requests.request(method=method, url=url, headers=headers, data=body)
else:
resultado = requests.request(method=method, url=url, headers=headers, json=json_body)
dados_json = resultado.json()
if not auto_pagination:
return AriaResponse(resultado.text, resultado.status_code)
if dados_json.get("page") != None and dados_json.get("pageSize") != None and dados_json.get("registros") != None:
if len(dados_json.get('registros', [])) < dados_json.get('pageSize') and data_append != None:
nova_response = AriaResponse(text=data_append, status_code=resultado.status_code)
return nova_response
elif len(dados_json.get('registros', [])) < dados_json.get('pageSize') and data_append == None:
return AriaResponse(resultado.text, resultado.status_code)
elif dados_json.get('page') and len(dados_json.get('registros', [])) == dados_json.get('pageSize'):
next_page_url = self.__set_query_param(url, 'page', str(dados_json.get('page') + 1))
if data_append != None:
dados_append_json = json.loads(data_append)
dados_json['registros'] += dados_append_json.get('registros', [])
return self.__fetch(method=method, body=body, json_body=json_body, headers=headers, url=next_page_url, data_append=json.dumps(dados_json), auto_pagination=auto_pagination)
else:
return AriaResponse(resultado.text, resultado.status_code)
else:
return AriaResponse(resultado.text, resultado.status_code)
[documentos]
def request(self, method : str, version : int, project : str, endpoint : str, body : str = "", json_body : dict = {}, headers : dict = {}, query_string_params : dict = {}, auto_pagination : bool = False, otp: Otp | None = None) -> AriaResponse:
"""
Realiza uma requisição ao ARIA.
Args:
method (string): método da requisição.
version (number): número da versão da api.
project (string): nome do projeto.
endpoint (string): código do endpoint.
body (str, optional): corpo da requisição em formato de string. Defaults to "".
json_body (dict, optional): corpo da requisição em formato de dict, enviado como json. Defaults to {}.
headers (dict, optional): headers da requisição. Defaults to {}.
query_string_params (dict, optional): parâmetros via querystring. Defaults to {}.
Returns:
any: resultado da requisição.
"""
if self.revalida_token:
try:
jwt.decode(self.token, algorithms=["HS512"], options={"verify_signature": False, "verify_exp": True})
except jwt.ExpiredSignatureError:
self.__autentica()
if self.token != None:
headers["Authorization"] = "Bearer " + self.token
# Fluxo OTP
# Caso interactive-console: duas etapas
interactive = False
if otp and isinstance(headers, dict):
interactive = getattr(otp, 'is_interactive', lambda: False)()
if not interactive: # modo manual padrão
if otp.type:
headers.setdefault('x-otp-type', otp.type)
headers.setdefault('Otp-Type', otp.type)
if otp.value:
headers.setdefault('x-otp-value', otp.value)
headers.setdefault('Otp-Value', otp.value)
if otp.flow_id:
headers.setdefault('x-otp-flow-id', otp.flow_id)
headers.setdefault('Otp-Flow-Id', otp.flow_id)
query_string = ""
for key in query_string_params:
if query_string == "":
query_string += "?"
else:
query_string += "&"
query_string += key + "=" + str(query_string_params.get(key))
url = self.base_url + ("/" if self.base_url[-1] != "/" else "") + "v" + str(version) + "/" + project + "/custom/" + endpoint + query_string
# Se não for interactive, fluxo normal
if not interactive:
resultado = self.__fetch(method=method, body=body, json_body=json_body, headers=headers, url=url, auto_pagination=auto_pagination)
return resultado
# Fluxo interactive-console / interactive-window
# Etapa 1: enviar apenas o tipo (sem value/flow) para obter flowId
headers_step1 = headers.copy() if isinstance(headers, dict) else {}
if otp and otp.type:
headers_step1.setdefault('x-otp-type', otp.type)
headers_step1.setdefault('Otp-Type', otp.type)
# Garantir que não enviamos value/flow nesta etapa
for k in ['x-otp-value', 'Otp-Value', 'x-otp-flow-id', 'Otp-Flow-Id']:
if k in headers_step1:
headers_step1.pop(k)
primeira_resposta = self.__fetch(method=method, body=body, json_body=json_body, headers=headers_step1, url=url, auto_pagination=False)
try:
primeira_json = primeira_resposta.json()
except Exception:
primeira_json = {}
flow_id = primeira_json.get('flowId') or primeira_json.get('flowID') or primeira_json.get('flow_id')
if flow_id:
otp.flow_id = flow_id
print("Fluxo OTP iniciado. Flow ID:", otp.flow_id, flush=True)
# Solicitar código ao usuário
if otp and getattr(otp, 'mode', '') == 'interactive-console':
try:
user_code = input("Informe o código OTP: ")
except EOFError:
user_code = None
if user_code:
otp.value = user_code.strip()
elif otp and getattr(otp, 'mode', '') == 'interactive-window':
if tk is None:
print("Tkinter indisponível - fallback para entrada via console.")
try:
user_code = input("Informe o código OTP: ")
except EOFError:
user_code = None
if user_code:
otp.value = user_code.strip()
else:
# Criar janela de forma bloqueante até que usuário confirme
code_container = {'value': None}
done_event = threading.Event()
def abrir_janela():
root = tk.Tk()
root.title("Autenticação OTP")
# Não forçar geometry fixa para evitar corte de componentes em alguns temas (macOS Aqua)
# root.geometry("420x260")
root.resizable(True, True)
# Tema / look and feel (tenta usar nativo, fallback para 'clam')
try:
style = ttk.Style()
preferidos = ["aqua", "vista", "xpnative", "clam", "default"]
disponiveis = style.theme_names()
for tema in preferidos:
if tema in disponiveis:
style.theme_use(tema)
break
except Exception:
pass
frame = ttk.Frame(root, padding=18)
frame.pack(fill='both', expand=True)
# Cabeçalho
titulo_lbl = ttk.Label(frame, text="Fluxo OTP em andamento", font=(None, 12, 'bold'))
titulo_lbl.pack(anchor='w', pady=(0,6))
# Monta mensagem base
info_txt = (
"Foi iniciado um fluxo OTP. Confirme o código enviado.\n"
"Você pode copiar o Flow ID abaixo caso precise informar em outro canal."
)
# Se o tipo de OTP for webauthn2 ou webauthn2steps, exibir URL de docs para usuário gerar/validar token
mostrar_docs = otp and (otp.type or "").lower() in {"webauthn2", "webauthn2steps"}
docs_url = None
if mostrar_docs:
# Construção da URL de docs: <BASE_URL>/v<version>/<project>/docs
base_docs = self.base_url.rstrip('/')
# version e project disponíveis no escopo externo (parâmetros do método request)
docs_url = f"{base_docs}/v{version}/{project}/docs"
info_txt += f"\n\nAcesse a URL de documentação para acompanhar o fluxo / gerar token:\n{docs_url}"
ttk.Label(frame, text=info_txt, wraplength=420, justify='left').pack(anchor='w', pady=(0,10))
# Se houver docs_url, adicionar botões para abrir/copiar
if mostrar_docs and docs_url:
docs_frame = ttk.Frame(frame)
docs_frame.pack(fill='x', pady=(0,10))
ttk.Label(docs_frame, text="Docs URL:").pack(side='left')
docs_var = tk.StringVar(value=docs_url)
docs_entry = ttk.Entry(docs_frame, textvariable=docs_var, state='readonly')
docs_entry.pack(side='left', fill='x', expand=True, padx=(6,6))
def abrir_docs(): # pragma: no cover - depende de ambiente gráfico
try:
webbrowser.open(docs_url)
except Exception:
messagebox.showwarning("Aviso", "Não foi possível abrir o navegador.")
def copiar_docs():
try:
root.clipboard_clear()
root.clipboard_append(docs_url)
root.update()
messagebox.showinfo("Copiado", "URL copiada para a área de transferência.")
except Exception:
pass
ttk.Button(docs_frame, text="Abrir", command=abrir_docs).pack(side='left')
ttk.Button(docs_frame, text="Copiar", command=copiar_docs).pack(side='left', padx=(4,0))
# Tipo
ttk.Label(frame, text=f"Tipo: {otp.type or '-'}").pack(anchor='w', pady=(0,4))
# Flow ID (readonly) + botão copiar
flow_frame = ttk.Frame(frame)
flow_frame.pack(fill='x', pady=(0,10))
ttk.Label(flow_frame, text="Flow ID:").pack(side='left')
flow_var = tk.StringVar(value=otp.flow_id or '')
flow_entry = ttk.Entry(flow_frame, textvariable=flow_var, state='readonly')
flow_entry.pack(side='left', fill='x', expand=True, padx=(6,6))
def copiar_flow():
try:
root.clipboard_clear()
root.clipboard_append(flow_var.get())
root.update() # garante que permanece no clipboard após fechar
messagebox.showinfo("Copiado", "Flow ID copiado para a área de transferência.")
except Exception:
pass
ttk.Button(flow_frame, text="Copiar", command=copiar_flow).pack(side='left')
# Campo código OTP
ttk.Label(frame, text="Código OTP:").pack(anchor='w')
entry = ttk.Entry(frame)
entry.pack(fill='x', pady=5)
entry.focus_set()
# Botões
def confirmar():
valor = entry.get().strip()
if not valor:
messagebox.showwarning("Atenção", "Informe um código OTP.")
return
code_container['value'] = valor
root.destroy()
done_event.set()
def cancelar():
root.destroy()
done_event.set()
botoes = ttk.Frame(frame)
botoes.pack(fill='x', pady=14)
ttk.Button(botoes, text="Confirmar", command=confirmar).pack(side='left', expand=True, fill='x')
ttk.Button(botoes, text="Cancelar", command=cancelar).pack(side='left', expand=True, fill='x', padx=(8,0))
# Permitir que Enter dentro do campo dispare a confirmação
try:
entry.bind('<Return>', lambda e: confirmar())
except Exception:
pass
# Ajustar tamanho mínimo após construir todos os widgets para evitar corte dos botões
root.update_idletasks()
req_w = root.winfo_reqwidth()
req_h = root.winfo_reqheight()
# Adicionar pequena folga vertical
root.minsize(req_w, req_h + 4)
# Opcional: centralizar na tela
try:
screen_w = root.winfo_screenwidth()
screen_h = root.winfo_screenheight()
x = max(0, (screen_w - req_w) // 2)
y = max(0, (screen_h - req_h) // 3)
root.geometry(f"{req_w}x{req_h}+{x}+{y}")
except Exception:
pass
root.mainloop()
abrir_janela() # roda no mesmo thread; se quiser não bloquear I/O futuro, poderíamos separar
if code_container['value']:
otp.value = code_container['value']
# Etapa 2: reenviar com type, value e flow_id
headers_step2 = headers.copy() if isinstance(headers, dict) else {}
if otp and otp.type:
headers_step2.setdefault('x-otp-type', otp.type)
headers_step2.setdefault('Otp-Type', otp.type)
if otp and otp.value:
headers_step2.setdefault('x-otp-value', otp.value)
headers_step2.setdefault('Otp-Value', otp.value)
if otp and otp.flow_id:
headers_step2.setdefault('x-otp-flow-id', otp.flow_id)
headers_step2.setdefault('Otp-Flow-Id', otp.flow_id)
segunda_resposta = self.__fetch(method=method, body=body, json_body=json_body, headers=headers_step2, url=url, auto_pagination=auto_pagination)
return segunda_resposta
[documentos]
def get_persona(self) -> Persona:
persona = Persona(self)
return persona
[documentos]
def get_sei(self) -> Sei:
sei = Sei(self)
return sei
[documentos]
def get_siafi(self, env, ug, projeto_aria) -> Siafi:
siafi = Siafi(self, env=env, ug=ug, projeto_aria=projeto_aria)
return siafi
[documentos]
def get_chatLogos(self) -> ChatLogos:
return ChatLogos(self)