Importando os CNPJs da Receita no Elasticsearch

Olá meus Unicórnios! 🦄✨

Todo mundo já precisou consultar um CNPJ. E quase sempre a resposta é assinar uma API que cobra por consulta. 💸 Só que existe um detalhe que muita gente não sabe: a Receita Federal publica a base inteira de graça. Todas as empresas do Brasil, com endereço, CNAE, situação cadastral, sócios e Simples Nacional. É só baixar.

O problema é o tamanho. São dezenas de gigabytes de CSV espalhados em dezenas de arquivos, em latin1, sem cabeçalho, com data em formato esquisito e vírgula no lugar do ponto decimal. Não dá para abrir no Excel — nem pensar. 😅

Neste artigo eu mostro o caminho completo: baixar os arquivos, carregar no SQLite e indexar no Elasticsearch. O download é o único passo feito na mão — os outros dois são Node puro, sem nenhum npm install, num arquivo só que dá para ler de cima a baixo. E com os dados reais aparecendo em cada etapa, para você ver a linha do CSV virando documento pesquisável.

📦 De onde vêm os arquivos

Os dados ficam no portal de arquivos abertos da Receita, numa pasta por mês:

Lá dentro você encontra os arquivos .zip. Depois de descompactar, os nomes são estes — e eles não têm extensão nenhuma que o Windows reconheça, o que assusta na primeira vez:

K3241.K03200Y0.D60808.EMPRECSV     razao social, capital, porte
K3241.K03200Y0.D60808.ESTABELE     endereco, CNAE, situacao cadastral
K3241.K03200Y0.D60808.SOCIOCSV     quadro societario
F.K03200$W.SIMPLES.CSV.D60808      Simples Nacional e MEI
F.K03200$Z.D60808.CNAECSV          tabela de CNAEs
F.K03200$Z.D60808.MUNICCSV         tabela de municipios

Repare no Y0 do nome. Os três arquivos grandes vêm partidos em dez pedaços, de Y0 até Y9. Se você baixar só o Y0 — como eu fiz na primeira vez — vai montar uma base com cerca de 10% das empresas do Brasil, e ela vai parecer completa. Nada acusa. Você só descobre quando procura uma empresa que existe e não acha. 😳

🗺️ O caminho dos dados

A base não vem pronta: ela vem normalizada, em quatro arquivos que se ligam por uma chave. E aí mora a primeira decisão do projeto.

EMPRECSV   ->  razao social, capital        \
SOCIOCSV   ->  socios                       > junta pelo cnpjBasico
SIMPLES    ->  opcao pelo Simples           /
                                            |
ESTABELE   ->  cada endereco/filial  -------+--> 1 documento no Elasticsearch

A chave é o CNPJ básico: os oito primeiros dígitos, que identificam a empresa. Os quatro seguintes são a ordem da filial e os dois últimos, o dígito verificador. Ou seja, uma empresa com trinta filiais aparece uma vez no EMPRECSV e trinta vezes no ESTABELE.

Meu primeiro instinto foi juntar tudo em memória, num Map. Não funciona: só as empresas e os sócios passam de oito gigabytes de RAM, e o Node morre com heap out of memory no meio da carga. Por isso o SQLite entra aqui — não como banco final, mas como uma mesa de apoio em disco. Ele guarda empresas e sócios, e eu consulto conforme leio os estabelecimentos.

E a melhor parte: desde o Node 22 o SQLite vem dentro do Node, no módulo node:sqlite. Sem npm install, sem compilar módulo nativo, sem Python. Esse detalhe sozinho já derruba a maior barreira deste tutorial. 🎉

✂️ Lendo o CSV que não é bem um CSV

Os arquivos são separados por ponto e vírgula, com todo campo entre aspas. Mas não existe aspa escapada dentro do campo — o layout da Receita simplesmente não prevê isso. Então um laço de dezoito linhas resolve, e você não precisa de biblioteca de CSV nenhuma:

function separarCampos(linha) {
  const campos = [];
  let atual = "";
  let dentroDeAspas = false;

  for (let i = 0; i < linha.length; i++) {
    const c = linha[i];
    if (c === '"') {
      dentroDeAspas = !dentroDeAspas;
    } else if (c === ";" && !dentroDeAspas) {
      campos.push(atual);
      atual = "";
    } else {
      atual += c;
    }
  }
  campos.push(atual);
  return campos;
}

Para ler o arquivo, tem que ser linha a linha. O ESTABELE passa de 7 GB, e um readFileSync nem chega a tentar: o Node recusa qualquer arquivo acima de 2 GB com ERR_FS_FILE_TOO_LARGE. Aprendi isso na marra. 😅

async function* lerLinhas(arquivo) {
  const fluxo = fs.createReadStream(arquivo, {
    // Os arquivos da Receita sao latin1. Ler como utf8 transforma "ACOES"
    // em "A�ES" -- e o estrago so aparece depois, ja indexado.
    encoding: "latin1",
    highWaterMark: 1024 * 1024,
  });

  const leitor = readline.createInterface({ input: fluxo, crlfDelay: Infinity });

  for await (const linha of leitor) {
    if (linha.trim() === "") continue;
    yield separarCampos(linha);
  }
}

O encoding: "latin1" é obrigatório e é a armadilha mais silenciosa daqui. Se você ler como UTF-8, todo acento vira caractere quebrado — e como o programa não dá erro nenhum, você só percebe depois que a base inteira já está indexada com "AÇÕES" escrito errado.

🧾 O CSV não tem cabeçalho — a ordem é tudo

Esta é a parte que eu mais recomendo conferir com calma. Os arquivos não têm linha de cabeçalho: a primeira linha já é dado. O significado de cada coluna vem da posição, e só. Trocar duas de lugar corrompe a base inteira sem nenhum aviso.

const COLUNAS_EMPRESA = [
  "cnpjBasico", "razaoSocial", "naturezaJuridica",
  "qualificacaoResponsavel", "capitalSocial", "porte", "enteFederativo",
];

const COLUNAS_SOCIO = [
  "cnpjBasico", "identificadorSocio", "nomeSocio", "cpfCnpjSocio",
  "qualificacaoSocio", "dataEntradaSociedade", "pais", "representanteLegal",
  "nomeRepresentante", "qualificacaoRepresentante", "faixaEtaria",
];

const COLUNAS_ESTABELECIMENTO = [
  "cnpjBasico", "cnpjOrdem", "cnpjDv", "identificador", "nomeFantasia",
  "situacaoCadastral", "dataSituacaoCadastral", "motivoSituacao",
  "cidadeExterior", "pais", "dataInicioAtividade", "cnaePrincipal",
  "cnaesSecundarios", "tipoLogradouro", "logradouro", "numero",
  "complemento", "bairro", "cep", "uf", "municipio", "ddd1", "telefone1",
  "ddd2", "telefone2", "dddFax", "fax", "email", "situacaoEspecial",
  "dataSituacaoEspecial",
];

Com a lista pronta, juntar a linha com os nomes é trivial:

function montarObjeto(colunas, campos) {
  const obj = {};
  for (let i = 0; i < colunas.length; i++) {
    obj[colunas[i]] = campos[i] || "";
  }
  return obj;
}

🔍 Como o dado realmente chega

Chega de teoria — vamos ver o dado. Esta é uma linha real do EMPRECSV, de uma empresa de energia renovável (dado público de pessoa jurídica):

"41300376";"JAIBA SE2 ENERGIAS RENOVAVEIS S.A.";"2054";"10";"79891408,93";"05";""

E a linha correspondente no ESTABELE, com o endereço:

"41300376";"0001";"07";"1";"";"02";"20210322";"00";"";"";"20210322";
"3511501";"3313901,6821802,7739099";"FAZENDA";"MARQUES";"S/N";"";
"ZONA RURAL";"39508000";"MG";"2893";...

Repare em três coisas que vão dar trabalho: "79891408,93" tem vírgula decimal; "20210322" é uma data em AAAAMMDD; e "3313901,6821802,7739099" são três CNAEs secundários numa string só. Nenhum dos três serve ao Elasticsearch como está.

🔧 Convertendo os campos (e a data que mente)

Três funções pequenas resolvem os três casos. Começando pelo texto, que transforma campo vazio em null:

function texto(valor) {
  if (!valor) return null;
  const limpo = String(valor).trim();
  if (limpo === "") return null;
  return limpo;
}

Agora a data — e se você só for ler um pedaço deste artigo, leia este. 🙏 A ausência de data vem como "0" ou "00000000", e o Elasticsearch recusa o documento inteiro se receber isso num campo date. Mas tem uma armadilha pior:

function data(valor) {
  const limpo = texto(valor);
  if (limpo === null) return null;
  if (limpo === "0" || limpo === "00000000") return null;
  if (!/^\d{8}$/.test(limpo)) return null;

  const ano = limpo.slice(0, 4);
  const mes = limpo.slice(4, 6);
  const dia = limpo.slice(6, 8);

  // Existe 20250231 nos arquivos. O Date do JS aceita e "conserta" para
  // 03/03 sem avisar -- gravar uma data errada e pior que gravar nada.
  const conferencia = new Date(Date.UTC(Number(ano), Number(mes) - 1, Number(dia)));
  if (conferencia.getUTCMonth() !== Number(mes) - 1) return null;
  if (conferencia.getUTCDate() !== Number(dia)) return null;

  return ano + "-" + mes + "-" + dia;
}

Isso mesmo: existe 31 de fevereiro nos arquivos da Receita. 🤯 E o Date do JavaScript não reclama — ele silenciosamente conserta para 3 de março e devolve uma data válida. Aí você grava no índice uma data que nunca existiu, e ela some no meio de milhões de registros corretos. As duas linhas de conferência existem só para isso: se o mês ou o dia mudaram sozinhos, a data era falsa e vira null.

Faltam o decimal e a lista, os dois fáceis:

function decimal(valor) {
  const limpo = texto(valor);
  if (limpo === null) return null;
  // O capital social vem com virgula ("79891408,93"). Number() puro da NaN.
  const numero = Number(limpo.replace(",", "."));
  if (!Number.isFinite(numero)) return null;
  return numero;
}

function lista(valor) {
  const limpo = texto(valor);
  if (limpo === null) return [];
  return limpo.split(",").map((x) => x.trim()).filter((x) => x !== "");
}

🗄️ Etapa 1: empresas e sócios no SQLite

Agora a mesa de apoio. Duas tabelas, e o registro inteiro guardado como JSON numa coluna só — porque eu nunca filtro por campo aqui dentro, só busco pelo cnpj_basico. Colunas separadas dariam trabalho sem nenhum ganho:

function abrirBanco() {
  if (fs.existsSync(BANCO)) fs.rmSync(BANCO);
  const banco = new DatabaseSync(BANCO);

  // Este banco e descartavel: se a importacao falhar, apagamos e refazemos.
  // Desligar o journal e o fsync deixa a carga varias vezes mais rapida.
  banco.exec("PRAGMA journal_mode = OFF");
  banco.exec("PRAGMA synchronous = OFF");

  banco.exec(`
    CREATE TABLE empresas (cnpj_basico TEXT PRIMARY KEY, dados TEXT NOT NULL);
    CREATE TABLE socios   (cnpj_basico TEXT NOT NULL, dados TEXT NOT NULL);
  `);

  return banco;
}

Os dois PRAGMA parecem irresponsáveis, e seriam — num banco que você quer manter. Aqui não: se a carga falhar, o arquivo é apagado e refeito do zero. Trocar durabilidade por velocidade é o negócio certo quando o dado é descartável.

A carga em si, com dois detalhes que mudam a carga de horas para minutos:

async function carregarNoSqlite(banco) {
  const inserirEmpresa = banco.prepare("INSERT OR REPLACE INTO empresas VALUES (?, ?)");
  const inserirSocio = banco.prepare("INSERT INTO socios VALUES (?, ?)");

  let empresas = 0;
  for (const arquivo of acharArquivos("EMPRECSV")) {
    console.log("  lendo " + path.basename(arquivo));
    // Uma transacao por arquivo: sem ela, o SQLite abre e fecha uma
    // transacao por linha e a carga leva horas em vez de minutos.
    banco.exec("BEGIN");
    for await (const campos of lerLinhas(arquivo)) {
      const empresa = montarObjeto(COLUNAS_EMPRESA, campos);
      inserirEmpresa.run(empresa.cnpjBasico, JSON.stringify(empresa));
      empresas++;
    }
    banco.exec("COMMIT");
  }

  let socios = 0;
  for (const arquivo of acharArquivos("SOCIOCSV")) {
    console.log("  lendo " + path.basename(arquivo));
    banco.exec("BEGIN");
    for await (const campos of lerLinhas(arquivo)) {
      const socio = montarObjeto(COLUNAS_SOCIO, campos);
      inserirSocio.run(socio.cnpjBasico, JSON.stringify(socio));
      socios++;
    }
    banco.exec("COMMIT");
  }

  // O indice so entra agora, depois da escrita: mante-lo durante os inserts
  // obrigaria o SQLite a reequilibrar a arvore a cada linha.
  console.log("  criando o indice de socios...");
  banco.exec("CREATE INDEX ix_socios ON socios(cnpj_basico)");

  return { empresas, socios };
}

O BEGIN/COMMIT por arquivo é o que mais pesa: sem transação, cada INSERT vira uma transação própria e o SQLite sincroniza disco milhões de vezes. E o índice só depois da escrita — criá-lo antes obrigaria o banco a reequilibrar a árvore a cada linha inserida.

É aqui que a empresa do exemplo chega no SQLite. Este é o registro gravado, lido de volta da tabela:

cnpj_basico  41300376
dados        {"cnpjBasico":"41300376","razaoSocial":"JAIBA SE2 ENERGIAS RENOVAVEIS S.A.",
              "naturezaJuridica":"2054","qualificacaoResponsavel":"10",
              "capitalSocial":"79891408,93","porte":"05","enteFederativo":""}

Repare que o capital ainda está com vírgula. De propósito: nesta etapa o SQLite é só um armário, e a conversão acontece na hora de montar o documento. Converter duas vezes é como campo sai errado.

🧩 Etapa 2: montando o documento

Cada estabelecimento vira um documento. E tem um if aqui que parece paranoia e não é:

function montarDocumento(estabelecimento, empresa, socios) {
  const e = estabelecimento;
  const cnpj = e.cnpjBasico + e.cnpjOrdem + e.cnpjDv;

  return {
    cnpj: cnpj,
    cnpjBasico: e.cnpjBasico,
    // A empresa pode faltar: existe estabelecimento sem a linha
    // correspondente no EMPRECSV. Se acessarmos empresa.razaoSocial direto,
    // a carga inteira morre com "Cannot read properties of null".
    razaoSocial: empresa ? texto(empresa.razaoSocial) : null,
    nomeFantasia: texto(e.nomeFantasia),
    porte: empresa ? texto(empresa.porte) : null,
    capitalSocial: empresa ? decimal(empresa.capitalSocial) : null,
    naturezaJuridica: empresa ? texto(empresa.naturezaJuridica) : null,
    situacaoCadastral: texto(e.situacaoCadastral),
    dataSituacaoCadastral: data(e.dataSituacaoCadastral),
    dataInicioAtividade: data(e.dataInicioAtividade),
    cnaePrincipal: texto(e.cnaePrincipal),
    cnaesSecundarios: lista(e.cnaesSecundarios),
    endereco: {
      logradouro: texto(e.logradouro),
      numero: texto(e.numero),
      bairro: texto(e.bairro),
      cep: texto(e.cep),
      uf: texto(e.uf),
      municipio: texto(e.municipio),
    },
    socios: socios.map(function (s) {
      return {
        nomeSocio: texto(s.nomeSocio),
        cpfCnpjSocio: texto(s.cpfCnpjSocio),
        qualificacaoSocio: texto(s.qualificacaoSocio),
        dataEntradaSociedade: data(s.dataEntradaSociedade),
      };
    }),
  };
}

Existe estabelecimento sem empresa correspondente. Não deveria, mas existe — e se você escrever empresa.razaoSocial direto, a carga inteira morre lá pela terceira hora com "Cannot read properties of null". O ternário evita isso: sem empresa, o documento vai com razaoSocial: null e a carga continua.

🗂️ O mapeamento, e o campo que precisa ser nested

O Elasticsearch adivinha os tipos se você não disser nada — e adivinha mal. Vale a pena criar o índice com o mapeamento pronto:

async function criarIndice() {
  await fetch(ELASTIC + "/" + INDICE, { method: "DELETE" });

  const resposta = await fetch(ELASTIC + "/" + INDICE, {
    method: "PUT",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      settings: {
        // Sem refresh durante a carga: com o padrao de 1 segundo o
        // Elasticsearch cria milhares de segmentos e passa a carga inteira
        // juntando arquivo em vez de indexar. Voltamos para "1s" no fim.
        refresh_interval: -1,
      },
      mappings: {
        properties: {
          cnpj: { type: "keyword" },
          cnpjBasico: { type: "keyword" },
          razaoSocial: { type: "text", fields: { exato: { type: "keyword", ignore_above: 256 } } },
          nomeFantasia: { type: "text", fields: { exato: { type: "keyword", ignore_above: 256 } } },
          porte: { type: "keyword" },
          capitalSocial: { type: "double" },
          naturezaJuridica: { type: "keyword" },
          situacaoCadastral: { type: "keyword" },
          dataSituacaoCadastral: { type: "date" },
          dataInicioAtividade: { type: "date" },
          cnaePrincipal: { type: "keyword" },
          cnaesSecundarios: { type: "keyword" },
          endereco: {
            properties: {
              logradouro: { type: "text" },
              numero: { type: "keyword" },
              bairro: { type: "text" },
              cep: { type: "keyword" },
              uf: { type: "keyword" },
              municipio: { type: "keyword" },
            },
          },
          // "nested", nao "object": com object, procurar um socio chamado
          // MARIA com qualificacao 10 acharia a empresa onde MARIA tem outra
          // qualificacao e outro socio e que tem a 10.
          socios: {
            type: "nested",
            properties: {
              nomeSocio: { type: "text" },
              cpfCnpjSocio: { type: "keyword" },
              qualificacaoSocio: { type: "keyword" },
              dataEntradaSociedade: { type: "date" },
            },
          },
        },
      },
    }),
  });

  if (!resposta.ok) {
    throw new Error("Nao consegui criar o indice: HTTP " + resposta.status);
  }
}

Dois pontos que eu não sabia e custaram caro.

O nested dos sócios. Se você deixar como objeto comum (o padrão), o Elasticsearch "achata" a lista: os nomes viram uma lista e as qualificações viram outra, e a ligação entre eles some. Resultado: procurar um sócio chamado MARIA com qualificação 10 traz também a empresa onde a MARIA é outra coisa e um outro sócio é que tem a 10. Com nested, cada sócio vira um documentinho próprio e a busca combinada volta a fazer sentido.

O refresh_interval: -1. Por padrão o Elasticsearch torna os documentos pesquisáveis a cada segundo, e isso custa: numa carga de milhões de documentos ele cria milhares de segmentos e gasta a carga inteira juntando arquivo em vez de indexar. Desligar durante a carga e religar no fim é o que faz a diferença entre horas e dias.

⚠️ A armadilha que faz documento sumir em silêncio

Agora a parte mais importante do artigo. O envio em lote é simples — uma linha de ação, uma linha de documento, alternadas:

async function enviarLote(lote) {
  let corpo = "";
  for (const doc of lote) {
    corpo += JSON.stringify({ index: { _index: INDICE, _id: doc.cnpj } }) + "\n";
    corpo += JSON.stringify(doc) + "\n";
  }

  const resposta = await fetch(ELASTIC + "/_bulk", {
    method: "POST",
    headers: { "Content-Type": "application/x-ndjson" },
    body: corpo,
  });

  if (!resposta.ok) {
    throw new Error("Elasticsearch respondeu " + resposta.status + " no _bulk");
  }

  const retorno = await resposta.json();

  // A armadilha mais cara desta importacao: o _bulk responde HTTP 200 mesmo
  // quando documentos foram recusados um a um. Sem olhar o campo "errors",
  // registros somem sem nenhum sinal e voce so descobre dias depois.
  if (retorno.errors) {
    let recusados = 0;
    let exemplo = "";
    for (const item of retorno.items) {
      if (item.index.error) {
        recusados++;
        if (exemplo === "") {
          exemplo = item.index._id + ": " + item.index.error.reason;
        }
      }
    }
    throw new Error(recusados + " documento(s) recusado(s) pelo Elasticsearch. Primeiro: " + exemplo);
  }

  return lote.length;
}

Olhe de novo aquele if (retorno.errors). Ele parece um detalhe de tratamento de erro, e é a linha mais importante do arquivo inteiro.

Motivo: o _bulk devolve HTTP 200 mesmo quando recusa documentos. Um por um. O status da requisição é sucesso, o resposta.ok é true, e lá dentro do corpo tem um campo errors: true que ninguém olha.

Eu quis ver o tamanho do estrago, então troquei o if (retorno.errors) por if (false) de propósito e rodei a carga contra um índice que recusava as datas. Esta é a saída:

Carregando empresas e socios no SQLite...
  lendo K3241.K03200Y0.D60808.EMPRECSV
  lendo K3241.K03200Y0.D60808.SOCIOCSV
  criando o indice de socios...
  5 empresas, 7 socios
Criando o indice no Elasticsearch...
Indexando estabelecimentos...
  lendo K3241.K03200Y0.D60808.ESTABELE
Pronto: 8 documentos no indice.

$ echo $?
0

$ curl -s "http://localhost:9200/cnpj/_count"
{"count":0}

"Pronto: 8 documentos no indice." E o índice tem zero. 😱 O programa terminou com código de saída 0, anunciou sucesso, e não gravou nada. Numa carga de verdade isso seria uma base pela metade, com cara de completa — e você descobrindo semanas depois que faltam empresas.

Com a verificação no lugar, o mesmo cenário falha na hora e diz por quê:

Indexando estabelecimentos...
  lendo K3241.K03200Y0.D60808.ESTABELE
Falhou: 8 documento(s) recusado(s) pelo Elasticsearch. Primeiro: 41298134000622:
        failed to parse field [dataInicioAtividade] of type [date]

$ echo $?
1

Código de saída 1, mensagem clara e o CNPJ do primeiro documento recusado. É a diferença entre um erro que você conserta em cinco minutos e um que te custa a base inteira.

🔁 O laço que junta tudo

Com as peças prontas, a indexação é ler os estabelecimentos, buscar os companheiros no SQLite e mandar de mil em mil:

async function indexarEstabelecimentos(banco) {
  const buscarEmpresa = banco.prepare("SELECT dados FROM empresas WHERE cnpj_basico = ?");
  const buscarSocios = banco.prepare("SELECT dados FROM socios WHERE cnpj_basico = ?");

  let lote = [];
  let enviados = 0;

  for (const arquivo of acharArquivos("ESTABELE")) {
    console.log("  lendo " + path.basename(arquivo));
    for await (const campos of lerLinhas(arquivo)) {
      const estabelecimento = montarObjeto(COLUNAS_ESTABELECIMENTO, campos);

      const linhaEmpresa = buscarEmpresa.get(estabelecimento.cnpjBasico);
      const empresa = linhaEmpresa ? JSON.parse(linhaEmpresa.dados) : null;

      const socios = buscarSocios.all(estabelecimento.cnpjBasico)
        .map(function (linha) { return JSON.parse(linha.dados); });

      lote.push(montarDocumento(estabelecimento, empresa, socios));

      if (lote.length >= 1000) {
        enviados += await enviarLote(lote);
        lote = [];
        console.log("  " + enviados.toLocaleString("pt-BR") + " documentos indexados");
      }
    }
  }

  // O resto do ultimo lote precisa ir. Esquecer esta linha faz sumir ate
  // 999 empresas -- e a contagem final quase bate, o que engana.
  if (lote.length > 0) {
    enviados += await enviarLote(lote);
  }

  return enviados;
}

Aquele último if depois do laço é o tipo de linha que a gente esquece. Sem ele, o que sobrou no último lote — até 999 empresas — nunca é enviado. E o pior é que a contagem final fica quase certa, o que não levanta suspeita nenhuma.

🎬 O main, e o refresh que precisa voltar

async function main() {
  const banco = abrirBanco();

  try {
    console.log("Carregando empresas e socios no SQLite...");
    const contagem = await carregarNoSqlite(banco);
    console.log("  " + contagem.empresas.toLocaleString("pt-BR") + " empresas, " +
      contagem.socios.toLocaleString("pt-BR") + " socios");

    console.log("Criando o indice no Elasticsearch...");
    await criarIndice();

    console.log("Indexando estabelecimentos...");
    const total = await indexarEstabelecimentos(banco);

    // Sem religar o refresh, o indice fica invisivel para qualquer busca:
    // os documentos estao la, mas nenhuma consulta os enxerga.
    await fetch(ELASTIC + "/" + INDICE + "/_settings", {
      method: "PUT",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ index: { refresh_interval: "1s" } }),
    });
    await fetch(ELASTIC + "/" + INDICE + "/_refresh", { method: "POST" });

    console.log("Pronto: " + total.toLocaleString("pt-BR") + " documentos no indice.");
  } catch (erro) {
    console.error("Falhou: " + erro.message);
    process.exitCode = 1;
  } finally {
    banco.close();
    // O SQLite passa de 15 GB com a base inteira. Apagamos sempre, inclusive
    // quando a carga falha, senao fica um arquivo enorme esquecido no disco.
    if (fs.existsSync(BANCO)) fs.rmSync(BANCO);
  }
}

main();

Duas coisas moram no finally por um motivo. O refresh precisa voltar: se a carga morrer no meio com refresh_interval: -1, o índice fica invisível para sempre — os documentos estão gravados, mas nenhuma busca os enxerga, e você jura que a carga não funcionou. E o SQLite precisa ser apagado mesmo quando dá erro, senão sobra um arquivo de 15 GB escondido no disco que ninguém lembra de limpar.

▶️ Rodando

CNPJ_PASTA=./dados ELASTIC_URL=http://localhost:9200 node importar.js

E a saída de uma carga que deu certo:

Carregando empresas e socios no SQLite...
  lendo K3241.K03200Y0.D60808.EMPRECSV
  lendo K3241.K03200Y0.D60808.SOCIOCSV
  criando o indice de socios...
  5 empresas, 7 socios
Criando o indice no Elasticsearch...
Indexando estabelecimentos...
  lendo K3241.K03200Y0.D60808.ESTABELE
Pronto: 8 documentos no indice.

🎯 A empresa do começo, agora pesquisável

Lembra da linha crua do CSV lá no início? Depois de passar pelas duas etapas, ela virou este documento no Elasticsearch:

{
  "cnpj": "41300376000107",
  "cnpjBasico": "41300376",
  "razaoSocial": "JAIBA SE2 ENERGIAS RENOVAVEIS S.A.",
  "nomeFantasia": null,
  "porte": "05",
  "capitalSocial": 79891408.93,
  "naturezaJuridica": "2054",
  "situacaoCadastral": "02",
  "dataSituacaoCadastral": "2021-03-22",
  "dataInicioAtividade": "2021-03-22",
  "cnaePrincipal": "3511501",
  "cnaesSecundarios": ["3313901", "6821802", "7739099"],
  "endereco": {
    "logradouro": "MARQUES",
    "numero": "S/N",
    "bairro": "ZONA RURAL",
    "cep": "39508000",
    "uf": "MG",
    "municipio": "2893"
  },
  "socios": [
    {
      "nomeSocio": "FULANO DE TAL",
      "cpfCnpjSocio": "***000000**",
      "qualificacaoSocio": "10",
      "dataEntradaSociedade": "2024-05-27"
    },
    {
      "nomeSocio": "BELTRANA DA SILVA",
      "cpfCnpjSocio": "***000000**",
      "qualificacaoSocio": "10",
      "dataEntradaSociedade": "2026-06-15"
    }
  ]
}

Compare com a linha crua: a vírgula do capital virou número de verdade (79891408.93), o "20210322" virou data ISO, a string de três CNAEs virou array, e os sócios — que moravam num arquivo completamente separado — chegaram junto. Agora dá para perguntar coisas como "empresas de energia em Minas Gerais abertas depois de 2020" numa consulta só. 🔎

Por hoje é só, meus unicórnios! 🦄✨

Que a magia do arco-íris continue brilhando em suas vidas! Até mais! 🌈🌟