← back to Nationalrealestate

src/enrich/broker_website_discovery.ts

305 lines

/**
 * Agent (broker) OWN-website discovery — free HTML search + LOCAL identity gate. $0.
 *
 * Sibling of firm_website_discovery.ts, but an agent search has a problem a firm
 * search doesn't: NAMESAKES. Two "John Smith" agents at different firms collide,
 * and a raw SERP will happily hand back the wrong one (or a directory/aggregator
 * profile). So after the free SERP scrape we run a LOCAL Ollama (qwen3:14b, $0)
 * identity gate: given the agent's name + firm + city/state and the candidate
 * results, the model picks the ONE result that is THIS agent's own professional
 * site — or none. Only a verified hit is written to broker.website.
 *
 * Priority queue: never-attempted agents, ACTIVE + firm-linked first (the ones
 * that actually get displayed), long tail after. A literal full sweep of ~2M
 * agents is a months-long grind, so this is meant to run as a continuous
 * background loop that covers the visible set fast and grinds the tail forever.
 *
 * Run:  npm run discover:brokers                         # default 100
 *       tsx src/enrich/broker_website_discovery.ts -- --limit=500
 *       OLLAMA_URL=http://192.168.1.133:11434 npm run discover:brokers   # Mac1 Ollama
 *
 * Polite: 2.5-4s jittered gap per query, 10s timeout, real-browser UA. Both SERP
 * engines bot-gate (Brave 429/challenge, DDG HTTP-202 anomaly page) — those are
 * NOT recorded as legit misses; escalating backoff, abort after 3 consecutive
 * throttled agents so we don't hammer a wall.
 */
import 'dotenv/config';
import { pool, query } from '../../db/pool.ts';

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 OLLAMA = (process.env.OLLAMA_URL || 'http://127.0.0.1:11434') + '/api/generate';
const MODEL = process.env.OLLAMA_MODEL || 'qwen3:14b';
const MIN_CONFIDENCE = 0.6; // below this, treat as "not this agent" → no_url

// Same block-list intent as the firm sweep: portals, MLS boards, socials,
// review/data-broker directories are NOT an agent's own site.
const BLOCK = new Set([
  'zillow.com', 'realtor.com', 'redfin.com', 'homes.com', 'trulia.com',
  'apartments.com', 'homesnap.com', 'movoto.com', 'estately.com',
  'homelight.com', 'realtytrac.com', 'auction.com', 'point2homes.com',
  'coldwellbankerhomes.com', 'har.com', 'onekeymls.com', 'streeteasy.com',
  'homefinder.com', 'landwatch.com', 'land.com', 'loopnet.com', 'crexi.com',
  'pbbor.com', 'cpar.us', 'miamire.com', 'onehome.com', 'realtyna.com',
  'facebook.com', 'instagram.com', 'linkedin.com', 'twitter.com', 'x.com',
  'youtube.com', 'tiktok.com', 'pinterest.com', 'reddit.com',
  'yelp.com', 'bbb.org', 'yellowpages.com', 'whitepages.com', 'manta.com',
  'bizapedia.com', 'opencorporates.com', 'dnb.com', 'zoominfo.com',
  'crunchbase.com', 'glassdoor.com', 'indeed.com', 'mapquest.com',
  'wikipedia.org', 'wikidata.org', 'google.com', 'duckduckgo.com', 'brave.com',
  'ratemyagent.com', 'fastexpert.com', 'realtyrates.com', 'homes.com',
]);

function host(u: string): string | null {
  try { return new URL(u).hostname.replace(/^www\./, '').toLowerCase(); } catch { return null; }
}
function blocked(h: string): boolean {
  for (const root of BLOCK) if (h === root || h.endsWith('.' + root)) return true;
  if (h.endsWith('.gov')) return true;
  if (h.includes('mls')) return true;
  // Realtor boards / associations are directories, never an agent's own site
  // (e.g. greenwichrealtors.com, *associationofrealtors, *boardofrealtors).
  if (/realtor/.test(h) && /(board|assoc|association)/.test(h)) return true;
  if (/(^|\.)realtors\.(com|org)$/.test(h)) return true;
  return false;
}

class ThrottleError extends Error {}

async function fetchHtml(url: string): Promise<string> {
  const res = await 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),
  });
  if (res.status === 429 || res.status === 403 || res.status === 202) throw new ThrottleError(`http ${res.status}`);
  if (!res.ok) throw new Error(`http ${res.status}`);
  return res.text();
}

interface Hit { url: string; title: string }

// Brave SERP: organic hits flagged data-type="web"; take the first href + any title text.
async function braveSearch(q: string): Promise<Hit[]> {
  const html = await fetchHtml('https://search.brave.com/search?q=' + encodeURIComponent(q));
  if (/verifying you are human|challenge-platform/i.test(html)) throw new ThrottleError('brave challenge');
  const out: Hit[] = [];
  for (const blk of html.split('data-type="web"').slice(1)) {
    const hm = blk.match(/href="(https?:\/\/[^"]+)"/);
    if (!hm) continue;
    const tm = blk.match(/<span[^>]*class="[^"]*title[^"]*"[^>]*>([^<]+)</i) || blk.match(/>([^<]{8,120})<\/a>/);
    out.push({ url: hm[1].replace(/&amp;/g, '&'), title: (tm ? tm[1] : '').trim() });
  }
  return out.slice(0, 10);
}

// DDG html endpoint: <a class="result__a" href="/l/?uddg=…">title</a> + result__snippet.
async function ddgSearch(q: string): Promise<Hit[]> {
  const html = await fetchHtml('https://html.duckduckgo.com/html/?q=' + encodeURIComponent(q));
  if (/anomaly-modal|anomaly\.js|challenge-form/i.test(html)) throw new ThrottleError('ddg anomaly page');
  const out: Hit[] = [];
  const re = /<a[^>]+class="[^"]*result__a[^"]*"[^>]*href="([^"]+)"[^>]*>([\s\S]*?)<\/a>/g;
  let m: RegExpExecArray | null;
  while ((m = re.exec(html)) !== null) {
    let abs = m[1].replace(/&amp;/g, '&');
    try {
      const u = new URL(abs, 'https://duckduckgo.com');
      const uddg = u.searchParams.get('uddg');
      if (uddg) abs = decodeURIComponent(uddg);
    } catch {}
    if (/^https?:\/\//i.test(abs)) out.push({ url: abs, title: m[2].replace(/<[^>]+>/g, '').trim() });
  }
  return out.slice(0, 10);
}

async function searchWeb(q: string): Promise<{ hits: Hit[]; engine: string }> {
  try {
    return { hits: await braveSearch(q), engine: 'brave' };
  } catch (e) {
    if (!(e instanceof ThrottleError)) throw e;
  }
  return { hits: await ddgSearch(q), engine: 'ddg' };
}

// LOCAL identity gate — pick the ONE candidate that is THIS agent's own site, or none.
async function verifyAgentSite(
  agent: { name: string; firm: string | null; city: string | null; license_state: string },
  candidates: Hit[],
): Promise<{ index: number; confidence: number }> {
  if (!candidates.length) return { index: -1, confidence: 0 };
  const list = candidates.map((c, i) => `${i}. ${host(c.url)} — "${(c.title || '').slice(0, 90)}"`).join('\n');
  const prompt = `You are verifying real-estate agent websites. The agent is:
  name: ${agent.name}
  firm: ${agent.firm || '(unknown)'}
  city/state: ${agent.city || '?'}, ${agent.license_state}

Here are candidate search results (index. domain — title):
${list}

Pick the single index that is THIS SPECIFIC agent's OWN professional website — their personal agent site OR their bio/profile page on their firm's own site. It must plausibly be THIS person (name matches), NOT: a different person with a similar name, a listing portal/aggregator, a REALTOR association/board/MLS directory, or a generic real-estate directory. Most agents have NO personal site — answering -1 is the common, correct answer. Only pick an index when you are genuinely confident it is this exact agent's own page.
Respond with ONLY compact JSON: {"index": <number>, "confidence": <0.0-1.0>}`;

  const res = await fetch(OLLAMA, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ model: MODEL, prompt, stream: false, options: { temperature: 0 } }),
    signal: AbortSignal.timeout(60_000),
  });
  if (!res.ok) throw new Error(`ollama ${res.status}`);
  const data = await res.json() as { response?: string };
  const m = (data.response || '').match(/\{[\s\S]*?\}/);
  if (!m) return { index: -1, confidence: 0 };
  try {
    const j = JSON.parse(m[0]);
    const idx = Number.isInteger(j.index) ? j.index : -1;
    const conf = typeof j.confidence === 'number' ? j.confidence : 0;
    if (idx < 0 || idx >= candidates.length) return { index: -1, confidence: 0 };
    return { index: idx, confidence: conf };
  } catch { return { index: -1, confidence: 0 }; }
}

interface BrokerRow { id: number; name: string; firm: string | null; city: string | null; license_state: string }

// Discover one agent's site right now (the $0 on-demand path the render calls when
// a specific agent record is opened — 1 query, never hits the batch throttle wall).
// Exported so the server can `import { discoverOne }` for lazy enrichment.
export async function discoverOne(brokerId: number): Promise<{ status: 'found' | 'no_url' | 'throttled' | 'missing'; website: string | null; confidence: number }> {
  const r = await query<BrokerRow>(`
    SELECT b.id, b.name, f.name AS firm, b.city, b.license_state
      FROM broker b LEFT JOIN firm f ON f.id = b.firm_id WHERE b.id = $1`, [brokerId]);
  if (!r.rows.length) return { status: 'missing', website: null, confidence: 0 };
  const b = r.rows[0];
  const q = `"${b.name}" ${b.firm || ''} ${b.city || ''} ${b.license_state} real estate agent`.replace(/\s+/g, ' ').trim();
  let hits: Hit[];
  try {
    ({ hits } = await searchWeb(q));
  } catch (e) {
    // Throttled (both free engines bot-gated) — DON'T record a no_url (it wasn't a
    // real miss, just a cooldown). Leave website_status NULL so it's retried later.
    if (e instanceof ThrottleError) return { status: 'throttled', website: null, confidence: 0 };
    throw e;
  }
  const credible = hits.filter(h => { const hh = host(h.url); return hh && !blocked(hh); }).slice(0, 8);
  const verdict = credible.length ? await verifyAgentSite(b, credible) : { index: -1, confidence: 0 };
  if (verdict.index >= 0 && verdict.confidence >= MIN_CONFIDENCE) {
    const winner = 'https://' + host(credible[verdict.index].url);
    await query(`UPDATE broker SET website=$2, website_status='found', website_confidence=$3, website_discovered_at=NOW() WHERE id=$1`, [b.id, winner, verdict.confidence]);
    return { status: 'found', website: winner, confidence: verdict.confidence };
  }
  await query(`UPDATE broker SET website_status='no_url', website_discovered_at=NOW() WHERE id=$1`, [b.id]);
  return { status: 'no_url', website: null, confidence: 0 };
}

async function main() {
  const argLimit = process.argv.find(a => a.startsWith('--limit='));
  const limit = argLimit ? parseInt(argLimit.split('=')[1], 10) : 100;

  // On-demand single-agent mode: `--broker-id=N` enriches just that agent and exits.
  const argId = process.argv.find(a => a.startsWith('--broker-id='));
  if (argId) {
    const id = parseInt(argId.split('=')[1], 10);
    const res = await discoverOne(id);
    console.log(`[broker-discover] on-demand #${id} → ${res.website || '(no verified site)'}`);
    await pool.end();
    return;
  }

  const run = await query<{ id: number }>(
    `INSERT INTO ingest_runs (source, notes) VALUES ('broker_discovery', $1) RETURNING id`,
    [`SERP + local ${MODEL} identity gate, batch limit ${limit}`],
  );
  const runId = run.rows[0].id;

  // Priority: never-attempted, ACTIVE + firm-linked first (these get displayed),
  // then everyone else. firm name joined for the query + the identity prompt.
  const r = await query<BrokerRow>(`
    SELECT b.id, b.name, f.name AS firm, b.city, b.license_state
      FROM broker b
      LEFT JOIN firm f ON f.id = b.firm_id
     WHERE b.website_status IS NULL
     ORDER BY (b.license_status ILIKE 'active%') DESC NULLS LAST,
              (b.firm_id IS NOT NULL) DESC,
              b.id
     LIMIT $1
  `, [limit]);
  const queue = r.rows;
  console.log(`[broker-discover] run #${runId} · ${queue.length} agents · SERP + ${MODEL} gate · jittered 2.5-4s/query`);

  let ok = 0, miss = 0, err = 0, consecThrottle = 0;
  let fatal: any = null;
  try {
    for (let i = 0; i < queue.length; i++) {
      const b = queue[i];
      const q = `"${b.name}" ${b.firm || ''} ${b.city || ''} ${b.license_state} real estate agent`.replace(/\s+/g, ' ').trim();
      let result: { hits: Hit[]; engine: string } | null = null;
      for (let attempt = 0; attempt < 3 && result === null; attempt++) {
        try {
          result = await searchWeb(q);
          consecThrottle = 0;
        } catch (e: any) {
          if (e instanceof ThrottleError) {
            const wait = [30_000, 90_000, 180_000][attempt];
            console.log(`  [throttle] ${e.message} — backing off ${wait / 1000}s (attempt ${attempt + 1}/3)`);
            await new Promise(res => setTimeout(res, wait));
          } else {
            console.log(`  [err] ${b.name.slice(0, 40)}: ${(e.message || e).slice(0, 80)}`);
            break;
          }
        }
      }

      if (result === null) {
        err++; consecThrottle++;
        console.log(`  [${i + 1}/${queue.length}] ⚠ ${b.name.slice(0, 40)} (search failed, not recorded)`);
        if (consecThrottle >= 3) { console.log('[broker-discover] 3 consecutive throttled — aborting run'); break; }
      } else {
        const credible = result.hits.filter(h => { const hh = host(h.url); return hh && !blocked(hh); }).slice(0, 8);
        let verdict = { index: -1, confidence: 0 };
        try {
          if (credible.length) verdict = await verifyAgentSite(b, credible);
        } catch (e: any) {
          console.log(`  [llm-err] ${b.name.slice(0, 30)}: ${(e.message || e).slice(0, 60)} — recording unverified`);
        }
        if (verdict.index >= 0 && verdict.confidence >= MIN_CONFIDENCE) {
          const winner = 'https://' + host(credible[verdict.index].url);
          ok++;
          await query(
            `UPDATE broker SET website=$2, website_status='found', website_confidence=$3, website_discovered_at=NOW() WHERE id=$1`,
            [b.id, winner, verdict.confidence],
          );
          console.log(`  [${i + 1}/${queue.length}] ✓ ${b.name.padEnd(30).slice(0, 30)} @ ${(b.firm || '').slice(0, 18).padEnd(18)} → ${winner} (${verdict.confidence.toFixed(2)})`);
        } else {
          miss++;
          await query(`UPDATE broker SET website_status='no_url', website_discovered_at=NOW() WHERE id=$1`, [b.id]);
          console.log(`  [${i + 1}/${queue.length}] · ${b.name.slice(0, 30)} (no verified site${credible.length ? `, ${credible.length} cand` : ''})`);
        }
      }
      await new Promise(res => setTimeout(res, 2500 + Math.random() * 1500));
    }
  } catch (e: any) {
    fatal = e;
    console.error(`[broker-discover] loop aborted, recording partial: ${String(e?.message || e).slice(0, 120)}`);
  }

  const status = (ok || miss) ? 'ok' : 'failed';
  await query(
    `UPDATE ingest_runs SET finished_at = NOW(), status = $5, rows_upserted = $2, rows_skipped = $3,
            notes = notes || $4 WHERE id = $1`,
    [runId, ok, miss, ` · ${ok} verified, ${miss} no-site, ${err} search-errors${fatal ? ` · partial(${String(fatal?.message || fatal).slice(0, 40)})` : ''}`, status],
  );
  console.log(`[broker-discover] done · ${ok} verified · ${miss} no-site · ${err} errors${fatal ? ' (partial)' : ''}`);
  await pool.end();
}

// Only run the batch when executed directly (`tsx …/broker_website_discovery.ts`),
// NOT when the server imports { discoverOne } for the on-demand lazy path.
import { pathToFileURL } from 'node:url';
const isEntry = process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href;
if (isEntry) {
  main().catch(async (e) => {
    console.error('[broker-discover]', e);
    try { await pool.end(); } catch {}
    process.exit(1);
  });
}