`agents/ingest-agent/importers/cms-care-compare.js`

225 lines
#!/usr/bin/env node
/**
 * CMS Care Compare — "Doctors and Clinicians National Downloadable File"
 * Stage 3 enrichment.
 *
 *   Source: https://data.cms.gov/provider-data/dataset/mj5m-pzi6
 *   CSV   : https://data.cms.gov/provider-data/sites/default/files/resources/<id>/DAC_NationalDownloadableFile.csv
 *
 * Joins to professionals by NPI. Adds:
 *   primary_specialty (CMS specialty)  →  professionals.primary_specialty
 *   hospital_affiliation_ccn / name    →  organizations + affiliation row
 *   group_practice_pac_id / name       →  organizations (medical_group)
 *   gender                             →  professionals.gender
 *   credentials                        →  professionals.title
 *
 * No PII exposed beyond what NPPES already publishes.
 */
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 = 'CMS Care Compare (Doctors and Clinicians)';
const SOURCE_URL  = 'https://data.cms.gov/provider-data/dataset/mj5m-pzi6';
const CSV_URL     = process.env.CMS_DAC_URL
  || 'https://data.cms.gov/provider-data/sites/default/files/resources/52c3f098d7e56028a298fd297cb0b38d_1774905035/DAC_NationalDownloadableFile.csv';
const CACHE       = path.resolve(__dirname, '../../../data/cms/DAC_NationalDownloadableFile.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-dac] using cache (${ageDays.toFixed(1)}d old)`);
      return;
    }
  }
  fs.mkdirSync(path.dirname(CACHE), { recursive: true });
  console.log(`[cms-dac] downloading ${CSV_URL}`);
  const res = await fetch(CSV_URL, { signal: AbortSignal.timeout(180000) });
  if (!res.ok) throw new Error(`CMS DAC download failed: HTTP ${res.status}`);
  const buf = Buffer.from(await res.arrayBuffer());
  fs.writeFileSync(CACHE, buf);
  console.log(`[cms-dac] 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-dac: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 findProfessional(client, npi) {
  if (!npi) return null;
  const r = await client.query(`SELECT id FROM professionals WHERE npi_number = $1 LIMIT 1`, [npi]);
  return r.rowCount ? r.rows[0].id : null;
}

async function ensureGroup(client, name, address, city, state, zip, ccn) {
  if (!name) return null;
  // Try CCN first, then name+zip.
  let r;
  if (ccn) {
    r = await client.query(`SELECT id FROM organizations WHERE cdph_license = $1 OR hcai_id = $1 LIMIT 1`, [ccn]);
    if (r.rowCount) return r.rows[0].id;
  }
  if (name && zip) {
    r = await client.query(`SELECT id FROM organizations WHERE LOWER(name)=LOWER($1) AND zip=$2 LIMIT 1`, [name, zip]);
    if (r.rowCount) return r.rows[0].id;
  }
  r = await client.query(`
    INSERT INTO organizations (name, type, address, city, state, zip, county)
    VALUES ($1,$2,$3,$4,$5,$6,$7)
    RETURNING id
  `, [name, 'medical_group', address, city, state || 'CA', zip, 'Los Angeles']);
  return r.rows[0].id;
}

async function applyDac(client, professionalId, row, sourceId) {
  const cmsSpecialty = pick(row, 'pri_spec', 'Primary specialty', 'Primary Specialty');
  const credentials  = pick(row, 'Cred', 'Credential', 'cred');
  const gender       = pick(row, 'gndr', 'Gender');

  await client.query(`
    UPDATE professionals SET
      primary_specialty       = COALESCE($2, primary_specialty),
      title                   = COALESCE($3, title),
      gender                  = COALESCE($4, gender),
      source_confidence_score = GREATEST(COALESCE(source_confidence_score, 0), 0.75),
      updated_at              = NOW()
    WHERE id = $1
  `, [professionalId, cmsSpecialty, credentials, gender ? gender.charAt(0).toUpperCase() : null]);

  // Group practice as an organization + affiliation.
  // NB: the CMS DAC CSV uses 'Facility Name' for the group name (not 'org_nm'); 'org_nm'
  // appears in the older DAC schema. Keep both for forward-compat.
  const groupName = pick(row, 'Facility Name', 'org_nm', 'Organization legal name', 'Group name');
  const groupZip  = zip5(pick(row, 'ZIP Code', 'adr_zip', 'org_zip'));
  const groupCity = pick(row, 'City/Town', 'cty', 'City', 'org_cty');
  const groupAddr = pick(row, 'adr_ln_1', 'Address Line 1', 'org_adr_ln_1');
  if (groupName) {
    const orgId = await ensureGroup(client, groupName, groupAddr, groupCity, 'CA', groupZip,
                                    pick(row, 'org_pac_id', 'Group PAC ID'));
    await client.query(`
      INSERT INTO professional_locations
        (professional_id, organization_id, role, address, source_url, last_verified_at)
      VALUES ($1,$2,'employed',$3,$4,NOW())
      ON CONFLICT DO NOTHING
    `, [professionalId, orgId, groupAddr, SOURCE_URL]);
  }

  // Hospital affiliation. NB: `IN (..., NULL)` does not match NULLs in SQL —
  // use OR ... IS NULL to also promote rows whose type was never set.
  const hospName = pick(row, 'hosp_afl_1', 'Hospital affiliation name 1');
  const hospCcn  = pick(row, 'hosp_afl_lbn_1', 'Hospital affiliation CCN 1');
  if (hospName) {
    const hospId = await ensureGroup(client, hospName, null, null, 'CA', null, hospCcn);
    await client.query(`
      UPDATE organizations SET type = 'hospital', updated_at = NOW()
       WHERE id = $1 AND (type IN ('medical_group','clinic') OR type IS NULL)
    `, [hospId]);
    await client.query(`
      INSERT INTO professional_locations
        (professional_id, organization_id, role, source_url, last_verified_at)
      VALUES ($1,$2,'privileges',$3,NOW())
      ON CONFLICT DO NOTHING
    `, [professionalId, hospId, SOURCE_URL]);
  }

  // Provenance.
  const json = JSON.stringify(row);
  const hash = sha256(json + '|professional|' + professionalId);
  await client.query(`
    INSERT INTO raw_records (source_id, source_url, entity_type, entity_id, raw_json, hash)
    VALUES ($1,$2,'professional',$3,$4::jsonb,$5)
    ON CONFLICT (hash) DO NOTHING
  `, [sourceId, SOURCE_URL, professionalId, json, hash]);
}

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

  let scanned = 0, matched = 0, skipped = 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 npi = pick(row, 'NPI', 'npi');
    const state = pick(row, 'st', 'State');
    const z = zip5(pick(row, 'adr_zip', 'ZIP Code'));

    if (state && state !== 'CA') { skipped++; continue; }
    if (z && !laZips.has(z)) { skipped++; continue; }
    if (!npi) { skipped++; continue; }

    try {
      await withTx(async (client) => {
        const id = await findProfessional(client, npi);
        if (!id) { skipped++; return; }
        await applyDac(client, id, row, sourceId);
        matched++;
      });
    } catch (e) {
      console.error(`[cms-dac] row error npi=${npi}: ${e.message}`);
    }

    if (scanned % 10000 === 0) console.log(`[cms-dac] scanned=${scanned} matched=${matched}`);
  }

  await finishJob(jobId, {
    status: 'done',
    records_found: scanned, records_updated: matched, records_skipped: skipped,
  });

  console.log(`[cms-dac] done. scanned=${scanned} matched=${matched} skipped=${skipped}`);
  await pool.end();
}

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