`scripts/ingest-la-sales.mjs`

85 lines
#!/usr/bin/env node
// TK-10708  LA (06037) sale-event backfill from the LA Assessor PAIS "Recent Sales
// Parcels" ArcGIS layer -- prices INFERRED from Documentary Transfer Tax (property-tax
// -> price), refreshed weekly, single-parcel sales, last ~24mo. This fills the gap that
// left recent_commercial_deals with 0 LA deals: LA had 153k commercial_parcel rows but
// 0 parcel_event rows because the wrong (parcel, no-price) layer was configured.
//
// Writes parcel_event rows joined AIN->commercial_parcel.ain, so recent_commercial_deals
// (parcel_event JOIN commercial_parcel, amount>250k, last 18mo) lights up for LA.
// Provenance is labelled source='assessor_dtt_inferred' (codex: label the inference).
// "no names, just price" (Steve) -- we pull AIN/address/date/price only, no owner names.
//
// Reversible: DELETE FROM parcel_event WHERE county_fips='06037' AND source='assessor_dtt_inferred';
// Usage: node scripts/ingest-la-sales.mjs [--dry]

import { execFileSync } from 'node:child_process';

const DRY = process.argv.includes('--dry');
const LAYER = 'https://egispais.gis.lacounty.gov/pais/rest/services/PAIS/pais_sales_parcels/MapServer/0/query';
const WHERE = "USETYPE='C/I' AND SALEPRICE>250000";
const PAGE = 1000;  // LA PAIS maxRecordCount is 1000
const psql = sql => execFileSync('psql', ['usre', '-t', '-A', '-c', sql], { encoding: 'utf8' }).trim();
const q = v => `'${String(v).replace(/'/g, "''")}'`;

async function pull() {
  const rows = [];
  for (let offset = 0; ; offset += PAGE) {
    const u = new URL(LAYER);
    u.search = new URLSearchParams({
      where: WHERE, outFields: 'AIN,SAADDR,SALEDATE,SALEPRICE,USETYPE',
      returnGeometry: 'false', orderByFields: 'AIN', resultRecordCount: String(PAGE),
      resultOffset: String(offset), f: 'json'
    }).toString();
    const r = await fetch(u).then(x => x.json());
    const feats = r.features || [];
    for (const f of feats) {
      const a = f.attributes;
      const ain = String(a.AIN ?? '').replace(/\D/g, '').padStart(10, '0');
      const price = Number(a.SALEPRICE);
      if (!ain || !(price > 250000)) continue;
      const d = a.SALEDATE ? new Date(a.SALEDATE) : null;
      const date = d && !isNaN(d) ? d.toISOString().slice(0, 10) : null;
      if (!date) continue;
      rows.push({ ain, price: Math.round(price), date, addr: (a.SAADDR || '').trim() });
    }
    if (feats.length < PAGE) break;
    if (offset > 20000) break; // safety cap
  }
  // dedup on (ain,date,price)
  const seen = new Set(), out = [];
  for (const r of rows) { const k = `${r.ain}|${r.date}|${r.price}`; if (!seen.has(k)) { seen.add(k); out.push(r); } }
  return out;
}

const rows = await pull();
console.log(`Pulled ${rows.length} LA C/I sale events (>250k, DTT-inferred).`);
console.log('sample:', rows.slice(0, 3).map(r => `${r.addr||'(no addr)'} $${r.price.toLocaleString()} ${r.date}`).join(' | '));

if (DRY) { console.log('DRY -- no writes.'); process.exit(0); }

// "be first": detect NEW deals vs what we already have, BEFORE reloading.
const prior = new Set(psql(`SELECT source_id||'|'||event_date||'|'||amount::bigint FROM parcel_event WHERE county_fips='06037' AND source='assessor_dtt_inferred'`).split('\n').filter(Boolean));
const fresh = rows.filter(r => !prior.has(`${r.ain}|${r.date}|${r.price}`));
if (fresh.length) {
  const stamp = new Date().toISOString().slice(0, 10);
  execFileSync('bash', ['-c', `cat > "${process.cwd()}/out/new-la-deals-${stamp}.json"`], { input: JSON.stringify(fresh, null, 2) });
  console.log(`NEW closed LA deals this run: ${fresh.length} (written to out/new-la-deals-${stamp}.json) -> feed 'tell people first'`);
} else {
  console.log(`NEW closed LA deals this run: 0 (source frozen at 2024-06 until a fresher DTT feed is connected).`);
}

// reversible reload: clear prior DTT-inferred LA events, then bulk insert in chunks
psql(`DELETE FROM parcel_event WHERE county_fips='06037' AND source='assessor_dtt_inferred'`);
let n = 0;
for (let i = 0; i < rows.length; i += 500) {
  const vals = rows.slice(i, i + 500).map(r =>
    `('06037',${q(r.ain)},'sale',${q(r.date)},${r.price},'deed',${q(JSON.stringify({ note: 'LA PAIS DTT-inferred', addr: r.addr || null }))}::jsonb,'assessor_dtt_inferred',${q(LAYER)})`
  ).join(',');
  psql(`INSERT INTO parcel_event (county_fips,source_id,event_type,event_date,amount,doc_type,detail,source,source_url) VALUES ${vals}`);
  n += Math.min(500, rows.length - i);
}
console.log(`Inserted ${n} LA parcel_event rows.`);
const deals = psql(`SELECT count(*) FROM recent_commercial_deals WHERE county_fips='06037'`);
console.log(`recent_commercial_deals now shows ${deals} LA deals (join x commercial_parcel, last 18mo).`);