`src/ingest/parcels/sandiego_ca.ts`

229 lines
/**
 * 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 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/
 * baths/sqft), zoning, and the LAST recorded document (doctype/docnmbr/docdate)
 * — the deed/sale reference. That doc becomes a parcel_event (event_type='deed')
 * with a source_url back to the public record. Sale PRICE + full deed CHAIN are
 * behind ParcelQuest/ARCC (AB 1785 removed online APN recorder search 2024-12) —
 * a paid/gated follow-up; this pilot lands the free authoritative base + links.
 *
 * Resumable: each run ingests the next MAX_PARCELS_PER_RUN by apn offset, so the
 * hourly loop paginates through Poway across ticks, then wraps to refresh.
 *
 *   npm run ingest:parcels -- sandiego
 */
import { query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { upsertParcels, normAddress, registerParcelSource, registerSourceFieldMap, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';

const FIPS = '06073';
const SOURCE = 'sandiego_ca';
const SOURCE_KEY = 'sandiego_ca';
// TK-50 field-level provenance: our parcel field → the exact source attribute it
// maps from. address is assembled from situs parts; use_desc/zoning are derived
// labels; last_sale_date is the last recorded document date (no price — AB-1785
// gated). SanGIS withholds owner name.
const FIELD_MAP: Record<string, string> = {
  source_id: 'apn', address: 'nucleus_situs_from_nbr+situs_pre_dir+situs_street+situs_suffix+situs_post_dir',
  city: 'situs_community', zip: 'situs_zip', year_built: 'year_effective', sqft: 'total_lvg_area',
  beds: 'bedrooms', baths: 'baths', use_desc: 'asr_landuse', land_value: 'asr_land',
  improvement_value: 'asr_impr', total_value: 'asr_total', zoning: 'asr_zone|nucleus_zone_cd',
  last_sale_date: 'docdate',
};
const LAYER = 'https://geo.sandag.org/server/rest/services/Hosted/Parcels/FeatureServer/0';
// 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 = [
  'apn', 'apn_8', 'parcelid', 'nucleus_situs_from_nbr', 'situs_pre_dir', 'situs_street',
  'situs_suffix', 'situs_post_dir', 'situs_zip', 'situs_community', 'asr_land', 'asr_impr',
  'asr_total', 'asr_zone', 'nucleus_zone_cd', 'asr_landuse', 'year_effective', 'total_lvg_area',
  'bedrooms', 'baths', 'garage_stalls', 'pool', 'doctype', 'docnmbr', 'docdate', 'taxstat', 'ownerocc',
].join(',');

const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const n = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
const intCode = (v: unknown): number | null => { const x = parseInt(String(v ?? ''), 10); return Number.isFinite(x) && x !== 0 ? x : null; };
// 2-digit year -> full year, pivoting at the CURRENT 2-digit year so we never
// emit a future date (yy=30 => 1930, not 2030; yy=24 => 2024). Shared by deeds + built-year.
const THIS_YY = new Date().getFullYear() % 100;
const fullYear = (yy: number): number => (yy <= THIS_YY ? 2000 + yy : 1900 + yy);
// SanGIS encodes baths in TENTHS for HOMES (45 => 4.5). On multi-unit/commercial
// the field is an aggregate whole count (547 total baths) — /10 there is nonsense,
// so treat raw>100 (=>10+ baths) as an un-decodable aggregate and null it.
const bathsVal = (v: unknown): number | null => { const x = intCode(v); return x == null || x > 100 ? null : x / 10; };
// Beds: literal for homes; a huge count is a multi-unit aggregate, not a home attr -> null.
const bedsVal = (v: unknown): number | null => { const x = intCode(v); return x == null || x > 12 ? null : x; };

/** SanGIS docdate is MMDDYY -> ISO. 2-digit year: >30 => 19xx else 20xx. */
function docDate(v: unknown): string | null {
  const t = String(v ?? '').trim();
  if (!/^\d{6}$/.test(t)) return null;
  const mm = +t.slice(0, 2), dd = +t.slice(2, 4), yy = +t.slice(4, 6);
  if (mm < 1 || mm > 12 || dd < 1 || dd > 31) return null;
  // shared guard rejects future/absurd + non-calendar (e.g. Feb 30) dates.
  return sanitizeEventDate(`${fullYear(yy)}-${String(mm).padStart(2, '0')}-${String(dd).padStart(2, '0')}`);
}
/** year_effective is a 2-digit year. */
function effYear(v: unknown): number | null {
  const t = String(v ?? '').trim();
  if (!/^\d{2}$/.test(t) || t === '00') return null;
  return fullYear(+t);
}
function composeAddr(a: any): string | null {
  const parts = [s(a.nucleus_situs_from_nbr), s(a.situs_pre_dir), s(a.situs_street), s(a.situs_suffix), s(a.situs_post_dir)]
    .filter(Boolean);
  const addr = parts.join(' ').replace(/\s+/g, ' ').trim();
  return addr && addr !== '0' ? addr : null;
}
/** rough polygon centroid (avg of first ring) -> [lng,lat] in 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)];
}

async function fetchPage(offset: number): Promise<any[]> {
  const u = new URL(LAYER + '/query');
  u.searchParams.set('where', SCOPE_WHERE);
  u.searchParams.set('outFields', OUT_FIELDS);
  u.searchParams.set('orderByFields', 'apn 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');
  const res = await fetch(u, { signal: AbortSignal.timeout(60_000) });
  if (!res.ok) throw new Error(`SanGIS ${res.status}: ${(await res.text()).slice(0, 160)}`);
  const j: any = await res.json();
  if (j.error) throw new Error(`SanGIS error: ${JSON.stringify(j.error).slice(0, 160)}`);
  return j.features || [];
}

// Only ONE verified per-parcel deep link exists (the SanGIS record — the county
// gated online APN search under AB 1785, and the assessor/recorder/tax portals
// 404/403/503 on every APN-parameterized URL we tested). So we persist ONLY the
// real deep link; the property API constructs honestly-labeled SEARCH portals
// from the county + APN (not stored 16k× as identical homepages).
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)' }];
}

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);
  // 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, 'San Diego County (SanGIS/SANDAG Parcels) — situs + assessed value + structure; deed via last recorded doc');
    const total = await (async () => {
      const u = new URL(LAYER + '/query');
      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`, [CURSOR_KEY]);
    let offset = cur.rows[0]?.next_offset ?? 0;
    if (offset >= total) offset = 0;
    console.log(`[${SOURCE}] scope=${scope} total=${total}, resume offset=${offset}`);

    const rows: ParcelUpsertRow[] = [];
    const events: any[] = [];
    const linkRows: any[] = [];
    let fetched = 0;

    while (fetched < MAX_PER_RUN) {
      const feats = await fetchPage(offset);
      if (!feats.length) break;
      for (const f of feats) {
        const a = f.attributes; const apn = s(a.apn); if (!apn) continue;
        const [lng, lat] = centroid(f.geometry);
        const addr = composeAddr(a);
        const sale = docDate(a.docdate);
        rows.push({
          county_fips: FIPS, source_id: apn,
          address: addr, norm_address: addr ? normAddress(addr) : null,
          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,
          use_desc: s(a.asr_landuse) ? `Land use ${s(a.asr_landuse)}` : null,
          land_value: n(a.asr_land), improvement_value: n(a.asr_impr), total_value: n(a.asr_total),
          tax_year: null, owner_name: null, zoning: s(a.asr_zone) || s(a.nucleus_zone_cd),
          last_sale_date: sale, last_sale_price: null,
          extra: JSON.stringify({ apn_8: s(a.apn_8), parcelid: s(a.parcelid), doctype: s(a.doctype),
            docnmbr: s(a.docnmbr), asr_landuse: s(a.asr_landuse), taxstat: s(a.taxstat),
            ownerocc: s(a.ownerocc), garage_stalls: s(a.garage_stalls), pool: s(a.pool), community: 'POWAY' }),
          // ── TK-50 field-level provenance (record-level: one feature → this row) ──
          sourceKey: SOURCE_KEY,
          sourceUrl: `${LAYER}/query?where=apn%3D%27${apn}%27&outFields=*&f=html`,
          fetchedAt,
          rawSource: JSON.stringify(a),
        });
        if (sale) events.push({ apn, date: sale, doctype: s(a.doctype), docnmbr: s(a.docnmbr) });
        for (const l of links(apn)) linkRows.push({ apn, ...l });
      }
      fetched += feats.length;
      offset += feats.length;
      if (feats.length < PAGE) break; // last page
    }

    const up = await upsertParcels(rows);

    // deed events (idempotent) + public-record links
    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, doc_type, doc_number, source, source_url, detail)
         VALUES ${chunk.map((_, j) => { const b = j * 8; return `($${b+1},$${b+2},'deed',$${b+3}::date,$${b+4},$${b+5},$${b+6},$${b+7},$${b+8}::jsonb)`; }).join(',')}
         ON CONFLICT (county_fips, source_id, event_type, event_date, doc_number) DO NOTHING`,
        chunk.flatMap(e => [FIPS, e.apn, e.date, e.doctype, e.docnmbr ?? '', SOURCE,
          `${LAYER}/query?where=apn%3D%27${e.apn}%27&outFields=*&f=html`, JSON.stringify({ recorded: e.date })]),
      );
    }
    for (let i = 0; i < linkRows.length; i += 400) {
      const chunk = linkRows.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(l => [FIPS, l.apn, l.kind, l.url, l.label]),
      );
    }

    // persist the forward cursor (wrap on full coverage)
    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()`,
      [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 };
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: e.message });
    throw e;
  }
}