← back to Nationalrealestate
src/ingest/parcels/nyc_pluto.ts
153 lines
/**
* NYC MapPLUTO parcel ingest (NYC Open Data, Socrata dataset 64uk-42ks) — ~860k lots.
* PLUTO is LOT-level (no beds/baths) but carries owner name, zoning, assessed values,
* lat/lng, year built — filling the Ownership + Zoning sections LA can't. The five
* boroughs map to five county FIPS; every lot is stored under its borough's county.
* $0 open data. Paginated JSON pull ($limit/$offset), batch upsert into `parcel`.
*
* Run: npx tsx src/ingest/parcels/nyc_pluto.ts (or via engine.ts)
*/
import { query, pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { sanitizeOwnerName } from './owner_guard.ts';
import { registerSourceFieldMap } from './upsert.ts';
const DATASET = 'https://data.cityofnewyork.us/resource/64uk-42ks.json';
const SOURCE_KEY = 'nyc_pluto';
const PAGE = 50000;
// TK-50 field-level provenance: our parcel field → the exact MapPLUTO source
// attribute it maps from in the batch mapper below. improvement_value is DERIVED
// (assesstot − assessland); use_desc from landuse label (fallback bldgclass).
const FIELD_MAP: Record<string, string> = {
source_id: 'bbl', county_fips: 'borocode', city: 'borocode (borough label)',
address: 'address', zip: 'zipcode', lat: 'latitude', lng: 'longitude',
year_built: 'yearbuilt', sqft: 'bldgarea', units: 'unitstotal',
use_desc: 'landuse (label)|bldgclass', land_value: 'assessland', total_value: 'assesstot',
improvement_value: 'assesstot-assessland (derived)', tax_year: 'version',
owner_name: 'ownername', zoning: 'zonedist1',
};
// borocode -> [county_fips, borough label]
const BORO: Record<string, [string, string]> = {
'1': ['36061', 'Manhattan'], '2': ['36005', 'Bronx'], '3': ['36047', 'Brooklyn'],
'4': ['36081', 'Queens'], '5': ['36085', 'Staten Island'],
};
const LANDUSE: Record<string, string> = {
'01': 'One & Two Family Buildings', '02': 'Multi-Family Walk-Up', '03': 'Multi-Family Elevator',
'04': 'Mixed Residential & Commercial', '05': 'Commercial & Office', '06': 'Industrial & Manufacturing',
'07': 'Transportation & Utility', '08': 'Public Facilities & Institutions', '09': 'Open Space & Outdoor',
'10': 'Parking Facilities', '11': 'Vacant Land',
};
const num = (v: unknown) => { const n = Number(v); return Number.isFinite(n) ? n : null; };
async function upsertBatch(all: any[]): Promise<number> {
if (!all.length) return 0;
// Postgres caps a statement at 65535 bind params; 23 cols (19 + 4 TK-50 provenance) ->
// keep chunks <= 2500 rows (23*2500=57,500 < 65,535). Was 3000 pre-provenance (23*3000=69k overflowed → 08P01).
let done = 0;
for (let i = 0; i < all.length; i += 2500) done += await upsertChunk(all.slice(i, i + 2500));
return done;
}
async function upsertChunk(rows: any[]): Promise<number> {
if (!rows.length) return 0;
const cols = ['county_fips', 'source_id', 'address', 'norm_address', 'city', 'zip', 'lat', 'lng',
'year_built', 'sqft', 'units', 'use_desc', 'land_value', 'improvement_value', 'total_value',
'tax_year', 'owner_name', 'zoning', 'extra',
// ── TK-50 field-level provenance columns (this ingester has its own upsert path) ──
'source_url', 'source_key', 'fetched_at', 'raw_source'];
const params: unknown[] = [];
const values = rows.map((r, j) => {
const b = j * cols.length;
params.push(r.county_fips, r.source_id, r.address, r.norm_address, r.city, r.zip, r.lat, r.lng,
r.year_built, r.sqft, r.units, r.use_desc, r.land_value, r.improvement_value, r.total_value,
r.tax_year, r.owner_name, r.zoning, r.extra,
r.source_url ?? null, r.source_key ?? null, r.fetched_at ?? null, r.raw_source ?? null);
return '(' + cols.map((_, k) => `$${b + k + 1}`).join(',') + ')';
});
await query(
`INSERT INTO parcel (${cols.join(',')}) VALUES ${values.join(',')}
ON CONFLICT (county_fips, source_id) DO UPDATE SET
address=EXCLUDED.address, norm_address=EXCLUDED.norm_address, city=EXCLUDED.city, zip=EXCLUDED.zip,
lat=EXCLUDED.lat, lng=EXCLUDED.lng, year_built=EXCLUDED.year_built, sqft=EXCLUDED.sqft,
units=EXCLUDED.units, use_desc=EXCLUDED.use_desc, land_value=EXCLUDED.land_value,
improvement_value=EXCLUDED.improvement_value, total_value=EXCLUDED.total_value,
tax_year=EXCLUDED.tax_year, owner_name=EXCLUDED.owner_name, zoning=EXCLUDED.zoning, extra=EXCLUDED.extra,
source_url=EXCLUDED.source_url, source_key=EXCLUDED.source_key,
fetched_at=EXCLUDED.fetched_at, raw_source=EXCLUDED.raw_source`,
params,
);
return rows.length;
}
export async function ingestNycPluto(): Promise<{ upserted: number }> {
const runId = await openRun('nyc_pluto', DATASET);
// TK-50: run start time — the fetched_at stamped on every row this run touches.
const fetchedAt = new Date().toISOString();
let upserted = 0, offset = 0, page = 0;
try {
await registerSourceFieldMap(SOURCE_KEY, FIELD_MAP, 'NYC MapPLUTO (NYC Open Data Socrata 64uk-42ks) — owner + zoning + assessed values + structure');
for (;;) {
const url = `${DATASET}?$limit=${PAGE}&$offset=${offset}&$order=bbl`;
const res = await fetch(url, { headers: { 'accept': 'application/json' } });
if (!res.ok) throw new Error(`PLUTO fetch ${res.status} at offset ${offset}: ${url}`);
const data = await res.json() as any[];
if (!data.length) break;
const batch = data.map(d => {
const boro = BORO[String(d.borocode)];
if (!boro || !d.bbl) return null;
const total = num(d.assesstot), land = num(d.assessland);
const addr = (d.address || '').trim();
return {
county_fips: boro[0], source_id: String(d.bbl),
address: addr || null, norm_address: addr ? addr.toUpperCase() : null,
city: boro[1], zip: d.zipcode || null,
lat: num(d.latitude), lng: num(d.longitude),
year_built: num(d.yearbuilt) || null,
sqft: num(d.bldgarea), units: num(d.unitstotal),
use_desc: LANDUSE[d.landuse] || (d.bldgclass ? `Class ${d.bldgclass}` : null),
land_value: land, improvement_value: total != null && land != null ? total - land : null,
total_value: total, tax_year: d.version ? String(d.version) : null,
owner_name: sanitizeOwnerName(d.ownername),
zoning: d.zonedist1 || null,
extra: JSON.stringify({ landuse: d.landuse, bldgclass: d.bldgclass, unitsres: d.unitsres,
lotarea: d.lotarea, numfloors: d.numfloors, borough: d.borough }),
// ── TK-50 field-level provenance (record-level: one PLUTO lot → this row) ──
source_url: `${DATASET}?bbl=${encodeURIComponent(String(d.bbl))}`,
source_key: SOURCE_KEY,
fetched_at: fetchedAt,
raw_source: JSON.stringify(d),
};
}).filter(Boolean) as any[];
upserted += await upsertBatch(batch);
page++; offset += PAGE;
console.log(`[nyc_pluto] page ${page}: ${upserted} upserted`);
if (data.length < PAGE) break;
}
if (upserted < 500000) throw new Error(`only ${upserted} PLUTO rows — expected ~860k, format/endpoint drift?`);
// register all five boroughs in parcel_source
for (const [fips, label] of Object.values(BORO)) {
const cnt = await query<{ n: string }>(`SELECT COUNT(*)::text AS n FROM parcel WHERE county_fips=$1`, [fips]);
await query(
`INSERT INTO parcel_source (county_fips, source_kind, path_or_url, parcel_count, notes, loaded_at)
VALUES ($1,'pg',$2,$3,$4,NOW())
ON CONFLICT (county_fips) DO UPDATE SET source_kind='pg', path_or_url=EXCLUDED.path_or_url,
parcel_count=EXCLUDED.parcel_count, notes=EXCLUDED.notes, loaded_at=NOW()`,
[fips, DATASET, Number(cnt.rows[0].n), `NYC MapPLUTO — ${label} (owner+zoning+assessed)`],
);
}
await closeRun(runId, 'ok', { upserted, notes: `NYC PLUTO five boroughs, ${page} pages` });
console.log(`[nyc_pluto] ok: ${upserted} lots across 5 boroughs`);
return { upserted };
} catch (e: any) {
await closeRun(runId, 'failed', { notes: String(e.message || e) });
throw e;
}
}
if (import.meta.url === `file://${process.argv[1]}`) {
ingestNycPluto().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}