`scripts/enrich-tier1-crexi.js`

120 lines
// enrich-tier1-crexi.js — Job B TIER 1: re-fetch Crexi /assets/<id>/brokers for our listings and
// persist the broker-block fields the original crawl dropped — brokerage office address, brokerage
// website, broker title/phone where present, and each broker's FULL book size (numberOfAssets).
//
// The original fetch-brokers.js only kept name/firm/total_assets; the same endpoint also carries
// brokerage.website + brokerage.location (office address) + per-broker title/phone. This re-parses that
// SAME crawl source (cheapest tier) — ~1-2 Browserbase sessions ($0.04-0.08) for all listings batched
// in-page. CCPA: business-contact fields only; source URL recorded per field. NO send.
//
// Usage: NODE_PATH=$HOME/.claude/skills/browserbase/node_modules node scripts/enrich-tier1-crexi.js
'use strict';
const fs = require('fs');
const path = require('path');
const { chromium } = require('playwright-core');
const Browserbase = require('@browserbasehq/sdk').default;
const db = require('./db/brokers-db');
const env = fs.readFileSync(process.env.HOME + '/.claude/skills/browserbase/.env', 'utf8');
const get = (t, k) => (t.match(new RegExp('^' + k + '=(.*)$', 'm')) || [])[1]?.replace(/['"]/g, '').trim();
const KEY = get(env, 'BROWSERBASE_API_KEY'), PROJECT = get(env, 'BROWSERBASE_PROJECT_ID');
const ROOT = path.join(__dirname, '..');
const SESSION_COST = 0.04;

// Record a per-field provenance row + set the broker column (only when not already set).
async function setField(brokerId, field, value, sourceUrl, tier) {
  if (value == null || value === '') return;
  await db.pool.query(
    `INSERT INTO broker_field_source(broker_id, field, value, source_url, tier)
     VALUES($1,$2,$3,$4,$5) ON CONFLICT(broker_id, field) DO NOTHING`,
    [brokerId, field, String(value).slice(0, 500), sourceUrl, tier]);
  const colMap = { phone: 'phone', email: 'email', website: 'website', linkedin: 'linkedin', office_addr: 'office_addr', title: 'title', specialties: 'specialties' };
  const col = colMap[field]; if (!col) return;
  await db.pool.query(
    `UPDATE broker SET ${col}=COALESCE(${col}, $2), enriched_at=now() WHERE id=$1`, [brokerId, String(value).slice(0, 500)]);
}

(async () => {
  // Listings whose crexi asset ids we can re-query.
  const listings = JSON.parse(fs.readFileSync(path.join(ROOT, 'data', 'listings.json'), 'utf8')).listings
    .filter(l => /^crx\d+/.test(l.id)).map(l => ({ id: l.id, assetId: l.id.replace(/^crx/, '') }));
  const COST_CAP = +(process.env.CC_MAX_COST || 10);

  let browser, sessions = 0;
  const collected = [];   // {assetId, brokers:[...]}
  try {
    const bb = new Browserbase({ apiKey: KEY });
    const session = await bb.sessions.create({ projectId: PROJECT, browserSettings: { solveCaptchas: true, viewport: { width: 1440, height: 1000 } } });
    sessions++;
    process.stderr.write(`[tier1 session ${sessions}] $${(sessions*SESSION_COST).toFixed(2)} running\n`);
    browser = await chromium.connectOverCDP(session.connectUrl);
    const ctx = browser.contexts()[0];
    const page = ctx.pages()[0] || await ctx.newPage();
    page.setDefaultTimeout(45000);

    let auth = null;
    page.on('request', req => { const h = req.headers(); if (!auth && h.authorization && /bearer/i.test(h.authorization) && req.url().includes('api.crexi.com')) auth = h.authorization; });
    await page.goto('https://www.crexi.com/properties/CA/Los-Angeles/multifamily', { waitUntil: 'domcontentloaded' });
    await page.waitForTimeout(6000); await page.mouse.wheel(0, 2500).catch(() => {}); await page.waitForTimeout(2500);

    const ids = listings.map(l => l.assetId);
    const BATCH = 25;
    for (let i = 0; i < ids.length; i += BATCH) {
      const batch = ids.slice(i, i + BATCH);
      const out = await page.evaluate(async ({ batch, auth }) => {
        const r = [];
        for (const id of batch) {
          try {
            const resp = await fetch(`https://api.crexi.com/assets/${id}/brokers`, { headers: auth ? { authorization: auth } : {} });
            if (resp.ok) r.push({ id, brokers: await resp.json() }); else r.push({ id, err: resp.status });
          } catch (e) { r.push({ id, err: String(e).slice(0, 50) }); }
        }
        return r;
      }, { batch, auth });
      collected.push(...out);
      process.stderr.write(`  fetched ${Math.min(i + BATCH, ids.length)}/${ids.length}\r`);
    }
  } catch (e) { process.stderr.write('\nFATAL ' + e.message + '\n'); }
  finally { if (browser) try { await browser.close(); } catch (_) {} }

  // Persist the richer broker-block fields.
  fs.writeFileSync(path.join(ROOT, 'data', 'raw', 'broker-blocks.json'), JSON.stringify(collected, null, 2));
  let filledOffice = 0, filledWebsite = 0, filledTitle = 0, filledPhone = 0, books = 0, brokersTouched = 0;
  const byId = Object.fromEntries(listings.map(l => [l.assetId, l]));

  for (const row of collected) {
    if (!Array.isArray(row.brokers) || !row.brokers.length) continue;
    const L = byId[row.id];
    const srcUrl = `https://api.crexi.com/assets/${row.id}/brokers`;
    for (const b of row.brokers) {
      const name = [b.firstName, b.lastName].filter(Boolean).join(' ').trim(); if (!name) continue;
      const firm = b.brokerage && b.brokerage.name;
      // Ensure broker exists (idempotent — matches the original ingest key name+firm).
      const brokerId = await db.upsertBroker({ name, firm, total_assets: b.numberOfAssets || null, crexi_id: String(b.id || b.globalId || ''), source: 'crexi' });
      brokersTouched++;

      // Brokerage office address + website + logo (business-contact, not consumer PII).
      const loc = b.brokerage && b.brokerage.location;
      if (loc && (loc.address || loc.city)) {
        const office = [loc.address, loc.city, loc.state && (loc.state.code || loc.state.name), loc.zip].filter(Boolean).join(', ');
        await setField(brokerId, 'office_addr', office, srcUrl, 'tier1-listing'); filledOffice++;
      }
      if (b.brokerage && b.brokerage.website) { await setField(brokerId, 'website', b.brokerage.website, srcUrl, 'tier1-listing'); filledWebsite++; }
      // Broker-level title / phone / email when the block exposes them.
      if (b.title || b.jobTitle) { await setField(brokerId, 'title', b.title || b.jobTitle, srcUrl, 'tier1-listing'); filledTitle++; }
      const phone = b.phone || b.phoneNumber || b.officePhone || (b.contact && b.contact.phone);
      if (phone) { await setField(brokerId, 'phone', phone, srcUrl, 'tier1-listing'); filledPhone++; }
      const email = b.email || (b.contact && b.contact.email);
      if (email) await setField(brokerId, 'email', email, srcUrl, 'tier1-listing');
      if (b.numberOfAssets) books++;
    }
  }

  process.stderr.write(`\nTier 1: touched ${brokersTouched} broker rows · office ${filledOffice} · website ${filledWebsite} · title ${filledTitle} · phone ${filledPhone} · book-sizes ${books}\n`);
  const stats = (await db.pool.query(
    `SELECT count(*) FILTER (WHERE phone IS NOT NULL) phone, count(*) FILTER (WHERE email IS NOT NULL) email,
            count(*) FILTER (WHERE website IS NOT NULL) website, count(*) FILTER (WHERE office_addr IS NOT NULL) office,
            count(*) total FROM broker`)).rows[0];
  console.log(JSON.stringify({ tier: 1, sessions, cost: '$' + (sessions * SESSION_COST).toFixed(2), dbState: stats }));
  await db.pool.end();
})();