`src/ingest/brokers/ct_dcp.ts`

70 lines
/**
 * CT — DCP "Real Estate Salesperson" full credential roster on data.ct.gov
 * (eqtn-rppv; 99,396 rows probed 2026-07-26). Wave-3 upgrade from the old
 * affiliation-only feed 6tja-6vdt (18,281 ACTIVE salesperson↔broker pairs, no
 * status, no city): eqtn-rppv is the FULL salesperson credential DB — every
 * status (ACTIVE 20.7k / INACTIVE 52k / EXPIRED APPLICATION 22.8k / LAPSED /
 * RETIRED / SUSPENDED / REVOKED …), mailing city+state, and the sponsoring
 * broker linkage (`supervising_broker` name + `broker_license` #, populated on
 * 22,812 rows). Same `source='ct_dcp'` + license_no = fullcredentialcode
 * ("RES.0000000"), which matches the old feed's source_id shape, so this
 * upserts over the existing CT rows in place (enriches status/city, adds the
 * inactive/expired long tail) — no double-count.
 *
 * Salespersons ONLY (credentialtype RES). The broker-side REB credential roster
 * is a separate DCP extract — a follow-up, not this dataset.
 */
import { download } from '../run.ts';
import { parseCsvObjects } from './csv.ts';
import type { BrokerRow, FetchedFile, StateAdapter } from './types.ts';

const URL = 'https://data.ct.gov/resource/eqtn-rppv.csv'
  + '?$order=fullcredentialcode&$limit=200000';

/** Non-individual credential holders (a few dozen) → brokerage firm rows. */
const ENTITY = /^(BUSINESS|CORPORATION|PARTNERSHIP|LIMITED LIABILITY COMPANY)$/i;

export const ctDcp: StateAdapter = {
  state: 'CT',
  source: 'ct_dcp',
  url: URL,
  async fetch() {
    return [await download(URL, 'ct_dcp_salesperson_full_roster.csv')];
  },
  *parse(files: FetchedFile[]): Generator<BrokerRow> {
    for (const o of parseCsvObjects(files[0].path)) {
      // Normalize the credential code to the canonical "RES.#######" shape: a
      // small tail of rows carry a legacy Socrata suffix ("RES.0769651.INACTIVE")
      // — strip it so source_id stays stable if upstream ever cleans the field
      // (else re-ingest would INSERT a duplicate instead of upserting). Require a
      // real credential number so the handful of digitless "RES" stubs are dropped
      // rather than collapsing to one shared key.
      const lic = (o.fullcredentialcode || '').replace(/\.[A-Za-z]+$/, '').trim();
      if (!lic || !/\d/.test(lic)) continue;
      const status = o.status || null;
      const city = o.city || null;
      const st = (o.state || '').toUpperCase() || null;
      const type = (o.type || '').toUpperCase();
      if (ENTITY.test(type)) {
        const name = o.businessname || o.name;
        if (!name) continue;
        yield {
          kind: 'firm', name, license_no: lic,
          license_type: o.credential || null, license_status: status,
          city, state_code: st,
        };
      } else {
        const name = o.name;
        if (!name) continue;
        yield {
          kind: 'broker', name, license_no: lic,
          license_type: o.credential || 'Salesperson', license_status: status,
          city, state_code: st,
          firm_name: o.supervising_broker || null,
          firm_license_no: o.supervising_broker ? (o.broker_license || null) : null,
        };
      }
    }
  },
};