`scripts/crawl-broker-listings.js`

118 lines
// crawl-broker-listings.js — $0 NO-AUTH broker ACTIVE-inventory expansion via Crexi's PUBLIC API.
//
// Companion to crawl-broker-comps.js (which pulls CLOSED deals). Discovery (2026-08-19) proved the
// broker profile's active-listing feed is ALSO a public, no-auth POST:
//   POST https://api.crexi.com/assets/search
//   {"brokerGlobalIds":["<globalId>"],"count":60,"includeUnpriced":true,"offset":0,
//    "sortDirection":"Descending","sortOrder":"activatedOn"}
//   -> {data:[...assets], totalCount}
//
// So for every broker we resolved (broker.global_id, backfilled by crawl-broker-comps.js), we pull
// their FULL current for-sale book, run it through the shared crexiExtract, and fold NEW listings
// into data/listings.json (broker attribution tagged) + broker_other_listing + the broker↔listing graph.
//
// COST $0 (public POST — no browser, no Browserbase, no bearer). GATE: writes data/listings.json
// (reversible — git-tracked) + local `cre` DB (reversible/internal). The listings.json growth reaches
// the LIVE grid only on the next `analyze.js` + deploy — that DEPLOY stays Steve-gated.
//
// Usage: BROKER_LIMIT=40 node scripts/crawl-broker-listings.js      (0 = all brokers w/ global_id)
'use strict';
const fs = require('fs');
const path = require('path');
const db = require('./db/brokers-db');
const { byKey } = require('./sources/firms');
const crexiExtract = byKey('crexi').extract;

const ROOT = path.join(__dirname, '..');
const LIMIT = parseInt(process.env.BROKER_LIMIT || '40', 10);
const CONC = parseInt(process.env.CONC || '6', 10);
const PAGE = 60, MAX_PAGES = 20;   // up to 1,200 active assets/broker
const API = 'https://api.crexi.com/assets/search';
const norm = (s) => String(s || '').toLowerCase().replace(/[^a-z0-9]+/g, ' ').trim();
const sleep = (ms) => new Promise(r => setTimeout(r, ms));

async function searchPage(gid, offset, tries = 3) {
  const body = JSON.stringify({ brokerGlobalIds: [gid], count: PAGE, includeUnpriced: true, offset, sortDirection: 'Descending', sortOrder: 'activatedOn' });
  for (let i = 0; i < tries; i++) {
    try {
      const r = await fetch(API, { method: 'POST', headers: { 'content-type': 'application/json', accept: 'application/json' }, body });
      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;
}
async function pool(items, n, fn) {
  let idx = 0; await Promise.all(Array.from({ length: n }, async () => {
    while (idx < items.length) { const i = idx++; await fn(items[i], i); if (i % 50 === 0) process.stderr.write(`  ${i}/${items.length}\r`); }
  }));
}

(async () => {
  const rows = (await db.pool.query(
    `SELECT b.id, b.name, b.global_id, f.name AS firm
       FROM broker b LEFT JOIN firm f ON f.id=b.firm_id
      WHERE b.global_id IS NOT NULL
      ORDER BY b.total_assets DESC NULLS LAST` + (LIMIT ? ` LIMIT ${LIMIT}` : ``))).rows;
  console.log(`brokers with global_id: ${rows.length} — pulling active inventory (POST /assets/search, $0)`);

  const listingsFile = path.join(ROOT, 'data', 'listings.json');
  const doc = JSON.parse(fs.readFileSync(listingsFile, 'utf8'));
  const have = new Set(doc.listings.map(l => norm(l.address)));
  const haveId = new Set(doc.listings.map(l => l.id));
  let added = 0, otherRows = 0, brokersWithActive = 0, assetsSeen = 0;
  const today = new Date().toISOString().slice(0, 10);

  await pool(rows, CONC, async (b) => {
    const assets = new Map();
    for (let p = 0; p < MAX_PAGES; p++) {
      const j = await searchPage(b.global_id, p * PAGE);
      const arr = j && (j.data || j.assets || []);
      if (!Array.isArray(arr) || !arr.length) break;
      for (const a of arr) { const id = a.id || a.assetId; if (id) assets.set(id, a); }
      if (assets.size >= (j.totalCount || 0)) break;
    }
    if (!assets.size) return;
    brokersWithActive++;
    const extracted = crexiExtract([...assets.values()]);
    assetsSeen += extracted.length;
    for (const a of extracted) {
      const url = 'https://www.crexi.com/properties/' + a.id + (a.urlSlug ? '/' + a.urlSlug : '');
      const type = /retail/i.test((a.types || []).join()) ? 'Retail' : /mixed/i.test((a.types || []).join()) ? 'Mixed-use' : 'Multifamily';
      const r = await db.pool.query(
        `INSERT INTO broker_other_listing(broker_id, ext_id, title, address, city, state, price, asset_type, url, source, found_at)
         VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,'crexi-active-book', now()) ON CONFLICT DO NOTHING`,
        [b.id, String(a.id), a.address, a.address, a.city, null, a.price || null, type, url]).catch(() => ({ rowCount: 0 }));
      otherRows += r.rowCount;
      await db.upsertListing({ id: 'crx' + a.id, address: a.address, city: a.city, type, price: a.price, cap_rate: a.cap, firm_name: b.firm, source: url }).catch(() => {});
      await db.link(b.id, 'crx' + a.id, 'book').catch(() => {});
      const key = norm(a.address);
      if (!haveId.has('crx' + a.id) && key && !have.has(key)) {
        have.add(key); haveId.add('crx' + a.id);
        doc.listings.push({
          id: 'crx' + a.id, address: a.address, city: a.city, zip: '', type,
          price: a.price, units: 1, sqft: null, cap_rate: a.cap, verified: false,
          year_built: null, rent_control: 'verify', status: 'Active',
          firm: 'Crexi', broker_firm: b.firm, broker_agent: b.name, broker_agents: [b.name],
          source_of_truth: 'broker-book',
          upside_note: `Sourced via ${b.name} (${b.firm || 'firm n/a'}) Crexi active book ${today}. ${a.cap ? 'Broker cap ' + a.cap + '% (unverified).' : 'Cap not disclosed.'} Verify status, rent roll & financials before acting.`,
          source: url, _sourced: 'crexi-active-book'
        });
        added++;
      }
    }
  });

  fs.writeFileSync(listingsFile, JSON.stringify(doc, null, 2));
  const report = {
    ran_at: new Date().toISOString(), cost_usd: 0, mode: LIMIT ? `PILOT (${LIMIT} brokers)` : 'FULL',
    brokers_scanned: rows.length, brokers_with_active: brokersWithActive, assets_seen: assetsSeen,
    new_active_listings_added: added, broker_other_listing_rows: otherRows, listings_total_now: doc.listings.length
  };
  fs.writeFileSync(path.join(ROOT, 'data', 'raw', `broker-active-${Date.now()}.json`), JSON.stringify(report, null, 2));
  console.log(`\n=== DONE ($0) ===`); console.log(report);
  console.log('NOTE: data/listings.json grew — run `node scripts/analyze.js` then deploy (deploy is Steve-gated).');
  await db.pool.end().catch(() => {});
  process.exit(0);
})().catch(e => { console.error('FATAL', e); process.exit(1); });