`src/ingest/commercial/arcgis_commercial.ts`

127 lines
/**
 * Live commercial-parcel ETL from a county's public ArcGIS assessor layer.
 * The West-Coast-first successor to the LA-sqlite ETL (la_assessor.ts, stale 2024):
 * pages a county's layer (config `where` filters to commercial/industrial), classifies
 * each via the shared classifyCommercial taxonomy, and materializes into
 * `commercial_parcel`. Reusable across metros — add a CONFIG entry per county.
 *
 * Generalization (Cody-flagged, hardened): field names AND the UseType/UseDesc values
 * differ per county, so every config supplies its OWN `where`, `outFields`, and `mapRow`,
 * and the UseType is normalized case-insensitively (+ optional per-county `useTypeMap`)
 * before classifyCommercial's 'Commercial'/'Industrial' gate — so county #2 can't
 * silently classify zero.
 *
 * APN normalization: ain is reduced to digits-only (canonical) so an ArcGIS
 * '2005-002-016' matches the existing sqlite key '2005002016' (else every parcel dupes).
 * Cursor-resumable + MAX_PER_RUN cap; $0 (free county open data).
 *
 *   npm run ingest:commercial -- la-arcgis
 *   SALES_MAX_PER_RUN=200000 npm run ingest:commercial -- la-arcgis   # bulk backfill
 */
import { query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { classifyCommercial } from '../../lib/commercial_types.ts';

const MAX_PER_RUN = Number(process.env.SALES_MAX_PER_RUN || 6000);
const PAGE = 1000;

const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
const digits = (v: unknown): string | null => { const t = (s(v) || '').replace(/\D/g, ''); return t || null; };
const titleCase = (v: string | null): string | null => v ? v.charAt(0).toUpperCase() + v.slice(1).toLowerCase() : null;

interface CommRow {
  ain: string | null; useType: string | null; useDesc: string | null;
  address: string | null; city: string | null; zip: string | null;
  assessedLand: number | null; assessedImp: number | null;
  sqft: number | null; yearBuilt: number | null; units: number | null;
}
interface CommercialConfig {
  key: string; label: string; fips: string; layer: string; maxRecordCount: number;
  where: string; outFields: string;
  useTypeMap?: Record<string, string>; // county-specific value → 'Commercial'/'Industrial'
  mapRow: (a: any) => CommRow | null;
}

const CONFIGS: Record<string, CommercialConfig> = {
  la: {
    key: 'la', label: 'Los Angeles County CA (live ArcGIS)', fips: '06037',
    layer: 'https://public.gis.lacounty.gov/public/rest/services/LACounty_Cache/LACounty_Parcel/MapServer/0',
    maxRecordCount: 1000,
    where: "UseType IN ('Commercial','Industrial')",
    outFields: 'APN,SitusFullAddress,SitusCity,SitusZIP,UseType,UseDescription,YearBuilt1,SQFTmain1,Units1,Roll_LandValue,Roll_ImpValue',
    mapRow: (a) => ({
      ain: digits(a.APN), useType: s(a.UseType), useDesc: s(a.UseDescription),
      address: s(a.SitusFullAddress), city: s(a.SitusCity), zip: s(a.SitusZIP),
      assessedLand: num(a.Roll_LandValue), assessedImp: num(a.Roll_ImpValue),
      sqft: num(a.SQFTmain1), yearBuilt: num(a.YearBuilt1), units: num(a.Units1),
    }),
  },
};

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 function ingestArcgisCommercial(key: string) {
  const cfg = CONFIGS[key];
  if (!cfg) throw new Error(`unknown commercial arcgis source: ${key}`);
  const normUseType = (raw: string | null): string | null => cfg.useTypeMap?.[raw ?? ''] ?? titleCase(raw);
  return async (): Promise<{ upserted: number }> => {
    const runId = await openRun(`commercial_${cfg.key}_arcgis`, cfg.layer);
    try {
      const cur = await query<{ next_offset: number }>(`SELECT next_offset FROM ingest_cursor WHERE source=$1`, [`commercial_${cfg.key}`]);
      let offset = cur.rows[0]?.next_offset ?? 0;
      const where = encodeURIComponent(cfg.where);
      const totalR = await fetch(`${cfg.layer}/query?where=${where}&returnCountOnly=true&f=json`).then(r => r.json() as any);
      const total = totalR.count ?? 0;
      const pageSize = Math.min(PAGE, cfg.maxRecordCount); // NEVER exceed server maxRecordCount, else the loop breaks after 1 page

      const rows: any[] = [];
      let fetched = 0;
      while (fetched < MAX_PER_RUN) {
        const url = `${cfg.layer}/query?where=${where}&outFields=${encodeURIComponent(cfg.outFields)}&orderByFields=OBJECTID&resultOffset=${offset}&resultRecordCount=${pageSize}&f=json`;
        const j = await fetch(url).then(r => r.json() as any);
        const feats = j.features || [];
        if (!feats.length) { offset = 0; break; }
        for (const f of feats) {
          const m = cfg.mapRow(f.attributes); if (!m || !m.ain) continue;
          const ctype = classifyCommercial(normUseType(m.useType), m.useDesc);
          if (!ctype) continue; // not commercial after normalization + classification
          rows.push({
            county_fips: cfg.fips, ain: m.ain, address: m.address, city: m.city, zip: m.zip,
            ctype, use_desc: m.useDesc, use_class: normUseType(m.useType),
            assessed_total: (m.assessedLand || m.assessedImp) ? (m.assessedLand || 0) + (m.assessedImp || 0) : null,
            assessed_land: m.assessedLand, assessed_imp: m.assessedImp,
            roll_year: null, recording_date: null, sqft: m.sqft, year_built: m.yearBuilt, units: m.units,
          });
        }
        fetched += feats.length; offset += feats.length;
        if (feats.length < pageSize) { offset = 0; break; }
      }

      const byAin = new Map<string, any>();
      for (const r of rows) byAin.set(r.ain, r);
      const dedup = [...byAin.values()];
      let up = 0;
      for (let i = 0; i < dedup.length; i += 1000) {
        const chunk = dedup.slice(i, i + 1000);
        const params: unknown[] = [];
        const vals = chunk.map((r, j) => { const b = j * COLS.length; COLS.forEach(c => params.push(r[c] ?? null)); return '(' + COLS.map((_, k) => `$${b + k + 1}`).join(',') + ')'; });
        await query(
          `INSERT INTO commercial_parcel (${COLS.join(',')}) VALUES ${vals.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);
        up += chunk.length;
      }
      await query(`INSERT INTO ingest_cursor (source, next_offset, total) VALUES ($1,$2,$3)
                   ON CONFLICT (source) DO UPDATE SET next_offset=$2, total=$3, updated_at=NOW()`, [`commercial_${cfg.key}`, offset, total]);
      await closeRun(runId, 'ok', { upserted: up, notes: `${cfg.label}: ${up} commercial parcels (offset ${offset}/${total})` });
      console.log(`[commercial:${cfg.key}] done: ${up} commercial parcels (offset ${offset}/${total})`);
      return { upserted: up };
    } catch (e: any) {
      await closeRun(runId, 'failed', { notes: e.message });
      throw e;
    }
  };
}