`scripts/crawl-hospitals.js`

172 lines
#!/usr/bin/env node
/**
 * Hospital directory crawl dispatcher — Stage 5.
 * Runs each hospital-system crawler module under crawlers/ and persists
 * results back into professionals + professional_locations.
 *
 * Usage:
 *   node scripts/crawl-hospitals.js                      # all 10
 *   node scripts/crawl-hospitals.js ucla-health,cedars-sinai
 *   MAX_PAGES=5 node scripts/crawl-hospitals.js ucla-health   # quick smoke
 */
const path = require('path');
const fs = require('fs');
const crypto = require('node:crypto');
const { pool, query, withTx } = require('../agents/shared/db');

const CRAWLER_DIR = path.resolve(__dirname, '../agents/crawler-agent/crawlers');
const MAX_PAGES = process.env.MAX_PAGES ? Number(process.env.MAX_PAGES) : Infinity;

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

async function getSourceId(name) {
  const r = await query('SELECT id FROM sources WHERE source_name = $1', [name]);
  return r.rowCount ? r.rows[0].id : null;
}

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

async function persistProvider(provider, sourceId) {
  if (!provider.name) return;

  await withTx(async (client) => {
    // Try to match an existing professional by NPI, then by name+specialty.
    let professionalId;
    if (provider.npi) {
      const r = await client.query(`SELECT id FROM professionals WHERE npi_number = $1`, [provider.npi]);
      if (r.rowCount) professionalId = r.rows[0].id;
    }
    if (!professionalId) {
      const r = await client.query(`
        SELECT id FROM professionals
         WHERE LOWER(full_name) = LOWER($1)
            OR (LOWER(last_name) = LOWER(SPLIT_PART($1, ' ', -1))
                AND LOWER(first_name) = LOWER(SPLIT_PART($1, ' ', 1)))
         LIMIT 2
      `, [provider.name]);
      if (r.rowCount === 1) professionalId = r.rows[0].id;
    }

    if (!professionalId) {
      // New professional — name only, low confidence until license joins.
      const r = await client.query(`
        INSERT INTO professionals (full_name, primary_specialty, profile_image_url, bio, source_confidence_score)
        VALUES ($1,$2,$3,$4,0.30)
        RETURNING id
      `, [provider.name, provider.specialty, provider.image_url, provider.bio]);
      professionalId = r.rows[0].id;
    } else {
      await client.query(`
        UPDATE professionals SET
          primary_specialty   = COALESCE(primary_specialty, $2),
          profile_image_url   = COALESCE(profile_image_url, $3),
          bio                 = COALESCE(bio, $4),
          medical_school      = COALESCE(medical_school, $5),
          source_confidence_score = GREATEST(COALESCE(source_confidence_score,0), 0.85),
          updated_at = NOW()
        WHERE id = $1
      `, [professionalId, provider.specialty, provider.image_url, provider.bio, provider.medical_school]);
    }

    // Hospital affiliation.
    const orgRes = await client.query(`SELECT id FROM organizations WHERE name = $1 LIMIT 1`, [provider.hospital_system]);
    let hospId;
    if (orgRes.rowCount) hospId = orgRes.rows[0].id;
    else {
      const ins = await client.query(`
        INSERT INTO organizations (name, type, state, county) VALUES ($1,'hospital','CA','Los Angeles') RETURNING id
      `, [provider.hospital_system]);
      hospId = ins.rows[0].id;
    }

    await client.query(`
      INSERT INTO professional_locations
        (professional_id, organization_id, role, address, phone, source_url, last_verified_at)
      VALUES ($1,$2,'privileges',$3,$4,$5,NOW())
      ON CONFLICT DO NOTHING
    `, [professionalId, hospId, provider.address, provider.phone, provider.source_url]);

    if (provider.email) {
      await client.query(`
        INSERT INTO emails (professional_id, email, email_type, source_url, discovered_at, verification_status)
        VALUES ($1, $2, 'office', $3, NOW(), 'unverified')
        ON CONFLICT DO NOTHING
      `, [professionalId, provider.email, provider.source_url]);
    }
    if (provider.phone) {
      await client.query(`
        INSERT INTO phones (professional_id, phone, phone_type, source_url, last_verified_at)
        VALUES ($1, $2, 'office', $3, NOW())
        ON CONFLICT DO NOTHING
      `, [professionalId, provider.phone, provider.source_url]);
    }

    // Provenance.
    const json = JSON.stringify(provider);
    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, provider.source_url, professionalId, json, hash]);
  });
}

async function runCrawler(modName) {
  const file = path.join(CRAWLER_DIR, `${modName}.js`);
  if (!fs.existsSync(file)) {
    console.warn(`[crawl] crawler not found: ${modName}`);
    return 0;
  }
  const mod = require(file);
  const sourceId = await getSourceId(mod.SOURCE_NAME);
  if (!sourceId) {
    console.warn(`[crawl] source not seeded for ${modName} — skipping`);
    return 0;
  }
  const jobId = await startJob(sourceId, `crawl:${modName}`);
  let inserted = 0, errors = 0;

  try {
    await mod.crawl({
      onProvider: async (p) => {
        try { await persistProvider(p, sourceId); inserted++; }
        catch (e) { errors++; console.error(`[crawl] persist error: ${e.message}`); }
      },
      maxPages: MAX_PAGES === Infinity ? Infinity : MAX_PAGES,
    });
    await finishJob(jobId, { status: 'done', records_inserted: inserted, records_skipped: errors });
  } catch (e) {
    await finishJob(jobId, { status: 'failed', error_message: e.message, records_inserted: inserted });
    throw e;
  }
  console.log(`[crawl:${modName}] inserted=${inserted} errors=${errors}`);
  return inserted;
}

async function main() {
  const targets = process.argv[2]
    ? process.argv[2].split(',')
    : fs.readdirSync(CRAWLER_DIR).filter(f => f.endsWith('.js')).map(f => f.replace(/\.js$/, ''));

  for (const t of targets) {
    try { await runCrawler(t); }
    catch (e) { console.error(`[crawl:${t}] fatal: ${e.message}`); }
  }
  await pool.end();
}

main().catch((e) => { console.error(e); process.exit(1); });