`src/enrich/firm_website_contacts.ts`

204 lines
/**
 * Firm-website contact-page enricher.
 *
 * For each org with a non-empty website:
 *   1. fetch homepage (robots-respected, rate-limited)
 *   2. find candidate contact pages (/contact, /contact-us, /about-us, /our-team, /people)
 *   3. fetch each
 *   4. regex-extract emails + phones
 *   5. upsert into emails / phones with email_type='office', phone_type='office'
 *   6. update organizations.phone if NULL
 *
 * Stays scoped to firms WITH websites (≈154 today). For firms without a website,
 * no free + ToS-clean enrichment path exists tonight; document and defer.
 */
import 'dotenv/config';
import { load } from 'cheerio';
import { fetchCompliant } from '../lib/compliance.ts';
import { pool, query } from '../db/pool.ts';

const SOURCE_NAME = 'Firm website contact pages';

const CONTACT_PATHS = ['/contact', '/contact-us', '/contact_us', '/contactus', '/about', '/about-us', '/our-team', '/team', '/attorneys', '/people', '/our-attorneys'];

const EMAIL_RX = /[\w.+-]+@[\w-]+(?:\.[\w-]+)+/g;
// Be conservative — accept (xxx) xxx-xxxx, xxx-xxx-xxxx, xxx.xxx.xxxx, +1 xxx-xxx-xxxx
const PHONE_RX = /(?:\+?1[\s.-]?)?\(?(\d{3})\)?[\s.-]?(\d{3})[\s.-]?(\d{4})\b/g;

const BAD_EMAIL_DOMAINS = ['example.com', 'sentry.io', 'godaddy.com', 'wixsite.com', 'wordpress.com', 'gmail.com', 'yahoo.com'];

function normPhone(area: string, prefix: string, line: string) {
  return `(${area}) ${prefix}-${line}`;
}

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, 'directory', '(per-firm websites)',
      'Per-firm website contact pages. Robots.txt respected per host. Polite-use 0.5 rps.',
      'crawl', 0.5)
    ON CONFLICT (source_name) DO NOTHING
  `, [SOURCE_NAME]);
  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) {
  const r = await query<{ id: number }>(`
    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: 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);
}

function extractEmails(html: string): string[] {
  const found = new Set<string>();
  const matches = html.match(EMAIL_RX) || [];
  for (const m of matches) {
    const lower = m.toLowerCase();
    const dom = lower.split('@')[1];
    if (BAD_EMAIL_DOMAINS.some(b => dom.endsWith(b))) continue;
    if (lower.includes('@sentry-next.wixpress.com')) continue;
    if (/\.(png|jpg|gif|svg|webp)$/.test(lower)) continue;
    found.add(lower);
  }
  return [...found];
}

function extractPhones(text: string): string[] {
  const found = new Set<string>();
  let m: RegExpExecArray | null;
  while ((m = PHONE_RX.exec(text)) !== null) {
    const area = m[1]; const prefix = m[2]; const line = m[3];
    if (area === '000' || prefix === '000') continue;
    if (area === '111' || area === '999') continue;
    if (area[0] === '0' || area[0] === '1') continue;     // not a real area code
    found.add(normPhone(area, prefix, line));
  }
  return [...found];
}

async function enrichOne(org: { id: number; name: string; website: string }, sourceId: number): Promise<{ phones: number; emails: number }> {
  const baseUrl = (() => {
    try { return new URL(org.website); } catch { return null; }
  })();
  if (!baseUrl) return { phones: 0, emails: 0 };

  // Build candidate URLs: homepage + N contact pages.
  const candidates = [baseUrl.href, ...CONTACT_PATHS.map(p => new URL(p, baseUrl).href)];
  const seenHtml = new Set<string>();
  const allEmails = new Set<string>();
  const allPhones = new Set<string>();

  for (const url of candidates.slice(0, 3)) {                  // hard cap 3 fetches/firm
    try {
      const res = await fetchCompliant(url, { timeoutMs: 15000 });
      if (!res || res.status >= 400) continue;
      const html = await res.text();
      if (seenHtml.has(html.slice(0, 4000))) continue;
      seenHtml.add(html.slice(0, 4000));

      const $ = load(html);
      $('script, style, noscript').remove();
      const body = $('body').text();

      for (const e of extractEmails(html)) allEmails.add(e);
      for (const p of extractPhones(body)) allPhones.add(p);
    } catch (e) {
      const code = (e as { code?: string }).code;
      if (code === 'ROBOTS_DISALLOWED' || code === 'CAPTCHA_DETECTED') return { phones: 0, emails: 0 };
      // transient error: try next candidate
    }
  }

  let phoneCount = 0;
  let emailCount = 0;

  for (const phone of allPhones) {
    const exists = await query(`SELECT 1 FROM phones WHERE organization_id=$1 AND phone=$2 LIMIT 1`, [org.id, phone]);
    if (exists.rowCount === 0) {
      await query(`INSERT INTO phones (organization_id, phone, phone_type, source_url, last_verified_at)
                   VALUES ($1,$2,'office',$3,NOW())`, [org.id, phone, org.website]);
      phoneCount++;
    }
  }
  if (allPhones.size > 0) {
    // Update primary phone on org if NULL.
    await query(`UPDATE organizations SET phone = COALESCE(phone, $2) WHERE id = $1`, [org.id, [...allPhones][0]]);
  }
  for (const email of allEmails) {
    const exists = await query(`SELECT 1 FROM emails WHERE organization_id=$1 AND LOWER(email)=LOWER($2) LIMIT 1`, [org.id, email]);
    if (exists.rowCount === 0) {
      await query(`INSERT INTO emails (organization_id, email, email_type, source_url, last_verified_at)
                   VALUES ($1,$2,'office',$3,NOW())`, [org.id, email, org.website]);
      emailCount++;
    }
  }

  return { phones: phoneCount, emails: emailCount };
}

async function main() {
  console.log('[enrich] firm-website contact crawl…');
  const sourceId = await ensureSource();
  const jobId = await startJob(sourceId, 'enrich:firm-website-contacts');

  const skipDoneDays = Number(process.env.SKIP_DONE_DAYS ?? '30');
  const limit = Number(process.env.LIMIT ?? '0');
  const r = await query<{ id: number; name: string; website: string }>(`
    SELECT id, name, website FROM organizations
    WHERE type='law_firm' AND website ~ '^https?://'
      AND (contact_crawled_at IS NULL OR contact_crawled_at < NOW() - ($1 || ' days')::interval)
    ORDER BY id
    ${limit > 0 ? `LIMIT ${limit}` : ''}
  `, [String(skipDoneDays)]);
  const firms = r.rows;
  console.log(`[enrich] ${firms.length} firms to crawl (skip-done-days=${skipDoneDays}${limit ? `, limit=${limit}` : ''})`);

  const concurrency = Number(process.env.CONCURRENCY ?? '8');
  let phoneTot = 0, emailTot = 0, processed = 0, errors = 0;
  let cursor = 0;
  async function worker() {
    while (cursor < firms.length) {
      const f = firms[cursor++];
      try {
        const { phones, emails } = await enrichOne(f, sourceId);
        phoneTot += phones; emailTot += emails;
        await query(`UPDATE organizations SET contact_crawled_at = NOW() WHERE id = $1`, [f.id]);
        processed++;
        if (processed % 25 === 0) console.log(`[enrich] ${processed}/${firms.length}: phones+=${phoneTot} emails+=${emailTot}`);
      } catch (e) {
        errors++;
        console.error(`[enrich] err ${f.name}: ${(e as Error).message}`);
      }
    }
  }
  console.log(`[enrich] concurrency=${concurrency}`);
  await Promise.all(Array.from({ length: concurrency }, () => worker()));

  await finishJob(jobId, {
    status: 'completed',
    records_found: firms.length,
    records_inserted: phoneTot + emailTot,
    records_skipped: errors,
  });
  console.log(`[enrich] done. firms=${firms.length} phones+=${phoneTot} emails+=${emailTot} errors=${errors}`);
  await pool.end();
}

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