`src/ingest/commercial/la_assessor.ts`

105 lines
/**
 * LA commercial layer ETL: read the CRCP assessor sqlite (read-only, 2.4M rows),
 * keep the Commercial + Industrial parcels (~154k), classify each into a canonical
 * commercial type, parse city/zip out of property_location, and materialize into the
 * pg `commercial_parcel` table so the feed / explorer / rankings query fast indexes.
 *
 * LA carries NO sale price and NO owner (assessor roll limits) — only assessed values
 * + last recording_date. We never fabricate those; prices come from the listings crawl.
 *
 * Run: npm run ingest:commercial -- la
 */
import { join } from 'node:path';
import { homedir } from 'node:os';
import { existsSync } from 'node:fs';
import { createRequire } from 'node:module';
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { classifyCommercial } from '../../lib/commercial_types.ts';

const require = createRequire(import.meta.url);
const FIPS = '06037';
const SQLITE = process.env.ASSESSOR_SQLITE || join(homedir(), 'Projects', 'commercialrealestate', 'data', 'assessor.sqlite');

interface Row {
  ain: string; property_location: string | null; year_built: number | null;
  sqft_main: number | null; units: number | null; use_desc1: string | null; use_desc2: string | null;
  roll_total_value: number | null; roll_land_value: number | null; roll_imp_value: number | null;
  roll_year: string | null; recording_date: string | null;
}

/** "8407 FALLBROOK AVE  WEST HILLS CA  91304" -> {street, city, zip}; nulls if unparseable. */
function parseLoc(loc: string | null): { street: string | null; city: string | null; zip: string | null } {
  if (!loc) return { street: null, city: null, zip: null };
  const m = loc.match(/^(.*?)\s{2,}(.+?)\s+CA\s+(\d{5})/);
  if (m) return { street: m[1].trim() || null, city: m[2].trim() || null, zip: m[3] };
  return { street: loc.trim() || null, city: null, zip: null };
}

const isoDate = (d: string | null) => (d && /^\d{4}-\d{2}-\d{2}$/.test(d.slice(0, 10)) ? d.slice(0, 10) : null);
const COLS = ['county_fips', 'ain', 'address', 'city', 'zip', 'ctype', 'use_desc', 'use_class',
  'assessed_total', 'assessed_land', 'assessed_imp', 'roll_year', 'recording_date', 'sqft', 'year_built', 'units'];

export async function ingestLaCommercial(): Promise<{ upserted: number }> {
  const runId = await openRun('commercial_la_assessor', SQLITE);
  try {
    if (!existsSync(SQLITE)) throw new Error(`LA assessor sqlite not found at ${SQLITE}`);
    const Database = require('better-sqlite3');
    const db = new Database(SQLITE, { readonly: true, fileMustExist: true });

    const rows = db.prepare(
      `SELECT ain, property_location, year_built, sqft_main, units, use_desc1, use_desc2,
              roll_total_value, roll_land_value, roll_imp_value, roll_year, recording_date
         FROM assessor_parcel WHERE use_desc1 IN ('Commercial','Industrial')`,
    ).all() as Row[];
    console.log(`[la-commercial] read ${rows.length} commercial+industrial parcels from sqlite`);

    // fresh authoritative load for this county
    await query(`DELETE FROM commercial_parcel WHERE county_fips = $1`, [FIPS]);

    let upserted = 0, skipped = 0;
    const batch: unknown[][] = [];
    const flush = async () => {
      if (!batch.length) return;
      const params: unknown[] = [];
      const values = batch.map((r, j) => {
        const b = j * COLS.length;
        params.push(...r);
        return '(' + COLS.map((_, k) => `$${b + k + 1}`).join(',') + ')';
      });
      await query(
        `INSERT INTO commercial_parcel (${COLS.join(',')}) VALUES ${values.join(',')}
         ON CONFLICT (county_fips, ain) DO UPDATE SET
           ${COLS.filter(c => c !== 'county_fips' && c !== 'ain').map(c => `${c}=EXCLUDED.${c}`).join(', ')},
           updated_at = now()`,
        params,
      );
      upserted += batch.length;
      batch.length = 0;
    };

    for (const r of rows) {
      const ctype = classifyCommercial(r.use_desc1, r.use_desc2);
      if (!ctype) { skipped++; continue; }
      const { street, city, zip } = parseLoc(r.property_location);
      batch.push([FIPS, r.ain, street, city, zip, ctype, (r.use_desc2 || r.use_desc1), r.use_desc1,
        r.roll_total_value, r.roll_land_value, r.roll_imp_value, r.roll_year, isoDate(r.recording_date),
        r.sqft_main, r.year_built, r.units]);
      if (batch.length >= 1000) await flush();
    }
    await flush();
    if (upserted < 100000) throw new Error(`only ${upserted} commercial parcels loaded — expected ~154k, sqlite drift?`);

    await closeRun(runId, 'ok', { upserted, skipped, notes: `LA commercial: ${upserted} parcels materialized (${skipped} unclassified)` });
    console.log(`[la-commercial] ok: ${upserted} parcels into commercial_parcel (${skipped} skipped)`);
    return { upserted };
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: String(e.message || e) });
    throw e;
  }
}

if (import.meta.url === `file://${process.argv[1]}`) {
  ingestLaCommercial().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}