`src/ingest/parcels/fulton_ga.ts`

184 lines
/**
 * Fulton County GA (Atlanta, FIPS 13121) parcel ingest — dedicated full-universe
 * adapter (applies the Wake/Miami-Dade/Tarrant/Bexar DTD verdict B precedent: a
 * rich full-universe keyless ArcGIS county gets a dedicated adapter, not the
 * sale-cursor generic one, so the real OWNER + physical-detail fields are
 * preserved).
 *
 * Source: the Fulton County Dept. of IT / GIS Division hosted "Tax Parcels"
 * FeatureServer (built from the Fulton County Board of Assessors tax digest;
 * keyless, maxRecordCount 1000, supportsPagination, $0):
 *   https://services1.arcgis.com/AQDHTHDrZzfsFsB5/arcgis/rest/services/Tax_Parcels/FeatureServer/0
 * The serviceDescription self-identifies as "the Fulton County Board of
 * Assessor's ... annual tax digest." The layer serves outSR=4326 server-side, so
 * parcel centroids come back as clean WGS84 lat/lng.
 *
 * Coverage (373,004-parcel universe, TaxYear 2026 — current): Owner (99.98%,
 * 372,938), Address (100% — assembled from the structured AddrNumber/PreDir/
 * Street/Suffix/PosDir/UnitType/Unit components, which is cleaner than the flat
 * `Address` string), LUCode (land-use code, 100%), ClassCode, LivUnits
 * (living/dwelling units, 86%), LandAcres, Neighborhood.
 *
 * This Fulton assessor open view carries NO market/assessed value, NO
 * year-built, NO building sqft, NO beds/baths, and NO sale/deed feed. Because
 * there is no priced-sale feed here, this county is parcel-DETAIL-only (like
 * Bexar TX / Salt Lake UT): zero parcel_event rows are written — never a
 * fabricated sale.
 *
 * source_id = ParcelID (the Fulton Board of Assessors parcel account id, e.g.
 * "07 410001590187"). We keep the raw ParcelID including its embedded space (it
 * is the canonical key on the Fulton digest); rows with a blank ParcelID are
 * skipped.
 *
 * Run: NODE_OPTIONS=--max-old-space-size=4096 npm run ingest:parcels fulton
 */
import { pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.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; };

/**
 * Assemble a clean street line from the structured Fulton address components,
 * in USPS-ish order: "<num> <predir> <street> <suffix> <posdir> <unittype> <unit>".
 * We prefer the components over the flat `Address` field because they are
 * separately reliable; fall back to the flat `Address` if no components exist.
 */
function buildStreet(a: any): string | null {
  const parts = [a.AddrNumber, a.AddrPreDir, a.AddrStreet, a.AddrSuffix, a.AddrPosDir, a.AddrUntTyp, a.AddrUnit]
    .map(s).filter(Boolean);
  const assembled = parts.join(' ').replace(/\s+/g, ' ').trim();
  // Fulton uses a literal "0" AddrNumber for unaddressed land parcels; when the
  // number is "0" and there's a real street, drop the leading 0 so the address
  // isn't "0 GULLATT RD" (keep the street though — it's still location info).
  const cleaned = assembled.replace(/^0\s+/, '').trim();
  return cleaned || s(a.Address);
}

/** plain vertex-average of ring[0] — NOT a true polygon centroid (fine for a map
 *  pin; multipart parcels only get the first ring). Geometry is already WGS84
 *  because we request outSR=4326. */
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 = ['ParcelID', 'TaxYear', 'Address', 'AddrNumber', 'AddrPreDir', 'AddrStreet',
  'AddrSuffix', 'AddrPosDir', 'AddrUntTyp', 'AddrUnit', 'Owner', 'LUCode', 'ClassCode',
  'ExCode', 'LivUnits', 'LandAcres', 'NbrHood', 'Subdiv'].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(`fulton ${res.status}: ${(await res.text()).slice(0, 140)}`);
      const j: any = await res.json();
      if (j.error) throw new Error(`fulton 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 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);
      if (!feats.length) break;
      const batch: ParcelUpsertRow[] = [];
      for (const f of feats) {
        const a = f.attributes || {};
        const pid = s(a.ParcelID);
        if (!pid) { skipped++; continue; }
        const owner = s(a.Owner);
        const street = buildStreet(a);
        // skip pure ghost polygons (no owner AND no street) so we don't store bare geometry rows
        if (!owner && !street) { skipped++; continue; }
        seen++;
        const [lng, lat] = centroid(f.geometry);
        const norm = street ? normAddress(street) : null;
        batch.push({
          county_fips: FIPS, source_id: pid,
          address: norm, norm_address: norm,
          // Fulton's tax-parcel view has no separate city column (the county
          // spans Atlanta + ~14 other municipalities); leave city null rather
          // than guess it from the neighborhood code.
          city: null, zip: null,
          lat, lng,
          year_built: null, sqft: null, beds: null, baths: null,
          units: num(a.LivUnits),
          use_desc: s(a.LUCode),
          land_value: null, improvement_value: null, total_value: null,
          tax_year: s(a.TaxYear), owner_name: owner, zoning: null,
          // no priced-sale / deed feed in this Fulton assessor open view — leave
          // both sale fields NULL (never fabricate a sale date or price).
          last_sale_date: null, last_sale_price: null,
          extra: JSON.stringify({
            class_code: s(a.ClassCode) || undefined,
            ex_code: s(a.ExCode) || undefined,
            land_acres: num(a.LandAcres) || undefined,
            neighborhood: s(a.NbrHood) || undefined,
            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);
      offset += feats.length;
      if (seen % 100000 < feats.length) console.log(`[fulton] ${upserted} upserted, ${skipped} stubs skipped`);
      if (feats.length < PAGE) break;
    }
    if (upserted < 340000) throw new Error(`only ${upserted} Fulton parcels (expected ~373k) — layer/paging/filter drift?`);

    const n = await registerParcelSource(FIPS, LAYER,
      `Fulton County GA (Atlanta) Board of Assessors tax digest via Fulton DoIT GIS Tax_Parcels view — ${upserted} parcels: owner (99.98%), property address assembled from structured components (100%), land-use code (LUCode, 100%), class code, living/dwelling units (LivUnits, 86%), land acres, neighborhood, Map (polygon centroid, WGS84 via outSR). TaxYear 2026 (current). No market/assessed value, year-built, building sqft, beds/baths, or sale/deed feed in this open assessor view (parcel-detail-only, like Bexar/Salt Lake — zero fabricated events). ${skipped} placeholder/geometry-only stubs skipped.`);
    await closeRun(runId, 'ok', { upserted, skipped, notes: `Fulton GA: ${upserted} parcels, ${skipped} stubs skipped` });
    console.log(`[fulton] ok: ${upserted} parcels, ${skipped} stubs (registry ${n})`);
    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]}`) {
  ingestFulton().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}