`scripts/fetch-assessor.js`

115 lines
#!/usr/bin/env node
// fetch-assessor.js — stream the LA County Assessor public parcel roll (all homes/parcels) to disk.
//
// Source: LA County eGIS "Assessor Parcel Data (Rolls 2021-2025)" hosted feature layer (public, FREE, $0).
//   https://services.arcgis.com/RmCCgQtiZLDCtblq/arcgis/rest/services/Parcel_Data_2021_Table/FeatureServer/0
// The full layer is 12.1M rows = ~2.4M parcels × 5 roll years (2021-2025). We pull ONLY the latest roll
// (RollYear=2025, ~2.43M rows = one record per parcel — the current snapshot of "all homes"). This is the
// disk-aware scoped subset; pulling all 5 years would 5× the size for redundant historical valuations we
// don't need. The cap is LOGGED, never silent. Override with CC_ROLL_YEAR= or CC_ALL_YEARS=1.
//
// Streams NDJSON to data/raw/assessor-parcels-<year>.ndjson (one parcel per line) — never holds the whole
// roll in RAM. Reports running row count + on-disk size. data/raw/ is gitignored (big raw file).
//
// Run: node scripts/fetch-assessor.js            # 2025 roll, all parcels
//      CC_LIMIT=5000 node scripts/fetch-assessor.js   # smoke test (first 5k)
'use strict';
const fs = require('fs');
const path = require('path');
const https = require('https');

const BASE = 'https://services.arcgis.com/RmCCgQtiZLDCtblq/arcgis/rest/services/Parcel_Data_2021_Table/FeatureServer/0';
const PAGE = 2000;                                  // = layer maxRecordCount
const ROLL_YEAR = process.env.CC_ROLL_YEAR || '2025';
const ALL_YEARS = process.env.CC_ALL_YEARS === '1';
const LIMIT = +(process.env.CC_LIMIT || 0) || 0;     // 0 = no cap (full roll)
const WHERE = ALL_YEARS ? '1=1' : `RollYear='${ROLL_YEAR}'`;

// Field subset: identity + location + the valuation/physical attrs that matter for CRE + broker work.
// (* would pull ~60 cols incl. redundant base-year strings; this keeps the file lean but complete.)
const FIELDS = [
  'AIN', 'RollYear', 'PropertyLocation', 'SitusHouseNo', 'SitusStreet', 'SitusUnit',
  'SitusCity', 'SitusZIP5', 'UseType', 'UseCode', 'UseCodeDescChar1', 'UseCodeDescChar2',
  'YearBuilt', 'EffectiveYearBuilt', 'SQFTmain', 'Bedrooms', 'Bathrooms', 'Units',
  'RecordingDate', 'Roll_LandValue', 'Roll_ImpValue', 'Roll_totLandImp', 'Roll_TotalValue',
  'netTaxableValue', 'isTaxableParcel', 'ParcelClassification', 'AdminRegion',
  'CENTER_LAT', 'CENTER_LON'
].join(',');

const RAW_DIR = path.join(__dirname, '..', 'data', 'raw');
fs.mkdirSync(RAW_DIR, { recursive: true });
const OUT = path.join(RAW_DIR, `assessor-parcels-${ALL_YEARS ? 'all' : ROLL_YEAR}.ndjson`);

function getJSON(url) {
  return new Promise((resolve, reject) => {
    const req = https.get(url, { timeout: 60000 }, res => {
      if (res.statusCode !== 200) { res.resume(); return reject(new Error('HTTP ' + res.statusCode)); }
      let buf = '';
      res.setEncoding('utf8');
      res.on('data', d => buf += d);
      res.on('end', () => { try { resolve(JSON.parse(buf)); } catch (e) { reject(new Error('bad JSON: ' + e.message)); } });
    });
    req.on('timeout', () => { req.destroy(new Error('timeout')); });
    req.on('error', reject);
  });
}

function fmtBytes(n) { return n > 1e9 ? (n / 1e9).toFixed(2) + ' GB' : (n / 1e6).toFixed(1) + ' MB'; }

async function main() {
  const t0 = Date.now();
  // total count for progress
  const cnt = await getJSON(`${BASE}/query?where=${encodeURIComponent(WHERE)}&returnCountOnly=true&f=json`);
  const total = cnt.count || 0;
  const target = LIMIT ? Math.min(LIMIT, total) : total;
  console.log(`[assessor] LA County Assessor Parcel Roll — WHERE ${WHERE} → ${total.toLocaleString()} parcels` +
    (LIMIT ? ` (CAPPED to ${LIMIT.toLocaleString()} via CC_LIMIT)` : '') +
    (!ALL_YEARS ? `\n[assessor] SCOPE NOTE: pulling RollYear=${ROLL_YEAR} only (current snapshot, 1 row/parcel). Historical rolls 2021-2024 intentionally skipped (5× redundant). Set CC_ALL_YEARS=1 to pull all.` : ''));
  console.log(`[assessor] streaming NDJSON → ${OUT}  ($0 — public open data)`);

  // Resume support: CC_START_OFFSET continues a partial pull (append) instead of re-fetching from 0.
  const START = +(process.env.CC_START_OFFSET || 0) || 0;
  const ws = fs.createWriteStream(OUT, { flags: START ? 'a' : 'w' });
  let offset = START, written = START, pages = 0;
  if (START) console.log(`[assessor] RESUME from offset ${START.toLocaleString()} (appending to existing file)`);
  while (written < target) {
    const url = `${BASE}/query?where=${encodeURIComponent(WHERE)}&outFields=${FIELDS}` +
      `&returnGeometry=false&orderByFields=AIN&resultOffset=${offset}&resultRecordCount=${PAGE}&f=json`;
    let data;
    for (let attempt = 1; ; attempt++) {
      try { data = await getJSON(url); break; }
      catch (e) {
        if (attempt >= 4) throw new Error(`page @offset ${offset} failed after ${attempt} tries: ${e.message}`);
        await new Promise(r => setTimeout(r, 1500 * attempt));
      }
    }
    const feats = data.features || [];
    if (!feats.length) break;
    for (const f of feats) {
      if (written >= target) break;
      ws.write(JSON.stringify(f.attributes) + '\n');
      written++;
    }
    offset += feats.length;
    pages++;
    if (pages % 25 === 0 || written >= target) {
      let sz = 0; try { sz = fs.statSync(OUT).size; } catch (_) {}
      const pct = total ? ((written / target) * 100).toFixed(1) : '?';
      process.stdout.write(`[assessor] ${written.toLocaleString()}/${target.toLocaleString()} (${pct}%) · ${fmtBytes(sz)} · ${pages} pages\n`);
    }
    // Terminate ONLY on the server's real end-of-data signal. This ArcGIS layer returns SHORT pages
    // (< maxRecordCount) mid-dataset while exceededTransferLimit is still true, so a short page is NOT
    // the end — breaking on feats.length<PAGE truncated the pull at ~294k/2.43M (the old bug).
    if (data.exceededTransferLimit === false) break;  // last (partial) page; no more rows after this
  }
  await new Promise(r => ws.end(r));
  const sz = fs.statSync(OUT).size;
  const secs = ((Date.now() - t0) / 1000).toFixed(0);
  console.log(`[assessor] DONE — wrote ${written.toLocaleString()} parcels · ${fmtBytes(sz)} · ${pages} pages · ${secs}s · $0 (public open data)`);
  console.log(`[assessor] file: ${OUT}`);
  // emit a small machine-readable result line (last line) for callers/ingest
  console.log(JSON.stringify({ ok: true, rows: written, bytes: sz, file: OUT, where: WHERE, rollYear: ALL_YEARS ? 'all' : ROLL_YEAR }));
}

main().catch(e => { console.error('[assessor] FAILED:', e.message); process.exit(1); });