← back to Nationalrealestate
Free priced-deed adapters (Oregon RLIS tri-county, Spokane WA, Alameda CA): config-driven ArcGIS sales ingest -> parcel_event(sale) w/ price+parties+links; wired to loop
a3e4169cf251095d8126ba9fca53d61c35e0214e · 2026-07-26 08:28:27 -0700 · Steve Abrams
Files touched
A src/ingest/parcels/arcgis_sales.tsM src/ingest/parcels/engine.tsM src/jobs/hourly_loop.ts
Diff
commit a3e4169cf251095d8126ba9fca53d61c35e0214e
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sun Jul 26 08:28:27 2026 -0700
Free priced-deed adapters (Oregon RLIS tri-county, Spokane WA, Alameda CA): config-driven ArcGIS sales ingest -> parcel_event(sale) w/ price+parties+links; wired to loop
---
src/ingest/parcels/arcgis_sales.ts | 216 +++++++++++++++++++++++++++++++++++++
src/ingest/parcels/engine.ts | 3 +
src/jobs/hourly_loop.ts | 4 +
3 files changed, 223 insertions(+)
diff --git a/src/ingest/parcels/arcgis_sales.ts b/src/ingest/parcels/arcgis_sales.ts
new file mode 100644
index 0000000..dff7f20
--- /dev/null
+++ b/src/ingest/parcels/arcgis_sales.ts
@@ -0,0 +1,216 @@
+/**
+ * Config-driven ArcGIS priced-DEED ingest — free/keyless county sales feeds that
+ * publish real recorded SALE PRICE (unlike SD, which AB-1785-gated it). Each
+ * source maps a page of ArcGIS features to normalized sale records → parcel row
+ * + a parcel_event(event_type='sale', amount=price) carrying grantor/grantee/doc
+ * where available, with a source_url back to the record.
+ *
+ * Verified 2026-07-26 (west-coast first):
+ * oregon-rlis — Portland metro tri-county (Multnomah/Washington/Clackamas),
+ * 428,872 priced (SALEPRICE + SALEDATE YYYYMM).
+ * spokane — Spokane County WA, 146,691 priced (gross_sale_price, buyer, deed type).
+ * alameda — Alameda County CA deed-transfer list, ~51,800 priced
+ * (value_from_trans_tax + grantor/grantee split across rows).
+ *
+ * npm run ingest:parcels -- oregon-rlis | spokane | alameda
+ */
+import { query } from '../../../db/pool.ts';
+import { openRun, closeRun } from '../run.ts';
+import { upsertParcels, normAddress, registerParcelSource, type ParcelUpsertRow } from './upsert.ts';
+
+const PAGE = 2000;
+const MAX_PER_RUN = Number(process.env.SALES_MAX_PER_RUN || 6000);
+
+interface Rec {
+ fips: string; apn: string;
+ address?: string | null; city?: string | null; zip?: string | null;
+ lat?: number | null; lng?: number | null;
+ beds?: number | null; baths?: number | null; sqft?: number | null; year?: number | null;
+ use?: string | null; totalValue?: number | null;
+ price?: number | null; saleDate?: string | null;
+ docNum?: string | null; docType?: string | null; grantor?: string | null; grantee?: string | null;
+}
+interface Source {
+ key: string; label: string; layer: string; where: string; outFields: string; cursorKey: string;
+ geometry: boolean; orderBy: string; // a valid field for stable resultOffset paging (OID name varies)
+ toRecords: (features: any[]) => Rec[];
+}
+
+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; };
+const epochToISO = (v: unknown): string | null => { const x = Number(v); if (!Number.isFinite(x) || x <= 0) return null; const d = new Date(x); return d.toISOString().slice(0, 10); };
+const yyyymm = (v: unknown): string | null => { const t = String(v ?? '').trim(); return /^\d{6}$/.test(t) ? `${t.slice(0, 4)}-${t.slice(4, 6)}-01` : null; };
+function centroid(geom: any): [number | null, number | null] {
+ const ring = geom?.rings?.[0]; if (!Array.isArray(ring) || !ring.length) return [null, null];
+ let sx = 0, sy = 0; for (const [x, y] of ring) { sx += x; sy += y; }
+ return [+(sx / ring.length).toFixed(6), +(sy / ring.length).toFixed(6)];
+}
+const clean = (v: unknown): string | null => { const t = s(v); return t ? t.replace(/\s+/g, ' ') : null; };
+
+// ── source configs ────────────────────────────────────────────────────────
+const OR_FIPS: Record<string, string> = { M: '41051', W: '41067', C: '41005' };
+const SOURCES: Record<string, Source> = {
+ 'oregon-rlis': {
+ key: 'oregon-rlis', label: 'Oregon Metro RLIS (Portland tri-county)',
+ layer: 'https://services2.arcgis.com/McQ0OlIABe29rJJy/arcgis/rest/services/Taxlots_(Public)/FeatureServer/3',
+ where: 'SALEPRICE>1000', cursorKey: 'oregon_rlis', geometry: true, orderBy: 'FID',
+ outFields: 'TLID,PRIMACCNUM,SITEADDR,SITECITY,SITEZIP,SALEDATE,SALEPRICE,COUNTY,YEARBUILT,BLDGSQFT,LANDUSE',
+ toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.PRIMACCNUM) || s(a.TLID); if (!apn) return null;
+ const [lng, lat] = centroid(f.geometry);
+ return { fips: OR_FIPS[s(a.COUNTY) || ''] || '41051', apn, address: clean(a.SITEADDR), city: clean(a.SITECITY), zip: s(a.SITEZIP),
+ lat, lng, sqft: num(a.BLDGSQFT), year: num(a.YEARBUILT), use: s(a.LANDUSE),
+ price: num(a.SALEPRICE), saleDate: yyyymm(a.SALEDATE) } as Rec; }).filter(Boolean) as Rec[],
+ },
+ spokane: {
+ key: 'spokane', label: 'Spokane County WA', cursorKey: 'spokane', geometry: false,
+ layer: 'https://gismo.spokanecounty.org/arcgis/rest/services/SCOUT/PropertyLookup/MapServer/0',
+ where: 'gross_sale_price>1000', orderBy: 'PID_NUM',
+ outFields: 'PID_NUM,gross_sale_price,document_date,excise_nbr,site_address,owner_name,transfer_type',
+ toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.PID_NUM); if (!apn) return null;
+ return { fips: '53063', apn, address: clean(a.site_address), city: 'Spokane', zip: null, lat: null, lng: null,
+ price: num(a.gross_sale_price), saleDate: epochToISO(a.document_date), docNum: s(a.excise_nbr),
+ docType: clean(a.transfer_type), grantee: clean(a.owner_name) } as Rec; }).filter(Boolean) as Rec[],
+ },
+ alameda: {
+ key: 'alameda', label: 'Alameda County CA (deed transfers)', cursorKey: 'alameda', geometry: false,
+ layer: 'https://services5.arcgis.com/ROBnTHSNjoZ2Wm1P/arcgis/rest/services/Assessor_Office_Ownership_Transfer_List/FeatureServer/0',
+ where: 'value_from_trans_tax IS NOT NULL', orderBy: 'OBJECTID',
+ outFields: 'apn,use_name,street_num,street_name,street_suffix,city_name,zip_cd,transfer_dt,doc_prefix,doc_series,value_from_trans_tax,name_type',
+ // grantor+grantee are separate rows sharing doc_series — collapse per doc.
+ toRecords: (fs) => {
+ const byDoc = new Map<string, Rec>();
+ for (const f of fs) {
+ const a = f.attributes; const apn = s(a.apn); const doc = (s(a.doc_prefix) || '') + (s(a.doc_series) || '');
+ if (!apn || !doc) continue;
+ const priceRaw = s(a.value_from_trans_tax); // zero-padded dollars e.g. '000001060000.'
+ const price = priceRaw ? num(priceRaw.replace(/[^\d]/g, '')) : null;
+ const addr = [s(a.street_num), clean(a.street_name), s(a.street_suffix)].filter(Boolean).join(' ').trim() || null;
+ const key = apn + '|' + doc;
+ const r = byDoc.get(key) || { fips: '06001', apn, address: addr, city: clean(a.city_name), zip: s(a.zip_cd),
+ lat: null, lng: null, use: clean(a.use_name), price, saleDate: epochToISO(a.transfer_dt), docNum: doc } as Rec;
+ const nm = clean(a.name_type)?.toUpperCase();
+ // name_type marks the party; the party NAME itself isn't in this field set, so record role presence.
+ if (nm === 'TRANSFEROR') r.grantor = r.grantor || 'recorded';
+ if (nm === 'TRANSFEREE') r.grantee = r.grantee || 'recorded';
+ byDoc.set(key, r);
+ }
+ return [...byDoc.values()];
+ },
+ },
+};
+
+async function fetchPage(src: Source, offset: number): Promise<any[]> {
+ const u = new URL(src.layer + '/query');
+ u.searchParams.set('where', src.where);
+ u.searchParams.set('outFields', src.outFields);
+ u.searchParams.set('orderByFields', src.orderBy + ' ASC');
+ u.searchParams.set('resultOffset', String(offset));
+ u.searchParams.set('resultRecordCount', String(PAGE));
+ u.searchParams.set('returnGeometry', src.geometry ? 'true' : 'false');
+ if (src.geometry) u.searchParams.set('outSR', '4326');
+ u.searchParams.set('f', 'json');
+ // some county servers (Spokane MapServer) are slow — retry transient timeouts.
+ let lastErr: any;
+ for (let attempt = 0; attempt < 3; attempt++) {
+ try {
+ const res = await fetch(u, { signal: AbortSignal.timeout(90_000) });
+ if (!res.ok) throw new Error(`${src.key} ${res.status}: ${(await res.text()).slice(0, 140)}`);
+ const j: any = await res.json();
+ if (j.error) throw new Error(`${src.key} error: ${JSON.stringify(j.error).slice(0, 140)}`);
+ return j.features || [];
+ } catch (e: any) { lastErr = e; if (attempt < 2) await new Promise(r => setTimeout(r, 2000 * (attempt + 1))); }
+ }
+ throw lastErr;
+}
+async function countTotal(src: Source): Promise<number> {
+ const u = new URL(src.layer + '/query');
+ u.searchParams.set('where', src.where); u.searchParams.set('returnCountOnly', 'true'); u.searchParams.set('f', 'json');
+ const j: any = await (await fetch(u, { signal: AbortSignal.timeout(30_000) })).json();
+ return Number(j.count || 0);
+}
+
+export function ingestArcgisSales(key: string) {
+ const src = SOURCES[key];
+ if (!src) throw new Error(`unknown sales source '${key}' — have: ${Object.keys(SOURCES).join(', ')}`);
+ return async (): Promise<{ upserted: number }> => {
+ const runId = await openRun(src.cursorKey, src.layer);
+ try {
+ const total = await countTotal(src);
+ const cur = await query<{ next_offset: number }>(`SELECT next_offset FROM ingest_cursor WHERE source=$1`, [src.cursorKey]);
+ let offset = cur.rows[0]?.next_offset ?? 0;
+ if (offset >= total) offset = 0;
+ console.log(`[${src.key}] ${src.label} total=${total}, resume offset=${offset}`);
+
+ const rows: ParcelUpsertRow[] = []; const events: Rec[] = []; const seenFips = new Set<string>();
+ let fetched = 0;
+ while (fetched < MAX_PER_RUN) {
+ const feats = await fetchPage(src, offset);
+ if (!feats.length) break;
+ for (const r of src.toRecords(feats)) {
+ if (!r.apn) continue;
+ seenFips.add(r.fips);
+ const gisUrl = `${src.layer}/query?where=${encodeURIComponent(src.outFields.split(',')[0] + "='" + r.apn + "'")}&outFields=*&f=html`;
+ rows.push({
+ county_fips: r.fips, source_id: r.apn,
+ address: r.address ?? null, norm_address: r.address ? normAddress(r.address) : null,
+ city: r.city ?? null, zip: r.zip ?? null, lat: r.lat ?? null, lng: r.lng ?? null,
+ year_built: r.year ?? null, sqft: r.sqft ?? null, beds: r.beds ?? null, baths: r.baths ?? null,
+ units: null, use_desc: r.use ?? null, land_value: null, improvement_value: null, total_value: r.totalValue ?? null,
+ tax_year: null, owner_name: r.grantee && r.grantee !== 'recorded' ? r.grantee : null, zoning: null,
+ last_sale_date: r.saleDate ?? null, last_sale_price: r.price ?? null,
+ extra: JSON.stringify({ source: src.key, doc: r.docNum, doc_type: r.docType }),
+ });
+ if (r.price && r.saleDate) events.push({ ...r, address: gisUrl }); // reuse address slot to carry the link
+ }
+ fetched += feats.length; offset += feats.length;
+ if (feats.length < PAGE) break;
+ }
+ // A parcel can have MANY sales in one batch (Alameda) — the parcel row must be
+ // unique per (fips,source_id) or ON CONFLICT DO UPDATE errors "cannot affect row
+ // a second time". Keep the most-recent-sale row per parcel; every sale still lands
+ // in parcel_event (its UNIQUE key includes doc_number).
+ const byParcel = new Map<string, ParcelUpsertRow>();
+ for (const r of rows) {
+ const k = r.county_fips + '|' + r.source_id; const prev = byParcel.get(k);
+ if (!prev || (r.last_sale_date || '') > (prev.last_sale_date || '')) byParcel.set(k, r);
+ }
+ const dedupRows = [...byParcel.values()];
+ const up = await upsertParcels(dedupRows);
+
+ // sale events with price + parties + public-record link
+ for (let i = 0; i < events.length; i += 500) {
+ const chunk = events.slice(i, i + 500);
+ await query(
+ `INSERT INTO parcel_event (county_fips, source_id, event_type, event_date, amount, doc_type, doc_number, source, source_url, detail)
+ VALUES ${chunk.map((_, j) => { const b = j * 9; return `($${b+1},$${b+2},'sale',$${b+3}::date,$${b+4},$${b+5},$${b+6},$${b+7},$${b+8},$${b+9}::jsonb)`; }).join(',')}
+ ON CONFLICT (county_fips, source_id, event_type, event_date, doc_number) DO NOTHING`,
+ chunk.flatMap(e => [e.fips, e.apn, e.saleDate, e.price, e.docType ?? null, e.docNum ?? '', src.cursorKey,
+ e.address /* gisUrl */, JSON.stringify({ grantor: e.grantor ?? null, grantee: e.grantee ?? null })]),
+ );
+ }
+ // per-parcel GIS deep link (deduped — one link per parcel)
+ for (let i = 0; i < dedupRows.length; i += 400) {
+ const chunk = dedupRows.slice(i, i + 400);
+ await query(
+ `INSERT INTO parcel_links (county_fips, source_id, kind, url, label) VALUES ${
+ chunk.map((_, j) => { const b = j * 5; return `($${b+1},$${b+2},$${b+3},$${b+4},$${b+5})`; }).join(',')
+ } ON CONFLICT (county_fips, source_id, kind) DO UPDATE SET url=EXCLUDED.url, label=EXCLUDED.label`,
+ chunk.flatMap(r => [r.county_fips, r.source_id, 'gis',
+ `${src.layer}/query?where=${encodeURIComponent(src.outFields.split(',')[0] + "='" + r.source_id + "'")}&outFields=*&f=html`,
+ `${src.label} record`]),
+ );
+ }
+ for (const fips of seenFips) await registerParcelSource(fips, src.layer, `${src.label} — free priced deeds (ArcGIS REST)`);
+
+ const nextOffset = offset >= total ? 0 : offset;
+ await query(`INSERT INTO ingest_cursor (source, next_offset, total) VALUES ($1,$2,$3)
+ ON CONFLICT (source) DO UPDATE SET next_offset=$2, total=$3, updated_at=NOW()`, [src.cursorKey, nextOffset, total]);
+ await closeRun(runId, 'ok', { upserted: up, notes: `${src.label}: ${up} parcels, ${events.length} sales (offset ${offset}/${total})` });
+ console.log(`[${src.key}] done: ${up} parcels, ${events.length} priced sales`);
+ return { upserted: up };
+ } catch (e: any) {
+ await closeRun(runId, 'failed', { notes: e.message });
+ throw e;
+ }
+ };
+}
diff --git a/src/ingest/parcels/engine.ts b/src/ingest/parcels/engine.ts
index bf570bd..9a98b11 100644
--- a/src/ingest/parcels/engine.ts
+++ b/src/ingest/parcels/engine.ts
@@ -13,6 +13,9 @@ const ADAPTERS: Record<string, () => Promise<{ run: () => Promise<{ upserted: nu
cook: async () => ({ run: (await import('./cook_il.ts')).ingestCook }),
sandiego: async () => { const m = await import('./sandiego_ca.ts'); return { run: () => m.ingestSanDiego('poway') }; },
'sandiego-county': async () => { const m = await import('./sandiego_ca.ts'); return { run: () => m.ingestSanDiego('county') }; },
+ 'oregon-rlis': async () => { const m = await import('./arcgis_sales.ts'); return { run: m.ingestArcgisSales('oregon-rlis') }; },
+ spokane: async () => { const m = await import('./arcgis_sales.ts'); return { run: m.ingestArcgisSales('spokane') }; },
+ alameda: async () => { const m = await import('./arcgis_sales.ts'); return { run: m.ingestArcgisSales('alameda') }; },
};
async function main() {
diff --git a/src/jobs/hourly_loop.ts b/src/jobs/hourly_loop.ts
index 40c4398..d0a9994 100644
--- a/src/jobs/hourly_loop.ts
+++ b/src/jobs/hourly_loop.ts
@@ -47,6 +47,10 @@ const JOBS: Job[] = [
{ name: 'brokers', script: 'src/jobs/broker_rotate.ts', args: [], minIntervalHours: 6, runSources: ['%_dre', '%_trec', '%_dos', '%_dbpr', '%_idfpr', '%_dcp', '%_dpr'] },
{ name: 'commercial', script: 'src/ingest/commercial/engine.ts', args: ['la'], minIntervalHours: 24, runSources: ['la_assessor'], stableVolume: true },
{ name: 'parcels-sd', script: 'src/ingest/parcels/engine.ts', args: ['sandiego-county'], minIntervalHours: 1, runSources: ['sandiego_ca'] },
+ // Free priced-DEED feeds (west-coast) — each paginates its county set over time.
+ { name: 'deeds-or', script: 'src/ingest/parcels/engine.ts', args: ['oregon-rlis'], minIntervalHours: 2, runSources: ['oregon_rlis'] },
+ { name: 'deeds-spok', script: 'src/ingest/parcels/engine.ts', args: ['spokane'], minIntervalHours: 3, runSources: ['spokane'] },
+ { name: 'deeds-alam', script: 'src/ingest/parcels/engine.ts', args: ['alameda'], minIntervalHours: 2, runSources: ['alameda'] },
];
/** hours since this job's freshest OK run across its runSources (∞ if never). */
← c452e2e ct_dcp: normalize credential code to canonical RES.# source_
·
back to Nationalrealestate
·
Spokane: per-source pageSize=500 (slow MapServer); all 3 dee 90f2a67 →