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

373 lines
#!/usr/bin/env node
/**
 * NPI bulk importer — Stage 1 of the Professional Directory build.
 *
 * Streams the NPPES NPI bulk CSV (downloaded separately to ./data/nppes/),
 * filters to state=CA + LA County ZIPs, and upserts:
 *   - Type 1 (individual)     → professionals
 *   - Type 2 (organization)   → organizations
 * Plus professional_locations, phones, professional_specialties, raw_records.
 *
 * Sources: https://download.cms.gov/nppes/NPI_Files.html  (public domain, no key)
 *
 * Smoke mode (SMOKE=1) reads agents/ingest-agent/importers/__fixtures__/npi_smoke.csv
 * — 5 synthetic LA-county rows — so the pipeline can be exercised end-to-end
 * without downloading the ~6 GB bulk file.
 */
const fs = require('node:fs');
const path = require('node:path');
const crypto = require('node:crypto');
const { parse } = require('csv-parse');
const { pool, query, withTx } = require('../../shared/db');
const { laZipSet } = require('../../shared/la-zips');

const SMOKE = process.env.SMOKE === '1';
const FIXTURE = path.join(__dirname, '__fixtures__/npi_smoke.csv');
const DATA_DIR = path.resolve(__dirname, '../../../data/nppes');

const SOURCE_NAME = 'NPPES NPI Registry (Bulk)';

// ─── Helpers ────────────────────────────────────────────────────────────────

function sha256(s) { return crypto.createHash('sha256').update(s).digest('hex'); }

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 fullName(row) {
  const parts = [
    row['Provider Name Prefix Text'],
    row['Provider First Name'],
    row['Provider Middle Name'],
    row['Provider Last Name (Legal Name)'],
    row['Provider Name Suffix Text'],
  ].map(clean).filter(Boolean);
  return parts.join(' ');
}

function pickLicense(row) {
  // Find the first CA license. NPI rows can have license_1 .. license_15.
  for (let i = 1; i <= 15; i++) {
    const n = clean(row[`Provider License Number_${i}`]);
    const s = clean(row[`Provider License Number State Code_${i}`]);
    if (n && s === 'CA') return { license_number: n };
  }
  return { license_number: null };
}

function pickTaxonomies(row) {
  const out = [];
  for (let i = 1; i <= 15; i++) {
    const code = clean(row[`Healthcare Provider Taxonomy Code_${i}`]);
    if (!code) continue;
    const isPrimary = clean(row[`Healthcare Provider Primary Taxonomy Switch_${i}`]) === 'Y';
    out.push({ code, is_primary: isPrimary });
  }
  // Prefer the primary; fall back to first.
  out.sort((a, b) => Number(b.is_primary) - Number(a.is_primary));
  return out;
}

function pickPracticeAddress(row) {
  return {
    address: [clean(row['Provider First Line Business Practice Location Address']),
              clean(row['Provider Second Line Business Practice Location Address'])]
              .filter(Boolean).join(', ') || null,
    city:    clean(row['Provider Business Practice Location Address City Name']),
    state:   clean(row['Provider Business Practice Location Address State Name']),
    zip:     zip5(row['Provider Business Practice Location Address Postal Code']),
    phone:   clean(row['Provider Business Practice Location Address Telephone Number']),
  };
}

function specialtyName(taxonomyCode, taxonomyMap) {
  return taxonomyMap.get(taxonomyCode) || null;
}

// ─── Upserts ────────────────────────────────────────────────────────────────

async function upsertProfessional(client, row, src) {
  const npi = clean(row['NPI']);
  if (!npi) return null;

  const license = pickLicense(row);
  const taxonomies = pickTaxonomies(row);
  const primaryTax = taxonomies[0]?.code || null;

  const enumDate = clean(row['Provider Enumeration Date']);
  // CMS NPI dates are MM/DD/YYYY
  const issueDate = enumDate ? toIsoDate(enumDate) : null;

  const params = [
    fullName(row),
    clean(row['Provider First Name']),
    clean(row['Provider Last Name (Legal Name)']),
    clean(row['Provider Middle Name']),
    clean(row['Provider Name Suffix Text']),
    clean(row['Provider Credential Text']),
    license.license_number,
    null, // license_type — filled by DCA/MBC stages
    null, // license_status
    issueDate,
    npi,
    clean(row['Provider Gender Code']),
    primaryTax,
    0.30, // initial NPI-only confidence; raised by later stages
  ];
  if (process.env.NPI_DEBUG === '1') {
    console.log('[npi] params lengths =', params.map(p => p === null ? 'null' : `'${p}'(${String(p).length})`));
  }
  const r = await client.query(`
    INSERT INTO professionals (
      full_name, first_name, last_name, middle_name, suffix, title,
      license_number, license_type, license_status, license_issue_date,
      npi_number, gender, primary_specialty,
      source_confidence_score
    ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
    ON CONFLICT (npi_number) DO UPDATE SET
      full_name           = EXCLUDED.full_name,
      first_name          = EXCLUDED.first_name,
      last_name           = EXCLUDED.last_name,
      middle_name         = EXCLUDED.middle_name,
      license_number      = COALESCE(EXCLUDED.license_number, professionals.license_number),
      primary_specialty   = COALESCE(EXCLUDED.primary_specialty, professionals.primary_specialty),
      updated_at          = NOW()
    RETURNING id
  `, params);
  return r.rows[0].id;
}

async function upsertOrganization(client, row, src) {
  const npi = clean(row['NPI']);
  if (!npi) return null;

  const orgName = clean(row['Provider Organization Name (Legal Business Name)']);
  if (!orgName) return null;

  const addr = pickPracticeAddress(row);

  const r = await client.query(`
    INSERT INTO organizations (
      name, type, address, city, state, zip, county, phone, npi_number
    ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
    ON CONFLICT (npi_number) DO UPDATE SET
      name    = EXCLUDED.name,
      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
  `, [
    orgName,
    'medical_group', // best guess until Stage 4 fills facility-type rows
    addr.address,
    addr.city,
    addr.state,
    addr.zip,
    'Los Angeles',
    addr.phone,
    npi,
  ]);
  return r.rows[0].id;
}

async function upsertProfessionalLocation(client, professionalId, row, sourceUrl) {
  if (!professionalId) return;
  const addr = pickPracticeAddress(row);
  if (!addr.address && !addr.zip) return;

  await client.query(`
    INSERT INTO professional_locations (
      professional_id, role, address, phone, source_url, is_primary, last_verified_at
    ) VALUES ($1,$2,$3,$4,$5,TRUE,NOW())
  `, [professionalId, 'attending', addr.address, addr.phone, sourceUrl]);

  if (addr.phone) {
    await client.query(`
      INSERT INTO phones (professional_id, phone, phone_type, source_url, last_verified_at)
      VALUES ($1,$2,'office',$3,NOW())
    `, [professionalId, addr.phone, sourceUrl]);
  }
}

async function attachSpecialties(client, professionalId, row, taxonomyMap) {
  if (!professionalId) return;
  const taxonomies = pickTaxonomies(row);
  for (const tax of taxonomies) {
    // Make sure a specialties row exists for this taxonomy code.
    const name = specialtyName(tax.code, taxonomyMap) || tax.code;
    const sp = await client.query(`
      INSERT INTO specialties (name, taxonomy_code)
      VALUES ($1,$2)
      ON CONFLICT (taxonomy_code) DO UPDATE SET name = EXCLUDED.name
      RETURNING id
    `, [name, tax.code]);
    await client.query(`
      INSERT INTO professional_specialties (professional_id, specialty_id, source, confidence_score)
      VALUES ($1,$2,'NPI',$3)
      ON CONFLICT (professional_id, specialty_id) DO UPDATE SET source = EXCLUDED.source
    `, [professionalId, sp.rows[0].id, tax.is_primary ? 0.90 : 0.60]);
  }
}

async function stashRaw(client, sourceId, row, entityType, entityId) {
  const json = JSON.stringify(row);
  const hash = sha256(json + '|' + entityType + '|' + (entityId || ''));
  await client.query(`
    INSERT INTO raw_records (source_id, source_url, entity_type, entity_id, raw_json, hash)
    VALUES ($1,$2,$3,$4,$5::jsonb,$6)
    ON CONFLICT (hash) DO NOTHING
  `, [
    sourceId,
    'https://download.cms.gov/nppes/NPI_Files.html',
    entityType,
    entityId,
    json,
    hash,
  ]);
}

function toIsoDate(s) {
  // Try MM/DD/YYYY first (NPPES default), then YYYY-MM-DD.
  const us = s.match(/^(\d{2})\/(\d{2})\/(\d{4})$/);
  if (us) return `${us[3]}-${us[1]}-${us[2]}`;
  const iso = s.match(/^(\d{4})-(\d{2})-(\d{2})$/);
  return iso ? s : null;
}

// ─── Main ───────────────────────────────────────────────────────────────────

async function loadTaxonomyMap() {
  // Populated by seed_specialties — fall back to empty map.
  const r = await query('SELECT taxonomy_code, name FROM specialties WHERE taxonomy_code IS NOT NULL', []);
  return new Map(r.rows.map(r => [r.taxonomy_code, r.name]));
}

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);
}

function pickInputCsv() {
  if (SMOKE) return FIXTURE;
  if (!fs.existsSync(DATA_DIR)) {
    throw new Error(`Bulk NPI dir missing: ${DATA_DIR}\n` +
      `Download the latest npidata_pfile_*.csv from\n` +
      `  https://download.cms.gov/nppes/NPI_Files.html\n` +
      `and place it in ${DATA_DIR}/.\n` +
      `For a quick smoke test instead, run:  SMOKE=1 npm run ingest:npi`);
  }
  const csvs = fs.readdirSync(DATA_DIR)
    .filter(f => /^npidata_pfile.*\.csv$/i.test(f))
    .filter(f => !/_fileheader\.csv$/i.test(f))   // exclude the 12KB header-only file
    .sort();
  if (csvs.length === 0) throw new Error(`No npidata_pfile_*.csv in ${DATA_DIR}`);
  return path.join(DATA_DIR, csvs[csvs.length - 1]);
}

async function main() {
  console.log(`[npi] mode=${SMOKE ? 'smoke' : 'bulk'}`);
  const inputCsv = pickInputCsv();
  console.log(`[npi] input = ${inputCsv}`);

  const sourceId = await getSourceId();
  const jobId = await startJob(sourceId, SMOKE ? 'npi:smoke' : 'npi:bulk');
  const taxonomyMap = await loadTaxonomyMap();
  const laZips = await laZipSet();

  let found = 0, inserted = 0, updated = 0, skipped = 0;

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

  for await (const row of parser) {
    found++;

    const entityType = clean(row['Entity Type Code']); // '1' or '2'
    const stateCode  = clean(row['Provider Business Practice Location Address State Name']);
    const z          = zip5(row['Provider Business Practice Location Address Postal Code']);

    if (stateCode && stateCode !== 'CA') { skipped++; continue; }
    if (z && !laZips.has(z))             { skipped++; continue; }
    if (!stateCode && !z)                { skipped++; continue; } // no usable address

    try {
      await withTx(async (client) => {
        if (entityType === '1') {
          const id = await upsertProfessional(client, row, SOURCE_NAME);
          await upsertProfessionalLocation(client, id, row, 'https://download.cms.gov/nppes/NPI_Files.html');
          await attachSpecialties(client, id, row, taxonomyMap);
          await stashRaw(client, sourceId, row, 'professional', id);
          inserted++;
        } else if (entityType === '2') {
          const id = await upsertOrganization(client, row, SOURCE_NAME);
          await stashRaw(client, sourceId, row, 'organization', id);
          inserted++;
        } else {
          skipped++;
        }
      });
    } catch (e) {
      console.error(`[npi] row error npi=${row['NPI']}: ${e.message} (col=${e.column}, schemaTable=${e.table})`);
      skipped++;
    }

    if (found % 5000 === 0) {
      console.log(`[npi] scanned=${found} kept=${inserted} skipped=${skipped}`);
    }
  }

  await finishJob(jobId, {
    status: 'done',
    records_found: found,
    records_inserted: inserted,
    records_updated: updated,
    records_skipped: skipped,
  });

  console.log(`[npi] done. scanned=${found} kept=${inserted} skipped=${skipped}`);
  await pool.end();
}

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