← back to Nationalrealestate

src/ingest/parcels/hennepin_mn.ts

202 lines

/**
 * Hennepin County MN (Minneapolis, FIPS 27053) parcel ingest — dedicated
 * full-universe adapter (applies the Wake/Miami-Dade DTD verdict B precedent).
 *
 * Source: the Minnesota Metropolitan Council "Metropolitan 7-County Parcels"
 * aggregate FeatureServer (keyless, supportsPagination, 2k/page, $0):
 *   https://arcgis.metc.state.mn.us/data1/rest/services/parcels/Parcels_Aggregate/FeatureServer/0
 * This is a 7-county metro aggregate — we filter server-side to CO_NAME='HENNEPIN'
 * (= 447,044 parcels). The Met Council standardizes every metro county's assessor
 * roll into one schema, so this is the authoritative full-universe Hennepin feed.
 *
 * A rich PRICED-SALE county: OWNER_NAME (100%), situs address (assembled from the
 * ANUMBER + ST_ name parts), CTU_NAME (city) + ZIP, EMV_LAND + EMV_BLDG + EMV_TOTAL
 * (Estimated Market Value — TRUE market, ~95%), TAX_YEAR, YEAR_BUILT (~94%),
 * NUM_UNITS, USECLASS1 (use), SALE_VALUE + SALE_DATE (a real priced last-sale,
 * ~87%). Note: the aggregate's FIN_SQ_FT is not populated for Hennepin (0 rows) —
 * sqft is left NULL rather than mapped from an empty column.
 *
 * source_id = PIN (unique within Hennepin — count(PIN)=447,044=total).
 *
 * Run: NODE_OPTIONS=--max-old-space-size=4096 npm run ingest:parcels hennepin
 */
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.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 = '27053';
const SOURCE_KEY = 'hennepin_mn';
const LAYER = 'https://arcgis.metc.state.mn.us/data1/rest/services/parcels/Parcels_Aggregate/FeatureServer/0';
const WHERE = "CO_NAME='HENNEPIN'";
const PAGE = 2000;
// TK-50 field-level provenance. total_value maps from EMV_TOTAL (Estimated Market
// Value — true market). This feed carries a priced sale (SALE_VALUE + SALE_DATE).
const FIELD_MAP: Record<string, string> = {
  source_id: 'PIN', address: 'ANUMBER+ST_NAME', city: 'CTU_NAME', zip: 'ZIP',
  year_built: 'YEAR_BUILT', units: 'NUM_UNITS', use_desc: 'USECLASS1', tax_year: 'TAX_YEAR',
  land_value: 'EMV_LAND', improvement_value: 'EMV_BLDG', total_value: 'EMV_TOTAL',
  owner_name: 'OWNER_NAME', last_sale_price: 'SALE_VALUE', 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; };
// SALE_DATE is an esri date (epoch-ms) — reject <=0 and sub-1971 near-epoch
// artifacts, then run the shared future/pre-1900/non-calendar guard.
const epochToISO = (v: unknown): string | null => { const x = Number(v); if (!Number.isFinite(x) || x < 31_536_000_000) return null; return sanitizeEventDate(new Date(x).toISOString().slice(0, 10)); };

/** Assemble the situs address from the standardized address parts, collapsing the
 *  gaps left by empty prefix/suffix/direction components. */
function situsAddr(a: any): string | null {
  const parts = [a.ANUMBERPRE, a.ANUMBER, a.ANUMBERSUF, a.ST_PRE_DIR, a.ST_PRE_TYP, a.ST_NAME, a.ST_POS_TYP, a.ST_POS_DIR]
    .map(s).filter((x): x is string => !!x);
  const joined = parts.join(' ').replace(/\s+/g, ' ').trim();
  return joined || null;
}

/** plain vertex-average of ring[0] — NOT a true polygon centroid (fine for a map
 *  pin; multipart parcels only get the first ring). */
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 OUT = ['PIN', 'OWNER_NAME', 'OWNER_MORE', 'ANUMBERPRE', 'ANUMBER', 'ANUMBERSUF',
  'ST_PRE_DIR', 'ST_PRE_TYP', 'ST_NAME', 'ST_POS_TYP', 'ST_POS_DIR', 'CTU_NAME', 'ZIP',
  'EMV_LAND', 'EMV_BLDG', 'EMV_TOTAL', 'TAX_YEAR', 'YEAR_BUILT', 'NUM_UNITS',
  'USECLASS1', 'HOME_STYLE', 'SALE_VALUE', 'SALE_DATE', 'ABB_LEGAL'].join(',');

async function fetchPage(offset: number): Promise<any[]> {
  const u = new URL(LAYER + '/query');
  u.searchParams.set('where', WHERE);
  u.searchParams.set('outFields', OUT);
  u.searchParams.set('orderByFields', 'OBJECTID ASC');
  u.searchParams.set('resultOffset', String(offset));
  u.searchParams.set('resultRecordCount', String(PAGE));
  u.searchParams.set('returnGeometry', 'true');
  u.searchParams.set('outSR', '4326');
  u.searchParams.set('f', 'json');
  let lastErr: any;
  for (let a = 0; a < 3; a++) {
    try {
      const res = await fetch(u, { headers: { 'User-Agent': 'Mozilla/5.0 (usre-parcel-ingest)' }, signal: AbortSignal.timeout(120_000) });
      if (!res.ok) throw new Error(`hennepin ${res.status}: ${(await res.text()).slice(0, 140)}`);
      const j: any = await res.json();
      if (j.error) throw new Error(`hennepin error: ${JSON.stringify(j.error).slice(0, 140)}`);
      return j.features || [];
    } catch (e) { lastErr = e; if (a < 2) await new Promise(r => setTimeout(r, 2000 * (a + 1))); }
  }
  throw lastErr;
}

export async function ingestHennepin(opts: { maxPages?: number } = {}): Promise<{ upserted: number }> {
  const runId = await openRun('parcel_hennepin_mn', LAYER);
  const fetchedAt = new Date().toISOString();
  try {
    await registerSourceFieldMap(SOURCE_KEY, FIELD_MAP, 'Hennepin County MN via MN Met Council Metropolitan 7-County Parcels aggregate');
    let offset = 0, page = 0, seen = 0, upserted = 0, sales = 0;
    for (;;) {
      const feats = await fetchPage(offset);
      if (!feats.length) break;
      const batch: ParcelUpsertRow[] = [];
      const events: { pin: string; price: number; date: string }[] = [];
      for (const f of feats) {
        const a = f.attributes || {};
        const pin = s(a.PIN);
        if (!pin) continue;
        seen++;
        const [lng, lat] = centroid(f.geometry);
        const addr = situsAddr(a);
        const norm = addr ? normAddress(addr) : null;
        const salePrice = sanitizeSalePrice(a.SALE_VALUE);
        const saleDate = epochToISO(a.SALE_DATE);
        batch.push({
          county_fips: FIPS, source_id: pin,
          address: norm, norm_address: norm,
          city: s(a.CTU_NAME), zip: s(a.ZIP),
          lat, lng,
          // FIN_SQ_FT is not populated for Hennepin in this aggregate — leave sqft NULL.
          year_built: num(a.YEAR_BUILT), sqft: null, beds: null, baths: null, units: num(a.NUM_UNITS),
          use_desc: s(a.USECLASS1),
          land_value: num(a.EMV_LAND), improvement_value: num(a.EMV_BLDG), total_value: num(a.EMV_TOTAL),
          tax_year: s(a.TAX_YEAR), owner_name: s(a.OWNER_NAME), zoning: null,
          last_sale_date: saleDate, last_sale_price: salePrice,
          extra: JSON.stringify({
            owner_more: s(a.OWNER_MORE) || undefined,
            home_style: s(a.HOME_STYLE) || undefined,
            legal: s(a.ABB_LEGAL) || undefined,
            note: 'Hennepin County MN via MN Met Council 7-county aggregate — EMV (estimated market value) + priced last-sale + year; sqft (FIN_SQ_FT) not populated for Hennepin in this feed',
          }),
          // ── TK-50 field-level provenance (record-level: one feature → this row) ──
          sourceKey: SOURCE_KEY,
          sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`PIN='${pin}'`)}&outFields=*&f=html`,
          fetchedAt,
          rawSource: JSON.stringify(a),
        });
        if (salePrice && saleDate) events.push({ pin, price: salePrice, date: saleDate });
      }
      upserted += await upsertParcels(batch);
      // priced last-sale events (amount = SALE_VALUE). doc_number = sale date (this
      // aggregate carries no deed book/page).
      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 => [FIPS, e.pin, e.date, e.price, 'Deed', e.date,
            'parcel_hennepin_mn', `${LAYER}/query?where=${encodeURIComponent(`PIN='${e.pin}'`)}&outFields=*&f=html`, JSON.stringify({})]),
        );
        sales += chunk.length;
      }
      offset += feats.length;
      page++;
      if (seen % 100000 < feats.length) console.log(`[hennepin] ${upserted} upserted, ${sales} sales`);
      if (opts.maxPages && page >= opts.maxPages) break;
      if (feats.length < PAGE) break;
    }
    let bulk = 0;
    if (!opts.maxPages) {
      if (upserted < 400000) throw new Error(`only ${upserted} Hennepin parcels (expected ~447k) — layer/paging/filter drift?`);
      // in-DB assemblage guard: this feed has no deed book/page, so a portfolio /
      // multi-parcel sale stamps the SAME (SALE_DATE, SALE_VALUE) onto every parcel
      // in the deal (verified: 25 Hennepin PINs share 2021-06 @ $2.3M). Without this
      // each parcel would record the full price → N-fold inflated sales. Any exact
      // (last_sale_date, last_sale_price) shared by >1 parcel is un-allocatable →
      // null the price + flag bulk.
      // ANALYZE first: after bulk-loading 447k rows into the multi-million-row prod
      // parcel table, stale planner stats made the `county_fips=$1` filter SEQ-SCAN
      // the whole table (a ~4h prod hang, 2026-08-11). Fresh stats let the
      // (county_fips, source_id) PK index range-scan just this county's rows, and
      // dropping the intermediate CTE removes a second full scan.
      await query(`ANALYZE parcel`);
      const b = await query(
        `WITH grp AS (
           SELECT last_sale_date, last_sale_price, COUNT(*)::int AS cnt
             FROM parcel WHERE county_fips=$1 AND last_sale_price IS NOT NULL AND last_sale_date IS NOT NULL
            GROUP BY 1,2 HAVING COUNT(*) > 1)
         UPDATE parcel p SET last_sale_price = NULL,
           extra = p.extra || jsonb_build_object('bulk_sale', true, 'bulk_parcel_count', grp.cnt)
           FROM grp WHERE p.county_fips=$1
             AND p.last_sale_date = grp.last_sale_date AND p.last_sale_price = grp.last_sale_price
           RETURNING 1`, [FIPS]);
      bulk = b.rowCount ?? 0;
      const n = await registerParcelSource(FIPS, LAYER,
        `Hennepin County MN (Minneapolis) via MN Met Council Metropolitan 7-County Parcels aggregate (CO_NAME='HENNEPIN' filter) — ${upserted} parcels: owner (100%), situs address + city + zip, EMV land+bldg+TOTAL (estimated MARKET value, ~95%), year built (~94%), units, use class, priced last-sale (SALE_VALUE + SALE_DATE, ~87%) → ${sales} parcel_event sale rows. sqft (FIN_SQ_FT) not populated for Hennepin in this aggregate. No beds/baths or zoning. ${bulk} multi-parcel (assemblage/portfolio) sales flagged bulk + price nulled.`);
      console.log(`[hennepin] registry ${n}, ${bulk} bulk-flagged`);
    }
    await closeRun(runId, 'ok', { upserted, notes: `Hennepin MN: ${upserted} parcels, ${sales} sales, ${bulk} bulk${opts.maxPages ? ' (bounded verify)' : ''}` });
    console.log(`[hennepin] ok: ${upserted} parcels, ${sales} sales, ${bulk} bulk${opts.maxPages ? ` (bounded ${opts.maxPages}p)` : ''}`);
    return { upserted };
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: String(e.message || e).slice(0, 500) });
    throw e;
  }
}

if (import.meta.url === `file://${process.argv[1]}`) {
  ingestHennepin().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}