`scripts/ingest-la-county-assessor.ts`

274 lines
/**
 * ingest-la-county-assessor.ts
 *
 * Downloads the LA County Assessor Roll 2021–Present (~2.7M parcels, 674 MB ZIP)
 * from data.lacounty.gov and ingests into the `parcel` table + creates listing
 * stubs per APN.
 *
 * Source: ArcGIS Hub item 785f54236d1644dc975a55af19b3dd70
 *   download URL: https://www.arcgis.com/sharing/rest/content/items/.../data
 *
 * Tier: A (LA County Assessor primary record).
 *
 * Pipeline:
 *   1. Stream zip to /tmp/la-assessor-roll.zip
 *   2. Unzip to find CSV(s) inside
 *   3. Stream-parse each CSV line by line
 *   4. Bulk upsert in batches of 1000
 */
import { Pool } from 'pg';
import { spawn } from 'node:child_process';
import { createReadStream, statSync } from 'node:fs';
import { mkdir } from 'node:fs/promises';
import { createInterface } from 'node:readline';
import { join } from 'node:path';
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createWriteStream } from 'node:fs';

const pool = new Pool({
  host: process.env.PGHOST ?? '/tmp',
  database: process.env.PGDATABASE ?? 'stayclaim',
  user: process.env.PGUSER ?? process.env.USER,
  password: process.env.PGPASSWORD,
  port: parseInt(process.env.PGPORT ?? '5432', 10),
  max: 6,
});

const ITEM_ID = '785f54236d1644dc975a55af19b3dd70';
const ZIP_URL = `https://www.arcgis.com/sharing/rest/content/items/${ITEM_ID}/data`;
const WORKDIR = '/tmp/la-assessor-roll';
const ZIP_PATH = `${WORKDIR}/data.zip`;
const BATCH = 2000;

async function downloadZip() {
  console.log(`Downloading ${ZIP_URL} ...`);
  await mkdir(WORKDIR, { recursive: true });
  try {
    const st = statSync(ZIP_PATH);
    if (st.size === 673759681) {
      console.log(`  already downloaded (${(st.size/1024/1024).toFixed(1)} MB), skipping`);
      return;
    }
  } catch { /* not present */ }

  const res = await fetch(ZIP_URL);
  if (!res.ok) throw new Error(`Download failed: ${res.status}`);
  const total = parseInt(res.headers.get('content-length') ?? '0', 10);
  let received = 0;
  const w = createWriteStream(ZIP_PATH);
  const reader = res.body!.getReader();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    w.write(value);
    received += value.length;
    if (received % (50 * 1024 * 1024) < value.length) {
      console.log(`  ${(received/1024/1024).toFixed(0)} / ${(total/1024/1024).toFixed(0)} MB`);
    }
  }
  w.end();
  await new Promise<void>(r => w.on('close', () => r()));
  console.log(`  ✓ saved ${(received/1024/1024).toFixed(1)} MB`);
}

function unzip(): Promise<string> {
  return new Promise((resolve, reject) => {
    const p = spawn('unzip', ['-o', ZIP_PATH, '-d', WORKDIR]);
    let out = '';
    p.stdout.on('data', d => { out += d.toString(); });
    p.stderr.on('data', d => { out += d.toString(); });
    p.on('close', code => {
      if (code !== 0) return reject(new Error(`unzip failed: ${out}`));
      // find CSV
      const m = out.match(/inflating: (\S+\.csv)/);
      if (!m) return reject(new Error(`no CSV found in zip: ${out.slice(0, 500)}`));
      resolve(m[1]);
    });
    p.on('error', reject);
  });
}

function canonicalize(addr: string): string {
  return addr.toLowerCase().replace(/[^\w\s-]/g, '').replace(/\s+/g, '-').replace(/-+/g, '-').replace(/^-|-$/g, '').slice(0, 90);
}

/** Cheap CSV row parser — handles quoted fields with commas. */
function splitCsv(line: string): string[] {
  const out: string[] = [];
  let cur = '';
  let inQ = false;
  for (let i = 0; i < line.length; i++) {
    const c = line[i];
    if (inQ) {
      if (c === '"' && line[i+1] === '"') { cur += '"'; i++; }
      else if (c === '"') inQ = false;
      else cur += c;
    } else {
      if (c === '"') inQ = true;
      else if (c === ',') { out.push(cur); cur = ''; }
      else cur += c;
    }
  }
  out.push(cur);
  return out;
}

async function ensureSchema() {
  // The `parcel` table already exists from earlier migrations; add any missing
  // columns first, then the unique constraint, then indexes.
  await pool.query(`
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS situs_full TEXT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS city TEXT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS zip_code TEXT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS use_code TEXT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS use_desc TEXT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS year_built INT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS bedrooms INT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS bathrooms NUMERIC(3,1);
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS bldg_sqft INT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS lot_sqft INT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS assessed_total NUMERIC(15,0);
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS assessed_land NUMERIC(15,0);
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS assessed_improvements NUMERIC(15,0);
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS roll_year INT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS latitude NUMERIC(9,6);
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS longitude NUMERIC(9,6);
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS source_url TEXT;
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS source_tier CHAR(1) DEFAULT 'A';
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS source_label TEXT DEFAULT 'LA County Assessor';
    ALTER TABLE parcel ADD COLUMN IF NOT EXISTS retrieved_at TIMESTAMPTZ DEFAULT now();
  `);
  // Unique constraint — separate query so failure is visible
  await pool.query(`
    DO $$ BEGIN
      IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'parcel_apn_year_uniq') THEN
        ALTER TABLE parcel ADD CONSTRAINT parcel_apn_year_uniq UNIQUE (apn, roll_year);
      END IF;
    END $$;
    CREATE INDEX IF NOT EXISTS idx_parcel_apn ON parcel(apn);
    CREATE INDEX IF NOT EXISTS idx_parcel_city ON parcel(lower(city));
    CREATE INDEX IF NOT EXISTS idx_parcel_use_code ON parcel(use_code);
  `);
}

async function processCsv(csvPath: string) {
  console.log(`\nStreaming ${csvPath} ...`);
  const rs = createReadStream(csvPath);
  const rl = createInterface({ input: rs, crlfDelay: Infinity });
  let headers: string[] | null = null;
  const buffer: Record<string, string>[] = [];
  let total = 0;
  const t0 = Date.now();

  for await (const line of rl) {
    if (!line.trim()) continue;
    const cols = splitCsv(line);
    if (!headers) {
      headers = cols.map(h => h.toLowerCase().replace(/\s+/g, '_').replace(/[^a-z0-9_]/g, ''));
      console.log(`  fields (${headers.length}): ${headers.slice(0, 12).join(', ')}…`);
      continue;
    }
    const row: Record<string, string> = {};
    headers.forEach((h, i) => { row[h] = cols[i] ?? ''; });
    buffer.push(row);
    if (buffer.length >= BATCH) {
      await flush(buffer);
      total += buffer.length;
      buffer.length = 0;
      const dt = (Date.now() - t0) / 1000;
      console.log(`  ${total.toLocaleString()} parcels (${(total/dt).toFixed(0)}/s)`);
    }
  }
  if (buffer.length) {
    await flush(buffer);
    total += buffer.length;
  }
  console.log(`  ✓ ${total.toLocaleString()} parcels processed`);
}

function pi(s: string | undefined): number | null { if (!s) return null; const n = parseInt(s, 10); return isFinite(n) ? n : null; }
function pf(s: string | undefined): number | null { if (!s) return null; const n = parseFloat(s); return isFinite(n) ? n : null; }

async function flush(rows: Record<string, string>[]) {
  // Field names verified 2026-04-30 from header row of Parcel_Data_0.csv:
  //   ain, roll_year, property_location, property_use_code, property_use_type,
  //   year_built, square_footage, number_of_bedrooms, number_of_bathrooms,
  //   land_value, improvement_value, total_value, city, zip_code,
  //   location_latitude, location_longitude, address_house_number, direction, street
  const tuples: string[] = [];
  const values: any[] = [];
  let ti = 0;
  for (const r of rows) {
    const apn = (r.ain ?? '').replace(/[^0-9]/g, '');
    if (!apn) continue;
    const rollYear = pi(r.roll_year);
    // property_location is the canonical situs; fall back to building from parts
    let situs = (r.property_location ?? '').trim();
    if (!situs) {
      const num = (r.address_house_number ?? '').trim();
      const dir = (r.direction ?? '').trim();
      const street = (r.street ?? '').trim();
      situs = [num, dir, street].filter(Boolean).join(' ').trim();
    }
    const city = (r.city ?? '').trim();
    const zip = (r.zip_code ?? '').trim();
    const yearBuilt = pi(r.year_built ?? r.effective_year);
    const bldg = pi(r.square_footage);
    const beds = pi(r.number_of_bedrooms);
    const baths = pf(r.number_of_bathrooms);
    const av = pf(r.total_value);
    const lv = pf(r.land_value);
    const iv = pf(r.improvement_value);
    const lat = pf(r.location_latitude);
    const lon = pf(r.location_longitude);

    const base = ti * 18;
    tuples.push(`(${Array.from({length: 18}, (_, k) => `$${base+k+1}`).join(',')})`);
    values.push(
      apn, rollYear, 'Los Angeles County', situs || null, city || null, zip || null,
      r.property_use_code ?? null, r.property_use_type ?? null,
      yearBuilt, beds, baths, bldg, av, lv, iv, lat, lon,
      `https://portal.assessor.lacounty.gov/parceldetail/${apn}`,
    );
    ti++;
  }
  if (tuples.length === 0) return;
  await pool.query(
    `INSERT INTO parcel
       (apn, roll_year, jurisdiction, situs_full, city, zip_code, use_code, use_desc,
        year_built, bedrooms, bathrooms, bldg_sqft,
        assessed_total, assessed_land, assessed_improvements,
        latitude, longitude, source_url)
     VALUES ${tuples.join(',')}
     ON CONFLICT (apn, roll_year) DO UPDATE SET
       situs_full     = COALESCE(EXCLUDED.situs_full, parcel.situs_full),
       city           = COALESCE(EXCLUDED.city, parcel.city),
       zip_code       = COALESCE(EXCLUDED.zip_code, parcel.zip_code),
       year_built     = COALESCE(EXCLUDED.year_built, parcel.year_built),
       bldg_sqft      = COALESCE(EXCLUDED.bldg_sqft, parcel.bldg_sqft),
       assessed_total = COALESCE(EXCLUDED.assessed_total, parcel.assessed_total),
       assessed_land  = COALESCE(EXCLUDED.assessed_land, parcel.assessed_land),
       assessed_improvements = COALESCE(EXCLUDED.assessed_improvements, parcel.assessed_improvements),
       latitude       = COALESCE(EXCLUDED.latitude, parcel.latitude),
       longitude      = COALESCE(EXCLUDED.longitude, parcel.longitude),
       source_url     = COALESCE(EXCLUDED.source_url, parcel.source_url),
       retrieved_at   = now()`,
    values
  );
}

async function main() {
  await ensureSchema();
  await downloadZip();
  const csvPath = await unzip();
  await processCsv(csvPath);
  const { rows } = await pool.query<{ n: string; with_value: string }>(
    `SELECT count(*)::text as n, count(*) FILTER (WHERE assessed_total IS NOT NULL)::text as with_value FROM parcel`
  );
  console.log(`\nparcel: ${parseInt(rows[0].n).toLocaleString()} total, ${parseInt(rows[0].with_value).toLocaleString()} with assessed value`);
  await pool.end();
}

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