`crawl/crexi-national-brokers.js`

98 lines
'use strict';
// National broker fetch: for each US state, load Crexi's property list, pull asset IDs from
// the DOM, then fetch /assets/<id>/brokers (in-page, cookie-authed). Upsert real brokers with
// their state. Metered Browserbase — HARD budget cap. $0.15/session-min.
const fs = require('fs');
const path = require('path');
const { chromium } = require('playwright-core');
const Browserbase = require('@browserbasehq/sdk').default || require('@browserbasehq/sdk');
const { Pool } = require('pg');
const pool = new Pool({ host: '/tmp', database: 'crunified' });
const envTxt = fs.readFileSync(path.join(process.env.HOME, '.claude/skills/browserbase/.env'), 'utf8');
const API_KEY = (envTxt.match(/BROWSERBASE_API_KEY=(.*)/) || [])[1]?.trim();
const PROJECT = (envTxt.match(/BROWSERBASE_PROJECT_ID=(.*)/) || [])[1]?.trim();

const STATES = ['AL','AK','AZ','AR','CA','CO','CT','DE','FL','GA','HI','ID','IL','IN','IA','KS','KY','LA','ME','MD','MA','MI','MN','MS','MO','MT','NE','NV','NH','NJ','NM','NY','NC','ND','OH','OK','OR','PA','RI','SC','SD','TN','TX','UT','VT','VA','WA','WV','WI','WY'];
const CAP_MIN = Number(process.env.CREXI_CAP_MIN || 90);   // hard budget cap in session-minutes ($0.15/min)
const PER_STATE = Number(process.env.CREXI_PER_STATE || 40); // assets sampled per state

async function upsertBroker(b, state) {
  if (!b.name) return false;
  const { rows } = await pool.query(
    `INSERT INTO brokers (name, firm, total_assets, crexi_id, source, state)
     VALUES ($1,$2,$3,$4,'crexi',$5)
     ON CONFLICT (crexi_id) WHERE crexi_id IS NOT NULL
     DO UPDATE SET state=COALESCE(brokers.state, EXCLUDED.state), firm=COALESCE(EXCLUDED.firm, brokers.firm),
       total_assets=GREATEST(COALESCE(brokers.total_assets,0), COALESCE(EXCLUDED.total_assets,0))
     RETURNING (xmax=0) AS inserted`,
    [b.name, b.firm || null, b.total_assets || null, b.crexi_id || null, state]);
  return rows[0]?.inserted === true;
}

async function doState(page, st) {
  await page.goto(`https://www.crexi.com/properties/${st}`, { waitUntil: 'domcontentloaded', timeout: 40000 });
  await page.waitForTimeout(4500);
  for (let s = 0; s < 3; s++) { await page.mouse.wheel(0, 4000).catch(() => {}); await page.waitForTimeout(1500); }
  const html = await page.content();
  const ids = [...new Set([...html.matchAll(/\/properties\/(\d{5,})/g)].map(m => m[1]))].slice(0, PER_STATE);
  let seen = 0, added = 0;
  for (let i = 0; i < ids.length; i += 25) {
    const batch = ids.slice(i, i + 25);
    const out = await page.evaluate(async (batch) => {
      const r = [];
      for (const id of batch) {
        try { const resp = await fetch(`https://api.crexi.com/assets/${id}/brokers`, { headers: {} });
          if (resp.ok) r.push({ id, brokers: await resp.json() }); } catch {}
      }
      return r;
    }, batch);
    for (const row of out) {
      if (!Array.isArray(row.brokers)) continue;
      for (const b of row.brokers) {
        const name = [b.firstName, b.lastName].filter(Boolean).join(' ').trim(); if (!name) continue;
        seen++;
        if (await upsertBroker({ name, firm: b.brokerage && b.brokerage.name, total_assets: b.numberOfAssets || null, crexi_id: String(b.id || b.globalId || '') || null }, st)) added++;
      }
    }
  }
  return { assets: ids.length, brokers: seen, new: added };
}

(async () => {
  const t0 = Date.now();
  const bb = new Browserbase({ apiKey: API_KEY });
  // Resume: skip states already covered (unless CREXI_FORCE=1)
  const done = process.env.CREXI_FORCE ? new Set()
    : new Set((await pool.query(`SELECT DISTINCT state FROM brokers WHERE source='crexi' AND state IS NOT NULL`)).rows.map(r => r.state));
  const todo = STATES.filter(st => !done.has(st));
  console.log(`${done.size} states already covered; ${todo.length} to do: ${todo.join(',')}`);
  const perState = {}; let totalNew = 0, totalSeen = 0;
  const BATCH_STATES = 8;  // fresh browser session per 8 states — avoids long-session death

  for (let bi = 0; bi < todo.length; bi += BATCH_STATES) {
    if ((Date.now() - t0) / 60000 > CAP_MIN) { console.log(`\n⏹ budget cap ${CAP_MIN}min reached`); break; }
    const batch = todo.slice(bi, bi + BATCH_STATES);
    let browser;
    try {
      const session = await bb.sessions.create({ projectId: PROJECT, browserSettings: { solveCaptchas: true, viewport: { width: 1440, height: 1000 } } });
      browser = await chromium.connectOverCDP(session.connectUrl);
      const ctx = browser.contexts()[0]; const page = ctx.pages()[0] || await ctx.newPage();
      page.setDefaultTimeout(45000);
      for (const st of batch) {
        try { const r = await doState(page, st); perState[st] = r; totalNew += r.new; totalSeen += r.brokers; process.stdout.write(`${st}:${r.new}new  `); }
        catch (e) { perState[st] = { err: String(e).slice(0, 40) }; process.stdout.write(`${st}:ERR  `); }
      }
    } catch (e) { console.log(`\nsession batch [${batch.join(',')}] failed: ${String(e).slice(0, 60)}`); }
    finally { if (browser) await browser.close().catch(() => {}); }
  }

  const mins = (Date.now() - t0) / 60000, cost = mins * 0.15;
  const byState = (await pool.query(`SELECT state, count(*) n FROM brokers WHERE source='crexi' AND state IS NOT NULL GROUP BY state ORDER BY n DESC`)).rows;
  console.log(`\n\n=== DONE · ${mins.toFixed(1)} min · $${cost.toFixed(2)} ===`);
  console.log(`new brokers this run: ${totalNew} (of ${totalSeen} seen)`);
  console.log('states with crexi brokers now:', byState.length);
  console.log('top:', byState.slice(0, 15).map(r => `${r.state}:${r.n}`).join(' '));
  fs.writeFileSync(path.join(__dirname, '..', 'data', 'crexi-national-run.json'), JSON.stringify({ mins, cost, perState }, null, 1));
  await pool.end();
})();