← 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); });
}