← back to Nationalrealestate

src/ingest/parcels/franklin_oh.ts

212 lines

/**
 * Franklin County OH (Columbus, FIPS 39049) parcel ingest.
 *
 * ── DTD verdict B (2026-07-26) + its dissent's guardrails ──────────────────
 * Franklin's ONLY open bulk surface is the auditor's GIS directory
 * (apps.franklincountyauditor.com/GIS_Shapefiles/YYYY/MM/) — the www data pages
 * are WAF-walled. The parcel POLYGON shapefile there is geometry-only (PARCELID +
 * Shape only). The rich CAMA attributes live ONLY inside the 603MB monthly
 * FileGeoDatabase (.gdb), which requires GDAL/ogr2ogr to read.
 *
 * This adapter is therefore the ONE documented exception to the "re-run on the
 * remote" model: it depends on GDAL (ogr2ogr), which is installed locally but
 * NOT on the Kamatera prod box. Run it on a GDAL-having machine; deploy the rows
 * to prod via pg_dump-subset -> restore (see FRANKLIN-RUNBOOK.md). Everything
 * downstream of the GDAL extract (dedupe, upsert, parcel_source registry) uses
 * the identical shared contract as every other county, so the "snowflake" is
 * confined to the single extraction step.
 *
 * Attribute source = the TaxParcelSale POINT layer (159,988 sale rows across
 * 122,047 distinct parcels — one row per recorded sale). We keep the LATEST sale
 * per parcel (current owner = its grantee) and lead with these 122k RICH parcels
 * (address / owner / use-class / res sqft / last sale + lat-lng) rather than
 * diluting the county with the ~341k geometry-only parcels (that full-coverage
 * pass is a documented follow-up). No assessed value / year-built / beds-baths in
 * this layer — recorded null, noted in extra.
 *
 * Run: FRANKLIN_GDB=/path/to/xxx.gdb npm run ingest:parcels franklin
 *   (if FRANKLIN_GDB unset, downloads the latest monthly .gdb zip — 603MB.)
 */
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { createReadStream } from 'node:fs';
import { mkdtemp, rm } from 'node:fs/promises';
import { createInterface } from 'node:readline';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';

const execFileP = promisify(execFile);
const FIPS = '39049';
const SOURCE_KEY = 'franklin_oh';
const GIS_BASE = 'https://apps.franklincountyauditor.com/GIS_Shapefiles';
const LAYER = 'TaxParcelSale';
// TK-50 field-level provenance: our parcel field → the exact TaxParcelSale layer
// attribute it maps from. owner is the LATEST recorded sale's grantee; sqft is the
// above-grade residential floor area. No assessed value/year/beds-baths in layer.
const FIELD_MAP: Record<string, string> = {
  source_id: 'PARCELID', address: 'SITEADDRESS', zip: 'ZIPCD', sqft: 'RESFLRAREA_AG',
  use_desc: 'CLASSDSCRP', owner_name: 'GranteeName1 (grantee of latest recorded sale)',
  last_sale_date: 'SALEDATE', last_sale_price: 'SalePrice (nulled on multi-parcel bulk deeds)',
};

const num = (v: unknown) => { const n = Number(v); return Number.isFinite(n) && n !== 0 ? n : null; };
/** "2021/12/02 00:00:00+00" -> "2021-12-02"; anything unparseable -> null */
const toIsoDate = (v: unknown): string | null => {
  const m = String(v ?? '').match(/(\d{4})[/-](\d{2})[/-](\d{2})/);
  // drop future/absurd bad-entry dates so they never reach parcel.last_sale_date (shared guard)
  return m ? sanitizeEventDate(`${m[1]}-${m[2]}-${m[3]}`) : null;
};
/** Full sortable timestamp "2021-12-02T00:00:01" — the SUB-DAY seconds matter:
 *  the county records a same-day $0 grantor-side transfer at HH:00 and the real
 *  arms-length sale one second later, so date-only dedupe picks the wrong row. */
const toTimestamp = (v: unknown): string => {
  const m = String(v ?? '').match(/(\d{4})[/-](\d{2})[/-](\d{2})[ T](\d{2}):(\d{2}):(\d{2})/);
  return m ? `${m[1]}-${m[2]}-${m[3]}T${m[4]}:${m[5]}:${m[6]}` : (toIsoDate(v) ?? '');
};

async function ogr2ogrExists(): Promise<boolean> {
  try { await execFileP('ogr2ogr', ['--version']); return true; } catch { return false; }
}

/** Find the newest YYYY/MM/*_FileGeoDataBase.zip in the auditor GIS directory. */
async function latestGdbUrl(): Promise<string> {
  const links = async (path: string): Promise<string[]> => {
    const html = await (await fetch(`${GIS_BASE}/${path}`)).text();
    return [...html.matchAll(/href="([^"]+)"/gi)].map(m => m[1]);
  };
  const years = (await links('')).map(h => h.match(/\/(\d{4})\/$/)?.[1]).filter(Boolean).sort().reverse();
  for (const y of years) {
    const months = (await links(`${y}/`)).map(h => h.match(new RegExp(`/${y}/(\\d{2})/$`))?.[1]).filter(Boolean).sort().reverse();
    for (const mo of months) {
      const zip = (await links(`${y}/${mo}/`)).find(h => /FileGeoDataBase\.zip$/i.test(h));
      if (zip) return zip.startsWith('http') ? zip : `https://apps.franklincountyauditor.com${zip}`;
    }
  }
  throw new Error('no FileGeoDataBase.zip found in Franklin GIS directory');
}

/** Resolve a local .gdb dir: env override, else download+unzip the latest monthly zip. */
async function resolveGdb(work: string): Promise<{ gdb: string; provenance: string }> {
  if (process.env.FRANKLIN_GDB) return { gdb: process.env.FRANKLIN_GDB, provenance: `local .gdb (${process.env.FRANKLIN_GDB})` };
  const url = await latestGdbUrl();
  console.log(`[franklin] downloading ${url} (~603MB)…`);
  const zip = join(work, 'fcgdb.zip');
  await execFileP('curl', ['-s', '-m', '900', '-o', zip, url]);
  await execFileP('unzip', ['-o', '-q', zip, '-d', join(work, 'gdb')]);
  const { stdout } = await execFileP('bash', ['-lc', `find ${join(work, 'gdb')} -maxdepth 4 -name '*.gdb' -type d | head -1`]);
  const gdb = stdout.trim();
  if (!gdb) throw new Error('unzip produced no .gdb');
  return { gdb, provenance: url };
}

export async function ingestFranklin(): Promise<{ upserted: number }> {
  if (!(await ogr2ogrExists())) {
    throw new Error('ogr2ogr (GDAL) not found — Franklin needs GDAL to read the .gdb. Run on a GDAL machine; see FRANKLIN-RUNBOOK.md.');
  }
  const work = await mkdtemp(join(tmpdir(), 'franklin-'));
  const runId = await openRun('parcel_franklin_oh', `${GIS_BASE}/ (TaxParcelSale)`);
  // TK-50: run start time — the fetched_at stamped on every row this run touches.
  const fetchedAt = new Date().toISOString();
  try {
    await registerSourceFieldMap(SOURCE_KEY, FIELD_MAP, 'Franklin County OH Auditor FileGeoDatabase — TaxParcelSale layer (GDAL-extracted)');
    const { gdb, provenance } = await resolveGdb(work);
    // GDAL extract: TaxParcelSale -> WGS84 NDJSON (point geom carries lat/lng)
    const ndjson = join(work, 'sales.ndjson');
    console.log(`[franklin] ogr2ogr extracting ${LAYER} -> WGS84 NDJSON`);
    await execFileP('ogr2ogr', ['-f', 'GeoJSONSeq', ndjson, gdb, LAYER, '-t_srs', 'EPSG:4326',
      '-select', 'PARCELID,SITEADDRESS,ZIPCD,CLASSDSCRP,GranteeName1,GrantorName1,SALEDATE,SalePrice,SaleType,ParcelCount,RESFLRAREA_AG,RESFLRAREA_BG',
    ], { maxBuffer: 1 << 30 });

    // Stream NDJSON, keep the LATEST sale per parcel. Rank by FULL timestamp
    // (sub-day seconds disambiguate same-date $0 transfer vs real sale), tie
    // broken by higher SalePrice so the arms-length sale wins a true tie.
    const best = new Map<string, ParcelUpsertRow>();
    const bestRank = new Map<string, string>(); // "<timestamp>|<zero-padded price>"
    let seen = 0, minSale = '9999', maxSale = '0000';
    const rl = createInterface({ input: createReadStream(ndjson), crlfDelay: Infinity });
    for await (const line of rl) {
      if (!line.trim()) continue;
      let f: any; try { f = JSON.parse(line); } catch { continue; }
      const p = f.properties || {};
      const pid = (p.PARCELID || '').trim();
      if (!pid) continue;
      seen++;
      const saleDate = toIsoDate(p.SALEDATE);
      if (saleDate) { if (saleDate < minSale) minSale = saleDate; if (saleDate > maxSale) maxSale = saleDate; }
      const priceRaw = Number(p.SalePrice) || 0;
      const rank = `${toTimestamp(p.SALEDATE)}|${String(Math.max(0, Math.min(priceRaw, 1e12))).padStart(13, '0')}`;
      const prev = bestRank.get(pid);
      if (prev !== undefined && rank <= prev) continue; // keep the higher-ranked (later, then pricier) sale
      const coords = f.geometry?.coordinates;
      const lng = Array.isArray(coords) ? num(coords[0]) : null;
      const lat = Array.isArray(coords) ? num(coords[1]) : null;
      const addr = (p.SITEADDRESS || '').trim();
      const norm = addr ? normAddress(addr) : null;
      const sqft = num(p.RESFLRAREA_AG);
      const parcelCount = Number(p.ParcelCount) || 1;
      const bulk = parcelCount > 1;
      // bulk sale rows carry the FULL bundle price on every parcel — can't allocate
      // per-parcel, so null the price and flag it rather than ship an inflated value.
      const salePrice = bulk ? null : num(p.SalePrice);
      best.set(pid, {
        county_fips: FIPS, source_id: pid,
        address: norm, norm_address: norm,
        city: null, zip: (p.ZIPCD || '').trim() || null,
        lat, lng,
        year_built: null, sqft, beds: null, baths: null, units: null,
        use_desc: (p.CLASSDSCRP || '').trim() || null,
        land_value: null, improvement_value: null, total_value: null, tax_year: null,
        owner_name: (p.GranteeName1 || '').trim() || null,
        zoning: null,
        last_sale_date: saleDate, last_sale_price: salePrice,
        extra: JSON.stringify({
          class_desc: (p.CLASSDSCRP || '').trim() || undefined,
          res_sqft_below_grade: num(p.RESFLRAREA_BG) || undefined,
          seller: (p.GrantorName1 || '').trim() || undefined,
          sale_type: (p.SaleType || '').trim() || undefined,
          bulk_sale: bulk || undefined,
          bulk_parcel_count: bulk ? parcelCount : undefined,
          owner_source: p.GranteeName1 ? 'grantee of latest recorded sale (see sale_window in coverage), Franklin County Auditor TaxParcelSale' : undefined,
          note: 'owner/sale from the open TaxParcelSale window (see coverage notes); no assessed value / year-built / beds-baths in this layer',
        }),
        // ── TK-50 field-level provenance (record-level: latest sale row → this parcel) ──
        // The bulk www data pages are WAF-walled, so the addressable public record is
        // the county auditor's parcel-detail page keyed by PARCELID.
        sourceKey: SOURCE_KEY,
        sourceUrl: `https://audr-apps.franklincountyohio.gov/Parcel/${encodeURIComponent(pid)}`,
        fetchedAt,
        rawSource: JSON.stringify(p),
      });
      bestRank.set(pid, rank);
    }
    rl.close();

    const rows = [...best.values()];
    if (rows.length < 90000) throw new Error(`only ${rows.length} Franklin parcels parsed (expected ~122k) — layer/schema drift?`);
    let upserted = 0;
    for (let i = 0; i < rows.length; i += 5000) {
      upserted += await upsertParcels(rows.slice(i, i + 5000));
      console.log(`[franklin] ${upserted}/${rows.length} upserted`);
    }

    const n = await registerParcelSource(FIPS, provenance,
      `Franklin County OH Auditor FileGeoDatabase (TaxParcelSale layer, GDAL-extracted) — ${rows.length} parcels with latest-sale attributes: site address, owner (latest grantee), use class, res sqft, last sale date+price, lat/lng from ${seen} sale rows. SALE WINDOW ${minSale}..${maxSale}: owner reflects the latest sale in that window (the open layer's fixed range), NOT current-day ownership. No assessed value/year-built/beds-baths in this open layer; ~341k geometry-only parcels not yet ingested (full-coverage follow-up).`);
    await closeRun(runId, 'ok', { upserted, notes: `Franklin OH: ${upserted} parcels from ${seen} sale rows` });
    console.log(`[franklin] ok: ${upserted} parcels (registry ${n})`);
    return { upserted };
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: String(e.message || e).slice(0, 500) });
    throw e;
  } finally {
    await rm(work, { recursive: true, force: true }).catch(() => {});
  }
}

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