`src/enrich/firm_email_crawl.ts`

159 lines
/**
 * firm_email_crawl — COLLECT public contact emails for commercial firms.
 *
 * We already have ~1,190 commercial firm websites (from the Places resolve pass).
 * This crawls each firm's OWN site (homepage + a few likely contact pages),
 * extracts the public contact email(s) they publish (info@, leasing@, sales@…),
 * and stores them in firm_contacts(kind='email') for DISPLAY on the directory
 * card. It is a collection-only pass: nothing is ever emailed. $0 (local fetch).
 *
 * Idempotent: firm_contacts has a UNIQUE (firm_id, kind, value) index, so
 * re-runs INSERT ... ON CONFLICT DO NOTHING and never duplicate.
 *
 * Usage:
 *   npx tsx src/enrich/firm_email_crawl.ts            # dry-run (no writes)
 *   npx tsx src/enrich/firm_email_crawl.ts --live     # write to firm_contacts
 *   ... --limit=200   --asset=commercial (default)    --concurrency=8
 */
import 'dotenv/config';
import { pool, query } from '../../db/pool.ts';

// Node's fetch (undici) can negotiate HTTP/2; a remote GOAWAY surfaces as a stray
// 'error' event on the Http2Session — OUTSIDE the promise chain, so a per-firm
// try/catch can't see it and Node escalates it to a process-killing uncaughtException.
// A background crawler must survive these; log and continue (only HTTP fetches +
// idempotent inserts are in flight, so there's no corruptible state to protect).
process.on('uncaughtException', (e) => console.log(`  [uncaught] ${(e as Error).message}`));
process.on('unhandledRejection', (e) => console.log(`  [unhandledRejection] ${String(e)}`));

const LIVE = process.argv.includes('--live');
const LIMIT = Number(process.argv.find(a => a.startsWith('--limit='))?.split('=')[1] || 0) || 0;
const CONCURRENCY = Number(process.argv.find(a => a.startsWith('--concurrency='))?.split('=')[1] || 8);
const ASSET_RAW = (process.argv.find(a => a.startsWith('--asset='))?.split('=')[1] || 'commercial').toLowerCase();
const ASSET = ['commercial', 'residential'].includes(ASSET_RAW) ? ASSET_RAW : 'commercial';

const UA = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36';
const TIMEOUT_MS = 10_000;
const MAX_EMAILS_PER_FIRM = 3;
const CONTACT_PATHS = ['/contact', '/contact-us', '/contactus', '/about', '/about-us'];

const EMAIL_RE = /[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,}/gi;

// Junk the regex will otherwise happily grab off a page.
const BAD_DOMAINS = /(example|yourdomain|domain\.com|email\.com|sentry|wixpress|squarespace|godaddy|cloudflare|w3\.org|schema\.org|sentry\.io|googleapis|gstatic)/i;
const ASSET_EXT = /\.(png|jpe?g|gif|svg|webp|ico|css|js|mp4|pdf|woff2?|ttf)$/i;
const NOISE_LOCAL = /^(no-?reply|do-?not-?reply|donotreply|postmaster|abuse|mailer-daemon)/i;
const ROLE_PREFIX = /^(info|contact|sales|hello|admin|leasing|office|mail|inquir(?:y|ies)|reception|team|hi|support)@/i;

interface FirmRow { id: number; name: string; website: string }

function registrable(host: string): string {
  const parts = host.replace(/^www\./, '').toLowerCase().split('.');
  return parts.slice(-2).join('.');
}

function fetchHtml(url: string): Promise<string> {
  return fetch(url, {
    headers: { 'User-Agent': UA, 'Accept': 'text/html,application/xhtml+xml', 'Accept-Language': 'en-US,en;q=0.9' },
    redirect: 'follow',
    signal: AbortSignal.timeout(TIMEOUT_MS),
  }).then(r => (r.ok ? r.text() : Promise.reject(new Error(`http ${r.status}`))));
}

// Pull emails from a page: mailto: hrefs first (highest quality), then raw text.
function extractEmails(html: string): string[] {
  const found = new Set<string>();
  for (const m of html.matchAll(/mailto:([^"'?>\s]+)/gi)) {
    try { found.add(decodeURIComponent(m[1])); } catch { found.add(m[1]); }  // malformed %-escape → keep raw
  }
  for (const m of html.matchAll(EMAIL_RE)) found.add(m[0]);
  return [...found]
    .map(e => e.trim().toLowerCase())
    .filter(e => e.includes('@') && !ASSET_EXT.test(e) && !BAD_DOMAINS.test(e) && !NOISE_LOCAL.test(e))
    .filter(e => { const d = e.split('@')[1] || ''; return d.includes('.') && d.split('.').pop()!.length >= 2 && d.length <= 60; });
}

// Rank: same-domain-as-website beats off-domain; role address beats a person's.
function rankEmails(emails: string[], siteDomain: string): string[] {
  return [...new Set(emails)]
    .map(e => {
      let score = 0;
      if (registrable(e.split('@')[1]) === siteDomain) score += 10;
      if (ROLE_PREFIX.test(e)) score += 5;
      return { e, score };
    })
    .sort((a, b) => b.score - a.score)
    .map(x => x.e)
    .slice(0, MAX_EMAILS_PER_FIRM);
}

async function crawlFirm(f: FirmRow): Promise<{ id: number; emails: string[]; via: string }> {
  let base: URL;
  try { base = new URL(f.website.startsWith('http') ? f.website : 'https://' + f.website); }
  catch { return { id: f.id, emails: [], via: 'bad-url' }; }
  const siteDomain = registrable(base.host);

  // Homepage first; only hit contact pages if it yields nothing.
  const pages = [base.href, ...CONTACT_PATHS.map(p => new URL(p, base.origin).href)];
  let via = '';
  const all: string[] = [];
  for (const url of pages) {
    let html: string;
    try { html = await fetchHtml(url); } catch { continue; }
    const hits = extractEmails(html);
    if (hits.length) { all.push(...hits); via = url; if (all.length) break; }
  }
  return { id: f.id, emails: rankEmails(all, siteDomain), via };
}

async function main() {
  // Website comes from EITHER the Places resolve (firm.website, commercial) OR the
  // free SERP discovery (firm_site.url, residential) — read both so this crawl chains
  // off whichever pass populated the site.
  const { rows } = await query<FirmRow>(
    `SELECT f.id, f.name, COALESCE(NULLIF(f.website,''), fs.url) AS website
       FROM firm f
       LEFT JOIN firm_site fs ON fs.firm_id = f.id AND NULLIF(fs.url,'') IS NOT NULL
      WHERE f.asset_class = $1
        AND COALESCE(NULLIF(f.website,''), fs.url) IS NOT NULL
        AND f.id NOT IN (SELECT firm_id FROM firm_contacts WHERE kind = 'email')
      ORDER BY f.agent_count DESC NULLS LAST, f.id ${LIMIT ? 'LIMIT ' + LIMIT : ''}`, [ASSET]);

  console.log(`[email-crawl] ${ASSET} · ${rows.length} firms with a site & no email yet · concurrency ${CONCURRENCY} · ${LIVE ? 'LIVE (writing)' : 'DRY-RUN'}`);

  let done = 0, withEmail = 0, emailsWritten = 0;
  const queue = [...rows];
  async function worker() {
    for (;;) {
      const f = queue.shift();
      if (!f) return;
      let r;
      try { r = await crawlFirm(f); }              // one bad firm must never kill the batch
      catch (e) { done++; console.log(`  [${done}/${rows.length}] ${f.name.slice(0,32)} → ERR ${(e as Error).message}`); continue; }
      done++;
      if (r.emails.length) {
        withEmail++;
        if (LIVE) {
          for (const e of r.emails) {
            const res = await query(
              `INSERT INTO firm_contacts (firm_id, kind, value, source_url)
               VALUES ($1, 'email', $2, $3)
               ON CONFLICT (firm_id, kind, value) DO NOTHING`, [f.id, e, r.via]);
            emailsWritten += res.rowCount || 0;
          }
        }
        if (done % 25 === 0 || r.emails.length)
          console.log(`  [${done}/${rows.length}] ${f.name.slice(0, 32).padEnd(32)} → ${r.emails.join(', ')}`);
      } else if (done % 100 === 0) {
        console.log(`  [${done}/${rows.length}] …`);
      }
    }
  }
  await Promise.all(Array.from({ length: CONCURRENCY }, worker));

  console.log(`\n[email-crawl] done. crawled ${done} · firms with ≥1 email ${withEmail} (${(withEmail / (done || 1) * 100).toFixed(1)}%) · rows written ${emailsWritten} · cost $0 (local fetch)`);
  await pool.end();
}

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