`src/ingest/listings/engine.ts`

122 lines
/**
 * M-B3 firm↔listing pilot ingest engine.
 *
 * Per site: openRun → robots gate → discover ≤CAP listing URLs from the public sitemap →
 * for each: robots-recheck → politeFetch (UA + 2–4s jitter) → parse FACTS ONLY from
 * JSON-LD → resolve firm_id by source host → resolve county region_id by centroid →
 * upsert into `listing` (UNIQUE(source, source_id)) → closeRun.
 *
 * Fail-loud: 0 parsed listings = failed run + non-zero exit. A robots-disallowed listings
 * path = the whole site is SKIPPED and recorded as a failed run with an explicit note.
 *
 *   npm run ingest:listings -- coldwellbanker
 *   npm run ingest:listings -- realtytexas
 *   npm run ingest:listings -- all
 */
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import type { ListingAdapter } from './types.ts';
import { coldwellbanker } from './coldwellbanker.ts';
import { realtytexas } from './realtytexas.ts';
import { politeFetch, robotsGate, resolveFirmByHost, resolveRegion } from './shared.ts';

const ADAPTERS: ListingAdapter[] = [coldwellbanker, realtytexas];
const CAP = 50; // pilot cap per site

async function upsertListing(
  source: string,
  f: import('./types.ts').ListingFacts,
  firmId: number | null,
  regionId: number | null,
): Promise<void> {
  await query(
    `INSERT INTO listing
       (source, source_id, firm_id, region_id, address, price, beds, baths, sqft, lat, lng, url, status, last_seen)
     VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13, NOW())
     ON CONFLICT (source, source_id) DO UPDATE SET
       firm_id  = COALESCE(EXCLUDED.firm_id, listing.firm_id),
       region_id= COALESCE(EXCLUDED.region_id, listing.region_id),
       address  = EXCLUDED.address,
       price    = EXCLUDED.price,
       beds     = COALESCE(EXCLUDED.beds, listing.beds),
       baths    = COALESCE(EXCLUDED.baths, listing.baths),
       sqft     = COALESCE(EXCLUDED.sqft, listing.sqft),
       lat      = COALESCE(EXCLUDED.lat, listing.lat),
       lng      = COALESCE(EXCLUDED.lng, listing.lng),
       url      = EXCLUDED.url,
       status   = EXCLUDED.status,
       last_seen= NOW()`,
    [source, f.sourceId, firmId, regionId, f.address, f.price, f.beds, f.baths, f.sqft, f.lat, f.lng, f.url, f.status],
  );
}

async function runAdapter(a: ListingAdapter): Promise<{ upserted: number; skipped: number }> {
  const runId = await openRun(a.source, `https://${a.host}`);
  console.log(`[${a.source}] robots gate…`);
  try {
    const allowed = await robotsGate(a.host);
    if (!allowed(a.listingsPathHint)) {
      const note = `robots.txt disallows listings path ${a.listingsPathHint} — SKIPPED`;
      console.log(`[${a.source}] ${note}`);
      await closeRun(runId, 'failed', { upserted: 0, skipped: 0, notes: note });
      return { upserted: 0, skipped: 0 };
    }

    const firmId = await resolveFirmByHost(a.host);
    console.log(`[${a.source}] firm_id resolved: ${firmId ?? 'NONE'}`);

    const urls = await a.discover(CAP);
    console.log(`[${a.source}] discovered ${urls.length} listing URLs (cap ${CAP})`);

    let upserted = 0;
    let skipped = 0;
    for (const url of urls) {
      const path = new URL(url).pathname;
      if (!allowed(path)) { skipped++; continue; }
      const { ok, status, body } = await politeFetch(url);
      if (!ok) { console.log(`[${a.source}] fetch ${status} ${url}`); skipped++; continue; }
      const facts = a.parse(url, body);
      if (!facts) { skipped++; continue; }
      const regionId = await resolveRegion(facts.state, facts.lat, facts.lng);
      await upsertListing(a.source, facts, firmId, regionId);
      upserted++;
      if (upserted % 5 === 0) console.log(`[${a.source}] upserted ${upserted}…`);
    }

    if (upserted === 0) {
      const note = `0 listings parsed from ${urls.length} URLs — structure drift or bot-block`;
      console.error(`[${a.source}] FAIL: ${note}`);
      await closeRun(runId, 'failed', { upserted: 0, skipped, notes: note });
      return { upserted: 0, skipped };
    }
    await closeRun(runId, 'ok', { upserted, skipped, notes: `firm_id=${firmId}` });
    console.log(`[${a.source}] DONE upserted=${upserted} skipped=${skipped} firm_id=${firmId}`);
    return { upserted, skipped };
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: String(e.message || e) });
    throw e;
  }
}

async function main() {
  const arg = (process.argv[2] || '').toLowerCase();
  const targets = arg === 'all' || !arg ? ADAPTERS : ADAPTERS.filter((a) => a.source === arg);
  if (!targets.length) {
    console.error(`unknown site '${arg}'. known: ${ADAPTERS.map((a) => a.source).join(', ')}, all`);
    process.exit(2);
  }
  let total = 0;
  for (const a of targets) {
    const r = await runAdapter(a);
    total += r.upserted;
  }
  console.log(`\n[listings] total upserted across ${targets.length} site(s): ${total}`);
  await pool.end();
  if (total === 0) process.exit(1);
}

main().catch((e) => {
  console.error(e);
  process.exit(1);
});