`src/ingest/commercial/from_parcel.ts`

113 lines
/**
 * Materialize commercial_parcel from the EXISTING parcel table — $0, NO new source.
 * For counties whose parcel.use_desc is already a plain description (King County WA:
 * 'Office Building', 'Warehouse', 'Retail Store', …). A SQL pre-filter narrows to
 * commercial-keyword candidates (so we don't load the whole county), then
 * classifyCommercialByDesc types each precisely; non-commercial rows are skipped.
 * APN normalized to digits. Idempotent upsert on (county_fips, ain).
 *
 *   npm run ingest:commercial -- king   # King County WA (53033) from loaded parcels
 */
import { query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { classifyCommercialByDesc, type CommercialType } from '../../lib/commercial_types.ts';

const digits = (v: unknown): string | null => { const t = (v == null ? '' : String(v)).replace(/\D/g, ''); return t || null; };

// Broad SQL pre-filter covering the classifier's keywords — precise typing happens in JS.
const KEYWORDS = 'commercial|office|ofc|professional|bank|savings|warehous|warehse|whse|distribution|storage|manufactur|industrial|store|shop|supermarket|department|restaurant|cocktail|hotel|motel|hospitality|lodging|parking|studio|motion picture|television|radio|retail|wholesale|outlet|nursery|greenhouse|kennel|laundry|repair|service station|auto|recreation';

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 ingestCommercialFromParcel(fips: string, label: string) {
  return async (): Promise<{ upserted: number }> => {
    const runId = await openRun(`commercial_${fips}_fromparcel`, 'parcel-table');
    try {
      const r = await query<any>(
        `SELECT county_fips, source_id, address, city, zip, use_desc, total_value, sqft, year_built, units
           FROM parcel WHERE county_fips = $1 AND use_desc ~* $2`, [fips, KEYWORDS]);
      const byAin = new Map<string, any>();
      for (const p of r.rows) {
        const ctype = classifyCommercialByDesc(p.use_desc);
        if (!ctype) continue;
        const ain = digits(p.source_id); if (!ain) continue;
        byAin.set(ain, {
          county_fips: p.county_fips, ain, address: p.address, city: p.city, zip: p.zip,
          ctype, use_desc: p.use_desc, use_class: null,
          assessed_total: p.total_value, assessed_land: null, assessed_imp: null,
          roll_year: null, recording_date: null, sqft: p.sqft, year_built: p.year_built, units: p.units,
        });
      }
      const rows = [...byAin.values()];
      let up = 0;
      for (let i = 0; i < rows.length; i += 1000) {
        const chunk = rows.slice(i, i + 1000);
        const params: unknown[] = [];
        const vals = chunk.map((row, j) => { const b = j * COLS.length; COLS.forEach(c => params.push(row[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 closeRun(runId, 'ok', { upserted: up, notes: `${label}: ${up} commercial parcels materialized from parcel table (${r.rows.length} candidates scanned)` });
      console.log(`[commercial:${label}] done: ${up} commercial parcels from ${r.rows.length} candidates`);
      return { upserted: up };
    } catch (e: any) {
      await closeRun(runId, 'failed', { notes: e.message });
      throw e;
    }
  };
}

/**
 * Coded-county variant: for counties whose parcel.use_desc is a CODE (e.g. NYC
 * 'Class O6', building-class letter O = Office). Pass an authoritative `classify`
 * (code string -> CommercialType|null) and a SQL `filter` regex that narrows to the
 * commercial-relevant codes so we don't scan the whole county. No guessing — the
 * classify fn must come from the county's published code table (Cook lesson).
 */
export function ingestCommercialFromParcelCoded(
  fips: string, label: string, filter: string, classify: (useDesc: string) => CommercialType | null,
) {
  return async (): Promise<{ upserted: number }> => {
    const runId = await openRun(`commercial_${fips}_coded`, 'parcel-table');
    try {
      const r = await query<any>(
        `SELECT county_fips, source_id, address, city, zip, use_desc, total_value, sqft, year_built, units
           FROM parcel WHERE county_fips = $1 AND use_desc ~* $2`, [fips, filter]);
      const byAin = new Map<string, any>();
      for (const p of r.rows) {
        const ctype = classify(p.use_desc || '');
        if (!ctype) continue;
        const ain = digits(p.source_id); if (!ain) continue;
        byAin.set(ain, {
          county_fips: p.county_fips, ain, address: p.address, city: p.city, zip: p.zip,
          ctype, use_desc: p.use_desc, use_class: null,
          assessed_total: p.total_value, assessed_land: null, assessed_imp: null,
          roll_year: null, recording_date: null, sqft: p.sqft, year_built: p.year_built, units: p.units,
        });
      }
      const rows = [...byAin.values()];
      let up = 0;
      for (let i = 0; i < rows.length; i += 1000) {
        const chunk = rows.slice(i, i + 1000);
        const params: unknown[] = [];
        const vals = chunk.map((row, j) => { const b = j * COLS.length; COLS.forEach(c => params.push(row[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 closeRun(runId, 'ok', { upserted: up, notes: `${label}: ${up} commercial parcels (coded, ${r.rows.length} candidates)` });
      console.log(`[commercial:${label}] done: ${up} commercial parcels from ${r.rows.length} coded candidates`);
      return { upserted: up };
    } catch (e: any) {
      await closeRun(runId, 'failed', { notes: e.message });
      throw e;
    }
  };
}