`scripts/ingest-gov-ny.js`

75 lines
#!/usr/bin/env node
/*
 * ingest-gov-ny.js — load licensed real-estate agents/brokers from the NEW YORK
 * government open-data API into cre.gov_licensed_agent. $0, no key, authoritative.
 *
 * Source: NY Dept. of State, Division of Licensing Services — "Active Real Estate
 *   Salespersons and Brokers", data.ny.gov dataset yg7h-zjbf (148,112 licensees).
 *   SODA API: https://data.ny.gov/resource/yg7h-zjbf.json  ($limit/$offset/$where/$select)
 *
 * Metro = a set of business_city values. Paginates the API, upserts by (source,license_number,firm).
 * Usage: node scripts/ingest-gov-ny.js <metro>   where metro ∈ hamptons | nyc
 */
'use strict';
const { pool } = require('./db/brokers-db');
const BASE = 'https://data.ny.gov/resource/yg7h-zjbf.json';

const METROS = {
  hamptons: ['SOUTHAMPTON','EAST HAMPTON','BRIDGEHAMPTON','SAG HARBOR','MONTAUK','WATER MILL',
    'WAINSCOTT','AMAGANSETT','SAGAPONACK','WESTHAMPTON','WESTHAMPTON BEACH','QUOGUE','REMSENBURG','HAMPTON BAYS'],
  nyc: ['NEW YORK','BROOKLYN','BRONX','STATEN ISLAND','LONG ISLAND CITY','ASTORIA','FLUSHING',
    'JAMAICA','JACKSON HEIGHTS','FOREST HILLS','RIDGEWOOD','WOODSIDE','CORONA','ELMHURST','REGO PARK'],
};

async function fetchPage(cities, offset, limit) {
  const inList = cities.map(c => `'${c.replace(/'/g, "''")}'`).join(',');
  // brokers + salespersons only — exclude pure office/branch records (Steve's scope)
  const where = encodeURIComponent(
    `business_city in(${inList}) AND license_type not like '%OFFICE%' AND license_type not like '%BRANCH%'`);
  const url = `${BASE}?%24where=${where}&%24limit=${limit}&%24offset=${offset}&%24order=license_number`;
  const r = await fetch(url, { headers: { 'User-Agent': 'crcp-gov-ingest' } });
  if (!r.ok) throw new Error(`SODA ${r.status}`);
  return r.json();
}

async function main() {
  const metro = (process.argv[2] || '').toLowerCase();
  const cities = METROS[metro];
  if (!cities) { console.error(`metro must be one of: ${Object.keys(METROS).join(', ')}`); process.exit(1); }

  const LIMIT = 1000; let offset = 0, total = 0, upserted = 0;
  for (;;) {
    const rows = await fetchPage(cities, offset, LIMIT);
    if (!rows.length) break;
    // batch upsert: one multi-row INSERT per page. Dedup within-page on (license_number,firm)
    // so a page can't trip "ON CONFLICT ... cannot affect row a second time".
    const seen = new Set(); const vals = []; const ph = [];
    for (const r of rows) {
      const lic = (r.license_number || '').trim(), firm = (r.business_name || '').trim();
      const key = lic + '|' + firm; if (seen.has(key)) continue; seen.add(key);
      const exp = r.license_expiration_date ? r.license_expiration_date.slice(0, 10) : null;
      const b = vals.length;
      ph.push(`('NY',$${b+1},$${b+2},$${b+3},$${b+4},$${b+5},$${b+6},$${b+7},$${b+8},$${b+9},$${b+10},'ny-dos')`);
      vals.push(metro, (r.license_holder_name || '').trim(), firm, lic, (r.license_type || '').trim(),
        (r.business_city || '').trim(), (r.business_address_1 || '').trim(),
        (r.business_zip || '').trim(), (r.county && r.county !== 'NONE' ? r.county : null), exp);
    }
    if (ph.length) {
      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);
      upserted += ph.length;
    }
    total += rows.length; offset += LIMIT;
    process.stdout.write(`\r  fetched ${total}  upserted ${upserted}   `);
    if (rows.length < LIMIT) break;
  }
  process.stdout.write('\n');
  const { rows: [c] } = await pool.query(`SELECT count(*) n, count(*) FILTER (WHERE license_type ILIKE '%broker%') brokers FROM gov_licensed_agent WHERE metro=$1`, [metro]);
  console.log(`✔ ${metro}: ${c.n} licensees in DB (${c.brokers} brokers). source: NY DOS gov open data. cost: $0`);
  await pool.end();
}
if (require.main === module) main().catch(e => { console.error(e); process.exit(1); });