`scripts/crawl-broker-comps.js`

122 lines
// crawl-broker-comps.js — $0 NO-AUTH firm/broker book-expansion via Crexi's PUBLIC API.
//
// Steve's ask (2026-08-19): "for every listing find the broker firm and pull listings from there…
// use crexi to find firms and brokers." Discovery (2026-08-19) proved two Crexi endpoints answer a
// plain GET with NO auth, NO browser, NO Browserbase — i.e. $0 and ungated:
//   • GET /assets/<assetId>/brokers            -> broker {id, globalId, publicProfileId, brokerage, numberOfAssets}
//   • GET /brokers/<globalId>/comps?count=<n>  -> that broker's FULL closed-transaction book {totalCount, data[]}
//
// So we can, for EVERY Crexi listing we hold, resolve its broker(s), learn each broker's globalId,
// and pull their entire closed-deal book — folding it into broker_closed_listing (purpose-built,
// currently 0 rows) + backfilling broker.global_id / public_profile_id. This is the CLOSED-DEAL half
// of the expansion; ACTIVE for-sale inventory needs POST /assets/search (bearer) — see crawl-broker-listings.js.
//
// COST $0 (public GET). GATE: read-only external GETs + local `cre` DB writes only (reversible/internal,
// git-tracked). LISTING_LIMIT caps the pilot; run unbounded once verified.
//
// Usage: LISTING_LIMIT=50 node scripts/crawl-broker-comps.js      (0 = all Crexi listings)
'use strict';
const fs = require('fs');
const path = require('path');
const db = require('./db/brokers-db');

const ROOT = path.join(__dirname, '..');
const LIMIT = parseInt(process.env.LISTING_LIMIT || '50', 10);
const CONC = parseInt(process.env.CONC || '6', 10);
const API = 'https://api.crexi.com';
const H = { accept: 'application/json' };
const sleep = (ms) => new Promise(r => setTimeout(r, ms));

async function getJSON(url, tries = 3) {
  for (let i = 0; i < tries; i++) {
    try { const r = await fetch(url, { headers: H });
      if (r.status === 429) { await sleep(1500 * (i + 1)); continue; }
      if (!r.ok) return null; return await r.json();
    } catch { await sleep(500 * (i + 1)); }
  }
  return null;
}
// tiny concurrency pool
async function pool(items, n, fn) {
  const out = []; let idx = 0;
  await Promise.all(Array.from({ length: n }, async () => {
    while (idx < items.length) { const i = idx++; out[i] = await fn(items[i], i); if (i % 50 === 0) process.stderr.write(`  ${i}/${items.length}\r`); }
  }));
  return out;
}

(async () => {
  // idempotent schema adds (reversible)
  await db.pool.query(`ALTER TABLE broker ADD COLUMN IF NOT EXISTS global_id text`);
  await db.pool.query(`ALTER TABLE broker ADD COLUMN IF NOT EXISTS public_profile_id text`);
  await db.pool.query(`ALTER TABLE broker_closed_listing ADD COLUMN IF NOT EXISTS crexi_asset_id text`);
  await db.pool.query(`CREATE UNIQUE INDEX IF NOT EXISTS ux_closed_broker_asset ON broker_closed_listing(broker_id, crexi_asset_id)`);

  const doc = JSON.parse(fs.readFileSync(path.join(ROOT, 'data', 'listings.json'), 'utf8'));
  let crx = doc.listings.filter(l => /^crx\d+/.test(l.id)).map(l => l.id.replace(/^crx/, ''));
  if (LIMIT) crx = crx.slice(0, LIMIT);
  console.log(`Phase 1 — resolving brokers for ${crx.length} Crexi listings via /assets/<id>/brokers ($0)`);

  // Phase 1: listing -> brokers, collect distinct brokers by globalId
  const brokers = new Map(); // globalId -> {globalId,numericId,ppid,name,firm,book}
  await pool(crx, CONC, async (assetId) => {
    const arr = await getJSON(`${API}/assets/${assetId}/brokers`);
    if (!Array.isArray(arr)) return;
    for (const b of arr) {
      if (!b.globalId) continue;
      const name = [b.firstName, b.lastName].filter(Boolean).join(' ').trim();
      if (!brokers.has(b.globalId)) brokers.set(b.globalId, {
        globalId: b.globalId, numericId: String(b.id || ''), ppid: b.publicProfileId || null,
        name, firm: (b.brokerage && b.brokerage.name) || null, book: b.numberOfAssets || 0
      });
    }
  });
  console.log(`\n  distinct brokers resolved: ${brokers.size} (book sum ${[...brokers.values()].reduce((s, b) => s + b.book, 0)})`);

  // Upsert broker identity (name+firm graph) + backfill global_id / public_profile_id
  for (const b of brokers.values()) {
    if (!b.name) continue;
    const bid = await db.upsertBroker({ name: b.name, firm: b.firm, crexi_id: b.numericId || null, total_assets: b.book || null, source: 'crexi' }).catch(() => null);
    if (bid) await db.pool.query(`UPDATE broker SET global_id=COALESCE(global_id,$2), public_profile_id=COALESCE(public_profile_id,$3) WHERE id=$1`, [bid, b.globalId, b.ppid]).catch(() => {});
    b.dbId = bid;
  }

  // Phase 2: per broker -> full closed-comps book (public GET, $0)
  console.log(`Phase 2 — pulling closed-comps books for ${brokers.size} brokers ($0)`);
  let compsInserted = 0, brokersWithComps = 0;
  const list = [...brokers.values()].filter(b => b.dbId);
  await pool(list, CONC, async (b) => {
    const want = Math.min(500, Math.max(3, b.book || 100));
    const j = await getJSON(`${API}/brokers/${b.globalId}/comps?brokerGlobalId=${b.globalId}&count=${want}`);
    const data = j && j.data; if (!Array.isArray(data) || !data.length) return;
    brokersWithComps++;
    for (const c of data) {
      const loc = c.location || {};
      const addr = c.name || loc.fullAddress || loc.address || '';
      const city = loc.city || null;
      const price = c.salePrice || null;
      const sold = c.closedDate ? String(c.closedDate).slice(0, 10) : null;
      const type = (c.assetTypes && c.assetTypes[0]) || null;
      const src = `https://api.crexi.com/brokers/${b.globalId}/comps`;
      const r = await db.pool.query(
        `INSERT INTO broker_closed_listing(broker_id, crexi_asset_id, address, city, sold_price, sold_date, type, source, created_at)
         VALUES($1,$2,$3,$4,$5,$6,$7,$8, now())
         ON CONFLICT(broker_id, crexi_asset_id) DO NOTHING`,
        [b.dbId, String(c.id || c.assetId || ''), addr, city, price, sold, type, src]).catch(() => ({ rowCount: 0 }));
      compsInserted += r.rowCount;
    }
  });

  const totalClosed = (await db.pool.query('select count(*) c from broker_closed_listing')).rows[0].c;
  const report = {
    ran_at: new Date().toISOString(), cost_usd: 0, mode: LIMIT ? `PILOT (${LIMIT} listings)` : 'FULL',
    listings_scanned: crx.length, brokers_resolved: brokers.size, brokers_with_globalid: [...brokers.values()].filter(b => b.globalId).length,
    brokers_with_comps: brokersWithComps, comps_inserted_this_run: compsInserted, broker_closed_listing_total: +totalClosed
  };
  fs.writeFileSync(path.join(ROOT, 'data', 'raw', `broker-comps-${Date.now()}.json`), JSON.stringify(report, null, 2));
  console.log(`\n=== DONE ($0) ===`);
  console.log(report);
  await db.pool.end().catch(() => {});
  process.exit(0);
})().catch(e => { console.error('FATAL', e); process.exit(1); });