chore: import maquina-comercial-pessoal history via subtree
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> git-subtree-dir: maquina-comercial-pessoal git-subtree-mainline:f5af038e19git-subtree-split:a9ef4d776d
This commit is contained in:
Gustavo
Claude Sonnet 5
@@ -0,0 +1,100 @@
|
||||
import path from 'node:path';
|
||||
import { loadDotEnv, paths } from '../env.mjs';
|
||||
import { openDb } from '../db.mjs';
|
||||
import { extractUf } from './rfb-extractor.mjs';
|
||||
import { downloadRfbDataset, validateRfbRawFiles } from './rfb-downloader.mjs';
|
||||
import { createJobIO } from '../jobs/protocol.mjs';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { createJobLease } from '../jobs/lease.mjs';
|
||||
|
||||
// Processo filho de extracao RFB: le a UF da linha em extractions e roda a
|
||||
// extracao completa, reportando progresso via stdout.
|
||||
|
||||
await loadDotEnv();
|
||||
|
||||
const args = new Map(
|
||||
process.argv.slice(2).map((arg) => {
|
||||
const [key, value = 'true'] = arg.replace(/^--/, '').split('=');
|
||||
return [key, value];
|
||||
})
|
||||
);
|
||||
|
||||
const io = createJobIO();
|
||||
|
||||
async function main() {
|
||||
const extractionId = Number(args.get('extraction-id'));
|
||||
if (!extractionId) throw new Error('extract-job: --extraction-id obrigatorio');
|
||||
|
||||
const db = openDb();
|
||||
let lease;
|
||||
try {
|
||||
const extraction = db.prepare('SELECT * FROM extractions WHERE id = ?').get(extractionId);
|
||||
if (!extraction) throw new Error(`extraction #${extractionId} nao encontrada`);
|
||||
const uf = extraction.uf;
|
||||
lease = createJobLease({ db, table: 'extractions', id: extractionId, ownerToken: args.get('owner-token') || randomUUID() });
|
||||
const outCsv = path.join(paths.cnpjDir, `estabelecimentos-${uf}.csv`);
|
||||
const checkpointPath = path.join(paths.cnpjDir, `checkpoint-estabelecimentos-${uf}.ndjson`);
|
||||
const download = await downloadRfbDataset({
|
||||
rawDir: paths.rfbRaw,
|
||||
isCancelled: io.isCancelled,
|
||||
onProgress: (progress) => {
|
||||
const mapped = { ...progress, pct: Math.round((progress.pct || 0) * 0.45) };
|
||||
io.emit(mapped);
|
||||
db.prepare('UPDATE extractions SET progress_json = ? WHERE id = ?')
|
||||
.run(JSON.stringify(mapped), extractionId);
|
||||
}
|
||||
});
|
||||
if (download.cancelled) {
|
||||
db.prepare("UPDATE extractions SET status = 'cancelled', finished_at = datetime('now') WHERE id = ?").run(extractionId);
|
||||
io.done({ cancelled: true, snapshot: download.snapshot, downloaded: download.downloaded.length });
|
||||
return;
|
||||
}
|
||||
await validateRfbRawFiles(paths.rfbRaw);
|
||||
const result = await extractUf({
|
||||
uf,
|
||||
rawDir: paths.rfbRaw,
|
||||
outCsv,
|
||||
checkpointPath,
|
||||
db,
|
||||
log: (m) => io.log(m),
|
||||
isCancelled: io.isCancelled,
|
||||
onProgress: (p) => {
|
||||
const mapped = { ...p, pct: 45 + Math.round((p.pct || 0) * 0.55) };
|
||||
io.emit(mapped);
|
||||
db.prepare('UPDATE extractions SET progress_json = ? WHERE id = ?')
|
||||
.run(JSON.stringify(mapped), extractionId);
|
||||
}
|
||||
});
|
||||
|
||||
if (result.cancelled) {
|
||||
db.prepare("UPDATE extractions SET status = 'cancelled', finished_at = datetime('now') WHERE id = ?")
|
||||
.run(extractionId);
|
||||
io.done({ cancelled: true });
|
||||
} 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);
|
||||
io.done({
|
||||
snapshot: download.snapshot,
|
||||
downloaded: download.downloaded.length,
|
||||
skipped: download.skipped.length,
|
||||
totalRows: result.totalRows,
|
||||
withPhone: result.withPhone,
|
||||
municipios: result.municipiosCount
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
db.prepare("UPDATE extractions SET status = 'error', error = ?, finished_at = datetime('now') WHERE id = ? AND status IN ('pending','running')")
|
||||
.run(String(error.message || error), extractionId);
|
||||
throw error;
|
||||
} finally {
|
||||
lease?.close();
|
||||
db.close();
|
||||
}
|
||||
}
|
||||
|
||||
main().catch((error) => {
|
||||
io.fail(error);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -0,0 +1,576 @@
|
||||
import fs from 'node:fs';
|
||||
import fsp from 'node:fs/promises';
|
||||
import path from 'node:path';
|
||||
import { Readable, Transform } from 'node:stream';
|
||||
import { pipeline } from 'node:stream/promises';
|
||||
import { fetchTextWithTimeout } from '../util/fetch.mjs';
|
||||
|
||||
export const REQUIRED_RFB_FILES = [
|
||||
'Municipios.zip',
|
||||
...Array.from({ length: 10 }, (_, index) => `Empresas${index}.zip`),
|
||||
...Array.from({ length: 10 }, (_, index) => `Estabelecimentos${index}.zip`)
|
||||
];
|
||||
|
||||
export const DEFAULT_RFB_INDEX_URL = 'https://dados-abertos-rf-cnpj.casadosdados.com.br/arquivos/';
|
||||
const DEFAULT_MIN_ZIP_BYTES = 1024;
|
||||
const MANIFEST_FILE = '.rfb-snapshot.json';
|
||||
const BACKUP_DIR = '.rfb-backup';
|
||||
const INCOMPLETE_DIR = '.rfb-incomplete';
|
||||
const EXTRACTION_MARGIN_BYTES = 2 * 1024 * 1024 * 1024;
|
||||
// Covers the compressed download, extraction checkpoints and generated CSV.
|
||||
export const DEFAULT_MINIMUM_FREE_BYTES = 8 * 1024 * 1024 * 1024;
|
||||
|
||||
class CancelledError extends Error {}
|
||||
|
||||
export function parseLatestSnapshot(html) {
|
||||
const dates = [...String(html).matchAll(/href\s*=\s*["'](\d{4}-\d{2}-\d{2})\/["']/gi)]
|
||||
.map((match) => match[1])
|
||||
.filter((date) => !Number.isNaN(Date.parse(`${date}T00:00:00Z`)));
|
||||
if (!dates.length) throw new Error('nenhum snapshot YYYY-MM-DD encontrado no indice da RFB');
|
||||
return dates.sort().at(-1);
|
||||
}
|
||||
|
||||
async function fetchResponse(fetchImpl, url, options = {}, timeoutMs = 25_000) {
|
||||
const controller = new AbortController();
|
||||
const signal = options.signal ? AbortSignal.any([options.signal, controller.signal]) : controller.signal;
|
||||
const timer = setTimeout(() => controller.abort(new Error(`timeout apos ${timeoutMs}ms`)), timeoutMs);
|
||||
timer.unref?.();
|
||||
try {
|
||||
return await fetchImpl(url, { ...options, signal });
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
export async function discoverLatestSnapshot(indexUrl = DEFAULT_RFB_INDEX_URL, { fetch: fetchImpl = globalThis.fetch } = {}) {
|
||||
const normalizedUrl = indexUrl.endsWith('/') ? indexUrl : `${indexUrl}/`;
|
||||
const { response, text } = await fetchTextWithTimeout(normalizedUrl, { fetch: fetchImpl }, 25_000, { maxBytes: 2_000_000 });
|
||||
if (!response.ok) throw new Error(`indice da RFB respondeu HTTP ${response.status}`);
|
||||
return parseLatestSnapshot(text);
|
||||
}
|
||||
|
||||
async function inspectZip(file, minZipBytes, expectedSize) {
|
||||
const stat = await fsp.stat(file);
|
||||
if (stat.size < minZipBytes) throw new Error(`${path.basename(file)} invalido: tamanho ${stat.size}, minimo ${minZipBytes}`);
|
||||
if (expectedSize != null && stat.size !== expectedSize) {
|
||||
throw new Error(`${path.basename(file)} incompleto: ${stat.size} bytes, esperado ${expectedSize}`);
|
||||
}
|
||||
const handle = await fsp.open(file, 'r');
|
||||
try {
|
||||
const signature = Buffer.alloc(2);
|
||||
await handle.read(signature, 0, 2, 0);
|
||||
if (signature.toString('ascii') !== 'PK') throw new Error(`${path.basename(file)} invalido: assinatura ZIP PK ausente`);
|
||||
const tailSize = Math.min(stat.size, 65_557);
|
||||
const tail = Buffer.alloc(tailSize);
|
||||
await handle.read(tail, 0, tailSize, stat.size - tailSize);
|
||||
let eocdOffset = -1;
|
||||
for (let offset = tail.length - 22; offset >= 0; offset -= 1) {
|
||||
if (tail.readUInt32LE(offset) !== 0x06054b50) continue;
|
||||
const commentLength = tail.readUInt16LE(offset + 20);
|
||||
if (offset + 22 + commentLength === tail.length) {
|
||||
eocdOffset = offset;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (eocdOffset < 0) throw new Error(`${path.basename(file)} invalido: registro EOCD ausente ou truncado`);
|
||||
|
||||
const disk = tail.readUInt16LE(eocdOffset + 4);
|
||||
const centralDisk = tail.readUInt16LE(eocdOffset + 6);
|
||||
const entriesOnDisk = tail.readUInt16LE(eocdOffset + 8);
|
||||
const entriesTotal = tail.readUInt16LE(eocdOffset + 10);
|
||||
const centralSize = tail.readUInt32LE(eocdOffset + 12);
|
||||
const centralOffset = tail.readUInt32LE(eocdOffset + 16);
|
||||
if (disk === 0xffff || centralDisk === 0xffff || entriesOnDisk === 0xffff || entriesTotal === 0xffff
|
||||
|| centralSize === 0xffffffff || centralOffset === 0xffffffff) {
|
||||
throw new Error(`${path.basename(file)} invalido: ZIP64 nao suportado`);
|
||||
}
|
||||
if (disk !== 0 || centralDisk !== 0) throw new Error(`${path.basename(file)} invalido: ZIP multi-disco nao suportado`);
|
||||
if (entriesOnDisk !== entriesTotal) throw new Error(`${path.basename(file)} invalido: contagem do diretorio central divergente`);
|
||||
const absoluteEocd = stat.size - tailSize + eocdOffset;
|
||||
if (centralOffset + centralSize > absoluteEocd) {
|
||||
throw new Error(`${path.basename(file)} invalido: diretorio central ultrapassa o EOCD`);
|
||||
}
|
||||
if (entriesTotal > 0) {
|
||||
const centralSignature = Buffer.alloc(4);
|
||||
const { bytesRead } = await handle.read(centralSignature, 0, 4, centralOffset);
|
||||
if (bytesRead !== 4 || centralSignature.readUInt32LE(0) !== 0x02014b50) {
|
||||
throw new Error(`${path.basename(file)} invalido: assinatura do diretorio central ausente`);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
return stat.size;
|
||||
}
|
||||
|
||||
export async function validateRfbRawFiles(rawDir, { minZipBytes = DEFAULT_MIN_ZIP_BYTES, sizes = {} } = {}) {
|
||||
const problems = [];
|
||||
const files = [];
|
||||
for (const name of REQUIRED_RFB_FILES) {
|
||||
try {
|
||||
const expected = typeof sizes[name] === 'number' ? sizes[name] : sizes[name]?.size;
|
||||
const size = await inspectZip(path.join(rawDir, name), minZipBytes, expected);
|
||||
files.push({ name, size });
|
||||
} catch (error) {
|
||||
problems.push(`${name}: ${error.code === 'ENOENT' ? 'ausente' : error.message}`);
|
||||
}
|
||||
}
|
||||
if (problems.length) {
|
||||
throw new Error(`base RFB invalida; exige os 21 arquivos ZIP: ${problems.join('; ')}`);
|
||||
}
|
||||
return files;
|
||||
}
|
||||
|
||||
async function readManifest(rawDir) {
|
||||
try {
|
||||
return JSON.parse(await fsp.readFile(path.join(rawDir, MANIFEST_FILE), 'utf8'));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function writeManifest(rawDir, manifest) {
|
||||
const manifestPath = path.join(rawDir, MANIFEST_FILE);
|
||||
const tempPath = `${manifestPath}.tmp`;
|
||||
await fsp.writeFile(tempPath, `${JSON.stringify(manifest, null, 2)}\n`, 'utf8');
|
||||
await fsp.rename(tempPath, manifestPath);
|
||||
}
|
||||
|
||||
function totalFromResponse(response, offset) {
|
||||
const range = response.headers.get('content-range');
|
||||
if (range) {
|
||||
const match = /\/([0-9]+)$/.exec(range);
|
||||
if (match) return Number(match[1]);
|
||||
}
|
||||
const length = Number(response.headers.get('content-length'));
|
||||
return Number.isFinite(length) && length >= 0 ? offset + length : null;
|
||||
}
|
||||
|
||||
function responseValidator(response) {
|
||||
const etag = response.headers.get('etag');
|
||||
const lastModified = response.headers.get('last-modified');
|
||||
return { validator: etag || lastModified || null, etag, lastModified };
|
||||
}
|
||||
|
||||
function numericContentLength(response) {
|
||||
const raw = response.headers.get('content-length');
|
||||
if (!/^\d+$/.test(raw || '')) return null;
|
||||
const value = Number(raw);
|
||||
return Number.isSafeInteger(value) && value >= 0 ? value : null;
|
||||
}
|
||||
|
||||
async function downloadOne({ rawDir, snapshotUrl, name, fetchImpl, minZipBytes, isCancelled, onProgress, idleTimeoutMs, manifest }) {
|
||||
const finalPath = path.join(rawDir, name);
|
||||
const partPath = `${finalPath}.part`;
|
||||
let offset = 0;
|
||||
try {
|
||||
offset = (await fsp.stat(partPath)).size;
|
||||
} catch (error) {
|
||||
if (error.code !== 'ENOENT') throw error;
|
||||
}
|
||||
|
||||
manifest.partials ||= {};
|
||||
const saved = manifest.partials[name];
|
||||
const mayResume = offset > 0 && saved?.validator && Number.isSafeInteger(saved?.totalSize)
|
||||
&& saved.totalSize > offset;
|
||||
if (!mayResume) offset = 0;
|
||||
|
||||
let response;
|
||||
let expectedSize;
|
||||
let metadata;
|
||||
for (let attempt = 0; attempt < 2; attempt += 1) {
|
||||
const headers = offset ? { Range: `bytes=${offset}-`, 'If-Range': saved.validator } : {};
|
||||
response = await fetchResponse(fetchImpl, new URL(name, snapshotUrl), { headers });
|
||||
if (!response.ok) throw new Error(`${name}: HTTP ${response.status}`);
|
||||
if (!response.body) throw new Error(`${name}: resposta sem corpo`);
|
||||
const current = responseValidator(response);
|
||||
let invalid = false;
|
||||
let reason = '';
|
||||
if (offset) {
|
||||
const range = response.headers.get('content-range');
|
||||
const match = /^bytes (\d+)-(\d+)\/(\d+)$/i.exec(range || '');
|
||||
const length = numericContentLength(response);
|
||||
expectedSize = match ? Number(match[3]) : null;
|
||||
invalid = response.status !== 206
|
||||
|| !match
|
||||
|| Number(match[1]) !== offset
|
||||
|| Number(match[2]) < Number(match[1])
|
||||
|| expectedSize !== saved.totalSize
|
||||
|| length !== Number(match[2]) - Number(match[1]) + 1
|
||||
|| current.validator !== saved.validator;
|
||||
reason = `Content-Range/validator invalido para retomada: ${range || 'ausente'}`;
|
||||
} else {
|
||||
const length = numericContentLength(response);
|
||||
invalid = response.status !== 200 || length === null;
|
||||
expectedSize = length;
|
||||
reason = response.status === 206
|
||||
? `${name}: Content-Range inesperado sem retomada`
|
||||
: `${name}: resposta final sem Content-Length numerico confiavel`;
|
||||
}
|
||||
if (!invalid) {
|
||||
metadata = { ...current, totalSize: expectedSize };
|
||||
break;
|
||||
}
|
||||
await response.body.cancel?.();
|
||||
if (!offset || attempt > 0) throw new Error(reason);
|
||||
await fsp.rm(partPath, { force: true });
|
||||
delete manifest.partials[name];
|
||||
await writeManifest(rawDir, manifest);
|
||||
offset = 0;
|
||||
}
|
||||
|
||||
manifest.partials[name] = metadata;
|
||||
await writeManifest(rawDir, manifest);
|
||||
let received = offset;
|
||||
let idleTimer;
|
||||
let lastProgressAt = 0;
|
||||
const emitProgress = (force = false) => {
|
||||
const now = Date.now();
|
||||
if (!force && now - lastProgressAt < 500) return;
|
||||
lastProgressAt = now;
|
||||
onProgress({ phase: 'download', file: name, bytes: received, totalBytes: expectedSize, message: `baixando ${name}` });
|
||||
};
|
||||
emitProgress(true);
|
||||
const guard = new Transform({
|
||||
transform(chunk, encoding, callback) {
|
||||
if (isCancelled()) return callback(new CancelledError('download cancelado'));
|
||||
clearTimeout(idleTimer);
|
||||
idleTimer = setTimeout(
|
||||
() => guard.destroy(new Error(`${name}: download sem dados por ${idleTimeoutMs}ms`)),
|
||||
idleTimeoutMs
|
||||
);
|
||||
idleTimer.unref?.();
|
||||
received += chunk.length;
|
||||
emitProgress();
|
||||
callback(null, chunk);
|
||||
}
|
||||
});
|
||||
try {
|
||||
idleTimer = setTimeout(
|
||||
() => guard.destroy(new Error(`${name}: download sem dados por ${idleTimeoutMs}ms`)),
|
||||
idleTimeoutMs
|
||||
);
|
||||
idleTimer.unref?.();
|
||||
await pipeline(Readable.fromWeb(response.body), guard, fs.createWriteStream(partPath, { flags: offset ? 'a' : 'w' }));
|
||||
} catch (error) {
|
||||
if (error instanceof CancelledError || isCancelled()) throw new CancelledError('download cancelado');
|
||||
throw error;
|
||||
} finally {
|
||||
clearTimeout(idleTimer);
|
||||
}
|
||||
await inspectZip(partPath, minZipBytes, expectedSize);
|
||||
emitProgress(true);
|
||||
await fsp.rename(partPath, finalPath);
|
||||
delete manifest.partials[name];
|
||||
return { name, size: received, etag: metadata.etag, lastModified: metadata.lastModified };
|
||||
}
|
||||
|
||||
async function ensureFreeSpace(rawDir, statfs, minimumFreeBytes) {
|
||||
const stats = await statfs(rawDir);
|
||||
const free = BigInt(stats.bavail) * BigInt(stats.bsize);
|
||||
if (free >= BigInt(minimumFreeBytes)) return;
|
||||
const error = new Error(`espaco livre insuficiente: exige ${minimumFreeBytes} bytes`);
|
||||
error.code = 'ENOSPC';
|
||||
throw error;
|
||||
}
|
||||
|
||||
async function readValidDataset(dir, minZipBytes) {
|
||||
const manifest = await readManifest(dir);
|
||||
if (!manifest?.complete) return null;
|
||||
const hasSizes = REQUIRED_RFB_FILES.every((name) => {
|
||||
const metadata = manifest.files?.[name];
|
||||
const size = typeof metadata === 'number' ? metadata : metadata?.size;
|
||||
return Number.isSafeInteger(size) && size >= minZipBytes;
|
||||
});
|
||||
if (!hasSizes) return null;
|
||||
try {
|
||||
await validateRfbRawFiles(dir, { minZipBytes, sizes: manifest.files });
|
||||
return manifest;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function archiveIncompleteDataset(rawDir, { includeParts = false } = {}) {
|
||||
let archivePath = path.join(rawDir, INCOMPLETE_DIR);
|
||||
try {
|
||||
await fsp.access(archivePath);
|
||||
archivePath = path.join(rawDir, `${INCOMPLETE_DIR}-${Date.now()}`);
|
||||
} catch {}
|
||||
await fsp.mkdir(archivePath);
|
||||
let moved = 0;
|
||||
for (const name of [...REQUIRED_RFB_FILES, MANIFEST_FILE]) {
|
||||
for (const suffix of ['', ...(includeParts && name !== MANIFEST_FILE ? ['.part'] : [])]) {
|
||||
try {
|
||||
await fsp.rename(path.join(rawDir, `${name}${suffix}`), path.join(archivePath, `${name}${suffix}`));
|
||||
moved += 1;
|
||||
} catch (error) {
|
||||
if (error.code !== 'ENOENT') throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!moved) await fsp.rm(archivePath, { recursive: true, force: true });
|
||||
return moved ? archivePath : null;
|
||||
}
|
||||
|
||||
async function backupPreviousDataset(rawDir, minZipBytes) {
|
||||
const backupPath = path.join(rawDir, BACKUP_DIR);
|
||||
if (await readValidDataset(backupPath, minZipBytes)) return backupPath;
|
||||
await fsp.rm(backupPath, { recursive: true, force: true });
|
||||
if (!await readValidDataset(rawDir, minZipBytes)) return null;
|
||||
const staging = `${backupPath}.tmp`;
|
||||
await fsp.rm(staging, { recursive: true, force: true });
|
||||
await fsp.mkdir(staging);
|
||||
const moved = [];
|
||||
try {
|
||||
for (const name of [...REQUIRED_RFB_FILES, MANIFEST_FILE]) {
|
||||
const source = path.join(rawDir, name);
|
||||
try {
|
||||
await fsp.rename(source, path.join(staging, name));
|
||||
moved.push(name);
|
||||
} catch (error) {
|
||||
if (error.code !== 'ENOENT') throw error;
|
||||
}
|
||||
}
|
||||
await fsp.rename(staging, backupPath);
|
||||
return backupPath;
|
||||
} catch (error) {
|
||||
for (const name of moved.reverse()) {
|
||||
await fsp.rename(path.join(staging, name), path.join(rawDir, name)).catch(() => {});
|
||||
}
|
||||
await fsp.rm(staging, { recursive: true, force: true });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function restoreBackup(rawDir, minZipBytes) {
|
||||
const backupPath = path.join(rawDir, BACKUP_DIR);
|
||||
const manifest = await readValidDataset(backupPath, minZipBytes);
|
||||
if (!manifest) return null;
|
||||
await archiveIncompleteDataset(rawDir);
|
||||
const staging = path.join(rawDir, `${BACKUP_DIR}-restore.tmp`);
|
||||
await fsp.rm(staging, { recursive: true, force: true });
|
||||
await fsp.mkdir(staging);
|
||||
try {
|
||||
for (const name of [...REQUIRED_RFB_FILES, MANIFEST_FILE]) {
|
||||
await fsp.copyFile(path.join(backupPath, name), path.join(staging, name));
|
||||
}
|
||||
for (const name of REQUIRED_RFB_FILES) {
|
||||
await fsp.rename(path.join(staging, name), path.join(rawDir, name));
|
||||
}
|
||||
await fsp.rename(path.join(staging, MANIFEST_FILE), path.join(rawDir, MANIFEST_FILE));
|
||||
} finally {
|
||||
await fsp.rm(staging, { recursive: true, force: true });
|
||||
}
|
||||
return manifest;
|
||||
}
|
||||
|
||||
async function revalidateCachedFiles({ previous, snapshotUrl, fetchImpl }) {
|
||||
for (const name of REQUIRED_RFB_FILES) {
|
||||
const metadata = previous.files[name];
|
||||
if (!metadata || typeof metadata !== 'object') continue;
|
||||
const validator = metadata.etag || metadata.lastModified;
|
||||
if (!validator) continue;
|
||||
const headers = metadata.etag
|
||||
? { 'If-None-Match': metadata.etag }
|
||||
: { 'If-Modified-Since': metadata.lastModified };
|
||||
const response = await fetchResponse(fetchImpl, new URL(name, snapshotUrl), { method: 'HEAD', headers });
|
||||
if (response.status === 304) continue;
|
||||
if (!response.ok) throw new Error(`${name}: HEAD respondeu HTTP ${response.status}`);
|
||||
const length = numericContentLength(response);
|
||||
const current = responseValidator(response);
|
||||
const changed = length !== metadata.size
|
||||
|| (metadata.etag && current.etag !== metadata.etag)
|
||||
|| (metadata.lastModified && current.lastModified !== metadata.lastModified);
|
||||
if (changed) return name;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
async function usefulPresentBytes(rawDir, previous, sameSnapshot, minZipBytes) {
|
||||
if (!sameSnapshot || !previous || previous.complete) return 0;
|
||||
let useful = 0;
|
||||
for (const name of REQUIRED_RFB_FILES) {
|
||||
const metadata = previous.files?.[name];
|
||||
const size = typeof metadata === 'number' ? metadata : metadata?.size;
|
||||
if (Number.isSafeInteger(size) && size >= minZipBytes) {
|
||||
try {
|
||||
await inspectZip(path.join(rawDir, name), minZipBytes, size);
|
||||
useful += size;
|
||||
} catch {}
|
||||
}
|
||||
const partial = previous.partials?.[name];
|
||||
if (!partial?.validator || !Number.isSafeInteger(partial.totalSize)) continue;
|
||||
try {
|
||||
const partSize = (await fsp.stat(path.join(rawDir, `${name}.part`))).size;
|
||||
if (partSize > 0 && partSize < partial.totalSize) useful += partSize;
|
||||
} catch {}
|
||||
}
|
||||
return useful;
|
||||
}
|
||||
|
||||
export async function downloadRfbDataset({
|
||||
rawDir,
|
||||
indexUrl = DEFAULT_RFB_INDEX_URL,
|
||||
fetch: fetchImpl = globalThis.fetch,
|
||||
onProgress = () => {},
|
||||
isCancelled = () => false,
|
||||
concurrency = 1,
|
||||
minZipBytes = DEFAULT_MIN_ZIP_BYTES,
|
||||
idleTimeoutMs = 60_000,
|
||||
statfs = fsp.statfs,
|
||||
minimumFreeBytes = DEFAULT_MINIMUM_FREE_BYTES
|
||||
}) {
|
||||
if (!rawDir) throw new Error('rawDir obrigatorio');
|
||||
if (concurrency < 1 || concurrency > 2) throw new Error('concurrency deve ser 1 ou 2');
|
||||
await fsp.mkdir(rawDir, { recursive: true });
|
||||
const previous = await readManifest(rawDir);
|
||||
let cachedFiles = null;
|
||||
const hasCompleteSizes = previous?.complete && REQUIRED_RFB_FILES.every((name) => {
|
||||
const metadata = previous.files?.[name];
|
||||
const size = typeof metadata === 'number' ? metadata : metadata?.size;
|
||||
return Number.isSafeInteger(size) && size >= minZipBytes;
|
||||
});
|
||||
if (hasCompleteSizes) {
|
||||
try {
|
||||
cachedFiles = await validateRfbRawFiles(rawDir, { minZipBytes, sizes: previous.files });
|
||||
} catch {
|
||||
cachedFiles = null;
|
||||
}
|
||||
}
|
||||
onProgress({ phase: 'download', pct: 0, message: 'verificando snapshot da RFB' });
|
||||
let snapshot;
|
||||
try {
|
||||
snapshot = await discoverLatestSnapshot(indexUrl, { fetch: fetchImpl });
|
||||
} catch (error) {
|
||||
if (!cachedFiles) {
|
||||
const restored = await restoreBackup(rawDir, minZipBytes);
|
||||
if (!restored) throw error;
|
||||
onProgress({ phase: 'download', pct: 100, message: `sem acesso ao indice; backup ${restored.snapshot} restaurado e validado` });
|
||||
return {
|
||||
cancelled: false,
|
||||
offline: true,
|
||||
snapshot: restored.snapshot,
|
||||
downloaded: [],
|
||||
skipped: REQUIRED_RFB_FILES.slice(),
|
||||
files: restored.files
|
||||
};
|
||||
}
|
||||
onProgress({ phase: 'download', pct: 100, message: `sem acesso ao indice; usando snapshot local validado ${previous.snapshot}` });
|
||||
return {
|
||||
cancelled: false,
|
||||
offline: true,
|
||||
snapshot: previous.snapshot,
|
||||
downloaded: [],
|
||||
skipped: REQUIRED_RFB_FILES.slice(),
|
||||
files: previous.files
|
||||
};
|
||||
}
|
||||
const normalizedIndexUrl = indexUrl.endsWith('/') ? indexUrl : `${indexUrl}/`;
|
||||
const snapshotUrl = new URL(`${snapshot}/`, normalizedIndexUrl);
|
||||
const sameSnapshot = previous?.snapshot === snapshot;
|
||||
let forceRefresh = false;
|
||||
if (sameSnapshot && cachedFiles) {
|
||||
try {
|
||||
forceRefresh = Boolean(await revalidateCachedFiles({ previous, snapshotUrl, fetchImpl }));
|
||||
} catch {
|
||||
onProgress({ phase: 'download', pct: 100, message: `revalidacao remota indisponivel; usando snapshot local validado ${snapshot}` });
|
||||
return {
|
||||
cancelled: false,
|
||||
offline: true,
|
||||
snapshot,
|
||||
downloaded: [],
|
||||
skipped: REQUIRED_RFB_FILES.slice(),
|
||||
files: previous.files
|
||||
};
|
||||
}
|
||||
}
|
||||
if (sameSnapshot && cachedFiles && !forceRefresh) {
|
||||
onProgress({ phase: 'download', pct: 100, message: `snapshot ${snapshot} ja esta completo e validado` });
|
||||
return {
|
||||
cancelled: false,
|
||||
offline: false,
|
||||
snapshot,
|
||||
downloaded: [],
|
||||
skipped: REQUIRED_RFB_FILES.slice(),
|
||||
files: previous.files
|
||||
};
|
||||
}
|
||||
const usefulBytes = await usefulPresentBytes(rawDir, previous, sameSnapshot && !forceRefresh, minZipBytes);
|
||||
const requiredFree = sameSnapshot && !forceRefresh
|
||||
? Math.max(EXTRACTION_MARGIN_BYTES, minimumFreeBytes - usefulBytes)
|
||||
: minimumFreeBytes;
|
||||
await ensureFreeSpace(rawDir, statfs, requiredFree);
|
||||
onProgress({ phase: 'download', pct: 0, message: `verificando espaco necessario (${requiredFree} bytes)` });
|
||||
const changed = !sameSnapshot || forceRefresh;
|
||||
if (changed) {
|
||||
await backupPreviousDataset(rawDir, minZipBytes);
|
||||
if (previous && !previous.complete) await archiveIncompleteDataset(rawDir, { includeParts: true });
|
||||
for (const name of REQUIRED_RFB_FILES) {
|
||||
await fsp.rm(path.join(rawDir, name), { force: true });
|
||||
await fsp.rm(path.join(rawDir, `${name}.part`), { force: true });
|
||||
}
|
||||
}
|
||||
|
||||
const manifest = {
|
||||
snapshot,
|
||||
complete: false,
|
||||
files: sameSnapshot && !forceRefresh && previous?.files && typeof previous.files === 'object' ? { ...previous.files } : {},
|
||||
partials: sameSnapshot && !forceRefresh && previous?.partials && typeof previous.partials === 'object' ? { ...previous.partials } : {}
|
||||
};
|
||||
await writeManifest(rawDir, manifest);
|
||||
|
||||
const downloaded = [];
|
||||
const skipped = [];
|
||||
try {
|
||||
for (let index = 0; index < REQUIRED_RFB_FILES.length; index += 1) {
|
||||
if (isCancelled()) throw new CancelledError('download cancelado');
|
||||
const name = REQUIRED_RFB_FILES[index];
|
||||
try {
|
||||
if (!sameSnapshot || forceRefresh || !manifest.files[name]) throw new Error('arquivo sem origem de snapshot confirmada');
|
||||
const expected = typeof manifest.files[name] === 'number' ? manifest.files[name] : manifest.files[name]?.size;
|
||||
if (!Number.isSafeInteger(expected) || expected < 0) throw new Error('arquivo sem tamanho esperado no manifesto');
|
||||
await inspectZip(path.join(rawDir, name), minZipBytes, expected);
|
||||
skipped.push(name);
|
||||
} catch {
|
||||
const file = await downloadOne({
|
||||
rawDir,
|
||||
snapshotUrl,
|
||||
name,
|
||||
fetchImpl,
|
||||
minZipBytes,
|
||||
isCancelled,
|
||||
onProgress,
|
||||
idleTimeoutMs,
|
||||
manifest
|
||||
});
|
||||
downloaded.push(file);
|
||||
manifest.files[name] = { size: file.size, etag: file.etag, lastModified: file.lastModified };
|
||||
await writeManifest(rawDir, manifest);
|
||||
}
|
||||
onProgress({
|
||||
phase: 'download',
|
||||
pct: Math.round(((index + 1) / REQUIRED_RFB_FILES.length) * 100),
|
||||
message: `${snapshot}: ${index + 1}/${REQUIRED_RFB_FILES.length} arquivos prontos`
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof CancelledError || isCancelled()) return { cancelled: true, snapshot, downloaded, skipped };
|
||||
throw error;
|
||||
}
|
||||
|
||||
const files = await validateRfbRawFiles(rawDir, { minZipBytes });
|
||||
const finalFiles = Object.fromEntries(files.map((file) => {
|
||||
const metadata = manifest.files[file.name];
|
||||
return [file.name, typeof metadata === 'object' ? { ...metadata, size: file.size } : { size: file.size, etag: null, lastModified: null }];
|
||||
}));
|
||||
await writeManifest(rawDir, { snapshot, complete: true, files: finalFiles, partials: {} });
|
||||
await fsp.rm(path.join(rawDir, BACKUP_DIR), { recursive: true, force: true });
|
||||
const checkpointDir = path.dirname(rawDir);
|
||||
for (const name of await fsp.readdir(checkpointDir)) {
|
||||
if (/^checkpoint-estabelecimentos-.*\.ndjson(?:\.tmp)?$/i.test(name)) {
|
||||
await fsp.rm(path.join(checkpointDir, name), { force: true });
|
||||
}
|
||||
}
|
||||
return { cancelled: false, snapshot, downloaded: downloaded.map((file) => file.name), skipped, files: finalFiles };
|
||||
}
|
||||
@@ -0,0 +1,423 @@
|
||||
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';
|
||||
import { validateRfbRawFiles } from './rfb-downloader.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.
|
||||
export function readZipLines(reader, zipPath, { spawn: spawnProcess = spawn } = {}) {
|
||||
const child = spawnProcess(reader.name, reader.argsFor(zipPath), { stdio: ['ignore', 'pipe', 'pipe'] });
|
||||
let stderr = '';
|
||||
let intentionalClose = false;
|
||||
let settled = false;
|
||||
let rl;
|
||||
let resolveDone;
|
||||
let rejectDone;
|
||||
const done = new Promise((resolve, reject) => {
|
||||
resolveDone = resolve;
|
||||
rejectDone = reject;
|
||||
});
|
||||
const settle = (error) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
if (error) rejectDone(error);
|
||||
else resolveDone();
|
||||
};
|
||||
child.stderr.setEncoding('utf8');
|
||||
child.stderr.on('data', (chunk) => {
|
||||
stderr = (stderr + chunk).slice(-4096);
|
||||
});
|
||||
child.on('error', (error) => {
|
||||
const wrapped = new Error(`falha ao iniciar leitor ZIP para ${path.basename(zipPath)}: ${error.message}`);
|
||||
settle(wrapped);
|
||||
rl?.close();
|
||||
});
|
||||
child.on('exit', (code, signal) => {
|
||||
if (intentionalClose) return settle();
|
||||
if (code === 0) return settle();
|
||||
const error = new Error(
|
||||
`leitor ZIP falhou em ${path.basename(zipPath)} (codigo ${code ?? 'n/a'}${signal ? `, sinal ${signal}` : ''}): ${stderr.slice(-1000)}`
|
||||
);
|
||||
settle(error);
|
||||
rl?.close();
|
||||
});
|
||||
child.stdout.setEncoding('latin1');
|
||||
rl = readline.createInterface({ input: child.stdout, crlfDelay: Infinity });
|
||||
const streamError = (error) => settle(new Error(`erro ao ler ${path.basename(zipPath)}: ${error.message}`));
|
||||
rl.on('error', streamError);
|
||||
child.stdout.on('error', streamError);
|
||||
return {
|
||||
rl,
|
||||
done,
|
||||
close() {
|
||||
intentionalClose = true;
|
||||
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));
|
||||
}
|
||||
await zip.done;
|
||||
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])
|
||||
};
|
||||
}
|
||||
|
||||
await zip.done;
|
||||
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;
|
||||
}
|
||||
|
||||
await zip.done;
|
||||
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();
|
||||
await validateRfbRawFiles(rawDir);
|
||||
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) && fs.statSync(checkpointPath).size > 0;
|
||||
if (checkpointDone) {
|
||||
log('checkpoint encontrado: pulando varredura de Estabelecimentos');
|
||||
const input = fs.createReadStream(checkpointPath, { encoding: 'utf8' });
|
||||
const checkpointLines = readline.createInterface({ input, crlfDelay: Infinity });
|
||||
try {
|
||||
for await (const line of checkpointLines) {
|
||||
if (!line.trim()) continue;
|
||||
const item = JSON.parse(line);
|
||||
if (!item.b || !item.m) throw new Error('checkpoint sem campos obrigatorios');
|
||||
cnpjBasicoSet.add(item.b);
|
||||
tomSeen.add(item.m);
|
||||
totalRows += 1;
|
||||
if (item.t1 || item.t2) withPhone += 1;
|
||||
}
|
||||
} catch (error) {
|
||||
await fsp.rm(checkpointPath, { force: true });
|
||||
throw new Error(`checkpoint invalido: ${error.message}`);
|
||||
}
|
||||
} else {
|
||||
const tmpPath = `${checkpointPath}.tmp`;
|
||||
const out = fs.createWriteStream(tmpPath, { encoding: 'utf8' });
|
||||
let outError = null;
|
||||
out.on('error', (error) => { outError = error; });
|
||||
|
||||
onProgress({ phase: 'estabelecimentos', pct: 0, message: 'varrendo Estabelecimentos0-9.zip' });
|
||||
|
||||
try {
|
||||
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')) {
|
||||
if (outError) throw outError;
|
||||
await new Promise((resolve, reject) => {
|
||||
out.once('drain', resolve);
|
||||
out.once('error', reject);
|
||||
});
|
||||
}
|
||||
}
|
||||
if (outError) throw outError;
|
||||
await new Promise((resolve, reject) => {
|
||||
out.once('error', reject);
|
||||
out.end(resolve);
|
||||
});
|
||||
} catch (error) {
|
||||
out.destroy();
|
||||
await fsp.rm(tmpPath, { force: true });
|
||||
throw error;
|
||||
}
|
||||
|
||||
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 (!totalRows && !isCancelled()) {
|
||||
await fsp.rm(checkpointPath, { force: true });
|
||||
throw new Error(`nenhum estabelecimento ativo encontrado para ${ufUpper}`);
|
||||
}
|
||||
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 csvTmp = `${outCsv}.tmp`;
|
||||
const csvOut = fs.createWriteStream(csvTmp, { encoding: 'utf8' });
|
||||
let csvError = null;
|
||||
csvOut.on('error', (error) => { csvError = error; });
|
||||
try {
|
||||
csvOut.write(BOM + headers.join(';') + '\n');
|
||||
const input = fs.createReadStream(checkpointPath, { encoding: 'utf8' });
|
||||
const rl = readline.createInterface({ input, crlfDelay: Infinity });
|
||||
for await (const line of rl) {
|
||||
if (!line.trim()) continue;
|
||||
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')) {
|
||||
if (csvError) throw csvError;
|
||||
await new Promise((resolve, reject) => {
|
||||
csvOut.once('drain', resolve);
|
||||
csvOut.once('error', reject);
|
||||
});
|
||||
}
|
||||
}
|
||||
if (csvError) throw csvError;
|
||||
await new Promise((resolve, reject) => {
|
||||
csvOut.once('error', reject);
|
||||
csvOut.end(resolve);
|
||||
});
|
||||
await fsp.rename(csvTmp, outCsv);
|
||||
} catch (error) {
|
||||
csvOut.destroy();
|
||||
await fsp.rm(csvTmp, { force: true });
|
||||
throw error;
|
||||
}
|
||||
|
||||
// ---- 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 };
|
||||
}
|
||||
Reference in New Issue
Block a user