import { processLead } from '../lead-engine.mjs'; import { createLeadInserter, transaction } from '../db.mjs'; import { createGeocoder } from './geocode.mjs'; import { leadRowFromProcessed } from './lead-row.mjs'; import { getSource } from '../sources/index.mjs'; import { envString, envNumber, paths } from '../env.mjs'; export function defaultEndpoints() { return { overpassUrls: envString( 'OVERPASS_API_URL', 'https://overpass-api.de/api/interpreter,https://overpass.kumi.systems/api/interpreter' ).split(',').map((url) => url.trim()).filter(Boolean), nominatimUrl: envString('NOMINATIM_API_URL', 'https://nominatim.openstreetmap.org/search'), wikidataUrl: envString('WIKIDATA_API_URL', 'https://query.wikidata.org/sparql'), brasilApiUrl: envString('BRASILAPI_URL', 'https://brasilapi.com.br/api').replace(/\/$/, '') }; } // Executa uma captura: percorre as fontes selecionadas, processa cada lead // (normalizacao + score) e insere no banco em lotes. A dedup acontece no // proprio INSERT (ON CONFLICT dedup_key DO NOTHING) — leads ja capturados // em qualquer execucao anterior contam como duplicados. export async function runCapture({ db, runId, params, emit = () => {}, log = () => {}, isCancelled = () => false }) { const insert = createLeadInserter(db); const endpoints = defaultEndpoints(); const ctx = { db, emit, log, isCancelled, cnpjDir: params.cnpjDir || paths.cnpjDir, endpoints, delays: { overpassDelayMs: envNumber('OVERPASS_REQUEST_DELAY_MS', 5000), overpassTimeoutMs: envNumber('OVERPASS_TIMEOUT_MS', 15000) }, overpassTextSearch: envString('OVERPASS_TEXT_SEARCH', 'false') === 'true', geocode: createGeocoder(db, { nominatimUrl: endpoints.nominatimUrl }) }; const counts = { raw: 0, new: 0, dup: 0 }; const updateRun = runId ? db.prepare( `UPDATE capture_runs SET raw_count = ?, new_count = ?, dup_count = ?, progress_json = ? WHERE id = ?` ) : null; const persistCounts = (progress) => { if (updateRun) { updateRun.run(counts.raw, counts.new, counts.dup, JSON.stringify(progress || {}), runId); } }; if (runId) { db.prepare( `UPDATE capture_runs SET status = 'running', pid = ?, started_at = datetime('now') WHERE id = ?` ).run(process.pid, runId); } let cancelled = false; try { for (const sourceId of params.sources) { if (isCancelled()) { cancelled = true; break; } const source = getSource(sourceId); if (!source) { log(`fonte desconhecida: ${sourceId} — pulada`); continue; } log(`fonte ${sourceId}: iniciando`); emit({ phase: sourceId, message: 'iniciando', counts: { ...counts } }); for await (const batch of source.collect(params, ctx)) { counts.raw += batch.leads.length; transaction(db, () => { for (const payload of batch.leads) { const result = processLead(payload); const row = leadRowFromProcessed({ payload, lead: result.lead }, sourceId); row.capture_run_id = runId || null; if (insert(row) === 'inserted') counts.new += 1; else counts.dup += 1; } }); const progress = { phase: sourceId, message: `${batch.label}: ${counts.new} novos, ${counts.dup} duplicados (${counts.raw} brutos)`, counts: { ...counts } }; emit(progress); persistCounts(progress); if (isCancelled()) { cancelled = true; break; } } if (cancelled) break; log(`fonte ${sourceId}: concluida`); } } catch (error) { if (runId) { db.prepare( `UPDATE capture_runs SET status = 'error', error = ?, raw_count = ?, new_count = ?, dup_count = ?, finished_at = datetime('now') WHERE id = ?` ).run(String(error.message || error), counts.raw, counts.new, counts.dup, runId); } throw error; } const finalStatus = cancelled ? 'cancelled' : 'done'; if (runId) { db.prepare( `UPDATE capture_runs SET status = ?, raw_count = ?, new_count = ?, dup_count = ?, finished_at = datetime('now') WHERE id = ?` ).run(finalStatus, counts.raw, counts.new, counts.dup, runId); } return { ...counts, status: finalStatus }; } export function createCaptureRun(db, params) { const result = db.prepare('INSERT INTO capture_runs (params_json, status) VALUES (?, ?)') .run(JSON.stringify(params), 'pending'); return Number(result.lastInsertRowid); }