← back to Lawyer Directory Builder

src/ingest/calbar_full_scrape.ts

343 lines

/**
 * CA State Bar — full attorney scrape via sequential Detail/{barNumber} URLs.
 *
 * Discovery: the SPA-shell-only theory in docs/blocked_sources.md was wrong.
 * The Detail/{n} URLs render the full attorney profile inline in the initial
 * HTML (no JS hydration needed). Plain fetch + regex parsing works.
 *
 * Scope: bar numbers 29960 → 360000 (per Steve, ~330k attorneys statewide).
 * Polite cadence ~4 RPS aggregate (concurrency 5, ~200ms per worker).
 *
 * Persistence:
 *   - professionals  (one row per bar_number, UNIQUE constraint dedups)
 *   - organizations  (firm extracted from address line, type='law_firm')
 *   - professional_locations (links pro → org by bar_number + address)
 *   - phones         (phone_type='office', source_url=detail page)
 *   - scrape_jobs.checkpoint = { last_id: <int> } for resume
 *
 * NOT extracted: email (page deliberately scatters 20 decoy emails behind
 * inline `<span id="eN">` elements — disambiguation needs JS rendering, which
 * would 30× the runtime). Skipped on purpose.
 *
 * CLI:
 *   npx tsx src/ingest/calbar_full_scrape.ts                  # full pass
 *   npx tsx src/ingest/calbar_full_scrape.ts --start 29960 --end 30100  # range
 *   npx tsx src/ingest/calbar_full_scrape.ts --resume         # use checkpoint
 *   npx tsx src/ingest/calbar_full_scrape.ts --smoke          # 10 attorneys
 */
import 'dotenv/config';
import { pool, query, withTx } from '../db/pool.ts';

const SOURCE_NAME = 'CA State Bar — License Search (Detail page scrape)';
const BASE_URL = 'https://apps.calbar.ca.gov/attorney/Licensee/Detail';
const USER_AGENT = process.env.USER_AGENT
  || 'LawyerDirectoryBuilder/0.1 (research; contact: steveabramsdesigns@gmail.com)';
const CONCURRENCY = Number(process.env.CALBAR_CONCURRENCY || 5);
const PER_REQ_DELAY_MS = Number(process.env.CALBAR_DELAY_MS || 250);

const argv = process.argv.slice(2);
const SMOKE = argv.includes('--smoke');
const RESUME = argv.includes('--resume');
const START_IDX = argv.indexOf('--start');
const END_IDX = argv.indexOf('--end');
const START = START_IDX >= 0 ? Number(argv[START_IDX + 1]) : 29960;
const END = END_IDX >= 0 ? Number(argv[END_IDX + 1]) : (SMOKE ? START + 9 : 360000);

type Parsed = {
  bar_number: string;
  full_name: string | null;
  first_name: string | null;
  last_name: string | null;
  middle_name: string | null;
  license_status: string | null;       // Active, Inactive, Disbarred, Resigned, etc.
  firm_name: string | null;             // First chunk of address up to first comma
  address: string | null;               // Address minus firm prefix
  city: string | null;
  state: string | null;
  zip: string | null;
  phone: string | null;
};

function decodeHtml(s: string): string {
  return s.replace(/&amp;/g, '&').replace(/&nbsp;/g, ' ').replace(/&#46;/g, '.').replace(/&#64;/g, '@')
          .replace(/&quot;/g, '"').replace(/&#39;/g, "'").replace(/<[^>]+>/g, '').replace(/\s+/g, ' ').trim();
}

function parseDetail(html: string, barNumber: number): Parsed | null {
  // Sanity-check the page actually has an attorney record (vs 404 / "no record found")
  if (!/Attorney Profile|License Status/i.test(html)) return null;
  if (/no record|not found|404/i.test(html.slice(0, 4000))) return null;

  // Name + bar number  e.g.  "Sandra J.  Shapiro" #29960
  // The header pattern: <h2>Sandra J. Shapiro</h2> followed by #29960
  // Loose regex — capture the visible name above "#<barNumber>"
  const nameRx = new RegExp(`([A-Z][A-Za-z'.,\\-\\s]{1,80}?)\\s*#\\s*${barNumber}\\b`);
  const nameM = html.match(nameRx);
  let full_name: string | null = nameM ? decodeHtml(nameM[1]).replace(/\s+/g, ' ').trim() : null;
  if (full_name && (full_name.length > 80 || /[<>]/.test(full_name))) full_name = null;

  // Status  e.g.  "License Status: <span ...>&nbsp;Active&nbsp;</span>"
  // Active attorneys' badges sometimes have empty inner text — assume Active if
  // the page exists and no other status was captured.
  let license_status: string | null = null;
  const statusM = html.match(/License Status:[\s\S]{0,400}?<span[^>]*>([^<]+)<\/span>/i);
  if (statusM) license_status = decodeHtml(statusM[1]).trim() || null;
  if (!license_status) license_status = 'Active';

  // Address line. Sample: "Address: Law Office of Sandra J Shapiro, 1700 Toledo Ave, Burlingame, CA 94010-5845"
  let firm_name: string | null = null;
  let address: string | null = null;
  let city: string | null = null;
  let state: string | null = null;
  let zip: string | null = null;
  const addrM = html.match(/Address:\s*([^<\n]+?)(?=<|\n|Phone:|Fax:|Email:)/i);
  if (addrM) {
    const full = decodeHtml(addrM[1]).trim();
    // Strip trailing "Public Email Address" or junk
    const cleaned = full.replace(/\s*Public Email Address\s*$/i, '').replace(/\s+/g, ' ').trim();
    const parts = cleaned.split(',').map(p => p.trim()).filter(Boolean);
    if (parts.length >= 3) {
      // Last part is "CA 94010-5845" or "CA  94010"
      const lastM = parts[parts.length - 1].match(/^([A-Z]{2})\s+(\d{5}(?:-\d{4})?)/);
      if (lastM) { state = lastM[1]; zip = lastM[2].slice(0, 5); }
      city = parts[parts.length - 2] || null;
      // Identify the firm-name prefix:
      // - if first part doesn't contain a number, treat it as firm name, rest as address
      // - if first part starts with a digit, probably no firm name (firm is solo, just street)
      if (parts.length >= 4 && !/^\d/.test(parts[0])) {
        firm_name = parts[0];
        address = parts.slice(1, parts.length - 2).join(', ');
      } else {
        address = parts.slice(0, parts.length - 2).join(', ');
      }
    } else {
      address = cleaned;
    }
  }

  // Phone — page is CRLF; capture digits up to whitespace
  let phone: string | null = null;
  const phoneM = html.match(/Phone:\s*([0-9()\-.\s]{10,25})/i);
  if (phoneM) {
    const digits = phoneM[1].replace(/\D/g, '');
    if (digits.length === 10) phone = `(${digits.slice(0,3)}) ${digits.slice(3,6)}-${digits.slice(6)}`;
    else if (digits.length === 11 && digits.startsWith('1')) {
      const d = digits.slice(1);
      phone = `(${d.slice(0,3)}) ${d.slice(3,6)}-${d.slice(6)}`;
    }
  }

  // First/last/middle from full_name
  let first_name: string | null = null, last_name: string | null = null, middle_name: string | null = null;
  if (full_name) {
    const tokens = full_name.replace(/\./g, '').split(/\s+/).filter(Boolean);
    if (tokens.length >= 2) {
      first_name = tokens[0];
      last_name = tokens[tokens.length - 1];
      if (tokens.length > 2) middle_name = tokens.slice(1, -1).join(' ');
    } else if (tokens.length === 1) {
      last_name = tokens[0];
    }
  }

  return { bar_number: String(barNumber), full_name, first_name, last_name, middle_name, license_status, firm_name, address, city, state, zip, phone };
}

async function fetchOne(barNumber: number): Promise<Parsed | null> {
  const { fetch } = await import('undici');
  const r = await fetch(`${BASE_URL}/${barNumber}`, {
    headers: { 'User-Agent': USER_AGENT, Accept: 'text/html' },
    signal: AbortSignal.timeout(15000),
  });
  if (r.status === 404) return null;
  if (!r.ok) {
    if (r.status === 429) throw new Error('CALBAR_429');
    return null;
  }
  const html = await r.text();
  return parseDetail(html, barNumber);
}

async function ensureSource(): Promise<number> {
  await query(`
    INSERT INTO sources (source_name, source_type, base_url, terms_notes, allowed_method, rate_limit_rps)
    VALUES ($1, 'official_registry', $2,
      'Public records by statute. Polite-use ~4 rps. Detail/{n} renders inline HTML.',
      'crawl', 4.0)
    ON CONFLICT (source_name) DO NOTHING
  `, [SOURCE_NAME, BASE_URL]);
  const r = await query<{ id: number }>(`SELECT id FROM sources WHERE source_name = $1`, [SOURCE_NAME]);
  return r.rows[0].id;
}

async function startJob(sourceId: number, label: string, checkpoint: any): Promise<number> {
  const r = await query<{ id: number }>(`
    INSERT INTO scrape_jobs (source_id, job_label, status, started_at, checkpoint)
    VALUES ($1, $2, 'running', NOW(), $3::jsonb) RETURNING id
  `, [sourceId, label, JSON.stringify(checkpoint)]);
  return r.rows[0].id;
}

async function updateCheckpoint(jobId: number, lastId: number, kept: number, missed: number) {
  await query(`UPDATE scrape_jobs SET checkpoint = $2::jsonb, records_inserted = $3, records_skipped = $4 WHERE id = $1`,
    [jobId, JSON.stringify({ last_id: lastId, kept, missed }), kept, missed]);
}

async function finishJob(jobId: number, fields: Record<string, unknown>) {
  const sets: string[] = []; const params: unknown[] = []; 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 loadResumeStart(): Promise<number | null> {
  // Take the highest last_id we've checkpointed across ANY prior run of this
  // source (running, failed, aborted, or completed but partial). Last-write
  // wins.
  const r = await query<{ checkpoint: any }>(`
    SELECT checkpoint FROM scrape_jobs
    WHERE source_id = (SELECT id FROM sources WHERE source_name = $1)
      AND checkpoint IS NOT NULL
    ORDER BY (checkpoint->>'last_id')::int DESC NULLS LAST
    LIMIT 1
  `, [SOURCE_NAME]);
  const last = r.rows[0]?.checkpoint?.last_id;
  return typeof last === 'number' ? last + 1 : null;
}

async function persist(p: Parsed, sourceUrl: string) {
  await withTx(async (client) => {
    // Upsert professional by bar_number
    const proR = await client.query<{ id: number }>(`
      INSERT INTO professionals (full_name, first_name, last_name, middle_name, bar_number, license_status, source_confidence_score, confidence_tier)
      VALUES ($1, $2, $3, $4, $5, $6, 0.95, 'high')
      ON CONFLICT (bar_number) DO UPDATE SET
        full_name = COALESCE(EXCLUDED.full_name, professionals.full_name),
        first_name = COALESCE(EXCLUDED.first_name, professionals.first_name),
        last_name = COALESCE(EXCLUDED.last_name, professionals.last_name),
        middle_name = COALESCE(EXCLUDED.middle_name, professionals.middle_name),
        license_status = COALESCE(EXCLUDED.license_status, professionals.license_status),
        updated_at = NOW()
      RETURNING id
    `, [p.full_name, p.first_name, p.last_name, p.middle_name, p.bar_number, p.license_status]);
    const proId = proR.rows[0].id;

    // Phone
    if (p.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
      `, [proId, p.phone, sourceUrl]);
    }

    // Firm — only create org if firm_name is reasonable (not solo + plain address)
    let orgId: number | null = null;
    if (p.firm_name && p.firm_name.length >= 4 && !/^\d/.test(p.firm_name)) {
      const addrLine = p.address || null;
      // Look up or insert org by name + city
      const found = await client.query<{ id: number }>(
        `SELECT id FROM organizations WHERE type='law_firm' AND LOWER(name) = LOWER($1) AND LOWER(COALESCE(city,'')) = LOWER(COALESCE($2,'')) LIMIT 1`,
        [p.firm_name, p.city || '']);
      if (found.rowCount && found.rowCount > 0) {
        orgId = found.rows[0].id;
      } else {
        const ins = await client.query<{ id: number }>(`
          INSERT INTO organizations (name, type, address, city, state, zip, source_url)
          VALUES ($1, 'law_firm', $2, $3, $4, $5, $6) RETURNING id
        `, [p.firm_name, addrLine, p.city, p.state, p.zip, sourceUrl]);
        orgId = ins.rows[0].id;
      }
    }

    // Location link
    if (orgId) {
      const fullAddr = [p.address, p.city, p.state, p.zip].filter(Boolean).join(', ');
      await client.query(`
        INSERT INTO professional_locations (professional_id, organization_id, address, phone, source_url, is_primary)
        VALUES ($1, $2, $3, $4, $5, true)
        ON CONFLICT (professional_id, organization_id) WHERE organization_id IS NOT NULL DO UPDATE SET
          address    = COALESCE(EXCLUDED.address,    professional_locations.address),
          phone      = COALESCE(EXCLUDED.phone,      professional_locations.phone),
          source_url = COALESCE(EXCLUDED.source_url, professional_locations.source_url),
          last_verified_at = NOW()
      `, [proId, orgId, fullAddr || null, p.phone, sourceUrl]);
    }
  });
}

async function main() {
  const sourceId = await ensureSource();
  let start = START;
  if (RESUME) {
    const resume = await loadResumeStart();
    if (resume && resume > start) start = resume;
  }
  const total = END - start + 1;
  console.log(`[calbar] scraping bar #${start} → #${END} (${total.toLocaleString()} attorneys, conc=${CONCURRENCY})…`);
  const jobId = await startJob(sourceId, `calbar:detail:${start}-${END}`, { last_id: start - 1, kept: 0, missed: 0 });

  let cursor = start;
  let kept = 0, missed = 0, errs = 0, last_id = start - 1;
  const lastIds = new Set<number>();

  async function processOne(barNumber: number) {
    const sourceUrl = `${BASE_URL}/${barNumber}`;
    try {
      const p = await fetchOne(barNumber);
      if (!p || !p.full_name) { missed++; return; }
      await persist(p, sourceUrl);
      kept++;
    } catch (e) {
      const msg = (e as Error).message;
      errs++;
      if (msg === 'CALBAR_429') {
        console.error('[calbar] hit 429 — backing off 60s');
        await new Promise(r => setTimeout(r, 60000));
      }
    } finally {
      lastIds.add(barNumber);
    }
  }

  const workers = Array.from({ length: CONCURRENCY }, async () => {
    while (true) {
      const id = cursor++;
      if (id > END) break;
      await processOne(id);
      await new Promise(r => setTimeout(r, PER_REQ_DELAY_MS));
      // Periodic progress + checkpoint
      if (lastIds.size >= 200) {
        const max = Math.max(...lastIds);
        lastIds.clear();
        last_id = max;
        await updateCheckpoint(jobId, last_id, kept, missed);
        const done = max - start + 1;
        const pct = ((done / total) * 100).toFixed(1);
        console.log(`[calbar] #${max}  done=${done.toLocaleString()}/${total.toLocaleString()} (${pct}%)  kept=${kept}  missed=${missed}  errs=${errs}`);
      }
    }
  });

  try {
    await Promise.all(workers);
  } finally {
    await finishJob(jobId, {
      status: 'completed',
      records_found: kept + missed,
      records_inserted: kept,
      records_skipped: missed,
      error_message: errs > 0 ? `${errs} per-id errors` : null,
    });
  }

  console.log(`[calbar] done. kept=${kept} missed=${missed} errs=${errs}`);
  await pool.end();
}

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