`scripts/ingest-usre-parcels.mjs`

172 lines
#!/usr/bin/env node
// TK-10708  SCALE-UP the per-property CRE index to the FULL property universe (re-usre).
// The news side (build-property-index.mjs) builds public/properties/property-index.json from news+RENTV
// (~2,340 properties). This step ENRICHES that index with usre's public-record commercial parcels:
// every commercial_parcel that has a recorded SALE event, merged in by the SAME canonical-address key
// build-property-index.mjs uses — so a news mention and its public-record twin land on ONE property.
//
// It is a POST-processing enrich layer: it READS the news-built property-index.json and re-writes it with
// usre layered on top. Run it AFTER build-property-index.mjs (which overwrites the file from news+RENTV each
// cycle). Fully IDEMPOTENT — it strips any prior usre entries first and re-derives from the DB, so re-runs
// and any run order are safe. ENRICH, DON'T OVERWRITE: news entries are always preserved; usre only fills
// null fields, appends its own sale entries, and adds parcel enrichment (county / sqft / year_built / units /
// assessed value). $0 local (usre reads via psql, same pattern as ingest-public-deals.mjs).
//
// Usage: node scripts/ingest-usre-parcels.mjs

import { execFileSync } from 'node:child_process';
import { readFileSync, writeFileSync, existsSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import { dirname, join } from 'node:path';

const ROOT = join(dirname(fileURLToPath(import.meta.url)), '..');
const INDEX = join(ROOT, 'public', 'properties', 'property-index.json');

// ── IDENTICAL canonical-address key to build-property-index.mjs (verbatim) so news mentions attach ─────────
const canon = a => a.toLowerCase().replace(/\b(ave|avenue|st|street|blvd|boulevard|dr|drive|rd|road|way|ln|lane|ct|court|pl|place|pkwy|parkway|hwy|highway|cir|circle|ter|terrace|sq|square)\b/g, m => m[0]).replace(/[^a-z0-9]+/g, ' ').trim();
const keyOf = (addr, city) => canon(addr || '') + '|' + (city || '').toLowerCase();

// money label consistent with ingest-public-deals.mjs / the news deal-store
const money = n => (n == null || n <= 0) ? null : n >= 1e9 ? '$' + (n / 1e9).toFixed(1) + 'B' : n >= 1e6 ? '$' + Math.round(n / 1e6) + 'M' : '$' + Math.round(n / 1e3) + 'K';
const tc = s => (s || '').toLowerCase().replace(/\b\w/g, c => c.toUpperCase());

// usre ctype (+ use_desc hint) -> the property_type vocabulary the /properties viewer filters on
function ptype(ctype, use) {
  const c = (ctype || '').toLowerCase(), u = (use || '').toLowerCase();
  if (c === 'office') return 'Office';
  if (c === 'industrial') return 'Industrial';
  if (c === 'retail') return 'Retail';
  if (c === 'hospitality') return 'Hotel';
  if (c === 'parking') return 'Parking';
  if (c === 'studio') return 'Studio';
  if (/apartment|multifamily|multi.?family|residential income|condo|dwelling/.test(u)) return 'Multifamily';
  if (/vacant|\bland\b/.test(u)) return 'Land';
  return null;
}

// per-entry signature — collapses the SAME sale whether it arrived via the news deal-store or via usre
// (same rounded price + same date = one transaction), else falls back to the title-dedup build uses.
const esig = e => (e.price && e.date) ? `p|${Math.round(e.price)}|${(e.date || '').slice(0, 10)}`
  : 't|' + (e.title || '').toLowerCase().replace(/[^a-z0-9]+/g, ' ').trim().slice(0, 50);

// ── 1. load the news-built index; strip any prior usre entries + enrichment for idempotency ────────────────
if (!existsSync(INDEX)) { console.error('property-index.json missing — run build-property-index.mjs first'); process.exit(1); }
const news = JSON.parse(readFileSync(INDEX, 'utf8'));
const props = new Map();
for (const p of news) {
  const entries = (p.entries || []).filter(e => e.origin !== 'usre');   // drop prior usre layer
  props.set(keyOf(p.address, p.city), {
    address: p.address, city: p.city, state: p.state, property_type: p.property_type,
    county: p.county || null, county_fips: p.county_fips || null, zip: p.zip || null,
    sqft: p.sqft || null, year_built: p.year_built || null, units: p.units || null,
    assessed_total: p.assessed_total || null,
    entries, sources: new Set((p.sources || []).filter(s => !/· Assessor$/.test(s))),
  });
}
const newsKeys = new Set(props.keys());

// ── 2. pull every ADDRESSED commercial parcel in the index counties, with its sale events aggregated ──────
// Steve 2026-08-24 (yoloforever re-CRE cycle 5): scope to the 4 counties usre has deal coverage for and
// include NO-SALE parcels too (not just EXISTS-a-sale) — this fills the LA hole (usre holds 132,582 LA
// commercial parcels but only 2,606 had a sale event = 2% coverage) WITHOUT the national 8× blowup that
// pulling all 20 commercial_parcel counties would cause. A no-sale parcel still carries address/type/
// assessed_total/sqft/year_built and sinks to the bottom (mentions:0); sales attach where they exist.
const INDEX_COUNTIES = ['12086', '06037', '37183', '41005'];   // Miami-Dade, LA, Wake, Clackamas
const sql = `SELECT row_to_json(t) FROM (
  SELECT cp.ain, cp.address, cp.city, cp.zip, cp.ctype, cp.use_desc, cp.sqft, cp.year_built, cp.units,
         cp.assessed_total::bigint AS assessed_total, cp.county_fips,
         r.name AS county, r.state_code AS state,
         (SELECT json_agg(json_build_object(
             'date', pe.event_date, 'amount', pe.amount::bigint, 'doc_type', pe.doc_type,
             'doc_number', pe.doc_number, 'source', pe.source, 'source_url', pe.source_url,
             'grantee', pe.detail->>'grantee', 'grantor', pe.detail->>'grantor')
             ORDER BY pe.event_date DESC NULLS LAST)
          FROM parcel_event pe
          WHERE pe.county_fips = cp.county_fips AND pe.source_id = cp.ain AND pe.event_type = 'sale') AS events
  FROM commercial_parcel cp
  LEFT JOIN region r ON r.fips = cp.county_fips AND r.region_type = 'county'
  WHERE cp.address IS NOT NULL AND cp.address <> ''
    AND cp.county_fips IN (${INDEX_COUNTIES.map(c => `'${c}'`).join(', ')})
) t`;
const rows = execFileSync('psql', ['usre', '-t', '-A', '-c', sql], { encoding: 'utf8', maxBuffer: 1024 * 1024 * 768 })
  .trim().split('\n').filter(Boolean).map(l => JSON.parse(l));
// deterministic parcel order (the SQL has no ORDER BY → Postgres scan order varies run-to-run) so the
// 54MB output is byte-identical across cycles and the refine loop doesn't commit reshuffled no-op diffs.
// county → address → city → AIN: AIN is the final tiebreaker so buildings where several parcel AINs (condo
// units) share one canonical address pick the SAME AIN's sqft/assessed each run (fill-null is first-wins).
rows.sort((a, b) => (a.county_fips || '').localeCompare(b.county_fips || '') || (a.address || '').localeCompare(b.address || '') || (a.city || '').localeCompare(b.city || '') || (a.ain || '').localeCompare(b.ain || ''));

// keep only per-folio deep links (miami-dade #/?folio=…); LA's endpoint is a generic query, not per-property
const usefulLink = u => u && /folio=|parcel=|ain=|pin=|[?#].*id=/.test(u) ? u : null;

let matched = 0, added = 0;
for (const r of rows) {
  const addr = (r.address || '').trim();
  if (!addr) continue;
  // skip parenthetical PLACEHOLDER "addresses" — assessor annotations like "(PYLON SIGN SITE)", not a real
  // street address (can't be shown/searched as a property). Rare (2 in the 4-county pull) but future-proofs
  // the no-sale-parcel inclusion (cycle 6). NOTE: a no-sale parcel with a REAL address but no assessed/sqft is
  // a legitimate property (e.g. "1 OCEAN DR" Miami Beach) and is KEPT — only the ( … ) placeholder form drops.
  if (/^\(.*\)$/.test(addr)) continue;
  const city = r.city ? tc(r.city) : null;
  const key = keyOf(addr, city);
  const pt = ptype(r.ctype, r.use_desc);
  const src = `${r.county || 'County'} · Assessor`;
  const title = `${addr}${city ? ', ' + city : ''}${pt ? ' — ' + pt : ''}`;

  let p = props.get(key);
  if (!p) {
    p = { address: addr, city, state: r.state || null, property_type: pt,
      county: r.county || null, county_fips: r.county_fips || null, zip: r.zip || null,
      sqft: r.sqft || null, year_built: r.year_built || null, units: r.units || null,
      assessed_total: r.assessed_total || null, entries: [], sources: new Set() };
    props.set(key, p);
    added++;
  } else {
    matched++;
    // ENRICH ONLY: fill nulls, never overwrite news-provided values
    if (!p.property_type) p.property_type = pt;
    if (!p.state) p.state = r.state || null;
    if (!p.county) p.county = r.county || null;
    if (!p.county_fips) p.county_fips = r.county_fips || null;
    if (!p.zip) p.zip = r.zip || null;
    if (!p.sqft) p.sqft = r.sqft || null;
    if (!p.year_built) p.year_built = r.year_built || null;
    if (!p.units) p.units = r.units || null;
    if (!p.assessed_total) p.assessed_total = r.assessed_total || null;
  }
  for (const ev of (r.events || [])) {
    p.entries.push({
      date: ev.date, txn: 'Sale', price: (ev.amount && ev.amount > 0) ? ev.amount : null,
      price_label: money(ev.amount), source: src, link: usefulLink(ev.source_url),
      title, buyer: ev.grantee || null, seller: ev.grantor || null, origin: 'usre',
    });
    p.sources.add(src);
  }
}

// ── 3. recompute aggregates over the UNION (news entries always retained), dedup same-sale across paths ────
const out = [...props.values()].map(p => {
  const sorted = p.entries.slice().sort((a, b) => (Date.parse(b.date) || 0) - (Date.parse(a.date) || 0) || (b.price || 0) - (a.price || 0) || esig(a).localeCompare(esig(b)));
  const seen = new Set(), ent = [];
  for (const x of sorted) { const s = esig(x); if (!seen.has(s)) { seen.add(s); ent.push(x); } }
  const priced = ent.filter(x => x.price).sort((a, b) => (b.price || 0) - (a.price || 0));
  const sources = new Set(ent.map(e => e.source).filter(Boolean));
  return {
    address: p.address, city: p.city, state: p.state, property_type: p.property_type,
    county: p.county, county_fips: p.county_fips, zip: p.zip,
    sqft: p.sqft, year_built: p.year_built, units: p.units, assessed_total: p.assessed_total,
    latest: ent[0]?.date, latest_txn: ent[0]?.txn, latest_price: ent[0]?.price_label,
    top_price: priced[0]?.price_label || null,
    mentions: ent.length, source_count: sources.size, sources: [...sources], entries: ent,
  };
}).sort((a, b) => b.mentions - a.mentions || (Date.parse(b.latest) || 0) - (Date.parse(a.latest) || 0)
  || (a.address + '|' + (a.city || '')).localeCompare(b.address + '|' + (b.city || '')));   // stable final tie-break

// drop null / empty-string keys to keep the full-universe file as lean as possible (viewer tolerates missing keys)
const compact = (k, v) => (v === null || v === '') ? undefined : v;
writeFileSync(INDEX, JSON.stringify(out, compact, 0));
const newRecords = out.length - newsKeys.size;
console.log(`usre parcel enrich: ${rows.length} addressed commercial parcels (${INDEX_COUNTIES.length} index counties, sale + no-sale) -> matched ${matched} existing news properties, added ${added} new -> ${out.length} total properties (${newsKeys.size} were news-built). -> public/properties/property-index.json`);
for (const p of out.filter(p => p.mentions > 1).slice(0, 6)) console.log(`  ${(p.top_price || p.latest_price || '—').padStart(7)}  ${(p.property_type || '?').padEnd(11)} ${(p.city || '').padEnd(15)} ${p.address}  [${p.mentions}× ${p.source_count} src]`);