COMERCIAL-GTI/scripts/extrair-base-cnpj.mjs

import path from 'node:path';
import { loadDotEnv, paths } from '../src/env.mjs';
import { openDb, ensureSeeds } from '../src/db.mjs';
import { extractUf } from '../src/extract/rfb-extractor.mjs';

// Extrai a base CNPJ de uma UF inteira a partir dos zips da Receita Federal:
//   node scripts/extrair-base-cnpj.mjs --uf=BA
// Pre-requisito: zips baixados em prospeccao/cnpj/rfb-raw (scripts/baixar-base-rfb.sh).

await loadDotEnv();

const args = new Map(
  process.argv.slice(2).map((arg) => {
    const [key, value = 'true'] = arg.replace(/^--/, '').split('=');
    return [key, value];
  })
);

const uf = (args.get('uf') || '').toUpperCase();
if (!/^[A-Z]{2}$/.test(uf)) {
  console.error('Uso: node scripts/extrair-base-cnpj.mjs --uf=BA');
  process.exit(1);
}

const rawDir = args.get('raw-dir') || paths.rfbRaw;
const outCsv = path.join(paths.cnpjDir, `estabelecimentos-${uf}.csv`);
const checkpointPath = path.join(paths.cnpjDir, `checkpoint-estabelecimentos-${uf}.ndjson`);

async function main() {
  const db = openDb();
  await ensureSeeds(db, { log: (m) => console.log(`[extrair] ${m}`) });

  const inserted = db.prepare(
    "INSERT INTO extractions (uf, status, started_at) VALUES (?, 'running', datetime('now'))"
  ).run(uf);
  const extractionId = Number(inserted.lastInsertRowid);
  db.prepare('UPDATE extractions SET pid = ? WHERE id = ?').run(process.pid, extractionId);

  try {
    const result = await extractUf({
      uf,
      rawDir,
      outCsv,
      checkpointPath,
      db,
      log: (m) => console.log(`[extrair] ${m}`),
      onProgress: (p) => {
        console.log(`[extrair] ${p.pct}% ${p.phase}: ${p.message}`);
        db.prepare('UPDATE extractions SET progress_json = ? WHERE id = ?')
          .run(JSON.stringify(p), extractionId);
      }
    });

    if (result.cancelled) {
      db.prepare("UPDATE extractions SET status = 'cancelled', finished_at = datetime('now') WHERE id = ?")
        .run(extractionId);
    } else {
      db.prepare(
        `UPDATE extractions SET status = 'done', csv_path = ?, total_rows = ?, with_phone = ?,
         municipios_count = ?, finished_at = datetime('now') WHERE id = ?`
      ).run(result.csvPath, result.totalRows, result.withPhone, result.municipiosCount, extractionId);
    }
  } catch (error) {
    db.prepare(
      "UPDATE extractions SET status = 'error', error = ?, finished_at = datetime('now') WHERE id = ?"
    ).run(String(error.message || error), extractionId);
    throw error;
  } finally {
    db.close();
  }
}

main().catch((error) => {
  console.error(`[extrair] erro: ${error.stack || error.message}`);
  process.exit(1);
});