`scripts/scrape-emails-llm.js`

302 lines
#!/usr/bin/env node
/**
 * LLM-driven email scraper.
 *
 * For every websited org with no email yet, fetch the homepage + several
 * staff/contact-shaped subpaths, then ask qwen3:14b (running on both
 * Mac Studio 2 [127.0.0.1:11434] and Mac Studio 1 [192.168.1.133:11434])
 * to identify every email on each page and which person it belongs to.
 *
 * Each extracted person email is matched against the professionals already
 * linked to that org via professional_locations. On match → emails row gets
 * professional_id. Otherwise → organization_id.
 *
 *   node scripts/scrape-emails-llm.js
 *   LIMIT=10 node ...                # smoke
 *   ONLY_DOCTORS=1 node ...          # restrict to doctor-type orgs
 */
const cheerio = require('cheerio');
const { fetch } = require('undici');
const { pool, query } = require('../agents/shared/db');
const { fetchCompliant, setHostRateLimit } = require('../agents/shared/compliance');

const LIMIT = process.env.LIMIT ? Number(process.env.LIMIT) : null;
const ONLY_DOCTORS = process.env.ONLY_DOCTORS === '1';

const DOCTOR_TYPES = ['medical_group','clinic','dental_office','optometrist_office','surgery_center','hospital','fqhc','tcm_clinic','acupuncture_clinic','chiropractor_office','tcm_herbalist','rcfe','ccrc','adult_residential','home_care_agency','home_health','hospice','nursing_facility'];

// Pages most likely to host staff emails.
const SUBPATHS = [
  '', '/contact', '/contact-us', '/about', '/about-us',
  '/team', '/our-team', '/meet-our-team', '/staff',
  '/providers', '/our-providers', '/doctors', '/our-doctors',
  '/physicians', '/dentists', '/optometrists',
  '/leadership', '/who-we-are',
];

// MS1 (192.168.1.133) is currently saturated by lawyer-directory-builder
// hammering qwen3:32b. We use MS2 only with gemma3:12b (already loaded,
// fast, free of contention). Override via OLLAMA_HOSTS / MODEL env.
const OLLAMA_HOSTS = (process.env.OLLAMA_HOSTS || 'http://127.0.0.1:11434|ms2')
  .split(',')
  .map(s => { const [url, label] = s.split('|'); return { url, label: label || url }; });
const MODEL = process.env.MODEL || 'gemma3:12b';

let hostIdx = 0;
function nextHost() { const h = OLLAMA_HOSTS[hostIdx % OLLAMA_HOSTS.length]; hostIdx++; return h; }

const EMAIL_RE = /\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b/gi;
const DECOY_RE = /^(noreply|no-reply|donotreply|do-not-reply|example|test|sentry|webmaster|security|abuse)@/i;
const VENDOR_DOMAINS = new Set([
  'sentry.io','sentry-next.wixpress.com','wix.com','squarespace.com','godaddy.com',
  'cloudflare.com','example.com','example.org','tld.com','site.com','domain.com','website.com','sample.com',
]);

function isPlausibleEmail(e) {
  if (!e || typeof e !== 'string' || e.length > 100 || !e.includes('@')) return false;
  if (DECOY_RE.test(e)) return false;
  const dom = e.split('@')[1];
  if (!dom || VENDOR_DOMAINS.has(dom.toLowerCase())) return false;
  if (/\.(png|jpe?g|gif|svg|webp|css|js)$/i.test(e)) return false;
  if (/[a-f0-9]{32,}/i.test(e.split('@')[0])) return false;
  return true;
}

function originOf(websiteRaw) {
  let s = String(websiteRaw || '').trim();
  if (!s) return null;
  if (!/^https?:\/\//i.test(s)) s = 'https://' + s;
  try { return new URL(s); } catch { return null; }
}

function plainText($) {
  $('script, style, noscript, header nav, footer nav').remove();
  return ($('body').text() || $.text() || '')
    .replace(/[ \t]+/g, ' ')
    .replace(/\n{3,}/g, '\n\n')
    .trim();
}

async function fetchPageText(url) {
  try {
    const res = await fetchCompliant(url, { accept: 'text/html' });
    if (!res.ok) { if (process.env.LLM_DEBUG) console.error(`  fetch ${res.status} ${url}`); return null; }
    const ct = res.headers.get('content-type') || '';
    if (!/html|text/i.test(ct)) { if (process.env.LLM_DEBUG) console.error(`  fetch non-html ${url}`); return null; }
    const html = await res.text();
    if (!html || !html.includes('@')) { if (process.env.LLM_DEBUG) console.error(`  fetch no@ ${url}`); return null; }
    const $ = cheerio.load(html);
    // Return the FULL plain text — emails are often in the footer, past where
    // a naive slice would cut. The LLM call truncates separately to fit ctx.
    return plainText($);
  } catch (e) {
    if (process.env.LLM_DEBUG) console.error(`  fetch err ${e.code || ''} ${e.message} ${url}`);
    if (e?.code === 'CAPTCHA_DETECTED') return null;
    return null;
  }
}

async function llmExtract(pageText, orgName, sourceUrl) {
  const host = nextHost();
  const prompt = `Extract every email address from this page of "${orgName}" and (when stated) the person each email belongs to.

Return ONLY valid JSON in this exact shape, no commentary:
{"emails":[{"email":"...","person":"Full Name or null","role":"Title or null","kind":"person|generic"}]}

If none, return: {"emails":[]}.

PAGE TEXT:
"""
${pageText.slice(0, 12000)}
"""`;

  // AbortSignal.timeout occasionally fails to fire under undici when a TCP
  // connection is established but reads stall — use an explicit AbortController
  // with our own setTimeout to guarantee abortion.
  const ac = new AbortController();
  const killer = setTimeout(() => ac.abort(), 45_000);

  try {
    const res = await fetch(host.url + '/api/generate', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        model: MODEL,
        prompt,
        stream: false,
        format: 'json',
        keep_alive: '10m',
        options: { temperature: 0.1, num_ctx: 6144 },
      }),
      signal: ac.signal,
    });
    clearTimeout(killer);
    if (!res.ok) return [];
    const j = await res.json();
    const out = JSON.parse(j.response || '{}');
    const emails = Array.isArray(out.emails) ? out.emails : [];
    return emails.filter(e => e && isPlausibleEmail(e.email)).map(e => ({
      email: String(e.email).toLowerCase().trim(),
      person: e.person && String(e.person).trim() || null,
      role: e.role && String(e.role).trim() || null,
      kind: e.kind === 'person' ? 'person' : 'generic',
      page: sourceUrl,
      via: host.label,
    }));
  } catch (e) {
    clearTimeout(killer);
    return [];
  }
}

function fallbackRegexExtract(pageText, sourceUrl) {
  if (!pageText) return [];
  const out = new Map();
  for (const m of pageText.matchAll(EMAIL_RE)) {
    const e = String(m[0]).toLowerCase().trim();
    if (isPlausibleEmail(e) && !out.has(e)) out.set(e, { email: e, person: null, role: null, kind: 'generic', page: sourceUrl, via: 'regex' });
  }
  return [...out.values()];
}

function tokens(s) {
  return String(s || '').toLowerCase().normalize('NFKD').replace(/[^a-z\s]/g, ' ').split(/\s+/).filter(Boolean);
}

function nameOverlap(personName, fullName) {
  if (!personName || !fullName) return 0;
  const a = new Set(tokens(personName));
  const b = new Set(tokens(fullName));
  let hit = 0;
  for (const t of a) if (b.has(t) && t.length >= 2) hit++;
  return hit;
}

async function matchPersonToProfessional(orgId, personName) {
  if (!personName) return null;
  const pros = (await query(
    `SELECT p.id, p.full_name FROM professional_locations pl
       JOIN professionals p ON p.id = pl.professional_id
      WHERE pl.organization_id = $1 AND p.opted_out = false`,
    [orgId]
  )).rows;
  if (pros.length === 0) return null;
  let best = null, bestScore = 0;
  for (const p of pros) {
    const s = nameOverlap(personName, p.full_name);
    if (s > bestScore) { bestScore = s; best = p; }
  }
  return bestScore >= 2 ? best : null;
}

async function processOrg(o) {
  const origin = originOf(o.website);
  if (!origin) return { pages: 0, found: 0, person_emails: 0, org_emails: 0 };

  setHostRateLimit(origin.hostname, 1.0);

  const seen = new Map();
  let pagesHit = 0;
  let llmCalls = 0;
  for (const sub of SUBPATHS) {
    const url = origin.origin + (sub === '' ? '/' : sub);
    const txt = await fetchPageText(url);
    if (!txt) continue;
    pagesHit++;
    if (process.env.LLM_DEBUG) console.error(`  fetched ${url} (${txt.length}B)`);

    // Always grab regex-found emails (cheap, never wrong).
    const regex = fallbackRegexExtract(txt, url);
    if (process.env.LLM_DEBUG && regex.length > 0) console.error(`  regex found ${regex.length} on ${url}: ${regex.map(e=>e.email).join(',')}`);
    for (const e of regex) {
      if (!seen.has(e.email)) seen.set(e.email, e);
    }

    // Only spend an LLM call when (a) the page has @ AND (b) the page reads
    // like a team/contact page (so we can attribute to a person). Skip the
    // LLM entirely on homepage pages where the regex usually suffices.
    const isStaffShape = /\b(team|staff|provider|doctor|physician|dentist|optometrist|nurse|leadership|about)\b/i.test(sub);
    if (isStaffShape && llmCalls < 3 && txt.includes('@')) {
      const extracted = await llmExtract(txt, o.name, url);
      llmCalls++;
      for (const e of extracted) {
        if (!seen.has(e.email)) seen.set(e.email, e);
        else if (e.kind === 'person' && seen.get(e.email).kind !== 'person') seen.set(e.email, e);
      }
    }
    if (seen.size >= 12 || pagesHit >= 8) break;
  }

  let personEmails = 0, orgEmails = 0;
  for (const e of seen.values()) {
    let proId = null;
    if (e.kind === 'person') {
      const m = await matchPersonToProfessional(o.id, e.person);
      if (m) proId = m.id;
    }
    try {
      if (proId) {
        await query(
          `INSERT INTO emails (professional_id, email, email_type, source_url, last_verified_at, verification_status)
           VALUES ($1, $2, $3, $4, NULL, 'unverified')
           ON CONFLICT DO NOTHING`,
          [proId, e.email, e.role || 'office', e.page]
        );
        personEmails++;
      } else {
        await query(
          `INSERT INTO emails (organization_id, email, email_type, source_url, last_verified_at, verification_status)
           VALUES ($1, $2, $3, $4, NULL, 'unverified')
           ON CONFLICT DO NOTHING`,
          [o.id, e.email, e.kind === 'person' ? 'staff_unmatched' : 'office', e.page]
        );
        orgEmails++;
      }
    } catch (_) { /* dup or fk - skip */ }
  }
  return { pages: pagesHit, found: seen.size, person_emails: personEmails, org_emails: orgEmails };
}

async function main() {
  const where = [
    `o.website IS NOT NULL`, `o.website <> ''`,
    `NOT EXISTS (SELECT 1 FROM emails e WHERE e.organization_id = o.id AND e.source_url LIKE 'http%')`,
  ];
  if (ONLY_DOCTORS) where.push(`o.type = ANY($1)`);
  const params = ONLY_DOCTORS ? [DOCTOR_TYPES] : [];
  const sql = `
    SELECT o.id, o.name, o.website, o.type
      FROM organizations o
     WHERE ${where.join(' AND ')}
     ORDER BY o.id
     ${LIMIT ? `LIMIT ${LIMIT}` : ''}
  `;
  const r = await query(sql, params);
  console.log(`[llm-email] candidates=${r.rowCount} hosts=${OLLAMA_HOSTS.map(h => h.label).join('+')}`);

  let orgs = 0, totalPersonEmails = 0, totalOrgEmails = 0, withAny = 0;
  for (const o of r.rows) {
    orgs++;
    try {
      const stat = await processOrg(o);
      totalPersonEmails += stat.person_emails;
      totalOrgEmails    += stat.org_emails;
      if (stat.person_emails + stat.org_emails > 0) withAny++;
      if (orgs % 5 === 0 || stat.found > 0) {
        console.log(`[llm-email] orgs=${orgs}/${r.rowCount} pages=${stat.pages} found=${stat.found} (person=${stat.person_emails}, org=${stat.org_emails}) | totals: p=${totalPersonEmails} o=${totalOrgEmails} with_any=${withAny}`);
      }
    } catch (e) {
      console.error(`[llm-email] error org=${o.id} (${o.name}): ${e.message}`);
    }
  }

  console.log(`[llm-email] DONE orgs=${orgs} person_emails=${totalPersonEmails} org_emails=${totalOrgEmails} with_any=${withAny}`);
  await pool.end();
}

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