← 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(/&/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(/&/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);
});
}