`src/ingest/parcels/dallas_tx.ts`

155 lines
/**
 * Dallas County TX (Dallas, FIPS 48113) parcel ingest — dedicated full-universe
 * adapter (applies the Wake/Tarrant DTD verdict B precedent).
 *
 * Source: the City of Dallas GIS mirror of the DCAD appraisal-account file — the
 * "Tax Account Points" layer (keyless, supportsPagination, 2k/page, POINT geom,
 * $0):
 *   https://services2.arcgis.com/rwnOSbfKSwyTBcwN/arcgis/rest/services/CRMHostedLayers/FeatureServer/0
 * (DCAD's own hosts — maps.dcad.org / gis.dcad.org / dallascad.org — 404 /
 * connection-refuse to headless; this City mirror is the reachable full-county
 * account universe, ~548,830 accounts.)
 *
 * Coverage: Owner1 (~95%), situs address (SiteAddrNum + SiteStreetname, ~95%),
 * LandVal + ImpVal + TotalVal (TRUE market/total value, ~93%), PropClass (use,
 * ~94%), AppraisalYr (tax year), CityJuris (city). This is a parcel-detail +
 * VALUE feed with NO sqft, NO year_built, and NO structured sale/deed feed (deed
 * instrument text lives unparsed inside Legal4) → like Bexar, ZERO parcel_event
 * rows are written; no sale is ever fabricated.
 *
 * source_id = AccountID (the 17-digit DCAD account). Rows with no AccountID
 * (mineral / personal-property shells) are skipped.
 *
 * Run: NODE_OPTIONS=--max-old-space-size=4096 npm run ingest:parcels dallas
 */
import { pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';

const FIPS = '48113';
const SOURCE_KEY = 'dallas_tx';
const LAYER = 'https://services2.arcgis.com/rwnOSbfKSwyTBcwN/arcgis/rest/services/CRMHostedLayers/FeatureServer/0';
const PAGE = 2000;
// TK-50 field-level provenance. total_value maps from TotalVal (TRUE market/total).
const FIELD_MAP: Record<string, string> = {
  source_id: 'AccountID', address: 'SiteAddrNum+SiteStreetname', city: 'CityJuris', tax_year: 'AppraisalYr',
  use_desc: 'PropClass', land_value: 'LandVal', improvement_value: 'ImpVal', total_value: 'TotalVal',
  owner_name: 'Owner1',
};

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; };

/** situs address = SiteAddrNum + ' ' + SiteStreetname (some SiteStreetname values
 *  carry a trailing space — trim/collapse). */
function situsAddr(a: any): string | null {
  const joined = [s(a.SiteAddrNum), s(a.SiteStreetname)].filter((x): x is string => !!x).join(' ').replace(/\s+/g, ' ').trim();
  return joined || null;
}

/** This layer is a POINT layer — geometry is {x,y} in WGS84 (outSR=4326), not
 *  polygon rings. Return [lng, lat]. */
function pointLngLat(geom: any): [number | null, number | null] {
  const x = Number(geom?.x), y = Number(geom?.y);
  if (!Number.isFinite(x) || !Number.isFinite(y) || (x === 0 && y === 0)) return [null, null];
  return [+x.toFixed(6), +y.toFixed(6)];
}

const OUT = ['AccountID', 'ParcelID', 'Owner1', 'Owner2', 'SiteAddrNum', 'SiteStreetname',
  'CityJuris', 'CountyJuris', 'AppraisalYr', 'LandVal', 'ImpVal', 'TotalVal',
  'PropClass', 'BldgClass', 'ResCom', 'NbhdCode'].join(',');

async function fetchPage(offset: number): Promise<any[]> {
  const u = new URL(LAYER + '/query');
  u.searchParams.set('where', '1=1');
  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(`dallas ${res.status}: ${(await res.text()).slice(0, 140)}`);
      const j: any = await res.json();
      if (j.error) throw new Error(`dallas 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 ingestDallas(opts: { maxPages?: number } = {}): Promise<{ upserted: number }> {
  const runId = await openRun('parcel_dallas_tx', LAYER);
  const fetchedAt = new Date().toISOString();
  try {
    await registerSourceFieldMap(SOURCE_KEY, FIELD_MAP, 'Dallas County TX (DCAD) via City of Dallas GIS Tax Account Points mirror');
    let offset = 0, page = 0, seen = 0, skipped = 0, upserted = 0;
    for (;;) {
      const feats = await fetchPage(offset);
      if (!feats.length) break;
      const batch: ParcelUpsertRow[] = [];
      for (const f of feats) {
        const a = f.attributes || {};
        const acct = s(a.AccountID);
        if (!acct) { skipped++; continue; }
        seen++;
        const [lng, lat] = pointLngLat(f.geometry);
        const addr = situsAddr(a);
        const norm = addr ? normAddress(addr) : null;
        batch.push({
          county_fips: FIPS, source_id: acct,
          address: norm, norm_address: norm,
          city: s(a.CityJuris), zip: null,
          lat, lng,
          year_built: null, sqft: null, beds: null, baths: null, units: null,
          use_desc: s(a.PropClass),
          land_value: num(a.LandVal), improvement_value: num(a.ImpVal), total_value: num(a.TotalVal),
          tax_year: s(a.AppraisalYr), owner_name: s(a.Owner1), zoning: null,
          // no sale/deed feed in this layer → no last-sale; never fabricated.
          last_sale_date: null, last_sale_price: null,
          extra: JSON.stringify({
            parcel_id: s(a.ParcelID) || undefined,
            owner_2: s(a.Owner2) || undefined,
            bldg_class: s(a.BldgClass) || undefined,
            res_com: s(a.ResCom) || undefined,
            nbhd: s(a.NbhdCode) || undefined,
            county_juris: s(a.CountyJuris) || undefined,
            note: 'Dallas County TX (DCAD) via City of Dallas GIS Tax Account Points mirror — TRUE market/total value; no sqft, year built, or sale feed in this layer (deed text lives unparsed in Legal4)',
          }),
          // ── TK-50 field-level provenance (record-level: one feature → this row) ──
          sourceKey: SOURCE_KEY,
          sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`AccountID='${acct}'`)}&outFields=*&f=html`,
          fetchedAt,
          rawSource: JSON.stringify(a),
        });
      }
      upserted += await upsertParcels(batch);
      offset += feats.length;
      page++;
      if (seen % 100000 < feats.length) console.log(`[dallas] ${upserted} upserted, ${skipped} stubs skipped`);
      if (opts.maxPages && page >= opts.maxPages) break;
      if (feats.length < PAGE) break;
    }
    if (!opts.maxPages) {
      if (upserted < 480000) throw new Error(`only ${upserted} Dallas parcels (expected ~549k) — layer/paging drift?`);
      const n = await registerParcelSource(FIPS, LAYER,
        `Dallas County TX (Dallas) DCAD appraisal accounts via City of Dallas GIS Tax Account Points mirror — ${upserted} parcels: owner (~95%), situs address, LAND+IMP + TRUE MARKET/total value (~93%), property-class use, appraisal year, jurisdiction city. No sqft, year built, beds/baths, or sale/deed feed in this layer (deed text unparsed in Legal4) → ZERO parcel_event rows (no fabricated sales). ${skipped} account-less (mineral/personal-property) rows skipped.`);
      console.log(`[dallas] registry ${n}`);
    }
    await closeRun(runId, 'ok', { upserted, skipped, notes: `Dallas TX: ${upserted} parcels, ${skipped} stubs${opts.maxPages ? ' (bounded verify)' : ''}` });
    console.log(`[dallas] ok: ${upserted} parcels, 0 events, ${skipped} stubs${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]}`) {
  ingestDallas().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}