← back to Nationalrealestate
src/ingest/parcels/king_wa.ts
222 lines
/**
* King County WA (Seattle, FIPS 53033) parcel ingest from the assessor's open
* EXTR bulk extracts (aqua.kingcounty.gov/extranet/assessor — no auth, $0):
* Parcel.zip -> EXTR_Parcel.csv (628k universe: zoning, district, present use)
* Residential Building.zip -> EXTR_ResBldg.csv (533k: address, beds/baths, sqft, yr built)
* Real Property Account.zip -> EXTR_RPAcct_NoName.csv (740k: appraised land/imps values, tax year)
* Real Property Sales.zip -> EXTR_RPSale.csv (2.4M: full deed history w/ prices — REAL sale records)
* Lookup.zip -> EXTR_LookUp.csv (LUType 102 = PresentUse descriptions)
*
* The account extract is the "_NoName" variant (taxpayer names withheld), so
* owner_name is the BUYER on the most recent recorded deed (public record),
* labeled as such in extra.owner_source.
*
* Run: npm run ingest:parcels king (NODE_OPTIONS=--max-old-space-size=6144 recommended)
*/
import { existsSync } from 'node:fs';
import { join } from 'node:path';
import { execFileSync } from 'node:child_process';
import { pool } from '../../../db/pool.ts';
import { openRun, closeRun, DOWNLOADS, download } from '../run.ts';
import { csvObjects } from './csv.ts';
import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';
import { sanitizeSalePrice } from './price_guard.ts';
const FIPS = '53033';
const SOURCE_KEY = 'king_wa';
const BASE = 'https://aqua.kingcounty.gov/extranet/assessor/';
// TK-50 field-level provenance: our parcel field → the source attribute + which
// EXTR extract it was joined from (by PIN = major+minor). This county is a JOIN of
// 4 extracts, so we name the file each attribute comes from.
const FIELD_MAP: Record<string, string> = {
source_id: 'EXTR_Parcel.major+minor (PIN)', city: 'EXTR_Parcel.districtname',
zoning: 'EXTR_Parcel.currentzoning', use_desc: 'EXTR_Parcel.presentuse (→ EXTR_LookUp LUType 102)',
address: 'EXTR_ResBldg.buildingnumber+directionprefix+streetname+streettype+directionsuffix',
zip: 'EXTR_ResBldg.zipcode', beds: 'EXTR_ResBldg.bedrooms',
baths: 'EXTR_ResBldg.bathfullcount+bath3qtrcount+bathhalfcount', sqft: 'EXTR_ResBldg.sqfttotliving',
year_built: 'EXTR_ResBldg.yrbuilt', units: 'EXTR_ResBldg.nbrlivingunits',
land_value: 'EXTR_RPAcct_NoName.apprlandval', improvement_value: 'EXTR_RPAcct_NoName.apprimpsval',
tax_year: 'EXTR_RPAcct_NoName.billyr',
owner_name: 'EXTR_RPSale.buyername (latest recorded deed buyer; names file withheld)',
last_sale_date: 'EXTR_RPSale.documentdate (newest priced sale)',
last_sale_price: 'EXTR_RPSale.saleprice (newest priced sale)',
};
const DIR = join(DOWNLOADS, 'king');
const FILES: [zip: string, csv: string][] = [
['Parcel.zip', 'EXTR_Parcel.csv'],
['Residential%20Building.zip', 'EXTR_ResBldg.csv'],
['Real%20Property%20Account.zip', 'EXTR_RPAcct_NoName.csv'],
['Real%20Property%20Sales.zip', 'EXTR_RPSale.csv'],
['Lookup.zip', 'EXTR_LookUp.csv'],
];
const num = (v: string | undefined) => { const n = Number(v); return Number.isFinite(n) && n !== 0 ? n : null; };
const num0 = (v: string | undefined) => { const n = Number(v); return Number.isFinite(n) ? n : 0; };
const pinOf = (o: Record<string, string>) => o.major.padStart(6, '0') + o.minor.padStart(4, '0');
/** MM/DD/YYYY -> YYYY-MM-DD (null if unparseable, future, or absurd) */
function isoDate(d: string): string | null {
const m = d.match(/^(\d{2})\/(\d{2})\/(\d{4})$/);
// drop future/absurd bad-entry dates so they never reach parcel.last_sale_date (shared guard)
return m ? sanitizeEventDate(`${m[3]}-${m[1]}-${m[2]}`) : null;
}
interface Sale { date: string; price: number; buyer: string; seller: string; instrument: string }
async function ensureFiles() {
for (const [zip, csv] of FILES) {
if (existsSync(join(DIR, csv))) continue;
const { path } = await download(BASE + zip, 'king/' + zip.replace(/%20/g, '_'), { reuseCached: true });
execFileSync('unzip', ['-o', '-q', path, '-d', DIR]);
}
}
export async function ingestKing(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_king_wa', BASE);
// 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, 'King County WA assessor EXTR bulk extracts (Parcel/ResBldg/RPAcct/RPSale/LookUp joined by PIN)');
await ensureFiles();
// PresentUse code -> description
const useDesc = new Map<string, string>();
for await (const o of csvObjects(join(DIR, 'EXTR_LookUp.csv'), 'latin1')) {
if (o.lutype.trim() === '102') useDesc.set(o.luitem.trim(), o.ludescription.trim());
}
// Residential buildings: address + physical facts (bldg 1 wins; units summed)
interface Res { address: string; zip: string; beds: number | null; baths: number | null;
sqft: number | null; year_built: number | null; units: number; stories: number | null; bldg: number }
const res = new Map<string, Res>();
for await (const o of csvObjects(join(DIR, 'EXTR_ResBldg.csv'), 'latin1')) {
const pin = pinOf(o);
const bldg = num0(o.bldgnbr);
const addr = [o.buildingnumber, o.fraction, o.directionprefix, o.streetname, o.streettype, o.directionsuffix]
.map(s => (s || '').trim()).filter(Boolean).join(' ');
const units = num0(o.nbrlivingunits);
const prev = res.get(pin);
if (prev) {
prev.units += units;
if (bldg < prev.bldg) Object.assign(prev, { bldg, address: addr || prev.address });
continue;
}
res.set(pin, {
address: addr, zip: (o.zipcode || '').trim() || '', bldg,
beds: num(o.bedrooms),
baths: (num0(o.bathfullcount) + 0.75 * num0(o.bath3qtrcount) + 0.5 * num0(o.bathhalfcount)) || null,
sqft: num(o.sqfttotliving), year_built: num(o.yrbuilt),
units: units || 1, stories: num(o.stories),
});
}
console.log(`[king] resbldg: ${res.size} pins`);
// Account values: latest bill year per pin (+ taxpayer mailing address, name withheld)
interface Acct { land: number | null; imps: number | null; year: number; mailing: string | null }
const acct = new Map<string, Acct>();
for await (const o of csvObjects(join(DIR, 'EXTR_RPAcct_NoName.csv'), 'latin1')) {
const pin = pinOf(o);
const year = num0(o.billyr);
const prev = acct.get(pin);
if (prev && prev.year >= year) continue;
const mailing = [o.addrline, o.citystate, o.zipcode].map(s => (s || '').trim()).filter(Boolean).join(', ');
acct.set(pin, { land: num(o.apprlandval), imps: num(o.apprimpsval), year, mailing: mailing || null });
}
console.log(`[king] rpacct: ${acct.size} pins`);
// Sales: keep 10 most recent priced sales per pin + latest deed buyer (any price)
const sales = new Map<string, Sale[]>();
const lastDeed = new Map<string, { date: string; buyer: string }>();
let saleRows = 0;
for await (const o of csvObjects(join(DIR, 'EXTR_RPSale.csv'), 'latin1')) {
saleRows++;
const pin = pinOf(o);
const date = isoDate(o.documentdate.trim());
if (!date) continue;
const buyer = o.buyername.trim();
if (buyer) {
const prev = lastDeed.get(pin);
if (!prev || date > prev.date) lastDeed.set(pin, { date, buyer });
}
// Route through the shared price guard: King's EXTR_RPSale.csv carries
// occasional NEGATIVE saleprice values (source data-entry artifacts) that
// the old `num()` (rejected only 0) let through into last_sale_price.
const price = sanitizeSalePrice(o.saleprice);
if (!price) continue;
const s: Sale = { date, price, buyer, seller: o.sellername.trim(), instrument: o.saleinstrument.trim() };
const list = sales.get(pin);
if (!list) { sales.set(pin, [s]); continue; }
list.push(s);
if (list.length > 14) { list.sort((a, b) => b.date.localeCompare(a.date)); list.length = 10; }
}
for (const list of sales.values()) list.sort((a, b) => b.date.localeCompare(a.date));
console.log(`[king] rpsale: ${saleRows} rows, ${sales.size} pins with priced sales`);
// Universe pass: EXTR_Parcel drives; join everything, batch upsert
let upserted = 0;
let batch: ParcelUpsertRow[] = [];
for await (const o of csvObjects(join(DIR, 'EXTR_Parcel.csv'), 'latin1')) {
const pin = pinOf(o);
const r = res.get(pin);
const a = acct.get(pin);
const sl = (sales.get(pin) || []).slice(0, 10);
const deed = lastDeed.get(pin);
const land = a?.land ?? null, imps = a?.imps ?? null;
const city = (o.districtname || '').trim() || null;
const address = r?.address ? normAddress(r.address) : null;
batch.push({
county_fips: FIPS, source_id: pin,
address, norm_address: address, city, zip: r?.zip || null,
lat: null, lng: null, // EXTR carries no coordinates; county centroid at render
year_built: r?.year_built ?? null, sqft: r?.sqft ?? null,
beds: r?.beds ?? null, baths: r?.baths ?? null, units: r?.units ?? null,
use_desc: useDesc.get((o.presentuse || '').trim()) || (o.presentuse ? `Use ${o.presentuse.trim()}` : null),
land_value: land, improvement_value: imps,
total_value: land != null || imps != null ? (land ?? 0) + (imps ?? 0) : null,
tax_year: a ? String(a.year) : null,
owner_name: deed?.buyer ?? null,
zoning: (o.currentzoning || '').trim() || null,
last_sale_date: sl[0]?.date ?? null, last_sale_price: sl[0]?.price ?? null,
extra: JSON.stringify({
sales: sl, prop_type: o.proptype, prop_name: (o.propname || '').trim() || undefined,
sqft_lot: num(o.sqftlot), stories: r?.stories ?? undefined,
taxpayer_mailing: a?.mailing ?? undefined,
owner_source: deed ? 'buyer on most recent recorded deed (EXTR_RPSale)' : undefined,
}),
// ── TK-50 field-level provenance (record-level: this PIN's joined extract row) ──
// KingCo per-parcel assessor detail page (addressable by PIN); raw_source
// captures the actual source attributes we mapped from across the 4 EXTR files.
sourceKey: SOURCE_KEY,
sourceUrl: `https://blue.kingcounty.com/Assessor/eRealProperty/Detail.aspx?ParcelNbr=${pin}`,
fetchedAt,
rawSource: JSON.stringify({
parcel: { major: o.major, minor: o.minor, presentuse: o.presentuse, districtname: o.districtname,
currentzoning: o.currentzoning, proptype: o.proptype, sqftlot: o.sqftlot },
resbldg: r ? { address: r.address, zipcode: r.zip, bedrooms: r.beds, baths: r.baths,
sqfttotliving: r.sqft, yrbuilt: r.year_built, nbrlivingunits: r.units } : null,
rpacct: a ? { apprlandval: a.land, apprimpsval: a.imps, billyr: a.year } : null,
rpsale_latest: sl[0] ?? null, deed_owner: deed ?? null,
}),
});
if (batch.length >= 2000) { upserted += await upsertParcels(batch); batch = []; }
if (upserted % 100000 < 2000 && upserted) console.log(`[king] ${upserted} upserted`);
}
upserted += await upsertParcels(batch);
if (upserted < 400000) throw new Error(`only ${upserted} King parcels — expected ~628k, extract drift?`);
const n = await registerParcelSource(FIPS, BASE,
'King County WA EXTR extracts — address/beds/baths/sqft (ResBldg), appraised values (RPAcct), zoning+use (Parcel), REAL sale history w/ prices (RPSale; owner = latest deed buyer, names file withheld)');
await closeRun(runId, 'ok', { upserted, notes: `King WA: ${upserted} parcels, ${sales.size} with priced sale history` });
console.log(`[king] ok: ${upserted} parcels (registry ${n})`);
return { upserted };
} catch (e: any) {
await closeRun(runId, 'failed', { notes: String(e.message || e) });
throw e;
}
}
if (import.meta.url === `file://${process.argv[1]}`) {
ingestKing().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}