`crawl/base-crawler.js`

93 lines
'use strict';
// Shared crawler toolkit for every firm agent. Public-pages-only, polite, $0 by default.
const { Pool } = require('pg');
const pool = new Pool({ host: process.env.PGHOST || '/tmp', database: process.env.PGDATABASE || 'crunified' });

const sleep = (ms) => new Promise(r => setTimeout(r, ms));
const UA = 'Mozilla/5.0 (compatible; SubleaseMarketplaceBot/0.1; +https://sublease.agentabrams.com)';

// Polite GET with timeout + one retry. Returns { ok, status, text }.
async function httpGet(url, { timeout = 20000, headers = {} } = {}) {
  for (let attempt = 0; attempt < 2; attempt++) {
    try {
      const r = await fetch(url, { headers: { 'User-Agent': UA, 'Accept-Language': 'en-US', ...headers }, signal: AbortSignal.timeout(timeout) });
      const text = await r.text();
      return { ok: r.ok, status: r.status, text };
    } catch (e) { if (attempt) return { ok: false, status: 0, text: '', error: String(e) }; await sleep(1500); }
  }
}

// robots.txt gate — returns true if the path is allowed for our UA (fail-open only on fetch error).
const robotsCache = new Map();
async function robotsAllowed(url) {
  try {
    const u = new URL(url);
    const key = u.origin;
    if (!robotsCache.has(key)) {
      const r = await httpGet(u.origin + '/robots.txt', { timeout: 8000 });
      robotsCache.set(key, r.ok ? r.text : '');
    }
    const body = robotsCache.get(key);
    if (!body) return true;
    // minimal parse: global (User-agent: *) Disallow rules
    const lines = body.split(/\r?\n/); let active = false; const dis = [];
    for (const ln of lines) {
      const m = ln.match(/^\s*User-agent:\s*(.+)/i);
      if (m) { active = m[1].trim() === '*'; continue; }
      const d = ln.match(/^\s*Disallow:\s*(.*)/i);
      if (active && d && d[1].trim()) dis.push(d[1].trim());
    }
    return !dis.some(p => u.pathname.startsWith(p));
  } catch { return true; }
}

// Free US address geocoder (Census Bureau — no key, no cost). Returns {lat,lng} or null.
async function geocodeCensus(address) {
  if (!address) return null;
  try {
    const url = 'https://geocoding.geo.census.gov/geocoder/locations/onelineaddress'
      + '?benchmark=Public_AR_Current&format=json&address=' + encodeURIComponent(address);
    const r = await httpGet(url, { timeout: 12000 });
    if (!r.ok) return null;
    const j = JSON.parse(r.text);
    const m = j?.result?.addressMatches?.[0]?.coordinates;
    return m ? { lat: m.y, lng: m.x } : null;
  } catch { return null; }
}

// Upsert one listing keyed on (sponsor_id, external_id). Bumps last_seen on repeat.
async function upsertListing(sponsorId, L) {
  const sql = `
    INSERT INTO listings
      (sponsor_id, external_id, title, address, city, state, zip, lat, lng, space_type,
       is_sublease, size_sf, price_amount, price_unit, term, available_on, source_url, image_url, description, raw, last_seen)
    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20, now())
    ON CONFLICT (sponsor_id, external_id) DO UPDATE SET
      title=EXCLUDED.title, address=EXCLUDED.address, size_sf=EXCLUDED.size_sf,
      price_amount=EXCLUDED.price_amount, price_unit=EXCLUDED.price_unit, term=EXCLUDED.term,
      source_url=EXCLUDED.source_url, image_url=EXCLUDED.image_url, description=EXCLUDED.description,
      raw=EXCLUDED.raw, last_seen=now(), status='active'
    RETURNING (xmax = 0) AS inserted`;
  const vals = [sponsorId, L.external_id, L.title, L.address, L.city, L.state || 'CA', L.zip,
    L.lat, L.lng, L.space_type, L.is_sublease !== false, L.size_sf, L.price_amount, L.price_unit,
    L.term, L.available_on, L.source_url, L.image_url, L.description, L.raw ? JSON.stringify(L.raw) : null];
  const { rows } = await pool.query(sql, vals);
  return rows[0]?.inserted === true;
}

async function getSponsor(slug) {
  const { rows } = await pool.query('SELECT * FROM sponsors WHERE slug=$1', [slug]);
  return rows[0];
}
async function startRun(sponsorId) {
  const { rows } = await pool.query('INSERT INTO crawl_runs (sponsor_id) VALUES ($1) RETURNING id', [sponsorId]);
  return rows[0].id;
}
async function finishRun(runId, { status, found = 0, added = 0, cost = 0, notes = '' }) {
  await pool.query(
    `UPDATE crawl_runs SET finished_at=now(), status=$2, listings_found=$3, listings_new=$4, cost_usd=$5, notes=$6 WHERE id=$1`,
    [runId, status, found, added, cost, notes]);
}

module.exports = { pool, sleep, httpGet, robotsAllowed, geocodeCensus, upsertListing, getSponsor, startRun, finishRun, UA };