`scripts/ingest-assessor.js`

85 lines
#!/usr/bin/env node
// ingest-assessor.js — apply assessor-schema.sql then COPY-stream the NDJSON parcel roll into cre.assessor_parcel.
// Streams the file line-by-line (never loads the whole roll in RAM), batch-upserts via multi-row INSERT
// ON CONFLICT. Idempotent: re-running re-upserts by AIN. $0 (local Postgres).
//
// Run: node scripts/ingest-assessor.js                       # ingests data/raw/assessor-parcels-2025.ndjson
//      node scripts/ingest-assessor.js path/to/file.ndjson   # explicit file
'use strict';
const fs = require('fs');
const path = require('path');
const readline = require('readline');
const { execFileSync } = require('child_process');
const { pool } = require('./db/brokers-db');

const FILE = process.argv[2] || path.join(__dirname, '..', 'data', 'raw', 'assessor-parcels-2025.ndjson');
const BATCH = 1000;

function toDate(ms) { return (ms == null || ms === '') ? null : new Date(+ms).toISOString().slice(0, 10); }
function toInt(v) { const n = parseInt(v, 10); return Number.isFinite(n) ? n : null; }
function s(v) { v = (v == null ? '' : String(v)).trim(); return v === '' ? null : v; }

const COLS = ['ain', 'roll_year', 'property_location', 'situs_house_no', 'situs_street', 'situs_unit',
  'situs_city', 'situs_zip5', 'use_type', 'use_code', 'use_desc1', 'use_desc2', 'year_built',
  'effective_year_built', 'sqft_main', 'bedrooms', 'bathrooms', 'units', 'recording_date',
  'roll_land_value', 'roll_imp_value', 'roll_tot_land_imp', 'roll_total_value', 'net_taxable_value',
  'is_taxable', 'parcel_classification', 'admin_region', 'lat', 'lng'];

function row(a) {
  return [
    s(a.AIN), s(a.RollYear), s(a.PropertyLocation), toInt(a.SitusHouseNo), s(a.SitusStreet), s(a.SitusUnit),
    s(a.SitusCity), s(a.SitusZIP5), s(a.UseType), s(a.UseCode), s(a.UseCodeDescChar1), s(a.UseCodeDescChar2),
    toInt(a.YearBuilt), toInt(a.EffectiveYearBuilt), toInt(a.SQFTmain), toInt(a.Bedrooms), toInt(a.Bathrooms),
    toInt(a.Units), toDate(a.RecordingDate), toInt(a.Roll_LandValue), toInt(a.Roll_ImpValue),
    toInt(a.Roll_totLandImp), toInt(a.Roll_TotalValue), toInt(a.netTaxableValue), s(a.isTaxableParcel),
    s(a.ParcelClassification), s(a.AdminRegion),
    (typeof a.CENTER_LAT === 'number' ? a.CENTER_LAT : null), (typeof a.CENTER_LON === 'number' ? a.CENTER_LON : null)
  ];
}

async function flush(batch) {
  if (!batch.length) return;
  const vals = [], params = [];
  let p = 0;
  for (const r of batch) {
    vals.push('(' + COLS.map(() => '$' + (++p)).join(',') + ')');
    params.push(...r);
  }
  const upd = COLS.filter(c => c !== 'ain').map(c => `${c}=EXCLUDED.${c}`).join(',');
  await pool.query(
    `INSERT INTO assessor_parcel(${COLS.join(',')}) VALUES ${vals.join(',')}
     ON CONFLICT(ain) DO UPDATE SET ${upd}, loaded_at=now()`, params);
}

async function main() {
  if (!fs.existsSync(FILE)) { console.error('[ingest] file not found:', FILE); process.exit(1); }
  const ddl = fs.readFileSync(path.join(__dirname, 'db', 'assessor-schema.sql'), 'utf8');
  await pool.query(ddl);
  console.log('[ingest] schema applied (cre.assessor_parcel). reading', FILE);

  const rl = readline.createInterface({ input: fs.createReadStream(FILE), crlfDelay: Infinity });
  let batch = [], n = 0, bad = 0;
  for await (const line of rl) {
    if (!line.trim()) continue;
    let a; try { a = JSON.parse(line); } catch { bad++; continue; }
    if (!a.AIN) { bad++; continue; }
    batch.push(row(a));
    if (batch.length >= BATCH) { await flush(batch); n += batch.length; batch = []; if (n % 50000 === 0) console.log(`[ingest] ${n.toLocaleString()} parcels…`); }
  }
  await flush(batch); n += batch.length;
  const cnt = (await pool.query('SELECT count(*)::int c FROM assessor_parcel')).rows[0].c;
  console.log(`[ingest] DONE — ${n.toLocaleString()} rows ingested (${bad} skipped) · table now ${cnt.toLocaleString()} parcels · $0 (local)`);
  await pool.end();

  // Rebuild the prod SQLite artifact so it can NEVER drift from the DB: every ingest -> fresh file.
  // (Prod has no Postgres; data/assessor.sqlite is what /api/property-history falls back to.)
  // Non-fatal: a failed export leaves the last good file in place and warns.
  try {
    console.log('[ingest] rebuilding data/assessor.sqlite …');
    execFileSync(process.execPath, [path.join(__dirname, 'export-assessor-sqlite.js')], { stdio: 'inherit' });
  } catch (e) {
    console.error('[ingest] WARN — SQLite rebuild failed; prod file is now STALE:', e.message);
  }
}
main().catch(e => { console.error('[ingest] FAILED:', e.message); process.exit(1); });