Gustavo
dad1abf30b
Testado contra a instancia real (devn8n.gtigrupo.com.br) com uma API key de verdade: 'take' retorna 400 'Unknown query parameter'; o nome certo e 'limit'. Confirmado end-to-end - /log agora mostra os envios reais (ERRO_ENVIO, CLIENTE_NAO_CADASTRADO etc.) da Data Table pagamentos_processados.
53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
"""
|
|
Cliente HTTP mínimo para ler a Data Table "pagamentos_processados" do n8n —
|
|
o histórico de envios automáticos de chave de reativação via WhatsApp,
|
|
mantido pelo workflow "Pagamento Confirmado Asaas..." (ver
|
|
licenciamento-n8n/docs/). Cada linha é um pagamento processado, nunca
|
|
sobrescrita — é o histórico que o sistema legado nunca teve.
|
|
|
|
Requer N8N_API_URL e N8N_API_KEY no ambiente (uma API key gerada em
|
|
Configurações > API do n8n, com o escopo de leitura de linhas de Data
|
|
Table). Se não estiverem configuradas, ou se a chamada falhar por
|
|
qualquer motivo (rede, chave errada, endpoint diferente do esperado),
|
|
listar_envios() devolve [] — a tela de log mostra um aviso em vez de
|
|
quebrar. Este módulo NUNCA escreve na Data Table, só lê.
|
|
"""
|
|
import os
|
|
|
|
import httpx
|
|
|
|
DATA_TABLE_ID = "bJHhFUDwFJ30YJVt" # pagamentos_processados
|
|
|
|
|
|
def configurado() -> bool:
|
|
return bool(os.environ.get("N8N_API_URL")) and bool(os.environ.get("N8N_API_KEY"))
|
|
|
|
|
|
def listar_envios(limite: int = 50) -> list[dict]:
|
|
"""
|
|
Últimas linhas de pagamentos_processados, mais recentes primeiro.
|
|
Lista vazia se a integração não estiver configurada ou a chamada falhar.
|
|
"""
|
|
if not configurado():
|
|
return []
|
|
|
|
base = os.environ["N8N_API_URL"].rstrip("/")
|
|
url = f"{base}/api/v1/data-tables/{DATA_TABLE_ID}/rows"
|
|
try:
|
|
resposta = httpx.get(
|
|
url,
|
|
headers={"X-N8N-API-KEY": os.environ["N8N_API_KEY"], "accept": "application/json"},
|
|
params={"limit": limite, "sortBy": "data_processamento:desc"},
|
|
timeout=5.0,
|
|
)
|
|
resposta.raise_for_status()
|
|
corpo = resposta.json()
|
|
except (httpx.HTTPError, ValueError):
|
|
return []
|
|
|
|
if isinstance(corpo, dict):
|
|
return corpo.get("data") or []
|
|
if isinstance(corpo, list):
|
|
return corpo
|
|
return []
|