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

154 lines
#!/usr/bin/env node
/**
 * HRSA FQHC / Health Center Look-Alike importer — Stage 4.
 *
 * HRSA publishes the federally qualified health center list as a public file:
 *   https://data.hrsa.gov/data/download
 *   "Health Center Service Delivery and LookAlike Sites" (CSV).
 *
 * For LA County we filter on State=CA + County=Los Angeles or LA-County ZIPs.
 * Marks org_type='fqhc' on matched rows (or inserts new ones).
 */
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 = 'HRSA FQHC Locator';
const SOURCE_URL  = 'https://data.hrsa.gov/data/download';
// HRSA's stable API endpoint for "Health Center Service Delivery Sites" CSV.
const CSV_URL     = process.env.HRSA_CSV_URL
  || 'https://data.hrsa.gov/DataDownload/DD_Files/Health_Center_Service_Delivery_and_LookAlike_Sites.csv';
const CACHE       = path.resolve(__dirname, '../../../data/hrsa/Health_Center_Service_Delivery_and_LookAlike_Sites.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 < 14) {
      console.log(`[hrsa] using cache (${ageDays.toFixed(1)}d old)`);
      return;
    }
  }
  fs.mkdirSync(path.dirname(CACHE), { recursive: true });
  console.log(`[hrsa] downloading ${CSV_URL}`);
  const res = await fetch(CSV_URL, { signal: AbortSignal.timeout(60000) });
  if (!res.ok) throw new Error(`HRSA download failed: HTTP ${res.status}`);
  const buf = Buffer.from(await res.arrayBuffer());
  fs.writeFileSync(CACHE, buf);
  console.log(`[hrsa] 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,'hrsa: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 upsertFqhc(client, row) {
  const r = await client.query(`
    INSERT INTO organizations (name, type, address, city, state, zip, county, phone)
    VALUES ($1, 'fqhc', $2, $3, $4, $5, $6, $7)
    RETURNING id
  `, [row.name, row.address, row.city, 'CA', row.zip, 'Los Angeles', row.phone]);
  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, inserted = 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, 'Site State Abbreviation', 'State Abbreviation', 'STATE', 'State');
    if (state && state !== 'CA') continue;
    const z = zip5(pick(row, 'Site Postal Code', 'ZIP', 'Postal Code'));
    const county = pick(row, 'Site County Name', 'County', 'COUNTY');
    const inLA = (z && laZips.has(z)) || /los angeles/i.test(county || '');
    if (!inLA) continue;
    kept++;

    const fac = {
      name:    pick(row, 'Site Name', 'BHCMISID Site Name', 'Health Center Name'),
      address: pick(row, 'Site Address', 'Address', 'Street Address'),
      city:    pick(row, 'Site City', 'City'),
      zip:     z,
      phone:   pick(row, 'Site Telephone Number', 'Phone Number', 'Telephone'),
    };
    if (!fac.name) continue;

    try {
      await withTx(async (client) => {
        const id = await upsertFqhc(client, fac);
        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]);
        inserted++;
      });
    } catch (e) {
      console.error(`[hrsa] row error: ${e.message}`);
    }
  }

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

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