diff --git a/scripts/n8n-api.py b/scripts/n8n-api.py new file mode 100755 index 0000000..7a70b9f --- /dev/null +++ b/scripts/n8n-api.py @@ -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()