`scripts/ingest-gov-ca.js`

74 lines
#!/usr/bin/env node
/*
 * ingest-gov-ca.js — load San Francisco + Silicon Valley (Santa Clara county) licensed
 * real-estate agents/brokers from the CALIFORNIA DRE public bulk file into gov_licensed_agent.
 *
 * Source: CA Dept. of Real Estate — public Licensee List (free bulk, $0)
 *   https://secure.dre.ca.gov/datafile/CurrList.zip  ->  CurrList.csv (431k rows, comma-delimited)
 * Cols (0-based): 1 lastname_primary · 2 firstname_secondary · 4 lic_number · 5 lic_type
 *   · 6 lic_status · 8 lic_expiration(YYYYMMDD) · 11 related_lastname(firm) · 15 addr1 · 17 city
 *   · 19 zip · 22 county_name
 *
 * Metros: SAN FRANCISCO county -> 'san-francisco'; SANTA CLARA county -> 'silicon-valley'.
 * Usage: node scripts/ingest-gov-ca.js   (expects /tmp/CurrList.csv already extracted)
 */
'use strict';
const fs = require('fs');
const readline = require('readline');
const { pool } = require('./db/brokers-db');
const FILE = '/tmp/CurrList.csv';
const METRO = { 'SAN FRANCISCO': 'san-francisco', 'SANTA CLARA': 'silicon-valley' };
const caDate = s => /^\d{8}$/.test(s || '') ? `${s.slice(0,4)}-${s.slice(4,6)}-${s.slice(6,8)}` : null;

async function flush(batch) {
  if (!batch.length) return 0;
  const seen = new Set(); const vals = []; const ph = [];
  for (const r of batch) {
    const key = r.lic + '|' + r.firm; if (seen.has(key)) continue; seen.add(key);
    const b = vals.length;
    ph.push(`('CA',$${b+1},$${b+2},$${b+3},$${b+4},$${b+5},$${b+6},$${b+7},$${b+8},$${b+9},$${b+10},'ca-dre')`);
    vals.push(r.metro, r.name, r.firm, r.lic, r.type, r.city, r.addr, r.zip, r.county, r.exp);
  }
  if (!ph.length) return 0;
  await pool.query(
    `INSERT INTO gov_licensed_agent (state,metro,name,firm,license_number,license_type,city,addr,zip,county,expiration,source)
     VALUES ${ph.join(',')}
     ON CONFLICT (source,license_number,firm) DO UPDATE
       SET license_type=EXCLUDED.license_type, expiration=EXCLUDED.expiration, city=EXCLUDED.city, fetched_at=now()`, vals);
  return ph.length;
}

async function main() {
  if (!fs.existsSync(FILE)) { console.error(`missing ${FILE}`); process.exit(1); }
  const rl = readline.createInterface({ input: fs.createReadStream(FILE), crlfDelay: Infinity });
  let n = 0, matched = 0, loaded = 0, batch = [], first = true;
  for await (const line of rl) {
    if (first) { first = false; continue; }              // header
    n++;
    if (!/SAN FRANCISCO|SANTA CLARA/.test(line)) continue; // fast prefilter
    const f = line.split(',');
    if (f.length < 23) continue;
    const county = (f[22] || '').trim().toUpperCase();
    const metro = METRO[county]; if (!metro) continue;
    const type = (f[5] || '').trim();
    if (type !== 'Salesperson' && type !== 'Broker') continue;   // persons only
    if ((f[6] || '').trim() !== 'Licensed') continue;
    const last = (f[1] || '').trim(), fn = (f[2] || '').trim();
    const name = (fn ? fn + ' ' : '') + last;
    if (!name || !(f[4] || '').trim()) continue;
    matched++;
    batch.push({ metro, name, firm: (f[11] || '').trim(), lic: (f[4] || '').trim(), type,
      city: (f[17] || '').trim(), addr: (f[15] || '').trim(), zip: (f[19] || '').trim(),
      county, exp: caDate((f[8] || '').trim()) });
    if (batch.length >= 500) { loaded += await flush(batch); batch = []; process.stdout.write(`\r  scanned ${n}  matched ${matched}  loaded ${loaded}   `); }
  }
  loaded += await flush(batch);
  process.stdout.write('\n');
  const { rows } = await pool.query(
    `SELECT metro, count(*) n, count(*) FILTER (WHERE license_type='Broker') br FROM gov_licensed_agent WHERE source='ca-dre' GROUP BY metro ORDER BY n DESC`);
  rows.forEach(r => console.log(`  ${r.metro}: ${r.n} (${r.br} brokers)`));
  console.log('  source: CA DRE public licensee list. cost: $0');
  await pool.end();
}
if (require.main === module) main().catch(e => { console.error(e); process.exit(1); });