← back to Professional Directory
agents/ingest-agent/importers/hcai.js
250 lines
#!/usr/bin/env node
/**
* HCAI / CDPH Licensed Healthcare Facility Locations — Stage 4.
*
* Source: https://data.chhs.ca.gov/dataset/healthcare-facility-listing
* Resource: "Licensed and Certified Healthcare Facility Locations"
* CSV (free, no key, public domain).
*
* Populates `organizations` for every licensed CA facility, then narrows the
* "in scope" set to LA County. We keep statewide rows so other counties can
* be added later by toggling the LA-only flag.
*
* Mapping:
* FACILITY_NAME → name
* FACILITY_TYPE → type (mapped to our enum)
* FACID / OSHPD_ID → hcai_id
* FAC_FEI / LICENSE_NUMBER → cdph_license
* ADDRESS / CITY / ZIP → address / city / zip
*/
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';
const SOURCE_URL = 'https://data.chhs.ca.gov/dataset/healthcare-facility-listing';
const CSV_URL = process.env.HCAI_CSV_URL
|| 'https://data.chhs.ca.gov/dataset/3b5b80e8-6b8d-4715-b3c0-2699af6e72e5/resource/f0ae5731-fef8-417f-839d-54a0ed3a126e/download/health_facility_locations.csv';
const CACHE = path.resolve(__dirname, '../../../data/hcai/health_facility_locations.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 mapType(facType) {
if (!facType) return 'clinic';
const f = facType.toLowerCase();
if (f.includes('hospital') || f.includes('acute')) return 'hospital';
if (f.includes('skilled nursing') || f.includes('nursing facility')) return 'nursing_facility';
if (f.includes('surgical') || f.includes('surgery')) return 'surgery_center';
if (f.includes('urgent')) return 'urgent_care';
if (f.includes('clinic') || f.includes('community')) return 'clinic';
if (f.includes('home health')) return 'home_health';
if (f.includes('hospice')) return 'hospice';
if (f.includes('dialysis')) return 'dialysis';
return 'clinic';
}
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 < 7) {
console.log(`[hcai] using cache (${ageDays.toFixed(1)}d old)`);
return;
}
}
console.log(`[hcai] downloading ${CSV_URL}`);
fs.mkdirSync(path.dirname(CACHE), { recursive: true });
const res = await fetch(CSV_URL, { signal: AbortSignal.timeout(120000) });
if (!res.ok) throw new Error(`HCAI download failed: HTTP ${res.status}`);
const buf = Buffer.from(await res.arrayBuffer());
fs.writeFileSync(CACHE, buf);
console.log(`[hcai] 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,'hcai: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 upsertOrg(client, fac) {
// Match priority: hcai_id, cdph_license, name+zip.
let r;
if (fac.hcai_id) {
r = await client.query(
`SELECT id FROM organizations WHERE hcai_id = $1 LIMIT 1`,
[fac.hcai_id]
);
if (r.rowCount > 0) return updateOrg(client, r.rows[0].id, fac);
}
if (fac.cdph_license) {
r = await client.query(
`SELECT id FROM organizations WHERE cdph_license = $1 LIMIT 1`,
[fac.cdph_license]
);
if (r.rowCount > 0) return updateOrg(client, r.rows[0].id, fac);
}
if (fac.name && fac.zip) {
r = await client.query(
`SELECT id FROM organizations
WHERE LOWER(name) = LOWER($1) AND zip = $2
LIMIT 1`,
[fac.name, fac.zip]
);
if (r.rowCount > 0) return updateOrg(client, r.rows[0].id, fac);
}
// Insert new.
r = await client.query(`
INSERT INTO organizations
(name, type, address, city, state, zip, county, phone,
hcai_id, cdph_license, lat, lng, geocoded_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,
$11::double precision, $12::double precision,
CASE WHEN $11::double precision IS NOT NULL
AND $12::double precision IS NOT NULL THEN NOW() END)
RETURNING id
`, [
fac.name, fac.type, fac.address, fac.city, 'CA', fac.zip,
fac.county || 'Los Angeles', fac.phone, fac.hcai_id, fac.cdph_license,
fac.lat, fac.lng,
]);
return r.rows[0].id;
}
async function updateOrg(client, orgId, fac) {
// HCAI is authoritative for facility type — overwrite (not COALESCE) when present.
await client.query(`
UPDATE organizations SET
name = COALESCE($2, name),
type = COALESCE($3, type),
address = COALESCE($4, address),
city = COALESCE($5, city),
zip = COALESCE($6, zip),
county = COALESCE($7, county),
phone = COALESCE($8, phone),
hcai_id = COALESCE($9, hcai_id),
cdph_license= COALESCE($10, cdph_license),
lat = COALESCE($11::double precision, lat),
lng = COALESCE($12::double precision, lng),
geocoded_at = CASE WHEN $11::double precision IS NOT NULL
AND $12::double precision IS NOT NULL THEN NOW()
ELSE geocoded_at END,
updated_at = NOW()
WHERE id = $1
`, [orgId, fac.name, fac.type, fac.address, fac.city, fac.zip,
fac.county, fac.phone, fac.hcai_id, fac.cdph_license,
fac.lat, fac.lng]);
return orgId;
}
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 fac = {
name: pick(row, 'FACNAME', 'FACILITY_NAME', 'FAC_NAME'),
facType: pick(row, 'FAC_FDR', 'FACILITY_TYPE', 'ENTITY_TYPE_DESCRIPTION'),
hcai_id: pick(row, 'HCAI_ID', 'OSHPD_ID', 'FACID'),
cdph_license: pick(row, 'LICENSE_NUMBER', 'FAC_FEI'),
address: pick(row, 'ADDRESS', 'FAC_ADDRESS', 'STREET_ADDRESS'),
city: pick(row, 'CITY', 'FAC_CITY'),
county: pick(row, 'COUNTY_NAME', 'COUNTY'),
zip: zip5(pick(row, 'ZIP', 'ZIP_CODE', 'POSTAL_CODE')),
phone: pick(row, 'CONTACT_PHONE_NUMBER', 'PHONE', 'FAC_PHONE'),
lat: parseFloat(pick(row, 'LATITUDE') || '') || null,
lng: parseFloat(pick(row, 'LONGITUDE') || '') || null,
status: pick(row, 'LICENSE_STATUS_DESCRIPTION'),
};
fac.type = mapType(fac.facType);
// LA County filter + active-license only
const inLA = (fac.zip && laZips.has(fac.zip)) ||
(fac.county && /los angeles/i.test(fac.county));
if (!inLA) continue;
if (!fac.name) continue;
if (fac.status && fac.status !== 'ACTIVE') continue;
kept++;
try {
await withTx(async (client) => {
const id = await upsertOrg(client, fac);
// Provenance.
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(`[hcai] row error ${fac.name}: ${e.message}`);
}
}
await finishJob(jobId, {
status: 'done',
records_found: scanned,
records_inserted: inserted,
records_skipped: scanned - kept,
});
console.log(`[hcai] done. scanned=${scanned} la_kept=${kept} upserted=${inserted}`);
await pool.end();
}
main().catch(async (err) => {
console.error('[hcai] fatal:', err);
try { await pool.end(); } catch (_) {}
process.exit(1);
});