← back to Commercialrealestate

scripts/fetch-sfr-redfin.js

251 lines

// fetch-sfr-redfin.js — FEED-FIRST Redfin LA County SFR (single-family) FOR-SALE scraper.
//
// Same gis-csv feed that cracked the condo sweep (fetch-condos-redfin.js) but uipt=1 (single-family
// detached) instead of uipt=3 (condo). status=9 = active for-sale. For each resolved LA region
// (data/redfin-regions.json cities + redfin-neighborhoods.json hoods, 229 region_ids) we warm ONE
// Browserbase session and fetch the gis-csv feed in-page, then upsert each SFR into cre.sfr.
//
//   /stingray/api/gis-csv?al=1&region_id=<id>&region_type=<6 city|1 hood>&uipt=1&num_homes=350&status=9
//
// Dense regions truncate at Redfin's ~350 cap, so a capped region is re-queried in PRICE BANDS and
// unioned (same as fetch-closed-sales.js) to beat the cap. A swept_region_sfr resume ledger marks
// finished regions so a re-run after a session timeout picks up the tail (idempotent regardless —
// sfr.id / sfr.source(url) are unique keys).
//
// HONEST LABELING: the gis-csv feed has no listing agent — broker_name/firm_name stay NULL here.
// Listing agents are captured separately by fetch-sfr-agents.js (mainHouseInfoPanelInfo), which
// honestly labels SFRs whose agent Redfin suppresses.
//
// HARD CAP (Steve-approved): stop at MAX_SESSIONS (30) sessions OR ~$1.50 spend, whichever first.
// Per-session + RUNNING BATCH TOTAL surfaced live. Redfin-ONLY (no Zillow/2captcha — separate gate).
//
// Usage: NODE_PATH=$HOME/.claude/skills/browserbase/node_modules node scripts/fetch-sfr-redfin.js
//   CC_MAX_SESSIONS=30  CC_MAX_COST=1.50  CC_PER_SESSION_REGIONS=10  CC_MAXREGIONS=0(all)  CC_FRESH=1
'use strict';
const fs = require('fs');
const path = require('path');
const { chromium } = require('playwright-core');
const Browserbase = require('@browserbasehq/sdk').default;
const { Pool } = require('pg');
const { inLACounty } = require('./sources/la-county');

const bbEnv = 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 ROOT = path.join(__dirname, '..');
const pool = new Pool({ host: '/tmp', port: 5432, database: 'cre', user: process.env.USER || 'stevestudio2' });

// CC_LOCAL=1 launches local Chrome (residential IP clears Redfin) instead of a paid Browserbase
// session — $0/run. Same in-page gis-csv fetch; only the browser transport differs.
const LOCAL = process.env.CC_LOCAL === '1';
const CHROME_PATH = process.env.CHROME_PATH || '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome';
const SESSION_COST = LOCAL ? 0 : 0.04;
const MAX_SESSIONS = +(process.env.CC_MAX_SESSIONS || (LOCAL ? 999 : 30));
const MAX_COST = +(process.env.CC_MAX_COST || 1.50); // in LOCAL mode SESSION_COST=0 so this cap never trips
const PER_SESSION_REGIONS = +(process.env.CC_PER_SESSION_REGIONS || (LOCAL ? 20 : 10));
const MAXR = process.env.CC_MAXREGIONS != null ? +process.env.CC_MAXREGIONS : 0; // 0 = all
const NUM_HOMES = 350;
const CAP_THRESHOLD = NUM_HOMES - 5;
const PRICE_BANDS = [
  [0, 500000], [500000, 750000], [750000, 1000000],
  [1000000, 1500000], [1500000, 3000000], [3000000, 0]
];

function parseCSV(text) {
  const rows = [];
  let i = 0, field = '', row = [], inQ = false;
  while (i < text.length) {
    const c = text[i];
    if (inQ) {
      if (c === '"') { if (text[i + 1] === '"') { field += '"'; i++; } else inQ = false; }
      else field += c;
    } else {
      if (c === '"') inQ = true;
      else if (c === ',') { row.push(field); field = ''; }
      else if (c === '\n') { row.push(field); rows.push(row); row = []; field = ''; }
      else if (c === '\r') { /* skip */ }
      else field += c;
    }
    i++;
  }
  if (field.length || row.length) { row.push(field); rows.push(row); }
  return rows;
}

// Extract SFR rows from a gis-csv response. Returns [] when no header (empty/blocked feed).
function sfrsFromCSV(text) {
  const rows = parseCSV(text);
  let hi = rows.findIndex(r => r.join(',').includes('PROPERTY TYPE') && r.join(',').includes('ADDRESS'));
  if (hi < 0) return [];
  const H = rows[hi].map(h => h.trim());
  const col = name => H.findIndex(h => h.toUpperCase() === name || h.toUpperCase().startsWith(name));
  const ci = {
    type: col('PROPERTY TYPE'), addr: col('ADDRESS'), city: col('CITY'), zip: col('ZIP OR POSTAL CODE'),
    price: col('PRICE'), beds: col('BEDS'), baths: col('BATHS'), sqft: col('SQUARE FEET'),
    year: col('YEAR BUILT'), status: col('STATUS'), dom: col('DAYS ON MARKET'),
    lat: col('LATITUDE'), lng: col('LONGITUDE'),
    url: H.findIndex(h => h.toUpperCase().startsWith('URL')), mls: col('MLS#')
  };
  const out = [];
  for (let r = hi + 1; r < rows.length; r++) {
    const row = rows[r];
    if (!row || row.length < 5) continue;
    const ptype = (row[ci.type] || '').trim();
    // uipt=1 returns detached single-family; guard against any stray attached types.
    if (!/single.?family|^house$/i.test(ptype)) continue;
    const price = Number((row[ci.price] || '').replace(/[^\d.]/g, ''));
    const addr = (row[ci.addr] || '').trim();
    const city = (row[ci.city] || '').trim();
    if (!addr || !price) continue;
    if (city && !inLACounty(city)) continue;
    const url = (row[ci.url] || '').trim();
    const mls = (row[ci.mls] || '').trim();
    const id = 'rdf' + (mls || (addr + (row[ci.zip] || '')).replace(/\s+/g, '').toLowerCase());
    out.push({
      id,
      address: addr.slice(0, 120),
      city,
      zip: (row[ci.zip] || '').trim(),
      price,
      beds: ci.beds >= 0 && row[ci.beds] ? Number(row[ci.beds]) || null : null,
      baths: ci.baths >= 0 && row[ci.baths] ? Number(row[ci.baths]) || null : null,
      sqft: ci.sqft >= 0 && row[ci.sqft] ? Math.round(Number(String(row[ci.sqft]).replace(/[^\d.]/g, ''))) || null : null,
      year_built: ci.year >= 0 && row[ci.year] ? Number(row[ci.year]) || null : null,
      days_on_market: ci.dom >= 0 && row[ci.dom] !== '' && row[ci.dom] != null ? (Number(String(row[ci.dom]).replace(/[^\d]/g, '')) ) : null,
      market_status: ci.status >= 0 ? ((row[ci.status] || '').trim() || null) : null,
      lat: ci.lat >= 0 && row[ci.lat] && isFinite(+row[ci.lat]) ? +row[ci.lat] : null,
      lng: ci.lng >= 0 && row[ci.lng] && isFinite(+row[ci.lng]) ? +row[ci.lng] : null,
      listing_text: ptype,
      source: url ? (url.startsWith('http') ? url : 'https://www.redfin.com' + url) : null
    });
  }
  return out;
}

function loadRegions() {
  const load = (f, type) => {
    try { return (JSON.parse(fs.readFileSync(path.join(ROOT, 'data', f), 'utf8')).regions || []).filter(r => r.region_id).map(r => ({ name: r.query || r.name, region_id: r.region_id, region_type: r.region_type || type })); }
    catch (_) { return []; }
  };
  const seen = new Set();
  return [...load('redfin-regions.json', 6), ...load('redfin-neighborhoods.json', 1)]
    .filter(r => { const k = r.region_id + ':' + r.region_type; if (seen.has(k)) return false; seen.add(k); return true; });
}

(async () => {
  await pool.query(`CREATE TABLE IF NOT EXISTS swept_region_sfr (
    region_id text, region_type int, name text, swept_at timestamptz DEFAULT now(),
    PRIMARY KEY (region_id, region_type))`);
  if (process.env.CC_FRESH === '1') await pool.query('TRUNCATE swept_region_sfr');
  const done = new Set((await pool.query('SELECT region_id, region_type FROM swept_region_sfr')).rows
    .map(r => r.region_id + ':' + r.region_type));

  const allRegions = loadRegions().filter(r => !done.has(r.region_id + ':' + r.region_type));
  const regions = MAXR ? allRegions.slice(0, MAXR) : allRegions;
  if (!regions.length) { console.log(JSON.stringify({ note: 'all resolved regions already swept (CC_FRESH=1 to re-sweep)', captured: 0 })); await pool.end(); return; }
  process.stderr.write(`SFR sweep: ${regions.length} regions to go (${done.size} already swept) · cap ${MAX_SESSIONS} sessions / $${MAX_COST.toFixed(2)}\n`);

  let added = 0, captured = 0;
  const num = v => { const n = +String(v == null ? '' : v).replace(/[^0-9.]/g, ''); return Number.isFinite(n) && n > 0 ? Math.round(n) : null; };

  // Upsert one SFR (idempotent on id; source/url is also UNIQUE).
  const upsert = async (s) => {
    if (!s.source) return;
    const ins = await pool.query(
      `INSERT INTO sfr(id,address,city,zip,price,beds,baths,sqft,year_built,listing_text,source,last_seen,status,days_on_market,market_status,listed_date,lat,lng)
       VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,now(),'active',$12::int,$13, CASE WHEN $12::int IS NOT NULL THEN CURRENT_DATE - $12::int ELSE NULL END,$14::float8,$15::float8)
       ON CONFLICT(id) DO UPDATE SET price=EXCLUDED.price, beds=EXCLUDED.beds, baths=EXCLUDED.baths,
         sqft=EXCLUDED.sqft, year_built=EXCLUDED.year_built,
         last_seen=now(), status='active', off_market_at=NULL, disposition=NULL, sold_price=NULL, sold_date=NULL,
         days_on_market=EXCLUDED.days_on_market, listed_date=EXCLUDED.listed_date,
         prev_market_status=sfr.market_status, market_status=EXCLUDED.market_status,
         prev_price=CASE WHEN EXCLUDED.price IS DISTINCT FROM sfr.price THEN sfr.price ELSE sfr.prev_price END,
         price_changed_at=CASE WHEN EXCLUDED.price IS DISTINCT FROM sfr.price THEN now() ELSE sfr.price_changed_at END,
         lat=COALESCE(EXCLUDED.lat, sfr.lat), lng=COALESCE(EXCLUDED.lng, sfr.lng)
       RETURNING (xmax = 0) AS inserted`,
      [s.id, s.address, s.city, s.zip, s.price, s.beds, s.baths, s.sqft, s.year_built, s.listing_text, s.source, s.days_on_market ?? null, s.market_status ?? null, s.lat ?? null, s.lng ?? null])
      .catch(() => null);
    if (ins && ins.rows[0] && ins.rows[0].inserted) added++;
  };
  const flush = async (recs, region) => {
    captured += recs.length;
    for (const s of recs) await upsert(s);
    if (region) await pool.query('INSERT INTO swept_region_sfr(region_id,region_type,name) VALUES($1,$2,$3) ON CONFLICT DO NOTHING',
      [region.region_id, region.region_type, region.name]);
  };

  let sessions = 0;
  for (let g = 0; g < regions.length; g += PER_SESSION_REGIONS) {
    if (sessions >= MAX_SESSIONS) { process.stderr.write(`\n[CAP] ${MAX_SESSIONS}-session cap reached; stopping.\n`); break; }
    if (sessions * SESSION_COST >= MAX_COST) { process.stderr.write(`\n[CAP] $${MAX_COST.toFixed(2)} cost cap reached; stopping.\n`); break; }
    const group = regions.slice(g, g + PER_SESSION_REGIONS);
    let browser, session;
    sessions++;
    const running = (sessions * SESSION_COST).toFixed(2);
    process.stderr.write(`\n[session ${sessions}/${MAX_SESSIONS}] ${group.length} regions  (per-session $${SESSION_COST.toFixed(2)} · running batch total $${running} / cap $${MAX_COST.toFixed(2)})\n`);

    try {
      if (LOCAL) {
        browser = await chromium.launch({ executablePath: CHROME_PATH, args: ['--disable-blink-features=AutomationControlled'] });
      } else {
        const bb = new Browserbase({ apiKey: get(bbEnv, 'BROWSERBASE_API_KEY') });
        session = await bb.sessions.create({ projectId: get(bbEnv, 'BROWSERBASE_PROJECT_ID'), browserSettings: { solveCaptchas: true, viewport: { width: 1440, height: 1000 } } });
        browser = await chromium.connectOverCDP(session.connectUrl);
      }
      const ctx = browser.contexts()[0] || await browser.newContext({ userAgent: 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', viewport: { width: 1440, height: 1000 } });
      const page = ctx.pages()[0] || await ctx.newPage();
      page.setDefaultTimeout(45000);
      await page.goto('https://www.redfin.com/city/11203/CA/Los-Angeles/filter/property-type=house', { waitUntil: 'domcontentloaded' }).catch(() => {});
      await page.waitForTimeout(2500);

      // status=9 = active for-sale; status=130 = pending/contingent (in escrow). We sweep both so the
      // catalog carries in-escrow homes (market_status Pending/Contingent) for the escrow filters.
      const fetchRegion = async (reg, band, status = 9) => {
        let extra = '';
        if (band) { if (band[0] > 0) extra += `&min_price=${band[0]}`; if (band[1] > 0) extra += `&max_price=${band[1]}`; }
        const url = `https://www.redfin.com/stingray/api/gis-csv?al=1&region_id=${reg.region_id}&region_type=${reg.region_type}&uipt=1&num_homes=${NUM_HOMES}&status=${status}&sf=1,2,3,5,6,7&v=8${extra}`;
        const res = await page.evaluate(async (u) => {
          try { const r = await fetch(u, { headers: { accept: 'text/csv' } }); return { status: r.status, text: r.ok ? await r.text() : '' }; }
          catch (e) { return { status: 0, text: '' }; }
        }, url);
        return (res.status === 200 && res.text && res.text.includes('PROPERTY TYPE')) ? sfrsFromCSV(res.text) : [];
      };

      for (const reg of group) {
        let base;
        try { base = await fetchRegion(reg); }
        catch (e) { process.stderr.write('  session closed mid-sweep — flushed regions persist, re-run to resume.\n'); throw e; }
        // Pending/contingent (in-escrow) — small set, fetch once, no price-banding.
        let pending = [];
        try { pending = await fetchRegion(reg, null, 130); await page.waitForTimeout(250); } catch (_) {}
        if (base.length < CAP_THRESHOLD) {
          await flush(base.concat(pending), reg);
          process.stderr.write(`  ${reg.name}: ${base.length} active + ${pending.length} escrow (cumulative +${added} new)\n`);
          await page.waitForTimeout(400);
          continue;
        }
        // Capped — re-query in price bands and union (dedupe on id).
        const seenId = new Set(); const unioned = [];
        const keep = arr => { for (const x of arr) if (x.id && !seenId.has(x.id)) { seenId.add(x.id); unioned.push(x); } };
        keep(base); keep(pending);
        let broke = false;
        for (const band of PRICE_BANDS) {
          try { keep(await fetchRegion(reg, band)); }
          catch (e) { broke = true; break; }
          await page.waitForTimeout(350);
        }
        await flush(unioned, broke ? null : reg);
        process.stderr.write(`  [banded] ${reg.name}: base ${base.length} -> ${seenId.size} unioned across ${PRICE_BANDS.length} bands${broke ? ' (PARTIAL)' : ''}\n`);
        if (broke) throw new Error('session closed mid-band');
      }
    } catch (e) {
      process.stderr.write('  session err: ' + e.message.split('\n')[0] + ' (re-run to resume from ledger)\n');
    } finally { if (browser) try { await browser.close(); } catch (_) {} }
  }

  const actualCost = (sessions * SESSION_COST).toFixed(2);
  const tot = (await pool.query('SELECT count(*)::int n, count(DISTINCT city)::int cities FROM sfr')).rows[0];
  process.stderr.write(`\nSessions ${sessions} · ACTUAL COST ~$${actualCost} · captured ${captured} rows (+${added} new) · sfr total ${tot.n} across ${tot.cities} cities\n`);
  console.log(JSON.stringify({ sessions, cost: '$' + actualCost, captured, added, sfr_total: tot.n, cities: tot.cities }));
  await pool.end();
})();