← back to Nationalrealestate
Generalize SD adapter Poway->county-wide sweep (1.09M parcels, per-scope cursor, per-parcel city); loop sweeps county
854bc7a609579fe6027fee7a3f3f8d705a52bdf7 · 2026-07-26 07:50:03 -0700 · Steve Abrams
Files touched
M src/ingest/parcels/engine.tsM src/ingest/parcels/sandiego_ca.tsM src/jobs/hourly_loop.ts
Diff
commit 854bc7a609579fe6027fee7a3f3f8d705a52bdf7
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sun Jul 26 07:50:03 2026 -0700
Generalize SD adapter Poway->county-wide sweep (1.09M parcels, per-scope cursor, per-parcel city); loop sweeps county
---
src/ingest/parcels/engine.ts | 3 ++-
src/ingest/parcels/sandiego_ca.ts | 31 +++++++++++++++++++++----------
src/jobs/hourly_loop.ts | 2 +-
3 files changed, 24 insertions(+), 12 deletions(-)
diff --git a/src/ingest/parcels/engine.ts b/src/ingest/parcels/engine.ts
index 397630c..bf570bd 100644
--- a/src/ingest/parcels/engine.ts
+++ b/src/ingest/parcels/engine.ts
@@ -11,7 +11,8 @@ const ADAPTERS: Record<string, () => Promise<{ run: () => Promise<{ upserted: nu
'nyc-acris': async () => ({ run: (await import('./nyc_acris.ts')).ingestNycAcris }),
king: async () => ({ run: (await import('./king_wa.ts')).ingestKing }),
cook: async () => ({ run: (await import('./cook_il.ts')).ingestCook }),
- sandiego: async () => ({ run: (await import('./sandiego_ca.ts')).ingestSanDiego }),
+ 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') }; },
};
async function main() {
diff --git a/src/ingest/parcels/sandiego_ca.ts b/src/ingest/parcels/sandiego_ca.ts
index fe4406f..d5a2b00 100644
--- a/src/ingest/parcels/sandiego_ca.ts
+++ b/src/ingest/parcels/sandiego_ca.ts
@@ -2,7 +2,7 @@
* San Diego County parcels + recorded-deed history from SanGIS/SANDAG ($0 public
* ArcGIS REST). Pilot scope = the city of POWAY (situs_community='POWAY',
* ~16,422 parcels); the same adapter generalizes to any SD community / all of
- * county_fips 06073 by widening POWAY_WHERE.
+ * county_fips 06073 by widening SCOPE_WHERE.
*
* Source: geo.sandag.org Hosted/Parcels FeatureServer/0. Per parcel we get the
* situs address parts, assessed values (land/impr/total), structure (year/beds/
@@ -24,7 +24,14 @@ import { upsertParcels, normAddress, registerParcelSource, type ParcelUpsertRow
const FIPS = '06073';
const SOURCE = 'sandiego_ca';
const LAYER = 'https://geo.sandag.org/server/rest/services/Hosted/Parcels/FeatureServer/0';
-const POWAY_WHERE = "situs_community='POWAY'";
+// 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.
+const SCOPES: Record<string, { where: string; cursor: string }> = {
+ poway: { where: "situs_community='POWAY'", cursor: 'sandiego_ca' },
+ county: { where: '1=1', cursor: 'sandiego_ca_county' },
+};
+let SCOPE_WHERE = SCOPES.poway.where;
+let CURSOR_KEY = SCOPES.poway.cursor;
const PAGE = 2000; // SanGIS maxRecordCount
const MAX_PER_RUN = Number(process.env.SD_MAX_PER_RUN || 4000);
const OUT_FIELDS = [
@@ -79,7 +86,7 @@ function centroid(geom: any): [number | null, number | null] {
async function fetchPage(offset: number): Promise<any[]> {
const u = new URL(LAYER + '/query');
- u.searchParams.set('where', POWAY_WHERE);
+ u.searchParams.set('where', SCOPE_WHERE);
u.searchParams.set('outFields', OUT_FIELDS);
u.searchParams.set('orderByFields', 'apn ASC');
u.searchParams.set('resultOffset', String(offset));
@@ -103,21 +110,25 @@ function links(apn: string): Array<{ kind: string; url: string; label: string }>
return [{ kind: 'gis', url: `${LAYER}/query?where=apn%3D%27${apn}%27&outFields=*&f=html`, label: 'SanGIS parcel record (authoritative)' }];
}
-export async function ingestSanDiego(): Promise<{ upserted: number }> {
+const titleCase = (v: string | null): string | null => v ? v.toLowerCase().replace(/\b\w/g, c => c.toUpperCase()) : null;
+
+export async function ingestSanDiego(scope: 'poway' | 'county' = 'poway'): Promise<{ upserted: number }> {
+ const sc = SCOPES[scope] || SCOPES.poway;
+ SCOPE_WHERE = sc.where; CURSOR_KEY = sc.cursor;
const runId = await openRun(SOURCE, LAYER);
try {
const total = await (async () => {
const u = new URL(LAYER + '/query');
- u.searchParams.set('where', POWAY_WHERE); u.searchParams.set('returnCountOnly', 'true'); u.searchParams.set('f', 'json');
+ u.searchParams.set('where', SCOPE_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);
})();
// Forward-advancing cursor (M-PH2): resume where the last run stopped; wrap to
// 0 only when the whole set is covered, so we never re-fetch page 0 every tick.
- const cur = await query<{ next_offset: number }>(`SELECT next_offset FROM ingest_cursor WHERE source=$1`, [SOURCE]);
+ const cur = await query<{ next_offset: number }>(`SELECT next_offset FROM ingest_cursor WHERE source=$1`, [CURSOR_KEY]);
let offset = cur.rows[0]?.next_offset ?? 0;
if (offset >= total) offset = 0;
- console.log(`[${SOURCE}] Poway total=${total}, resume offset=${offset}`);
+ console.log(`[${SOURCE}] scope=${scope} total=${total}, resume offset=${offset}`);
const rows: ParcelUpsertRow[] = [];
const events: any[] = [];
@@ -135,7 +146,7 @@ export async function ingestSanDiego(): Promise<{ upserted: number }> {
rows.push({
county_fips: FIPS, source_id: apn,
address: addr, norm_address: addr ? normAddress(addr) : null,
- city: 'Poway', zip: s(a.situs_zip),
+ city: titleCase(s(a.situs_community)) || 'Poway', zip: s(a.situs_zip),
lat, lng,
year_built: effYear(a.year_effective), sqft: n(a.total_lvg_area),
beds: bedsVal(a.bedrooms), baths: bathsVal(a.baths), units: null,
@@ -183,8 +194,8 @@ export async function ingestSanDiego(): Promise<{ upserted: number }> {
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()`,
- [SOURCE, nextOffset, total]);
- await registerParcelSource(FIPS, LAYER, 'San Diego (SanGIS) — Poway pilot; deeds via SanGIS last-doc, price/chain gated (ParcelQuest/ARCC)');
+ [CURSOR_KEY, nextOffset, total]);
+ await registerParcelSource(FIPS, LAYER, 'San Diego (SanGIS) — Poway pilot + county sweep; deeds via SanGIS last-doc, price/chain gated (ParcelQuest/ARCC)');
await closeRun(runId, 'ok', { upserted: up, notes: `Poway: ${up} parcels, ${events.length} deed events, ${linkRows.length} links (offset ${offset})` });
console.log(`[${SOURCE}] done: ${up} parcels, ${events.length} deeds, ${linkRows.length} links`);
return { upserted: up };
diff --git a/src/jobs/hourly_loop.ts b/src/jobs/hourly_loop.ts
index 876e04e..591e5db 100644
--- a/src/jobs/hourly_loop.ts
+++ b/src/jobs/hourly_loop.ts
@@ -43,7 +43,7 @@ const JOBS: Job[] = [
{ name: 'discover', script: 'src/enrich/firm_website_discovery.ts', args: [], minIntervalHours: 4, runSources: ['firm_discovery'] },
{ 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'] },
- { name: 'parcels-sd', script: 'src/ingest/parcels/engine.ts', args: ['sandiego'], minIntervalHours: 1, runSources: ['sandiego_ca'] },
+ { name: 'parcels-sd', script: 'src/ingest/parcels/engine.ts', args: ['sandiego-county'], minIntervalHours: 1, runSources: ['sandiego_ca'] },
];
/** hours since this job's freshest OK run across its runSources (∞ if never). */
← f195028 Property page: render deed events + public-record links (dee
·
back to Nationalrealestate
·
Self-correct: gate silent-break detection behind stableVolum 9e65b77 →