`agents/ingest-agent/importers/cms-hospitals.js`

179 lines
#!/usr/bin/env node
/**
 * CMS Hospital General Information — Stage 4 enrichment.
 *
 *   Source: https://data.cms.gov/provider-data/dataset/xubh-q36u
 *   CSV   : Hospital_General_Information.csv (every Medicare-certified hospital
 *           in the country, with CCN, name, address, type, ownership, ratings).
 *
 * For LA County, joins by CDPH license / name / ZIP into organizations and
 * upgrades the row's type to 'hospital'.
 */
const fs = require('node:fs');
const path = require('node:path');
const crypto = require('node:crypto');
const { parse } = require('csv-parse');
const { fetch } = require('undici');
const { pool, query, withTx } = require('../../shared/db');
const { laZipSet } = require('../../shared/la-zips');

const SOURCE_NAME = 'CA HCAI Hospital Annual Utilization';   // existing seeded source
const SOURCE_URL  = 'https://data.cms.gov/provider-data/dataset/xubh-q36u';
const CSV_URL     = process.env.CMS_HOSP_URL
  || 'https://data.cms.gov/provider-data/sites/default/files/resources/893c372430d9d71a1c52737d01239d47_1770163599/Hospital_General_Information.csv';
const CACHE       = path.resolve(__dirname, '../../../data/cms/Hospital_General_Information.csv');

function clean(s) {
  if (s === undefined || s === null) return null;
  const t = String(s).trim();
  return t.length === 0 ? null : t;
}
function pick(row, ...keys) {
  for (const k of keys) {
    const v = row[k];
    if (v !== undefined && v !== null && String(v).trim() !== '') return clean(v);
  }
  return null;
}
function zip5(z) {
  if (!z) return null;
  const m = String(z).match(/\d{5}/);
  return m ? m[0] : null;
}
function sha256(s) { return crypto.createHash('sha256').update(s).digest('hex'); }

async function downloadIfNeeded() {
  if (fs.existsSync(CACHE)) {
    const ageDays = (Date.now() - fs.statSync(CACHE).mtimeMs) / 86400000;
    if (ageDays < 30) {
      console.log(`[cms-hosp] using cache (${ageDays.toFixed(1)}d old)`);
      return;
    }
  }
  fs.mkdirSync(path.dirname(CACHE), { recursive: true });
  console.log(`[cms-hosp] downloading ${CSV_URL}`);
  const res = await fetch(CSV_URL, { signal: AbortSignal.timeout(60000) });
  if (!res.ok) throw new Error(`CMS hospitals download failed: HTTP ${res.status}`);
  const buf = Buffer.from(await res.arrayBuffer());
  fs.writeFileSync(CACHE, buf);
  console.log(`[cms-hosp] cached → ${CACHE} (${(buf.length/1e6).toFixed(1)} MB)`);
}

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) {
  const r = await query(
    `INSERT INTO scrape_jobs (source_id, job_label, status, started_at)
     VALUES ($1,'cms-hosp:bulk','running',NOW()) RETURNING id`, [sourceId]);
  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 upsertHospital(client, h) {
  // Match by CCN (Provider ID) → cdph_license, then name+zip.
  let r;
  if (h.ccn) {
    r = await client.query(`SELECT id FROM organizations WHERE cdph_license=$1 OR hcai_id=$1 LIMIT 1`, [h.ccn]);
    if (r.rowCount) {
      await client.query(`
        UPDATE organizations SET
          name=COALESCE($2,name), type='hospital',
          address=COALESCE($3,address), city=COALESCE($4,city),
          zip=COALESCE($5,zip), county=COALESCE($6,county),
          phone=COALESCE($7,phone), updated_at=NOW()
        WHERE id=$1
      `, [r.rows[0].id, h.name, h.address, h.city, h.zip, h.county, h.phone]);
      return r.rows[0].id;
    }
  }
  if (h.name && h.zip) {
    r = await client.query(`SELECT id FROM organizations WHERE LOWER(name)=LOWER($1) AND zip=$2 LIMIT 1`, [h.name, h.zip]);
    if (r.rowCount) {
      await client.query(`UPDATE organizations SET type='hospital', cdph_license=COALESCE($2, cdph_license), phone=COALESCE($3, phone), updated_at=NOW() WHERE id=$1`,
                         [r.rows[0].id, h.ccn, h.phone]);
      return r.rows[0].id;
    }
  }
  r = await client.query(`
    INSERT INTO organizations
      (name, type, address, city, state, zip, county, phone, cdph_license)
    VALUES ($1,'hospital',$2,$3,'CA',$4,$5,$6,$7)
    RETURNING id
  `, [h.name, h.address, h.city, h.zip, h.county, h.phone, h.ccn]);
  return r.rows[0].id;
}

async function main() {
  await downloadIfNeeded();
  const sourceId = await getSourceId();
  const jobId = await startJob(sourceId);
  const laZips = await laZipSet();

  let scanned = 0, kept = 0, upserted = 0;

  const parser = fs.createReadStream(CACHE).pipe(parse({
    columns: true, skip_empty_lines: true, bom: true, relax_column_count: true,
  }));

  for await (const row of parser) {
    scanned++;
    const state = pick(row, 'State');
    if (state && state !== 'CA') continue;
    const z = zip5(pick(row, 'ZIP Code'));
    const county = pick(row, 'County/Parish', 'County');
    if (!((z && laZips.has(z)) || /los angeles/i.test(county || ''))) continue;
    kept++;

    const h = {
      ccn:     pick(row, 'CMS Certification Number (CCN)', 'Facility ID', 'Provider ID'),
      name:    pick(row, 'Hospital Name', 'Facility Name'),
      address: pick(row, 'Address'),
      city:    pick(row, 'City', 'City/Town', 'CITY'),
      zip:     z,
      county:  county || 'Los Angeles',
      phone:   pick(row, 'Phone Number', 'Telephone'),
    };
    if (!h.name) continue;

    try {
      await withTx(async (client) => {
        const id = await upsertHospital(client, h);
        const json = JSON.stringify(row);
        const hash = sha256(json + '|organization|' + id);
        await client.query(`
          INSERT INTO raw_records
            (source_id, source_url, entity_type, entity_id, raw_json, hash)
          VALUES ($1,$2,'organization',$3,$4::jsonb,$5)
          ON CONFLICT (hash) DO NOTHING
        `, [sourceId, SOURCE_URL, id, json, hash]);
        upserted++;
      });
    } catch (e) {
      console.error(`[cms-hosp] error ${h.name}: ${e.message}`);
    }
  }

  await finishJob(jobId, {
    status: 'done', records_found: scanned,
    records_inserted: upserted, records_skipped: scanned - kept,
  });
  console.log(`[cms-hosp] done. scanned=${scanned} la_kept=${kept} upserted=${upserted}`);
  await pool.end();
}

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