COMERCIAL-GTI/src/extract/rfb-extractor.mjs

import { spawn, spawnSync } from 'node:child_process';
import readline from 'node:readline';
import fs from 'node:fs';
import fsp from 'node:fs/promises';
import path from 'node:path';
import { cleanText, stripAccents } from '../lead-engine.mjs';
import { copyStr } from '../util/collections.mjs';
import { csvEscape, BOM } from '../util/csv.mjs';

// Extrai a base de estabelecimentos ativos de UMA UF inteira (todos os
// municipios) a partir dos zips brutos da Receita Federal, gerando
// prospeccao/cnpj/estabelecimentos-<UF>.csv.
//
// Cuidados de memoria (aprendidos nas execucoes da Bahia):
//  - checkpoint em NDJSON streamado para disco: nunca guarda o array de
//    matches em memoria (uma UF inteira passa de 1 milhao de linhas);
//  - copyStr (Buffer.from) quebra o vinculo SlicedString do V8 entre campos
//    retidos e a linha CSV original;
//  - firstField evita o split completo das ~50 milhoes de linhas descartadas
//    dos Empresas*.zip.

const CANDIDATE_READERS = [
  { name: 'unzip', probeArgs: ['-v'], argsFor: (zipPath) => ['-p', zipPath] },
  { name: 'C:\\Program Files\\Git\\usr\\bin\\unzip.exe', probeArgs: ['-v'], argsFor: (zipPath) => ['-p', zipPath] },
  { name: 'tar', probeArgs: ['--version'], argsFor: (zipPath) => ['-xOf', zipPath] },
  { name: 'C:\\Windows\\System32\\tar.exe', probeArgs: ['--version'], argsFor: (zipPath) => ['-xOf', zipPath] }
];

export function resolveZipReader() {
  for (const candidate of CANDIDATE_READERS) {
    try {
      const probe = spawnSync(candidate.name, candidate.probeArgs, { stdio: 'ignore' });
      if (!probe.error && probe.status === 0) return candidate;
    } catch {
      // tenta o proximo
    }
  }
  throw new Error(
    'Nenhum leitor de ZIP encontrado (unzip ou tar). Instale o Git para Windows ou verifique o PATH.'
  );
}

// Abre um zip da RFB e devolve as linhas ja decodificadas de ISO-8859-1.
function readZipLines(reader, zipPath) {
  const child = spawn(reader.name, reader.argsFor(zipPath), { stdio: ['ignore', 'pipe', 'pipe'] });
  child.stderr.on('data', () => {});
  child.stdout.setEncoding('latin1');
  const rl = readline.createInterface({ input: child.stdout, crlfDelay: Infinity });
  return {
    rl,
    close() {
      rl.close();
      child.kill();
    }
  };
}

// So o primeiro campo (antes do primeiro '";"'), sem split completo.
function firstField(line) {
  const end = line.indexOf('";"', 1);
  if (end === -1) return '';
  return line.slice(1, end);
}

function parseQuotedRow(line) {
  if (line.startsWith('"') && line.endsWith('"')) {
    return line.slice(1, -1).split('";"');
  }
  return line.split(';').map((value) => value.replace(/^"|"$/g, ''));
}

function titleCase(name) {
  return String(name).toLowerCase().replace(/(^|\s)([a-z])/g, (m, sep, ch) => sep + ch.toUpperCase());
}

async function buildTomMap(reader, rawDir) {
  const zipPath = path.join(rawDir, 'Municipios.zip');
  if (!fs.existsSync(zipPath)) {
    throw new Error(`Municipios.zip nao encontrado em ${rawDir}`);
  }

  const tomToNome = new Map();
  const zip = readZipLines(reader, zipPath);
  for await (const line of zip.rl) {
    if (!line.trim()) continue;
    const [codigo, nome] = parseQuotedRow(line);
    if (codigo) tomToNome.set(copyStr(codigo), copyStr(nome));
  }
  zip.close();
  return tomToNome;
}

async function* estabelecimentoRows(reader, rawDir, uf, { onFileProgress, isCancelled }) {
  for (let i = 0; i <= 9; i += 1) {
    const zipPath = path.join(rawDir, `Estabelecimentos${i}.zip`);
    if (!fs.existsSync(zipPath)) continue;

    const zip = readZipLines(reader, zipPath);
    let lineCount = 0;
    let matchCount = 0;

    for await (const line of zip.rl) {
      lineCount += 1;
      if (lineCount % 500000 === 0 && isCancelled()) {
        zip.close();
        return;
      }
      if (!line.trim()) continue;

      // Filtro barato por UF antes do parse completo: a UF aparece como
      // campo '";UF";"' — o indexOf evita o split das ~29M linhas de outras UFs.
      if (!line.includes(`"${uf}"`)) continue;

      const cols = parseQuotedRow(line);
      if (cols[19] !== uf) continue;
      if (cols[5] !== '02') continue;

      matchCount += 1;
      yield {
        b: copyStr(cols[0]),   // cnpj basico
        o: copyStr(cols[1]),   // ordem
        d: copyStr(cols[2]),   // dv
        f: copyStr(cols[4]),   // nome fantasia
        c: copyStr(cols[11]),  // cnae
        l: copyStr([cols[13], cols[14]].filter(Boolean).join(' ')), // logradouro
        n: copyStr(cols[15]),  // numero
        ba: copyStr(cols[17]), // bairro
        m: copyStr(cols[20]),  // codigo municipio (TOM)
        d1: copyStr(cols[21]),
        t1: copyStr(cols[22]),
        d2: copyStr(cols[23]),
        t2: copyStr(cols[24])
      };
    }

    zip.close();
    onFileProgress(`Estabelecimentos${i}.zip`, lineCount, matchCount, i);
    if (global.gc) global.gc();
    if (isCancelled()) return;
  }
}

async function buildRazaoMap(reader, rawDir, cnpjBasicoSet, { onFileProgress, isCancelled }) {
  const razaoByCnpj = new Map();

  for (let i = 0; i <= 9; i += 1) {
    const zipPath = path.join(rawDir, `Empresas${i}.zip`);
    if (!fs.existsSync(zipPath)) continue;

    const zip = readZipLines(reader, zipPath);
    let lineCount = 0;
    let matchCount = 0;

    for await (const line of zip.rl) {
      lineCount += 1;
      if (lineCount % 500000 === 0 && isCancelled()) {
        zip.close();
        return razaoByCnpj;
      }
      if (!line.trim()) continue;
      const cnpjBasico = firstField(line);
      if (!cnpjBasico || !cnpjBasicoSet.has(cnpjBasico)) continue;
      const cols = parseQuotedRow(line);
      razaoByCnpj.set(copyStr(cnpjBasico), copyStr(cols[1]));
      matchCount += 1;
    }

    zip.close();
    onFileProgress(`Empresas${i}.zip`, lineCount, matchCount, i);
    if (global.gc) global.gc();
    if (isCancelled()) break;
  }

  return razaoByCnpj;
}

export async function extractUf({
  uf,
  rawDir,
  outCsv,
  checkpointPath,
  db = null,
  onProgress = () => {},
  isCancelled = () => false,
  log = () => {}
}) {
  const ufUpper = String(uf).toUpperCase();
  const reader = resolveZipReader();
  log(`leitor de zip: ${reader.name}`);

  const tomToNome = await buildTomMap(reader, rawDir);
  log(`${tomToNome.size} municipios no indice da RFB`);

  // Nome "bonito" (IBGE, sem acento, capitalizado) por codigo TOM, para o CSV
  // ficar consistente com os leads ja migrados ('Santo Antonio de Jesus').
  const prettyByClean = new Map();
  if (db) {
    for (const row of db.prepare('SELECT nome, nome_clean FROM municipios WHERE uf = ?').all(ufUpper)) {
      prettyByClean.set(row.nome_clean, stripAccents(row.nome));
    }
  }
  const prettyName = (tomCode) => {
    const tomNome = tomToNome.get(tomCode) || '';
    return prettyByClean.get(cleanText(tomNome)) || titleCase(tomNome);
  };

  // ---- Fase 1: Estabelecimentos -> checkpoint NDJSON em disco ----
  const cnpjBasicoSet = new Set();
  const tomSeen = new Set();
  let totalRows = 0;
  let withPhone = 0;

  const checkpointDone = fs.existsSync(checkpointPath);
  if (checkpointDone) {
    log('checkpoint encontrado: pulando varredura de Estabelecimentos');
    await new Promise((resolve, reject) => {
      const rl = readline.createInterface({
        input: fs.createReadStream(checkpointPath, { encoding: 'utf8' }),
        crlfDelay: Infinity
      });
      rl.on('line', (line) => {
        if (!line.trim()) return;
        const item = JSON.parse(line);
        cnpjBasicoSet.add(item.b);
        tomSeen.add(item.m);
        totalRows += 1;
        if (item.t1 || item.t2) withPhone += 1;
      });
      rl.on('close', resolve);
      rl.on('error', reject);
    });
  } else {
    const tmpPath = `${checkpointPath}.tmp`;
    const out = fs.createWriteStream(tmpPath, { encoding: 'utf8' });

    onProgress({ phase: 'estabelecimentos', pct: 0, message: 'varrendo Estabelecimentos0-9.zip' });

    for await (const item of estabelecimentoRows(reader, rawDir, ufUpper, {
      isCancelled,
      onFileProgress: (file, lines, matches, index) => {
        log(`${file}: ${lines} linhas, ${matches} ativas em ${ufUpper}`);
        onProgress({
          phase: 'estabelecimentos',
          pct: Math.round(((index + 1) / 10) * 55),
          message: `${file}: ${totalRows} estabelecimentos ate agora`
        });
      }
    })) {
      cnpjBasicoSet.add(item.b);
      tomSeen.add(item.m);
      totalRows += 1;
      if (item.t1 || item.t2) withPhone += 1;
      if (!out.write(JSON.stringify(item) + '\n')) {
        await new Promise((resolve) => out.once('drain', resolve));
      }
    }

    await new Promise((resolve, reject) => out.end((e) => (e ? reject(e) : resolve())));

    if (isCancelled()) {
      await fsp.rm(tmpPath, { force: true });
      return { cancelled: true };
    }
    await fsp.rename(tmpPath, checkpointPath);
    log(`checkpoint salvo: ${checkpointPath}`);
  }

  log(`total de estabelecimentos ativos em ${ufUpper}: ${totalRows}`);
  if (isCancelled()) return { cancelled: true };

  // ---- Fase 2: Empresas -> razao social ----
  onProgress({ phase: 'empresas', pct: 55, message: 'varrendo Empresas0-9.zip (razao social)' });
  let razoesRecuperadas = 0;
  const razaoByCnpj = await buildRazaoMap(reader, rawDir, cnpjBasicoSet, {
    isCancelled,
    onFileProgress: (file, lines, matches, index) => {
      razoesRecuperadas += matches;
      const heapMb = (process.memoryUsage().heapUsed / 1024 / 1024).toFixed(0);
      log(`${file}: ${lines} linhas, ${matches} razoes sociais, heap=${heapMb}MB`);
      onProgress({
        phase: 'empresas',
        pct: 55 + Math.round(((index + 1) / 10) * 35),
        message: `${file}: ${razoesRecuperadas} razoes sociais`
      });
    }
  });
  if (isCancelled()) return { cancelled: true };

  // ---- Fase 3: CSV final (streaming do checkpoint) ----
  onProgress({ phase: 'csv', pct: 90, message: 'gravando CSV final' });
  const headers = [
    'cnpj', 'razao_social', 'nome_fantasia', 'ddd_1', 'telefone_1', 'ddd_2', 'telefone_2',
    'municipio', 'municipio_codigo', 'uf', 'cnae', 'logradouro', 'numero', 'bairro'
  ];
  const csvOut = fs.createWriteStream(outCsv, { encoding: 'utf8' });
  csvOut.write(BOM + headers.join(';') + '\n');

  await new Promise((resolve, reject) => {
    const rl = readline.createInterface({
      input: fs.createReadStream(checkpointPath, { encoding: 'utf8' }),
      crlfDelay: Infinity
    });
    // readline ainda entrega linhas ja em buffer depois do pause(); o flag
    // evita empilhar um listener de drain por linha excedente.
    let waitingDrain = false;
    rl.on('line', (line) => {
      if (!line.trim()) return;
      const item = JSON.parse(line);
      const row = [
        `${item.b}${item.o}${item.d}`,
        razaoByCnpj.get(item.b) || '',
        item.f,
        item.d1, item.t1, item.d2, item.t2,
        prettyName(item.m),
        item.m,
        ufUpper,
        item.c,
        item.l, item.n, item.ba
      ].map(csvEscape).join(';');
      if (!csvOut.write(row + '\n') && !waitingDrain) {
        waitingDrain = true;
        rl.pause();
        csvOut.once('drain', () => {
          waitingDrain = false;
          rl.resume();
        });
      }
    });
    rl.on('close', resolve);
    rl.on('error', reject);
  });
  await new Promise((resolve, reject) => csvOut.end((e) => (e ? reject(e) : resolve())));

  // ---- Fase 4: aprende codigo TOM dos municipios no banco ----
  let municipiosCount = tomSeen.size;
  if (db) {
    const update = db.prepare('UPDATE municipios SET codigo_tom = ? WHERE uf = ? AND nome_clean = ?');
    db.exec('BEGIN');
    try {
      for (const tom of tomSeen) {
        const nome = tomToNome.get(tom);
        if (nome) update.run(tom, ufUpper, cleanText(nome));
      }
      db.exec('COMMIT');
    } catch (error) {
      db.exec('ROLLBACK');
      throw error;
    }
  }

  // O checkpoint so serve para retomar uma extracao interrompida. Mantido apos
  // concluir, faria uma re-extracao futura (base nova da Receita) reutilizar
  // dados velhos da fase 1 — entao e apagado no sucesso.
  await fsp.rm(checkpointPath, { force: true });

  onProgress({ phase: 'done', pct: 100, message: `${totalRows} estabelecimentos, ${withPhone} com telefone` });
  log(`CSV gravado: ${outCsv}`);
  log(`com telefone: ${withPhone} (${totalRows ? ((withPhone / totalRows) * 100).toFixed(1) : 0}%)`);

  return { cancelled: false, totalRows, withPhone, municipiosCount, csvPath: outCsv };
}