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

184 lines
#!/usr/bin/env node
/**
 * DCA License Search enricher — Stage 2.
 *
 *   Source page: https://search.dca.ca.gov/
 *   Public, no key, robots-aware via lib compliance.
 *
 *   The public DCA "results" endpoint accepts a license number + board ID
 *   and returns HTML; we parse the structured fields. We rate-limit to
 *   0.5 rps per host and back off on any non-200.
 *
 * Run:
 *   node agents/ingest-agent/importers/dca.js                # all CA-licensed pros
 *   LIMIT=50 node agents/ingest-agent/importers/dca.js       # smoke
 */
const cheerio = require('cheerio');
const { pool, query, withTx } = require('../../shared/db');
const { fetchCompliant, setHostRateLimit } = require('../../shared/compliance');

const SOURCE_NAME = 'DCA License Search';
const BOARD_MD = '800';   // Medical Board of California — physicians & surgeons (M.D.)
const BOARD_DO = '870';   // Osteopathic Medical Board of California (D.O.)
const SEARCH_URL = 'https://search.dca.ca.gov/results';

setHostRateLimit('search.dca.ca.gov', 0.5);

function buildResultsUrl(licenseType, licenseNumber) {
  const board = licenseType === 'DO' ? BOARD_DO : BOARD_MD;
  // The site accepts a GET form post style URL.
  const qs = new URLSearchParams({
    boardCode: board,
    licenseType: 'A',     // any
    licenseNumber: licenseNumber,
  });
  return `${SEARCH_URL}?${qs.toString()}`;
}

function parseDcaPage(html) {
  const $ = cheerio.load(html);
  // Each license card has labels in <strong> with values following.
  const out = {};
  $('div.search_result, table tr').each((_, el) => {
    const $el = $(el);
    const text = $el.text().replace(/\s+/g, ' ').trim();
    const m = text.match(/License (?:Status|Type|Number|Issue Date|Expiration Date|Original Issue Date)[:\s]+([^\n]+?)(?=\s+License|\s+Address|\s+School|$)/gi);
    if (!m) return;
    for (const piece of m) {
      const [, key, val] = piece.match(/(License (?:Status|Type|Number|Issue Date|Expiration Date|Original Issue Date))[:\s]+(.+)/i) || [];
      if (key) out[key.toLowerCase().replace(/\s+/g, '_')] = val.trim();
    }
  });
  // Robust fallback: scan label/value pairs.
  $('td,div,span,li').each((_, el) => {
    const t = $(el).text().replace(/\s+/g,' ').trim();
    let m;
    if ((m = t.match(/^License Status:?\s*(.+)$/i)))         out.status = m[1].trim();
    if ((m = t.match(/^License Type:?\s*(.+)$/i)))           out.license_type_full = m[1].trim();
    if ((m = t.match(/^Original Issue Date:?\s*(.+)$/i)))    out.issue_date = m[1].trim();
    if ((m = t.match(/^Expiration Date:?\s*(.+)$/i)))        out.expiration_date = m[1].trim();
    if ((m = t.match(/^School Name:?\s*(.+)$/i)))            out.school = m[1].trim();
    if ((m = t.match(/^Graduation Year:?\s*(.+)$/i)))        out.graduation_year = m[1].trim();
    if ((m = t.match(/^Address of Record:?\s*(.+)$/i)))      out.address = m[1].trim();
  });
  return out;
}

function toIso(d) {
  if (!d) return null;
  const us = d.match(/(\d{1,2})\/(\d{1,2})\/(\d{4})/);
  if (us) return `${us[3]}-${us[1].padStart(2,'0')}-${us[2].padStart(2,'0')}`;
  const iso = d.match(/(\d{4})-(\d{2})-(\d{2})/);
  return iso ? `${iso[1]}-${iso[2]}-${iso[3]}` : null;
}

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 loadCandidates(limit) {
  const sql = `
    SELECT id, license_number, title
      FROM professionals
     WHERE license_number IS NOT NULL
       AND title IN ('MD','DO','M.D.','D.O.')
       AND (license_status IS NULL OR license_status NOT IN ('Active','Current'))
     ORDER BY id
     ${limit ? `LIMIT ${Number(limit)}` : ''}
  `;
  const r = await query(sql, []);
  return r.rows;
}

async function startJob(sourceId) {
  const r = await query(
    `INSERT INTO scrape_jobs (source_id, job_label, status, started_at)
     VALUES ($1,'dca:enrich','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 applyDca(professionalId, dca, sourceUrl, sourceId) {
  await withTx(async (client) => {
    await client.query(`
      UPDATE professionals SET
        license_status          = COALESCE($2, license_status),
        license_issue_date      = COALESCE($3, license_issue_date),
        license_expiration_date = COALESCE($4, license_expiration_date),
        medical_school          = COALESCE($5, medical_school),
        graduation_year         = COALESCE($6, graduation_year),
        source_confidence_score = GREATEST(COALESCE(source_confidence_score,0), 0.80),
        updated_at = NOW()
      WHERE id = $1
    `, [
      professionalId,
      dca.status || null,
      toIso(dca.issue_date),
      toIso(dca.expiration_date),
      dca.school || null,
      dca.graduation_year ? Number(String(dca.graduation_year).match(/\d{4}/)?.[0]) || null : null,
    ]);

    const json = JSON.stringify(dca);
    const crypto = require('node:crypto');
    const hash = crypto.createHash('sha256').update(json + '|professional|' + professionalId).digest('hex');
    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, sourceUrl, professionalId, json, hash]);
  });
}

async function main() {
  const limit = process.env.LIMIT ? Number(process.env.LIMIT) : null;
  const sourceId = await getSourceId();
  const jobId = await startJob(sourceId);
  const candidates = await loadCandidates(limit);
  console.log(`[dca] candidates=${candidates.length}`);

  let ok = 0, fail = 0;
  for (const cand of candidates) {
    const url = buildResultsUrl(cand.title, cand.license_number);
    try {
      const res = await fetchCompliant(url);
      if (!res.ok) { fail++; continue; }
      const html = await res.text();
      const dca = parseDcaPage(html);
      if (Object.keys(dca).length === 0) { fail++; continue; }
      await applyDca(cand.id, dca, url, sourceId);
      ok++;
    } catch (e) {
      if (e.code === 'ROBOTS_DISALLOWED' || e.code === 'CAPTCHA_DETECTED') {
        console.error(`[dca] aborted by compliance: ${e.code}`);
        break;
      }
      fail++;
    }
    if ((ok + fail) % 100 === 0) console.log(`[dca] ok=${ok} fail=${fail}`);
  }

  await finishJob(jobId, {
    status: 'done',
    records_found: candidates.length,
    records_updated: ok,
    records_skipped: fail,
  });
  console.log(`[dca] done. ok=${ok} fail=${fail}`);
  await pool.end();
}

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