← back to Nationalrealestate
TK-50: field-level provenance across ALL remaining parcel ingesters
6ff7a8171946ff3af82108d4240d40f14db7c603 · 2026-07-28 08:14:42 -0700 · Steve Abrams
Wire source_url + source_key + fetched_at + raw_source onto every parcel row,
and registerSourceFieldMap(our_field -> source_attr) for each source, replicating
the arcgis_sales reference pattern. Now 100% of the pipeline that writes `parcel`
carries full field-level provenance.
Wired: nyc_pluto (own upsert path — provenance cols added to its bespoke
upsertChunk), king_wa (4 EXTR extracts joined by PIN), cook_il (4 Socrata datasets
by PIN), franklin_oh (GDAL .gdb), miami_dade, wake_nc, tarrant_tx, bexar_tx,
fulton_ga, saltlake_ut, sandiego_ca. Each gets a per-record addressable sourceUrl
(ArcGIS OBJECTID/id query, Socrata ?pin/?bbl, or county detail page), fetchedAt =
run start, rawSource = the raw feature/row attrs before normalization, and a
source_field_map declaring the REAL attribute each field maps from (read from the
actual mapping, joined-file/dataset named for the multi-source counties).
Behavior-preserving otherwise. tsc --noEmit clean. Verified end-to-end via bounded
real runs (San Diego Poway, Wake NC, Bexar TX): all four columns populate + the
field-map resolves a field to its source attr + raw value; test rows purged, dev
DB left consistent. No existing-row backfill (gated prod step — see
TK-50-field-provenance.md).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Files touched
A TK-50-field-provenance.mdM src/ingest/parcels/bexar_tx.tsM src/ingest/parcels/cook_il.tsM src/ingest/parcels/franklin_oh.tsM src/ingest/parcels/fulton_ga.tsM src/ingest/parcels/king_wa.tsM src/ingest/parcels/miami_dade.tsM src/ingest/parcels/nyc_pluto.tsM src/ingest/parcels/saltlake_ut.tsM src/ingest/parcels/sandiego_ca.tsM src/ingest/parcels/tarrant_tx.tsM src/ingest/parcels/wake_nc.ts
Diff
commit 6ff7a8171946ff3af82108d4240d40f14db7c603
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Tue Jul 28 08:14:42 2026 -0700
TK-50: field-level provenance across ALL remaining parcel ingesters
Wire source_url + source_key + fetched_at + raw_source onto every parcel row,
and registerSourceFieldMap(our_field -> source_attr) for each source, replicating
the arcgis_sales reference pattern. Now 100% of the pipeline that writes `parcel`
carries full field-level provenance.
Wired: nyc_pluto (own upsert path — provenance cols added to its bespoke
upsertChunk), king_wa (4 EXTR extracts joined by PIN), cook_il (4 Socrata datasets
by PIN), franklin_oh (GDAL .gdb), miami_dade, wake_nc, tarrant_tx, bexar_tx,
fulton_ga, saltlake_ut, sandiego_ca. Each gets a per-record addressable sourceUrl
(ArcGIS OBJECTID/id query, Socrata ?pin/?bbl, or county detail page), fetchedAt =
run start, rawSource = the raw feature/row attrs before normalization, and a
source_field_map declaring the REAL attribute each field maps from (read from the
actual mapping, joined-file/dataset named for the multi-source counties).
Behavior-preserving otherwise. tsc --noEmit clean. Verified end-to-end via bounded
real runs (San Diego Poway, Wake NC, Bexar TX): all four columns populate + the
field-map resolves a field to its source attr + raw value; test rows purged, dev
DB left consistent. No existing-row backfill (gated prod step — see
TK-50-field-provenance.md).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
TK-50-field-provenance.md | 55 +++++++++++++++++++++++++++++++++++++++
src/ingest/parcels/bexar_tx.ts | 18 ++++++++++++-
src/ingest/parcels/cook_il.ts | 34 +++++++++++++++++++++++-
src/ingest/parcels/franklin_oh.ts | 21 ++++++++++++++-
src/ingest/parcels/fulton_ga.ts | 20 +++++++++++++-
src/ingest/parcels/king_wa.ts | 36 ++++++++++++++++++++++++-
src/ingest/parcels/miami_dade.ts | 20 +++++++++++++-
src/ingest/parcels/nyc_pluto.ts | 32 ++++++++++++++++++++---
src/ingest/parcels/saltlake_ut.ts | 22 +++++++++++++++-
src/ingest/parcels/sandiego_ca.ts | 22 +++++++++++++++-
src/ingest/parcels/tarrant_tx.ts | 20 +++++++++++++-
src/ingest/parcels/wake_nc.ts | 19 +++++++++++++-
12 files changed, 306 insertions(+), 13 deletions(-)
diff --git a/TK-50-field-provenance.md b/TK-50-field-provenance.md
new file mode 100644
index 0000000..373a801
--- /dev/null
+++ b/TK-50-field-provenance.md
@@ -0,0 +1,55 @@
+# TK-50 — Field-Level Data Provenance ("source every bit of info")
+
+Every parcel ingester that writes to the `parcel` table now stamps **record-level
+provenance** on each row and **declares its field→attribute mapping** once per
+source. This memo is the coverage record + the (gated) backfill note. It supersedes
+the earlier arcgis_sales-only draft: **provenance now covers the entire pipeline.**
+
+## The model (see `db/migrations/012_parcel_provenance.sql`)
+
+One source record (an ArcGIS feature, a joined CSV/Socrata row) → one `parcel`
+row, so all of a parcel's fields share the same source. Provenance is therefore:
+
+- On the parcel row: `source_url` (addressable record link), `source_key`
+ (ingest source id), `fetched_at` (run start time), `raw_source` (JSONB of the
+ raw source attributes we mapped from).
+- Once per source, in `source_field_map(source_key, our_field, source_attr, notes)`:
+ which raw attribute each of our fields came from — registered at ingest time by
+ `registerSourceFieldMap()`.
+
+"Where did field X come from?" resolves to:
+`parcel.source_url` + `source_field_map[source_key][X].source_attr`
++ `parcel.raw_source ->> source_attr` (the raw value). Verified by
+`src/ingest/parcels/provenance_selftest.ts`.
+
+## Coverage — every ingester that calls the parcel upsert
+
+| source_key | file | sourceUrl strategy | fields mapped | sale? |
+|---|---|---|---|---|
+| (14 arcgis_sales configs) | `arcgis_sales.ts` | per-record layer `?where=<idField>='<apn>'&f=html` | per-config `fieldMap` | priced deeds |
+| `nyc_pluto` | `nyc_pluto.ts` (own upsert path — provenance cols wired into its bespoke `upsertChunk`) | Socrata `64uk-42ks.json?bbl=<bbl>` | 15 (owner, zoning, assessed land/total, derived impr, structure) | none |
+| `king_wa` | `king_wa.ts` (4 EXTR extracts joined by PIN) | KingCo `eRealProperty/Detail.aspx?ParcelNbr=<pin>` | 15 (attr + which EXTR file) | REAL priced sales |
+| `cook_il` | `cook_il.ts` (4 Socrata datasets joined by PIN) | Socrata `pabr-t5kh.json?pin=<pin>` | 14 (attr + which dataset) | none (assessed values) |
+| `franklin_oh` | `franklin_oh.ts` (GDAL `.gdb` extract) | Auditor `Parcel/<parcelid>` | 8 | latest recorded sale |
+| `miami_dade` | `miami_dade.ts` | layer `?where=FOLIO='<folio>'&f=html` | 8 (sale = newest of 3 DOS_n/PRICE_n slots) | up to 3 recorded sales |
+| `wake_nc` | `wake_nc.ts` | layer `?where=REID='<reid>'&f=html` | 14 (owner, assessed L/B/T, structure, sale) | priced sales |
+| `tarrant_tx` | `tarrant_tx.ts` | layer `?where=ACCOUNT='<acct>'&f=html` | 12 (owner, TRUE market value, deed date; no price) | deed date only |
+| `bexar_tx` | `bexar_tx.ts` | layer `?where=PropID=<pid>&f=html` | 7 (owner, situs-parsed addr, sqft, use) | none |
+| `fulton_ga` | `fulton_ga.ts` | layer `?where=ParcelID='<pid>'&f=html` (space-encoded) | 5 (owner, structured addr, LUCode, units) | none |
+| `saltlake_ut` | `saltlake_ut.ts` | layer `?where=PARCEL_ID='<pid>'&f=html` (or SERIAL_NUM) | 9 (MARKET value, structure; derived impr) | none |
+| `sandiego_ca` | `sandiego_ca.ts` | layer `?where=apn='<apn>'&f=html` | 14 (assessed value, structure, zoning, deed date) | deed date only |
+
+`nyc_acris.ts` writes to `parcel_event`, not `parcel` — outside this row-provenance scope.
+
+## HARD RAIL: no backfill of existing rows (gated)
+
+Migration 012 is additive; existing rows keep `source_url = NULL` until they are
+**re-ingested**. A one-time bulk backfill of the ~5M already-loaded prod rows is a
+**separate, Steve-gated prod step** — it is NOT performed here and must not be run
+autonomously (it is a prod write to Kamatera-canonical `dw_unified`/`usre`). Going
+forward, every fresh ingest / re-ingest carries full provenance automatically.
+
+Verification: `tsc --noEmit` clean; `provenance_selftest.ts` green; bounded real
+runs of San Diego (Poway), Wake NC, Bexar TX confirmed all four columns populate
+and the field-map resolves a real field to its source attr + raw value (test rows
+purged, dev DB left consistent).
diff --git a/src/ingest/parcels/bexar_tx.ts b/src/ingest/parcels/bexar_tx.ts
index a7672f3..f1b6cc3 100644
--- a/src/ingest/parcels/bexar_tx.ts
+++ b/src/ingest/parcels/bexar_tx.ts
@@ -38,12 +38,20 @@
*/
import { pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, registerParcelSource, normAddress, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
const FIPS = '48029';
+const SOURCE_KEY = 'bexar_tx';
const LAYER = 'https://services.arcgis.com/g1fRTDLeMgspWrYp/arcgis/rest/services/BCAD_Parcels/FeatureServer/0';
const WHERE = 'PropID>0';
const PAGE = 2000;
+// TK-50 field-level provenance: our parcel field → the exact source attribute it
+// maps from. address/city/zip are PARSED out of the single Situs string. No
+// appraised value / year / beds-baths / sale in this CoSA-hosted BCAD view.
+const FIELD_MAP: Record<string, string> = {
+ source_id: 'PropID', address: 'Situs (street parsed)', city: 'Situs (city parsed)',
+ zip: 'Situs (zip parsed)', sqft: 'GBA_Living', use_desc: 'state_cd', owner_name: 'Owner_Name',
+};
const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
@@ -113,7 +121,10 @@ async function fetchPage(offset: number): Promise<any[]> {
export async function ingestBexar(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_bexar_tx', LAYER);
+ // 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, 'Bexar County TX (BCAD) via City of San Antonio open-data parcel view');
let offset = 0, seen = 0, skipped = 0, upserted = 0;
for (;;) {
const feats = await fetchPage(offset);
@@ -151,6 +162,11 @@ export async function ingestBexar(): Promise<{ upserted: number }> {
exemptions: s(a.Exemptions) || undefined,
note: 'Bexar County TX (BCAD) via City of San Antonio open-data parcel view — owner + situs + living-area sqft + use code; no appraised value / year-built / beds-baths / sale in this feed',
}),
+ // ── TK-50 field-level provenance (record-level: one feature → this row) ──
+ sourceKey: SOURCE_KEY,
+ sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`PropID=${pid}`)}&outFields=*&f=html`,
+ fetchedAt,
+ rawSource: JSON.stringify(a),
});
}
upserted += await upsertParcels(batch);
diff --git a/src/ingest/parcels/cook_il.ts b/src/ingest/parcels/cook_il.ts
index 393795a..945c92b 100644
--- a/src/ingest/parcels/cook_il.ts
+++ b/src/ingest/parcels/cook_il.ts
@@ -14,11 +14,26 @@
*/
import { pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, registerParcelSource, normAddress, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
const FIPS = '17031';
+const SOURCE_KEY = 'cook_il';
const BASE = 'https://datacatalog.cookcountyil.gov/resource/';
const PAGE = 200000;
+// TK-50 field-level provenance: our parcel field → the source attribute + which
+// Socrata dataset it was joined from (by PIN). Values are ASSESSED (10%/25% of
+// market), not market. No zoning or sale-price feed here.
+const FIELD_MAP: Record<string, string> = {
+ source_id: 'pabr-t5kh.pin', lat: 'pabr-t5kh.lat', lng: 'pabr-t5kh.lon',
+ use_desc: 'pabr-t5kh.class (→ class-group label)',
+ address: '3723-97qp.prop_address_full', city: '3723-97qp.prop_address_city_name|pabr-t5kh.cook_municipality_name',
+ zip: '3723-97qp.prop_address_zipcode_1|pabr-t5kh.zip_code',
+ owner_name: '3723-97qp.owner_address_name|mail_address_name',
+ land_value: 'uzyt-m557.board_land|certified_land', improvement_value: 'uzyt-m557.board_bldg|certified_bldg',
+ total_value: 'uzyt-m557.board_tot|certified_tot',
+ year_built: 'x54s-btds.char_yrblt', sqft: 'x54s-btds.char_bldg_sf', beds: 'x54s-btds.char_beds',
+ baths: 'x54s-btds.char_fbath+char_hbath',
+};
const CLASS_GROUP: Record<string, string> = {
'1': 'Vacant Land', '2': 'Residential (<7 units)', '3': 'Multi-Family (7+ units)',
@@ -57,7 +72,10 @@ async function* pages(dataset: string, select: string, where?: string): AsyncGen
export async function ingestCook(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_cook_il', BASE + 'pabr-t5kh.json');
+ // 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, 'Cook County IL Assessor open data (parcel-universe pabr-t5kh joined with addresses 3723-97qp, assessed values uzyt-m557, res characteristics x54s-btds by PIN)');
// 1) situs address + owner/taxpayer name (current assessment year)
const addr = new Map<string, [addr: string, city: string, zip: string, owner: string, mailing: string]>();
for await (const page of pages('3723-97qp',
@@ -129,6 +147,20 @@ export async function ingestCook(): Promise<{ upserted: number }> {
owner_source: a?.[3] ? 'owner/taxpayer name, Cook County Assessor parcel addresses (2026)' : undefined,
value_basis: v ? 'ASSESSED value (Cook County 2025 certified/BOR — 10% res / 25% comm of market value)' : undefined,
}),
+ // ── TK-50 field-level provenance (record-level: this PIN's joined datasets) ──
+ // Socrata parcel-universe record link (addressable by PIN); raw_source
+ // captures the actual source values we mapped from across the 4 datasets.
+ sourceKey: SOURCE_KEY,
+ sourceUrl: `${BASE}pabr-t5kh.json?pin=${encodeURIComponent(o.pin)}`,
+ fetchedAt,
+ rawSource: JSON.stringify({
+ universe: { pin: o.pin, class: cls, township_name: o.township_name, zip_code: o.zip_code,
+ lat: o.lat, lon: o.lon, cook_municipality_name: o.cook_municipality_name },
+ addresses: a ? { prop_address_full: a[0], prop_address_city_name: a[1],
+ prop_address_zipcode_1: a[2], owner_address_name: a[3], mailing: a[4] } : null,
+ values: v ? { land: v[0], bldg: v[1], tot: v[2], year: '2025' } : null,
+ characteristics: c ? { char_yrblt: c[0], char_bldg_sf: c[1], char_beds: c[2], baths: c[3] } : null,
+ }),
});
}
upserted += await upsertParcels(batch);
diff --git a/src/ingest/parcels/franklin_oh.ts b/src/ingest/parcels/franklin_oh.ts
index f863b1a..e82dab9 100644
--- a/src/ingest/parcels/franklin_oh.ts
+++ b/src/ingest/parcels/franklin_oh.ts
@@ -36,13 +36,22 @@ 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, normAddress, type ParcelUpsertRow } from './upsert.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 */
@@ -100,7 +109,10 @@ export async function ingestFranklin(): Promise<{ upserted: number }> {
}
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');
@@ -161,6 +173,13 @@ export async function ingestFranklin(): Promise<{ upserted: number }> {
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);
}
diff --git a/src/ingest/parcels/fulton_ga.ts b/src/ingest/parcels/fulton_ga.ts
index 2cb36fd..2bc0e0c 100644
--- a/src/ingest/parcels/fulton_ga.ts
+++ b/src/ingest/parcels/fulton_ga.ts
@@ -34,12 +34,21 @@
*/
import { pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, registerParcelSource, normAddress, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
const FIPS = '13121';
+const SOURCE_KEY = 'fulton_ga';
const LAYER = 'https://services1.arcgis.com/AQDHTHDrZzfsFsB5/arcgis/rest/services/Tax_Parcels/FeatureServer/0';
const WHERE = "ParcelID IS NOT NULL AND ParcelID <> ''";
const PAGE = 1000; // layer maxRecordCount is 1000
+// TK-50 field-level provenance: our parcel field → the exact source attribute it
+// maps from. address is assembled from the structured Addr* components (falling
+// back to the flat Address). No value/year/sqft/beds/baths/sale in this view.
+const FIELD_MAP: Record<string, string> = {
+ source_id: 'ParcelID',
+ address: 'AddrNumber+AddrPreDir+AddrStreet+AddrSuffix+AddrPosDir+AddrUntTyp+AddrUnit|Address',
+ units: 'LivUnits', use_desc: 'LUCode', tax_year: 'TaxYear', owner_name: 'Owner',
+};
const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
@@ -99,7 +108,10 @@ async function fetchPage(offset: number): Promise<any[]> {
export async function ingestFulton(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_fulton_ga', LAYER);
+ // 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, 'Fulton County GA (Board of Assessors tax digest) via Fulton DoIT GIS Tax_Parcels view');
let offset = 0, seen = 0, skipped = 0, upserted = 0;
for (;;) {
const feats = await fetchPage(offset);
@@ -140,6 +152,12 @@ export async function ingestFulton(): Promise<{ upserted: number }> {
subdivision: s(a.Subdiv) || undefined,
note: 'Fulton County GA (Board of Assessors tax digest) via Fulton DoIT GIS Tax_Parcels view — owner + structured address + land-use code + living units + land acres; no market value / year-built / building sqft / beds-baths / sale-or-deed feed in this open view',
}),
+ // ── TK-50 field-level provenance (record-level: one feature → this row) ──
+ // ParcelID contains an embedded space — encodeURIComponent handles it.
+ sourceKey: SOURCE_KEY,
+ sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`ParcelID='${pid}'`)}&outFields=*&f=html`,
+ fetchedAt,
+ rawSource: JSON.stringify(a),
});
}
upserted += await upsertParcels(batch);
diff --git a/src/ingest/parcels/king_wa.ts b/src/ingest/parcels/king_wa.ts
index 5c0c4cc..335e03e 100644
--- a/src/ingest/parcels/king_wa.ts
+++ b/src/ingest/parcels/king_wa.ts
@@ -19,12 +19,29 @@ 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, normAddress, type ParcelUpsertRow } from './upsert.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'],
@@ -57,7 +74,10 @@ async function ensureFiles() {
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
@@ -164,6 +184,20 @@ export async function ingestKing(): Promise<{ upserted: number }> {
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`);
diff --git a/src/ingest/parcels/miami_dade.ts b/src/ingest/parcels/miami_dade.ts
index 2bc4e80..d006421 100644
--- a/src/ingest/parcels/miami_dade.ts
+++ b/src/ingest/parcels/miami_dade.ts
@@ -24,12 +24,22 @@
*/
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, registerParcelSource, normAddress, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';
const FIPS = '12086';
+const SOURCE_KEY = 'miami_dade';
const LAYER = 'https://gisweb.miamidade.gov/arcgis/rest/services/MD_ComparableSales/MapServer/5';
const PAGE = 20000;
+// TK-50 field-level provenance: our parcel field → the exact source attribute it
+// maps from. last_sale_* come from the NEWEST of the up-to-3 recorded sales
+// (DOS_n/PRICE_n slots, n=1 the county's newest) — declared as the slot family.
+const FIELD_MAP: Record<string, string> = {
+ source_id: 'FOLIO', address: 'TRUE_SITE_ADDR', city: 'TRUE_SITE_CITY', zip: 'TRUE_SITE_ZIP_CODE',
+ use_desc: 'DOR_DESC', owner_name: 'TRUE_OWNER1', zoning: 'PRIMARY_ZONE',
+ last_sale_date: 'DOS_1|DOS_2|DOS_3 (newest recorded sale)',
+ last_sale_price: 'PRICE_1|PRICE_2|PRICE_3 (newest recorded sale)',
+};
const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
@@ -91,7 +101,10 @@ function salesOf(a: any): Sale[] {
export async function ingestMiami(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_miami_dade', LAYER);
+ // 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, 'Miami-Dade County FL Property Appraiser GIS (MDC.PaGis, keyless ArcGIS)');
// Stream: fetch a page -> upsert its parcels + insert its sale events -> release.
// Memory-bounded (never holds the whole county); partial progress persists.
// Bulk-sale detection is deferred to an in-DB post-pass on the stored latest_deed.
@@ -126,6 +139,11 @@ export async function ingestMiami(): Promise<{ upserted: number }> {
latest_sale_qualified: latest ? latest.qualified : undefined,
note: 'no assessed value / beds / baths / sqft / year-built in the public PaGis layer',
}),
+ // ── TK-50 field-level provenance (record-level: one feature → this row) ──
+ sourceKey: SOURCE_KEY,
+ sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`FOLIO='${folio}'`)}&outFields=*&f=html`,
+ fetchedAt,
+ rawSource: JSON.stringify(a),
});
// every dated transfer becomes an event (incl. $0 non-arms-length) so the
// parcel's last_sale_date never points to a missing event row; price/doc_type
diff --git a/src/ingest/parcels/nyc_pluto.ts b/src/ingest/parcels/nyc_pluto.ts
index 8e4e2f8..261d4cf 100644
--- a/src/ingest/parcels/nyc_pluto.ts
+++ b/src/ingest/parcels/nyc_pluto.ts
@@ -10,9 +10,22 @@
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]> = {
@@ -40,13 +53,16 @@ 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'];
+ '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.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(
@@ -56,7 +72,9 @@ async function upsertChunk(rows: any[]): Promise<number> {
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`,
+ 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;
@@ -64,8 +82,11 @@ async function upsertChunk(rows: any[]): Promise<number> {
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' } });
@@ -91,6 +112,11 @@ export async function ingestNycPluto(): Promise<{ upserted: number }> {
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);
diff --git a/src/ingest/parcels/saltlake_ut.ts b/src/ingest/parcels/saltlake_ut.ts
index dd4fce2..c6f5baf 100644
--- a/src/ingest/parcels/saltlake_ut.ts
+++ b/src/ingest/parcels/saltlake_ut.ts
@@ -16,11 +16,21 @@
*/
import { pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, registerParcelSource, normAddress, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
const FIPS = '49035';
+const SOURCE_KEY = 'saltlake_ut';
const LAYER = 'https://services1.arcgis.com/99lidPhWCzftIe9K/arcgis/rest/services/Parcels_SaltLake_LIR/FeatureServer/0';
const PAGE = 2000;
+// TK-50 field-level provenance: our parcel field → the exact source attribute it
+// maps from. source_id prefers PARCEL_ID, else SERIAL_NUM; improvement_value is
+// DERIVED (TOTAL_MKT_VALUE − LAND_MKT_VALUE). Utah LIR carries no owner or sale.
+const FIELD_MAP: Record<string, string> = {
+ source_id: 'PARCEL_ID|SERIAL_NUM', address: 'PARCEL_ADD', city: 'PARCEL_CITY',
+ year_built: 'BUILT_YR', sqft: 'BLDG_SQFT', units: 'HOUSE_CNT', use_desc: 'PROP_CLASS|PROP_TYPE',
+ land_value: 'LAND_MKT_VALUE', total_value: 'TOTAL_MKT_VALUE',
+ improvement_value: 'TOTAL_MKT_VALUE-LAND_MKT_VALUE (derived)',
+};
const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
@@ -60,7 +70,10 @@ async function fetchPage(offset: number): Promise<any[]> {
export async function ingestSaltLake(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_saltlake_ut', LAYER);
+ // 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, 'Salt Lake County UT — Utah LIR standardized parcel FeatureServer (keyless ArcGIS)');
let offset = 0, seen = 0, upserted = 0;
for (;;) {
const feats = await fetchPage(offset);
@@ -99,6 +112,13 @@ export async function ingestSaltLake(): Promise<{ upserted: number }> {
value_basis: total != null ? 'MARKET value (Utah LIR)' : undefined,
note: 'Utah LIR — market value + characteristics; no owner name or sale feed in the public statewide layer',
}),
+ // ── TK-50 field-level provenance (record-level: one feature → this row) ──
+ // address the record by whichever id field supplied source_id (PARCEL_ID
+ // preferred, else SERIAL_NUM) so the deep link resolves to this parcel.
+ sourceKey: SOURCE_KEY,
+ sourceUrl: `${LAYER}/query?where=${encodeURIComponent((s(a.PARCEL_ID) ? 'PARCEL_ID' : 'SERIAL_NUM') + `='${pid}'`)}&outFields=*&f=html`,
+ fetchedAt,
+ rawSource: JSON.stringify(a),
};
// skip attribute-less ghost features (ROW/easement polygons UT LIR exports with
// no value/address/sqft/use) — coords only, nothing a property lookup can use.
diff --git a/src/ingest/parcels/sandiego_ca.ts b/src/ingest/parcels/sandiego_ca.ts
index 8e3f366..ced4a6f 100644
--- a/src/ingest/parcels/sandiego_ca.ts
+++ b/src/ingest/parcels/sandiego_ca.ts
@@ -19,11 +19,23 @@
*/
import { query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, normAddress, registerParcelSource, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, normAddress, registerParcelSource, registerSourceFieldMap, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';
const FIPS = '06073';
const SOURCE = 'sandiego_ca';
+const SOURCE_KEY = 'sandiego_ca';
+// TK-50 field-level provenance: our parcel field → the exact source attribute it
+// maps from. address is assembled from situs parts; use_desc/zoning are derived
+// labels; last_sale_date is the last recorded document date (no price — AB-1785
+// gated). SanGIS withholds owner name.
+const FIELD_MAP: Record<string, string> = {
+ source_id: 'apn', address: 'nucleus_situs_from_nbr+situs_pre_dir+situs_street+situs_suffix+situs_post_dir',
+ city: 'situs_community', zip: 'situs_zip', year_built: 'year_effective', sqft: 'total_lvg_area',
+ beds: 'bedrooms', baths: 'baths', use_desc: 'asr_landuse', land_value: 'asr_land',
+ improvement_value: 'asr_impr', total_value: 'asr_total', zoning: 'asr_zone|nucleus_zone_cd',
+ last_sale_date: 'docdate',
+};
const LAYER = 'https://geo.sandag.org/server/rest/services/Hosted/Parcels/FeatureServer/0';
// Scope: pilot=Poway; 'county' widens to all of San Diego County (west-coast
// expansion). Cursor is keyed per-scope so the two sweeps don't fight.
@@ -118,7 +130,10 @@ export async function ingestSanDiego(scope: 'poway' | 'county' = 'poway'): Promi
const sc = SCOPES[scope] || SCOPES.poway;
SCOPE_WHERE = sc.where; CURSOR_KEY = sc.cursor;
const runId = await openRun(SOURCE, LAYER);
+ // 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, 'San Diego County (SanGIS/SANDAG Parcels) — situs + assessed value + structure; deed via last recorded doc');
const total = await (async () => {
const u = new URL(LAYER + '/query');
u.searchParams.set('where', SCOPE_WHERE); u.searchParams.set('returnCountOnly', 'true'); u.searchParams.set('f', 'json');
@@ -159,6 +174,11 @@ export async function ingestSanDiego(scope: 'poway' | 'county' = 'poway'): Promi
extra: JSON.stringify({ apn_8: s(a.apn_8), parcelid: s(a.parcelid), doctype: s(a.doctype),
docnmbr: s(a.docnmbr), asr_landuse: s(a.asr_landuse), taxstat: s(a.taxstat),
ownerocc: s(a.ownerocc), garage_stalls: s(a.garage_stalls), pool: s(a.pool), community: 'POWAY' }),
+ // ── TK-50 field-level provenance (record-level: one feature → this row) ──
+ sourceKey: SOURCE_KEY,
+ sourceUrl: `${LAYER}/query?where=apn%3D%27${apn}%27&outFields=*&f=html`,
+ fetchedAt,
+ rawSource: JSON.stringify(a),
});
if (sale) events.push({ apn, date: sale, doctype: s(a.doctype), docnmbr: s(a.docnmbr) });
for (const l of links(apn)) linkRows.push({ apn, ...l });
diff --git a/src/ingest/parcels/tarrant_tx.ts b/src/ingest/parcels/tarrant_tx.ts
index 391bf60..5de07fe 100644
--- a/src/ingest/parcels/tarrant_tx.ts
+++ b/src/ingest/parcels/tarrant_tx.ts
@@ -28,13 +28,23 @@
*/
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, registerParcelSource, normAddress, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';
const FIPS = '48439';
+const SOURCE_KEY = 'tarrant_tx';
const LAYER = 'https://services5.arcgis.com/3ddLCBXe1bRt7mzj/arcgis/rest/services/Parcels_Public_Vview/FeatureServer/0';
const WHERE = "COUNTYNAME='Tarrant'";
const PAGE = 2000;
+// TK-50 field-level provenance: our parcel field → the exact source attribute it
+// maps from. total_value maps from MARKET_VALUE (true market, not the assessed
+// fraction); this feed carries a deed DATE but no sale price.
+const FIELD_MAP: Record<string, string> = {
+ source_id: 'ACCOUNT', address: 'SITUS_ADDR', city: 'CITYNAME', tax_year: 'TAX_YEAR',
+ year_built: 'YR_BUILT', sqft: 'LIVING_AREA', use_desc: 'STATE_USE_CODE',
+ land_value: 'LAND_VAL', improvement_value: 'IMPR_VAL', total_value: 'MARKET_VALUE',
+ owner_name: 'OWNER_NAME', last_sale_date: 'DEED_DATE',
+};
const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
@@ -95,7 +105,10 @@ async function fetchPage(offset: number): Promise<any[]> {
export async function ingestTarrant(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_tarrant_tx', LAYER);
+ // 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, 'Tarrant County TX (TAD) via City of Fort Worth public parcel view');
let offset = 0, seen = 0, skipped = 0, upserted = 0, deeds = 0;
for (;;) {
const feats = await fetchPage(offset);
@@ -134,6 +147,11 @@ export async function ingestTarrant(): Promise<{ upserted: number }> {
land_sqft: num(a.LAND_SQFT) || undefined,
note: 'Tarrant County TX (TAD) via City of Fort Worth public parcel view — TRUE market value; no beds/baths, no sale price (deed date only) in this feed',
}),
+ // ── TK-50 field-level provenance (record-level: one feature → this row) ──
+ sourceKey: SOURCE_KEY,
+ sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`ACCOUNT='${acct}'`)}&outFields=*&f=html`,
+ fetchedAt,
+ rawSource: JSON.stringify(a),
});
if (deedDate) events.push({ acct, date: deedDate });
}
diff --git a/src/ingest/parcels/wake_nc.ts b/src/ingest/parcels/wake_nc.ts
index 472dcbc..ae46c97 100644
--- a/src/ingest/parcels/wake_nc.ts
+++ b/src/ingest/parcels/wake_nc.ts
@@ -21,12 +21,21 @@
*/
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, registerParcelSource, normAddress, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';
const FIPS = '37183';
+const SOURCE_KEY = 'wake_nc';
const LAYER = 'https://maps.wakegov.com/arcgis/rest/services/Property/Parcels/MapServer/0';
const PAGE = 2000;
+// TK-50 field-level provenance: our parcel field → the exact source attribute it
+// was mapped from in the loop below. Registered into source_field_map at ingest.
+const FIELD_MAP: Record<string, string> = {
+ source_id: 'REID', address: 'SITE_ADDRESS', city: 'CITY_DECODE', zip: 'ZIPNUM',
+ year_built: 'YEAR_BUILT', sqft: 'HEATEDAREA', units: 'UNITS', use_desc: 'TYPE_USE_DECODE',
+ land_value: 'LAND_VAL', improvement_value: 'BLDG_VAL', total_value: 'TOTAL_VALUE_ASSD',
+ owner_name: 'OWNER', last_sale_price: 'TOTSALPRICE', last_sale_date: 'SALE_DATE',
+};
const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
@@ -70,7 +79,10 @@ async function fetchPage(offset: number): Promise<any[]> {
export async function ingestWake(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_wake_nc', LAYER);
+ // 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, 'Wake County NC Assessor (authoritative ArcGIS MapServer)');
let offset = 0, seen = 0, upserted = 0, sales = 0;
for (;;) {
const feats = await fetchPage(offset);
@@ -104,6 +116,11 @@ export async function ingestWake(): Promise<{ upserted: number }> {
latest_deed: /^[-0]*$/.test(deed) ? undefined : deed,
note: 'Wake County NC Assessor — assessed value; no beds/baths or zoning in this layer',
}),
+ // ── TK-50 field-level provenance (record-level: one feature → this row) ──
+ sourceKey: SOURCE_KEY,
+ sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`REID='${reid}'`)}&outFields=*&f=html`,
+ fetchedAt,
+ rawSource: JSON.stringify(a),
});
if (salePrice && saleDate) events.push({ reid, price: salePrice, date: saleDate, deed });
}
← 536b408 parcels: field-level data provenance — source every parcel f
·
back to Nationalrealestate
·
(newest)