Japlog · ETL · Sispro → Analytics

The workflow JSON

Copy or download the full n8n JSON below. Paste it into a new n8n workflow, add your credentials, activate. Full import guide →

{
  "name": "JAPLog \u00b7 ETL \u00b7 Sispro \u2192 analytics",
  "nodes": [
    {
      "parameters": {
        "rule": {
          "interval": [
            {
              "field": "hours",
              "hoursInterval": 1
            }
          ]
        }
      },
      "id": "17dbad2c-8054-4730-afd3-4a0ed1fa561c",
      "name": "Every 1 hour",
      "type": "n8n-nodes-base.scheduleTrigger",
      "typeVersion": 1.2,
      "position": [
        200,
        300
      ]
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "INSERT INTO analytics.sync_cursor (source, ultimo_id, ultima_sync) VALUES ('sispro', NULL, NOW()) ON CONFLICT (source) DO UPDATE SET ultima_sync = NOW();"
      },
      "id": "e6d2396c-5d12-46ef-a0af-5d51d6605d71",
      "name": "Heartbeat",
      "type": "n8n-nodes-base.postgres",
      "typeVersion": 2.5,
      "position": [
        320,
        140
      ],
      "credentials": {
        "postgres": {
          "name": "<your credential>"
        }
      }
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "SELECT COALESCE(\n  (SELECT MAX(created_at) FROM analytics.logistica_movimentacoes WHERE source = 'sispro'),\n  '2025-01-01T00:00:00Z'::timestamptz\n) AS cursor_ts;"
      },
      "id": "3b526989-0fd6-471b-a2bd-d442f6e54551",
      "name": "Read Cursor",
      "type": "n8n-nodes-base.postgres",
      "typeVersion": 2.5,
      "position": [
        420,
        300
      ]
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "=SELECT\n  c.CD_DOCUM_NR_NENT      AS numero_nf,\n  c.CD_DOCUM_NR_CONH      AS numero_cte,\n  c.CD_PESSOA_TRANS       AS cnpj_transportadora,\n  c.CD_ESTAB_NENT         AS estab_destino,\n  n.DT_NOTA_SAI           AS data_emissao,\n  n.DT_NOTA_SAI_SAIDA     AS data_saida,\n  n.VL_NOTA_SAI           AS valor_nf,\n  n.VL_FRETE              AS valor_frete_nf,\n  n.QT_VOLUME             AS qtd_volumes,\n  n.AN_PLACA              AS placa,\n  n.IN_SIT_NFE            AS status_nfe,\n  n.NR_CHV_NFE            AS chave_nfe,\n  n.CD_UF_DESTINAT        AS uf_destino,\n  SUM(c.QT_NENT_ITEM_PROD) AS qtd_itens,\n  SUM(c.VL_RATEIO_FRETE)   AS valor_frete_cte,\n  SUM(c.QT_PESO_BRUTO)     AS peso_bruto\nFROM GS_NENT_CONH_NOTA c\nLEFT JOIN GC_NSAI n\n  ON n.CD_DOCUM_NR_NOTA_SAI = c.CD_DOCUM_NR_NENT\nWHERE c.CD_PESSOA_REMET = '04214987000106'\n  AND c.CD_DOCUM_CONH = 'CTE'\n  AND n.DT_NOTA_SAI >= CONVERT(\n    varchar,\n    DATEADD(day, -1, CAST('{{ $('Read Cursor').item.json.cursor_ts }}' AS datetime)),\n    23\n  )\nGROUP BY\n  c.CD_DOCUM_NR_NENT, c.CD_DOCUM_NR_CONH,\n  c.CD_PESSOA_TRANS, c.CD_ESTAB_NENT,\n  n.DT_NOTA_SAI, n.DT_NOTA_SAI_SAIDA,\n  n.VL_NOTA_SAI, n.VL_FRETE, n.QT_VOLUME,\n  n.AN_PLACA, n.IN_SIT_NFE, n.NR_CHV_NFE,\n  n.CD_UF_DESTINAT\nORDER BY n.DT_NOTA_SAI DESC;"
      },
      "id": "ac2c387a-a8f4-46c0-a12d-ea96315513d2",
      "name": "Query Sispro",
      "type": "n8n-nodes-base.microsoftSql",
      "typeVersion": 1.1,
      "position": [
        640,
        300
      ]
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "SELECT cnpj, nome_norm, variacao, ativo\nFROM analytics.transportadora_dict\nWHERE ativo = TRUE AND cnpj IS NOT NULL;"
      },
      "id": "2769c3a2-dac1-44b1-b040-9f7675edcc18",
      "name": "Read Transportadora Dict",
      "type": "n8n-nodes-base.postgres",
      "typeVersion": 2.5,
      "position": [
        640,
        460
      ]
    },
    {
      "parameters": {
        "mode": "runOnceForAllItems",
        "language": "javaScript",
        "jsCode": "// =============================================================================\n// JAPLog \u00b7 Dashboard de KPIs Log\u00edsticos \u2014 M\u00f3dulo 2 (ETL n8n)\n// Arquivo: n8n/lib/sispro_transform.js\n// Respons\u00e1vel: Silvio Italo + Vinicius (Automa\u00e7\u00f5es)\n// =============================================================================\n// Fonte \u00fanica da l\u00f3gica de transforma\u00e7\u00e3o das linhas do Sispro (SQL Server)\n// para analytics.logistica_movimentacoes.\n//\n// Sem\u00e2ntica: fonte INDEPENDENTE (source='sispro'), n\u00e3o enrichment.\n// Chave composta: numero_nf + '_' + numero_cte = source_id.\n//\n// Refer\u00eancia: JAPLog_Dashboard_KPIs_Arquitetura.docx \u00a74.3 + query SQL Server\n//             fornecida pela equipe de Pesquisa e Inova\u00e7\u00e3o (Abril 2026).\n// =============================================================================\n\n// Remetente fixo (Japur\u00e1 Pneus \u2014 Manaus). Confirmado pelo CD_PESSOA_REMET\n// no WHERE da query Sispro: '04214987000106'.\nconst REMETENTE_CNPJ = '04214987000106';\nconst UNIDADE_SAIDA_FIXA = 'CD_MANAUS';\n\n// -----------------------------------------------------------------------------\n// slugUnidade \u2014 mesmo comportamento dos outros transforms (duplicado pois\n// n8n Code Nodes n\u00e3o compartilham m\u00f3dulos).\n// -----------------------------------------------------------------------------\nfunction slugUnidade(raw) {\n  if (raw == null) return null;\n  const s = String(raw).trim();\n  if (!s) return null;\n  return s\n    .normalize('NFD').replace(/[\\u0300-\\u036f]/g, '')\n    .toUpperCase()\n    .replace(/[^A-Z0-9]+/g, '_')\n    .replace(/^_+|_+$/g, '');\n}\n\n// -----------------------------------------------------------------------------\n// normalizeCnpj \u2014 remove pontua\u00e7\u00e3o e mant\u00e9m s\u00f3 d\u00edgitos\n// -----------------------------------------------------------------------------\nfunction normalizeCnpj(raw) {\n  if (raw == null) return null;\n  const digits = String(raw).replace(/\\D+/g, '');\n  return digits || null;\n}\n\n// -----------------------------------------------------------------------------\n// toNumber \u2014 lida com Decimals, strings com v\u00edrgula decimal, etc.\n// -----------------------------------------------------------------------------\nfunction toNumber(raw) {\n  if (raw == null || raw === '') return null;\n  if (typeof raw === 'number') return raw;\n  let s = String(raw).trim();\n  // Padr\u00e3o BR: \"1.234,56\" \u2192 remove pontos, troca v\u00edrgula por ponto\n  if (/,\\d{1,3}$/.test(s) && /\\.\\d{3}/.test(s)) {\n    s = s.replace(/\\./g, '').replace(',', '.');\n  } else if (/,\\d{1,3}$/.test(s)) {\n    s = s.replace(',', '.');\n  } else {\n    s = s.replace(/,/g, '');\n  }\n  const n = parseFloat(s);\n  return isNaN(n) ? null : n;\n}\n\n// -----------------------------------------------------------------------------\n// toIsoDate \u2014 normaliza timestamps vindos do SQL Server (node mssql devolve\n// objetos Date em JS na maioria dos casos; fallback pra string).\n// -----------------------------------------------------------------------------\nfunction toIsoDate(raw) {\n  if (raw == null || raw === '') return null;\n  if (raw instanceof Date) return isNaN(raw.getTime()) ? null : raw.toISOString();\n  const s = String(raw).trim();\n  const d = new Date(s);\n  return isNaN(d.getTime()) ? null : d.toISOString();\n}\n\n// -----------------------------------------------------------------------------\n// lookupTransportadoraByCnpj \u2014 usa dict pr\u00e9-carregado do Postgres\n// -----------------------------------------------------------------------------\nfunction lookupTransportadoraByCnpj(cnpj, dictRows) {\n  if (!cnpj) return { nome_norm: null };\n  const hit = (dictRows || []).find(\n    (r) => normalizeCnpj(r.cnpj) === cnpj\n  );\n  return { nome_norm: hit ? (hit.nome_norm ?? null) : null };\n}\n\n// -----------------------------------------------------------------------------\n// buildRow \u2014 transforma uma linha do SELECT da query Sispro em linha analytics.\n//\n// sisRow: objeto com as 16 colunas do SELECT:\n//   numero_nf, numero_cte, cnpj_transportadora, estab_destino,\n//   data_emissao, data_saida, valor_nf, valor_frete_nf, qtd_volumes,\n//   placa, status_nfe, chave_nfe, uf_destino,\n//   qtd_itens, valor_frete_cte, peso_bruto\n// -----------------------------------------------------------------------------\nfunction buildRow(sisRow, dictRows) {\n  const numeroNf  = sisRow.numero_nf != null ? String(sisRow.numero_nf).trim() : null;\n  const numeroCte = sisRow.numero_cte != null ? String(sisRow.numero_cte).trim() : null;\n  if (!numeroNf || !numeroCte) return null;  // linhas sem chave s\u00e3o descartadas\n\n  const cnpj = normalizeCnpj(sisRow.cnpj_transportadora);\n  const { nome_norm } = lookupTransportadoraByCnpj(cnpj, dictRows);\n\n  return {\n    source:              'sispro',\n    source_id:           `${numeroNf}_${numeroCte}`,\n    created_at:          toIsoDate(sisRow.data_emissao),\n    data_prevista:       toIsoDate(sisRow.data_saida),\n    data_status_final:   null,\n    tipo_operacao:       'interestadual',\n    modalidade:          null,\n    status_normalizado:  null,   // Sispro n\u00e3o traz status operacional\n    unidade_saida:       UNIDADE_SAIDA_FIXA,\n    unidade_destino:     slugUnidade(sisRow.estab_destino),\n    destino_cidade:      null,   // Sispro tem UF, n\u00e3o cidade; deixa null\n    vendedor:            null,\n    motorista:           null,\n    submitted_by:        null,\n    nota_fiscal:         numeroNf,\n    transportadora:      cnpj,   // guardamos o CNPJ bruto como identidade\n    transportadora_norm: nome_norm,\n    numero_cte:          numeroCte,\n    valor_cte:           toNumber(sisRow.valor_frete_cte),\n    qtd_itens:           Math.round(toNumber(sisRow.qtd_itens) ?? 0) || null,\n    valor_estimado:      toNumber(sisRow.valor_nf),\n    status_nfe:          sisRow.status_nfe != null ? String(sisRow.status_nfe).trim() : null,\n    chave_nfe:           sisRow.chave_nfe  != null ? String(sisRow.chave_nfe).trim() : null,\n    peso_bruto:          toNumber(sisRow.peso_bruto),\n  };\n}\n\n// Chain linear: Query Sispro \u2192 Read Transportadora Dict \u2192 Transform.\n// $input aqui vem do Dict; os sisItems saem via refer\u00eancia expl\u00edcita.\nconst sisItems  = $('Query Sispro').all();\nconst dictItems = $input.all();\nconst dictRows  = dictItems.map(i => i.json);\n\nconst out = [];\nfor (const it of sisItems) {\n  const row = buildRow(it.json, dictRows);\n  if (row) out.push({ json: row });\n}\nreturn out;"
      },
      "id": "2f187f55-0ff9-4386-a635-c4de9b24cf4b",
      "name": "Transform",
      "type": "n8n-nodes-base.code",
      "typeVersion": 2,
      "position": [
        880,
        300
      ]
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "INSERT INTO analytics.logistica_movimentacoes (\n  source, source_id, created_at, data_prevista, data_status_final,\n  tipo_operacao, modalidade, status_normalizado,\n  unidade_saida, unidade_destino, destino_cidade,\n  vendedor, motorista, submitted_by,\n  nota_fiscal, transportadora, transportadora_norm,\n  numero_cte, valor_cte, qtd_itens, valor_estimado,\n  status_nfe, chave_nfe, peso_bruto,\n  processado_em\n) VALUES (\n  $1, $2, NULLIF($3,'null')::timestamptz, NULLIF($4,'null')::timestamptz, NULLIF($5,'null')::timestamptz,\n  $6, NULLIF($7,'null'), NULLIF($8,'null'),\n  NULLIF($9,'null'), NULLIF($10,'null'), NULLIF($11,'null'),\n  NULLIF($12,'null'), NULLIF($13,'null'), NULLIF($14,'null'),\n  NULLIF($15,'null'), NULLIF($16,'null'), NULLIF($17,'null'),\n  NULLIF($18,'null'), NULLIF($19,'null')::numeric, NULLIF($20,'null')::integer, NULLIF($21,'null')::numeric,\n  NULLIF($22,'null'), NULLIF($23,'null'), NULLIF($24,'null')::numeric,\n  NOW()\n)\nON CONFLICT (source, source_id) DO UPDATE SET\n  created_at          = COALESCE(EXCLUDED.created_at, analytics.logistica_movimentacoes.created_at),\n  data_prevista       = COALESCE(EXCLUDED.data_prevista, analytics.logistica_movimentacoes.data_prevista),\n  transportadora_norm = EXCLUDED.transportadora_norm,\n  valor_cte           = EXCLUDED.valor_cte,\n  qtd_itens           = EXCLUDED.qtd_itens,\n  peso_bruto          = EXCLUDED.peso_bruto,\n  chave_nfe           = EXCLUDED.chave_nfe,\n  status_nfe          = EXCLUDED.status_nfe,\n  processado_em       = NOW();",
        "options": {
          "queryReplacement": "={{ $json.source }},={{ $json.source_id }},={{ $json.created_at }},={{ $json.data_prevista }},={{ $json.data_status_final }},={{ $json.tipo_operacao }},={{ $json.modalidade }},={{ $json.status_normalizado }},={{ $json.unidade_saida }},={{ $json.unidade_destino }},={{ $json.destino_cidade }},={{ $json.vendedor }},={{ $json.motorista }},={{ $json.submitted_by }},={{ $json.nota_fiscal }},={{ $json.transportadora }},={{ $json.transportadora_norm }},={{ $json.numero_cte }},={{ $json.valor_cte }},={{ $json.qtd_itens }},={{ $json.valor_estimado }},={{ $json.status_nfe }},={{ $json.chave_nfe }},={{ $json.peso_bruto }}"
        }
      },
      "id": "2fd3e5fd-c239-4355-acc3-04b6ff6c6b95",
      "name": "Upsert Analytics",
      "type": "n8n-nodes-base.postgres",
      "typeVersion": 2.5,
      "position": [
        1140,
        300
      ]
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "WITH cursor_upsert AS (\n  INSERT INTO analytics.sync_cursor (source, ultimo_id, ultima_sync)\n  VALUES ($1, $2, NOW())\n  ON CONFLICT (source) DO UPDATE SET\n    ultimo_id   = EXCLUDED.ultimo_id,\n    ultima_sync = EXCLUDED.ultima_sync\n  RETURNING source\n)\nSELECT analytics.refresh_kpi_diario() FROM cursor_upsert;",
        "options": {
          "queryReplacement": "={{ 'sispro' }},={{ ($items('Transform').map(i => i.json.source_id).sort().slice(-1)[0]) || null }}"
        }
      },
      "id": "b3a6a902-d6d4-47e9-abdc-84ee25beb1d9",
      "name": "Update Cursor + Refresh",
      "type": "n8n-nodes-base.postgres",
      "typeVersion": 2.5,
      "position": [
        1400,
        300
      ]
    }
  ],
  "connections": {
    "Every 1 hour": {
      "main": [
        [
          {
            "node": "Heartbeat",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Read Cursor": {
      "main": [
        [
          {
            "node": "Query Sispro",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Query Sispro": {
      "main": [
        [
          {
            "node": "Read Transportadora Dict",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Read Transportadora Dict": {
      "main": [
        [
          {
            "node": "Transform",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Transform": {
      "main": [
        [
          {
            "node": "Upsert Analytics",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Upsert Analytics": {
      "main": [
        [
          {
            "node": "Update Cursor + Refresh",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Heartbeat": {
      "main": [
        [
          {
            "node": "Read Cursor",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  },
  "settings": {
    "executionOrder": "v1",
    "saveDataErrorExecution": "all",
    "saveDataSuccessExecution": "all",
    "timezone": "America/Manaus"
  }
}

Credentials you'll need

Each integration node will prompt for credentials when you import. We strip credential IDs before publishing — you'll add your own.

• postgres

About this workflow

JAPLog · ETL · Sispro → analytics. Uses postgres, microsoftSql. Scheduled trigger; 8 nodes.

Source: https://github.com/mapxmbs/montec-hub/blob/f3e78eefaff2cd1ee36fc00635695f58fa2526de/n8n/workflows/07_sispro.json — original creator credit. Request a take-down →

More Data & Sheets workflows → · Browse all categories →

Related workflows

Workflows that share integrations, category, or trigger type with this one. All free to copy and import.