`agents/ingest-agent/importers/ccld.js`

189 lines
#!/usr/bin/env node
/**
 * CA DSS Community Care Licensing — facility importer.
 *
 * Pulls public CSVs from data.ca.gov for the three resident/elder-relevant
 * datasets:
 *   - Residential Care Facilities for the Elderly (RCFEs + CCRCs) → 'rcfe' / 'ccrc'
 *   - Adult Residential Facilities                                → 'adult_residential' + variants
 *   - Home Care Organizations                                     → 'home_care_agency'
 *
 * Filters to LA County. Upserts on `cdph_license` (= facility_number).
 *
 * Run:
 *   node agents/ingest-agent/importers/ccld.js          # all three
 *   DATASET=rcfe node ccld.js                           # one only
 */
const fs = require('node:fs');
const path = require('node:path');
const { parse } = require('csv-parse');
const { pool, query, withTx } = require('../../shared/db');

const SOURCE_NAME = 'CA DSS Community Care Licensing';
const DATA_DIR = path.resolve(__dirname, '../../../data/ccld');

const FEEDS = {
  rcfe: {
    file: 'rcfe.csv',
    url:  'https://data.chhs.ca.gov/dataset/46ffcbdf-4874-4cc1-92c2-fb715e3ad014/resource/744d1583-f9eb-45b6-b0f8-b9a9dab936a6/download/tmpacjmwy9v.csv',
  },
  arf:  {
    file: 'arf.csv',
    url:  'https://data.chhs.ca.gov/dataset/46ffcbdf-4874-4cc1-92c2-fb715e3ad014/resource/9f5d1d00-6b24-4f44-a158-9cbe4b43f117/download/tmpx8kml5z4.csv',
  },
  hco:  {
    file: 'home_care.csv',
    url:  'https://data.chhs.ca.gov/dataset/46ffcbdf-4874-4cc1-92c2-fb715e3ad014/resource/b4d78b7f-12df-4b0c-a81a-ff40b949bc75/download/tmpbgsqj_4n.csv',
  },
};

const FACILITY_TYPE_MAP = {
  'RESIDENTIAL CARE ELDERLY':                                          'rcfe',
  'RCFE-CONTINUING CARE RETIREMENT COMMUNITY':                         'ccrc',
  'ADULT RESIDENTIAL':                                                 'adult_residential',
  'ADULT DAY PROGRAM':                                                 'adult_day_program',
  'SOCIAL REHABILITATION FACILITY':                                    'social_rehab',
  'ENHANCED BEHAVIORAL SUPPORTS HOME - ARF':                           'arf_behavioral',
  'ADULT RESIDENTIAL FACILITY FOR PERSONS WITH SPECIAL HEALTH CARE NEEDS': 'arf_special_needs',
  'COMMUNITY CRISIS HOME - ARF':                                       'arf_crisis',
  'RESIDENTIAL FACILITY CHRONICALLY ILL':                              'arf_chronic_ill',
  'HOME CARE':                                                         'home_care_agency',
};

function clean(s) {
  if (s === undefined || s === null) return null;
  const t = String(s).trim();
  return t.length === 0 ? null : t;
}

function zip5(z) {
  if (!z) return null;
  const m = String(z).match(/\d{5}/);
  return m ? m[0] : null;
}

function digits(s) { return s ? String(s).replace(/[^0-9]/g, '') : null; }

async function getSourceId() {
  const r = await query('SELECT id FROM sources WHERE source_name = $1', [SOURCE_NAME]);
  if (r.rowCount === 0) throw new Error(`Source not seeded: ${SOURCE_NAME}`);
  return r.rows[0].id;
}

async function startJob(sourceId, label) {
  const r = await query(`
    INSERT INTO scrape_jobs (source_id, job_label, status, started_at)
    VALUES ($1,$2,'running',NOW()) RETURNING id`,
    [sourceId, label]);
  return r.rows[0].id;
}

async function finishJob(jobId, fields) {
  const sets = []; const params = []; let i = 1;
  for (const [k, v] of Object.entries(fields)) { sets.push(`${k} = $${i++}`); params.push(v); }
  sets.push(`finished_at = NOW()`);
  params.push(jobId);
  await query(`UPDATE scrape_jobs SET ${sets.join(', ')} WHERE id = $${i}`, params);
}

async function ensureFile(feed) {
  const localPath = path.join(DATA_DIR, feed.file);
  if (fs.existsSync(localPath) && fs.statSync(localPath).size > 0) return localPath;
  fs.mkdirSync(DATA_DIR, { recursive: true });
  console.log(`[ccld] downloading ${feed.url}`);
  const { fetch } = require('undici');
  const res = await fetch(feed.url);
  if (!res.ok) throw new Error(`download failed ${res.status}`);
  const buf = Buffer.from(await res.arrayBuffer());
  fs.writeFileSync(localPath, buf);
  return localPath;
}

async function upsertOrg(client, row, sourceUrl) {
  const facilityType = clean(row.facility_type);
  const orgType = FACILITY_TYPE_MAP[facilityType?.toUpperCase()] || 'care_facility';
  const name = clean(row.facility_name);
  const license = clean(row.facility_number);
  if (!name || !license) return null;

  const phone = digits(row.facility_telephone_number);
  const r = await client.query(`
    INSERT INTO organizations (
      name, type, address, city, state, zip, county, phone,
      cdph_license
    ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
    ON CONFLICT (cdph_license) WHERE cdph_license IS NOT NULL DO UPDATE SET
      name    = EXCLUDED.name,
      type    = EXCLUDED.type,
      address = COALESCE(EXCLUDED.address, organizations.address),
      city    = COALESCE(EXCLUDED.city,    organizations.city),
      zip     = COALESCE(EXCLUDED.zip,     organizations.zip),
      phone   = COALESCE(EXCLUDED.phone,   organizations.phone),
      updated_at = NOW()
    RETURNING id, (xmax = 0) AS inserted
  `, [
    name, orgType,
    clean(row.facility_address),
    clean(row.facility_city),
    clean(row.facility_state) || 'CA',
    zip5(row.facility_zip),
    'Los Angeles',
    phone,
    license,
  ]);
  return { id: r.rows[0].id, inserted: r.rows[0].inserted };
}

async function importFeed(feedKey, feed, sourceId) {
  const file = await ensureFile(feed);
  console.log(`[ccld:${feedKey}] file=${file}`);
  const jobId = await startJob(sourceId, `ccld:${feedKey}`);

  let scanned = 0, kept = 0, inserted = 0, updated = 0, skipped = 0;
  const parser = fs.createReadStream(file).pipe(parse({
    columns: true, skip_empty_lines: true, relax_column_count: true, bom: true,
  }));

  for await (const row of parser) {
    scanned++;
    const county = (row.county_name || '').trim().toUpperCase();
    const status = (row.facility_status || '').trim().toUpperCase();
    if (county !== 'LOS ANGELES') { skipped++; continue; }
    if (status && status !== 'LICENSED') { skipped++; continue; }
    kept++;
    try {
      await withTx(async (client) => {
        const r = await upsertOrg(client, row, feed.url);
        if (!r) { skipped++; return; }
        if (r.inserted) inserted++; else updated++;
      });
    } catch (e) {
      console.error(`[ccld:${feedKey}] row error fac=${row.facility_number}: ${e.message}`);
      skipped++;
    }
    if (kept % 500 === 0) console.log(`[ccld:${feedKey}] scanned=${scanned} kept=${kept} ins=${inserted} upd=${updated} skip=${skipped}`);
  }

  await finishJob(jobId, {
    status: 'done',
    records_found: scanned,
    records_inserted: inserted,
    records_updated: updated,
    records_skipped: skipped,
  });
  console.log(`[ccld:${feedKey}] done. scanned=${scanned} kept=${kept} ins=${inserted} upd=${updated} skip=${skipped}`);
}

async function main() {
  const sourceId = await getSourceId();
  const which = process.env.DATASET ? [process.env.DATASET] : Object.keys(FEEDS);
  for (const k of which) {
    const feed = FEEDS[k];
    if (!feed) { console.error(`[ccld] unknown DATASET=${k} (allowed: ${Object.keys(FEEDS).join(',')})`); continue; }
    await importFeed(k, feed, sourceId);
  }
  await pool.end();
}

main().catch(async (e) => { console.error('[ccld] fatal:', e); try { await pool.end(); } catch (_) {} process.exit(1); });