feat(n8n): helper CLI para automacao via cookie de sessao

Item 4 da sugestão B — substituir uso manual de UI por CLI.

scripts/n8n-api.py: cliente Python via /rest/login (cookie n8n-auth JWT)

Subcomandos:
- login         — força novo login (autoexecuta antes de cada chamada)
- workflows     — lista workflows (id, name, active, archived)
- executions    — ultimas execucoes (status, workflowId, startedAt)
- activate      — ativa workflow (precisa versionId)
- deactivate    — desativa workflow
- webhook URL   — dispara webhook publico (sem auth)
- raw PATH      — chamada REST arbitraria (debug)

Credenciais lidas de scripts/.env:
- N8N_USER_EMAIL=admin@m5online.com.br
- N8N_LOGIN_PASSWORD=M5Online@2026

Testado:
- login: OK (role global:owner)
- workflows: 2 ativos (OpenWA → Twenty, Twenty → OpenWA)
- executions: 2 erros recentes (esperado, workspace Twenty nao criado)
- webhook openwa-incoming: HTTP 200

Nota: API key antiga N8N_USER/N8N_PASSWORD no .env e' legado
(N8N_USER=admin, N8N_PASSWORD=lKs9...) — mantida pra retrocompat.
Nova auth usa /rest/login + cookie (mais confiavel).
This commit is contained in:
Iris Florencio 2026-07-06 23:21:17 -03:00
parent 7ebe0529c5
commit dcca4bf3db

180
scripts/n8n-api.py Executable file
View File

@ -0,0 +1,180 @@
#!/usr/bin/env python3
"""
n8n API Helper automacao via cookie de sessao (login + REST)
Carrega credenciais de scripts/.env (N8N_USER, N8N_PASSWORD)
"""
import os
import sys
import json
import argparse
import urllib.request
import urllib.error
import http.cookiejar
from pathlib import Path
ENV_FILE = Path(__file__).parent / ".env"
if ENV_FILE.exists():
for line in ENV_FILE.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
k, v = line.split("=", 1)
v = v.strip().strip('"').strip("'")
os.environ.setdefault(k, v)
N8N_URL = os.environ.get("N8N_URL", "https://n8n.m5digital.net.br")
N8N_EMAIL = os.environ.get("N8N_USER_EMAIL", os.environ.get("N8N_USER", "admin@m5online.com.br"))
N8N_PASSWORD = os.environ.get("N8N_LOGIN_PASSWORD", "M5Online@2026")
# Cookie jar global
COOKIES = http.cookiejar.CookieJar()
OPENER = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(COOKIES))
def login() -> dict:
"""Faz login e guarda cookie n8n-auth."""
payload = json.dumps({
"emailOrLdapLoginId": N8N_EMAIL,
"password": N8N_PASSWORD,
}).encode()
req = urllib.request.Request(
f"{N8N_URL}/rest/login",
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
with OPENER.open(req, timeout=30) as resp:
return json.loads(resp.read().decode())
def rest(path: str, method: str = "GET", data: dict = None) -> dict:
"""Executa chamada REST autenticada."""
url = f"{N8N_URL}{path}"
body = json.dumps(data).encode() if data is not None else None
req = urllib.request.Request(url, data=body, method=method)
req.add_header("Content-Type", "application/json")
req.add_header("Accept", "application/json")
try:
with OPENER.open(req, timeout=30) as resp:
raw = resp.read().decode()
return json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
raw = e.read().decode()
try:
return json.loads(raw)
except Exception:
return {"_error": True, "_status": e.code, "_raw": raw[:500]}
def cmd_login(args):
r = login()
email = r.get("data", {}).get("email", "?")
role = r.get("data", {}).get("role", "?")
print(f"OK: logado como {email} ({role})")
def cmd_workflows(args):
r = rest("/rest/workflows")
items = r.get("data", [])
print(f"{len(items)} workflows:")
for wf in items:
active = "ATIVO" if wf.get("active") else " "
archived = "[ARQ]" if wf.get("isArchived") else " "
print(f" {active} {archived} {wf.get('id','?'):20} {wf.get('name','?')}")
def cmd_executions(args):
r = rest(f"/rest/executions?limit={args.limit}")
items = r.get("data", {}).get("results", [])
print(f"{len(items)} execucoes recentes:")
for ex in items:
status = ex.get("status", "?")
wf_id = ex.get("workflowId", "?")
started = ex.get("startedAt", "?")
print(f" [{status:10}] {started} wf={wf_id}")
def cmd_activate(args):
wf_id = args.id
# Pega versionId
r = rest(f"/rest/workflows/{wf_id}")
data = r.get("data") or {}
version_id = data.get("versionId")
if not version_id:
print(f"ERRO: nao achei versionId de {wf_id}")
print(json.dumps(r, indent=2)[:500])
return
r2 = rest(f"/rest/workflows/{wf_id}/activate", "POST", {"versionId": version_id})
print(f"Ativar {wf_id}: {r2}")
def cmd_deactivate(args):
r = rest(f"/rest/workflows/{args.id}/deactivate", "POST")
print(f"Desativar {args.id}: {r}")
def cmd_exec_webhook(args):
"""Dispara webhook publico (sem auth)."""
url = args.url
payload = json.dumps(args.data or {}).encode()
req = urllib.request.Request(url, data=payload, method="POST")
req.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
print(f"HTTP {resp.status}: {resp.read().decode()[:500]}")
except urllib.error.HTTPError as e:
print(f"HTTP {e.code}: {e.read().decode()[:500]}")
def cmd_raw(args):
r = rest(args.path, args.method, json.loads(args.data) if args.data else None)
print(json.dumps(r, indent=2, ensure_ascii=False)[:2000])
def main():
parser = argparse.ArgumentParser(description="n8n API Helper (cookie-based)")
sub = parser.add_subparsers(dest="cmd", required=True)
sub.add_parser("login", help="forca novo login").set_defaults(func=cmd_login)
sub.add_parser("workflows", help="lista workflows").set_defaults(func=cmd_workflows)
p = sub.add_parser("executions", help="ultimas execucoes")
p.add_argument("--limit", type=int, default=10)
p.set_defaults(func=cmd_executions)
p = sub.add_parser("activate", help="ativa workflow")
p.add_argument("id", help="workflow ID")
p.set_defaults(func=cmd_activate)
p = sub.add_parser("deactivate", help="desativa workflow")
p.add_argument("id", help="workflow ID")
p.set_defaults(func=cmd_deactivate)
p = sub.add_parser("webhook", help="dispara webhook publico")
p.add_argument("url")
p.add_argument("--data", help="JSON payload")
p.set_defaults(func=cmd_exec_webhook)
p = sub.add_parser("raw", help="chamada REST arbitraria")
p.add_argument("path", help="ex: /rest/workflows")
p.add_argument("--method", default="GET")
p.add_argument("--data", help='JSON ex: \'{"key":"value"}\'')
p.set_defaults(func=cmd_raw)
args = parser.parse_args()
# login automatico se nao for webhook (que e' publico)
if args.cmd != "webhook":
try:
login()
except Exception as e:
print(f"ERRO login: {e}", file=sys.stderr)
sys.exit(1)
args.func(args)
if __name__ == "__main__":
main()