← back to Nationalrealestate
src/ingest/brokers/engine.ts
196 lines
/**
* M-B1 broker-registry ingest engine. Runs one StateAdapter (or all) through the
* shared harness: openRun → fetch files → parse → upsert firms first (keyed on
* (source, source_id) where source_id = the firm's license number when the roster
* prints one, else 'n:'+normalized name) → upsert brokers with firm_id linkage →
* refresh firm.agent_count → closeRun. region_id = the state region ('state:XX');
* county-level geocoding is a later pass. Fail-loud: 0 parsed brokers = failed run.
*
* npm run ingest:brokers -- tx # one state
* npm run ingest:brokers -- all # every registered adapter
*/
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import type { BrokerRow, StateAdapter } from './types.ts';
import { txTrec } from './tx_trec.ts';
import { flDbpr } from './fl_dbpr.ts';
import { nyDos } from './ny_dos.ts';
import { coDre } from './co_dre.ts';
import { ctDcp } from './ct_dcp.ts';
import { ctDcpReb } from './ct_dcp_reb.ts';
import { ilIdfpr } from './il_idfpr.ts';
import { deDpr } from './de_dpr.ts';
import { caDre } from './ca_dre.ts';
const ADAPTERS: StateAdapter[] = [txTrec, flDbpr, nyDos, coDre, ctDcp, ctDcpReb, ilIdfpr, deDpr, caDre];
const BATCH = 5000;
export function normalizeName(name: string): string {
return name.toUpperCase().replace(/[^A-Z0-9]+/g, ' ').trim()
.replace(/\s+(LLC|L L C|INC|CORP|CO|LTD|LP|LLP|PLLC|PC|PA)$/,'').trim();
}
interface FirmEntity {
sourceId: string;
name: string;
license_no: string | null;
license_status: string | null;
city: string | null;
state_code: string | null;
fromRow: boolean; // true = real firm row from the roster, not just an employer reference
}
async function stateRegionId(state: string): Promise<number | null> {
const r = await query<{ id: number }>(`SELECT id FROM region WHERE canonical_key = $1`, ['state:' + state]);
return r.rows[0]?.id ?? null;
}
function firmKey(licenseNo: string | null | undefined, name: string): string {
return licenseNo && licenseNo.trim() ? licenseNo.trim() : 'n:' + normalizeName(name);
}
async function runAdapter(a: StateAdapter): Promise<void> {
const t0 = Date.now();
console.log(`[${a.source}] fetching ${a.url}`);
const runId = await openRun(a.source, a.url);
try {
const files = await a.fetch();
console.log(`[${a.source}] ${files.length} file(s) downloaded`);
const regionId = await stateRegionId(a.state);
// ---- pass 1: collect firm entities + broker rows (deduped by source_id) ----
const firms = new Map<string, FirmEntity>();
const brokers = new Map<string, BrokerRow>();
let skipped = 0;
for (const row of a.parse(files)) {
if (row.kind === 'firm') {
const key = firmKey(row.firm_license_no ?? row.license_no, row.name);
const prev = firms.get(key);
if (!prev || !prev.fromRow) {
firms.set(key, {
sourceId: key, name: row.name, license_no: row.license_no,
license_status: row.license_status ?? null,
city: row.city ?? null, state_code: row.state_code ?? null, fromRow: true,
});
}
} else {
if (!row.license_no) { skipped++; continue; }
if (brokers.has(row.license_no)) skipped++;
brokers.set(row.license_no, row);
if (row.firm_name) {
const key = firmKey(row.firm_license_no, row.firm_name);
if (!firms.has(key)) {
firms.set(key, {
sourceId: key, name: row.firm_name, license_no: row.firm_license_no ?? null,
license_status: null, city: null, state_code: a.state, fromRow: false,
});
}
}
}
}
if (!brokers.size) throw new Error(`0 broker rows parsed for ${a.source} — layout drift?`);
console.log(`[${a.source}] parsed ${brokers.size.toLocaleString()} brokers, ${firms.size.toLocaleString()} firms (${skipped} skipped)`);
// ---- pass 2: upsert firms, build source_id -> id + normalized name -> id maps ----
const firmIdByKey = new Map<string, number>();
const firmIdByName = new Map<string, number>();
const firmList = [...firms.values()];
for (let i = 0; i < firmList.length; i += BATCH) {
const b = firmList.slice(i, i + BATCH);
const r = await query<{ id: number; source_id: string; normalized_name: string }>(
`INSERT INTO firm (name, normalized_name, license_no, license_state, hq_city, hq_state, source, source_id, region_id)
SELECT * FROM UNNEST($1::text[],$2::text[],$3::text[],$4::text[],$5::text[],$6::text[],$7::text[],$8::text[],$9::int[])
ON CONFLICT (source, source_id) DO UPDATE SET
name = EXCLUDED.name, normalized_name = EXCLUDED.normalized_name,
license_no = COALESCE(EXCLUDED.license_no, firm.license_no),
hq_city = COALESCE(EXCLUDED.hq_city, firm.hq_city),
hq_state = COALESCE(EXCLUDED.hq_state, firm.hq_state),
region_id = COALESCE(EXCLUDED.region_id, firm.region_id)
RETURNING id, source_id, normalized_name`,
[
b.map(f => f.name), b.map(f => normalizeName(f.name)), b.map(f => f.license_no),
b.map(() => a.state), b.map(f => f.city), b.map(f => f.state_code),
b.map(() => a.source), b.map(f => f.sourceId), b.map(() => regionId),
],
);
for (const row of r.rows) {
firmIdByKey.set(row.source_id, row.id);
if (!firmIdByName.has(row.normalized_name)) firmIdByName.set(row.normalized_name, row.id);
}
}
// ---- pass 3: upsert brokers with firm linkage ----
const brokerList = [...brokers.values()];
let upserted = 0;
for (let i = 0; i < brokerList.length; i += BATCH) {
const b = brokerList.slice(i, i + BATCH);
const firmIds = b.map(row => {
if (!row.firm_name) return null;
return firmIdByKey.get(firmKey(row.firm_license_no, row.firm_name))
?? firmIdByName.get(normalizeName(row.firm_name)) ?? null;
});
const r = await query(
`INSERT INTO broker (firm_id, name, license_no, license_state, license_type, license_status,
email, phone, city, state_code, source, source_id, region_id)
SELECT * FROM UNNEST($1::int[],$2::text[],$3::text[],$4::text[],$5::text[],$6::text[],
$7::text[],$8::text[],$9::text[],$10::text[],$11::text[],$12::text[],$13::int[])
ON CONFLICT (source, source_id) DO UPDATE SET
firm_id = EXCLUDED.firm_id, name = EXCLUDED.name,
license_type = EXCLUDED.license_type, license_status = EXCLUDED.license_status,
email = COALESCE(EXCLUDED.email, broker.email),
phone = COALESCE(EXCLUDED.phone, broker.phone),
city = COALESCE(EXCLUDED.city, broker.city),
state_code = COALESCE(EXCLUDED.state_code, broker.state_code),
region_id = COALESCE(EXCLUDED.region_id, broker.region_id)`,
[
firmIds, b.map(r2 => r2.name), b.map(r2 => r2.license_no),
b.map(() => a.state), b.map(r2 => r2.license_type), b.map(r2 => r2.license_status),
b.map(r2 => r2.email ?? null), b.map(r2 => r2.phone ?? null),
b.map(r2 => r2.city ?? null), b.map(r2 => r2.state_code ?? null),
b.map(() => a.source), b.map(r2 => r2.license_no), b.map(() => regionId),
],
);
upserted += r.rowCount ?? 0;
if ((i / BATCH) % 10 === 0) process.stdout.write(`\r[${a.source}] brokers ${Math.min(i + BATCH, brokerList.length).toLocaleString()}/${brokerList.length.toLocaleString()}`);
}
process.stdout.write('\n');
await query(
`UPDATE firm f SET agent_count = c.n
FROM (SELECT firm_id, COUNT(*)::int AS n FROM broker WHERE source = $1 AND firm_id IS NOT NULL GROUP BY firm_id) c
WHERE f.id = c.firm_id`,
[a.source],
);
await closeRun(runId, 'ok', {
upserted, skipped,
fileHash: files[0]?.sha256,
notes: `${a.state}: ${brokers.size} brokers, ${firms.size} firms, ${files.length} file(s), ${((Date.now() - t0) / 1000).toFixed(0)}s`,
});
console.log(`[${a.source}] OK — ${upserted.toLocaleString()} brokers upserted, ${firms.size.toLocaleString()} firms, ${((Date.now() - t0) / 1000).toFixed(0)}s`);
} catch (e: any) {
await closeRun(runId, 'failed', { notes: String(e.message || e).slice(0, 500) });
throw e;
}
}
async function main() {
const arg = (process.argv[2] || 'all').toLowerCase();
const picked = arg === 'all'
? ADAPTERS
: ADAPTERS.filter(x => x.state.toLowerCase() === arg || x.source === arg);
if (!picked.length) {
console.error(`unknown state '${arg}' — have: ${ADAPTERS.map(x => x.state).join(', ')} (or 'all')`);
process.exit(2);
}
let failed = 0;
for (const a of picked) {
try { await runAdapter(a); }
catch (e) { failed++; console.error(`[${a.source}] FAILED:`, e); }
}
await pool.end();
if (failed) process.exit(1);
}
main();