`src/ingest/parcels/philadelphia_pa.ts`

168 lines
/**
 * Philadelphia PA (FIPS 42101) parcel ingest — dedicated full-universe adapter
 * (applies the Wake/Miami-Dade DTD verdict B precedent: a rich full-universe
 * keyless source gets a dedicated adapter so the real OWNER + market value +
 * priced sale are preserved).
 *
 * Source: the Philadelphia OPA (Office of Property Assessment) "opa_properties_
 * public" dataset on OpenDataPhilly, served by the keyless Carto SQL API ($0):
 *   https://phl.carto.com/api/v2/sql
 * This is the FIRST adapter over a Carto SQL feed (all prior adapters are ArcGIS
 * or Socrata). Pagination is plain SQL LIMIT/OFFSET over a stable ORDER BY
 * parcel_number; geometry is a PostGIS point, so lat/lng come back as computed
 * ST_Y/ST_X columns (no centroid math needed).
 *
 * Richest county in the system: owner_1 (~100%), location (situs address),
 * market_value (~100%, TRUE market value), sale_price + sale_date (~99%, a real
 * priced last-sale — 73% arm's-length >$1, the rest nominal $1 family/deed
 * transfers which the shared price guard keeps as legitimate low-consideration
 * sales), year_built (92%), total_livable_area (sqft, 92%), number_of_bedrooms
 * + number_of_bathrooms (the ONLY full-universe county so far with beds/baths),
 * zoning, category_code_description (use). taxable_land/taxable_building (the
 * post-abatement taxable split) are carried in extra.
 *
 * source_id = parcel_number (the canonical OPA account, unique + stable).
 *
 * Run: NODE_OPTIONS=--max-old-space-size=4096 npm run ingest:parcels philadelphia
 */
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 = '42101';
const SOURCE_KEY = 'philadelphia_pa';
const SQL_API = 'https://phl.carto.com/api/v2/sql';
const TABLE = 'opa_properties_public';
const PAGE = 2000;
// TK-50 field-level provenance: our parcel field → the exact source column.
const FIELD_MAP: Record<string, string> = {
  source_id: 'parcel_number', address: 'location', zip: 'zip_code',
  year_built: 'year_built', sqft: 'total_livable_area', beds: 'number_of_bedrooms',
  baths: 'number_of_bathrooms', use_desc: 'category_code_description', zoning: 'zoning',
  total_value: 'market_value', owner_name: 'owner_1',
  last_sale_price: 'sale_price', 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; };
// year_built is a varchar in OPA ("1960", occasionally "" / "0") — accept a plain
// 3-4 digit year, reject everything else.
const yearNum = (v: unknown): number | null => { const t = s(v); if (!t || !/^\d{3,4}$/.test(t)) return null; const y = Number(t); return y > 0 ? y : null; };
// sale_date is a timestamptz ("2024-06-17T04:00:00Z") — take the calendar date,
// then run it through the shared future/pre-1900/non-calendar guard.
const saleDateToISO = (v: unknown): string | null => { const t = s(v); if (!t) return null; return sanitizeEventDate(t.slice(0, 10)); };

// The exact columns pulled per row (lat/lng are computed from the_geom).
const COLS = ['parcel_number', 'owner_1', 'owner_2', 'location', 'zip_code',
  'market_value', 'taxable_land', 'taxable_building', 'sale_price', 'sale_date',
  'year_built', 'total_livable_area', 'total_area', 'category_code_description',
  'number_of_bedrooms', 'number_of_bathrooms', 'zoning',
  'ST_Y(the_geom) AS lat', 'ST_X(the_geom) AS lng'].join(',');

async function fetchPage(offset: number): Promise<any[]> {
  const sql = `SELECT ${COLS} FROM ${TABLE} ORDER BY parcel_number LIMIT ${PAGE} OFFSET ${offset}`;
  const u = new URL(SQL_API);
  u.searchParams.set('q', sql);
  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(`philly ${res.status}: ${(await res.text()).slice(0, 140)}`);
      const j: any = await res.json();
      if (j.error) throw new Error(`philly error: ${JSON.stringify(j.error).slice(0, 140)}`);
      return j.rows || [];
    } catch (e) { lastErr = e; if (a < 2) await new Promise(r => setTimeout(r, 2000 * (a + 1))); }
  }
  throw lastErr;
}

export async function ingestPhiladelphia(opts: { maxPages?: number } = {}): Promise<{ upserted: number }> {
  const runId = await openRun('parcel_philadelphia_pa', SQL_API);
  const fetchedAt = new Date().toISOString();
  try {
    await registerSourceFieldMap(SOURCE_KEY, FIELD_MAP, 'Philadelphia OPA (opa_properties_public) via keyless Carto SQL API');
    let offset = 0, page = 0, seen = 0, upserted = 0, sales = 0;
    for (;;) {
      const rows = await fetchPage(offset);
      if (!rows.length) break;
      const batch: ParcelUpsertRow[] = [];
      const events: { id: string; price: number; date: string }[] = [];
      for (const a of rows) {
        const id = s(a.parcel_number);
        if (!id) continue;
        seen++;
        const addr = s(a.location);
        const norm = addr ? normAddress(addr) : null;
        const salePrice = sanitizeSalePrice(a.sale_price);
        const saleDate = saleDateToISO(a.sale_date);
        const lat = num(a.lat), lng = num(a.lng);
        batch.push({
          county_fips: FIPS, source_id: id,
          address: norm, norm_address: norm,
          // Philadelphia is a consolidated city-county; the feed carries no situs
          // city column (every parcel is Philadelphia) — set it explicitly.
          city: 'PHILADELPHIA', zip: s(a.zip_code),
          lat, lng,
          year_built: yearNum(a.year_built), sqft: num(a.total_livable_area),
          beds: num(a.number_of_bedrooms), baths: num(a.number_of_bathrooms), units: null,
          use_desc: s(a.category_code_description),
          land_value: null, improvement_value: null, total_value: num(a.market_value),
          tax_year: null, owner_name: s(a.owner_1), zoning: s(a.zoning),
          last_sale_date: saleDate, last_sale_price: salePrice,
          extra: JSON.stringify({
            owner_2: s(a.owner_2) || undefined,
            taxable_land: num(a.taxable_land) || undefined,
            taxable_building: num(a.taxable_building) || undefined,
            land_sqft: num(a.total_area) || undefined,
            note: 'Philadelphia OPA (opa_properties_public) via Carto SQL — market value + priced last-sale + beds/baths + zoning; taxable_* are the post-abatement taxable split, not the market land/building split',
          }),
          // ── TK-50 field-level provenance (record-level: one row → this parcel) ──
          sourceKey: SOURCE_KEY,
          sourceUrl: `${SQL_API}?q=${encodeURIComponent(`SELECT * FROM ${TABLE} WHERE parcel_number='${id}'`)}`,
          fetchedAt,
          rawSource: JSON.stringify(a),
        });
        if (salePrice && saleDate) events.push({ id, price: salePrice, date: saleDate });
      }
      upserted += await upsertParcels(batch);
      // record priced last-sale events (amount = sale_price). doc_number = sale_date
      // (this feed carries no deed book/page in the public dataset), matching the
      // (county,source_id,event_type,event_date,doc_number) dedup key.
      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.id, e.date, e.price, 'Deed', e.date,
            'parcel_philadelphia_pa', `https://property.phila.gov/?p=${e.id}`, JSON.stringify({ nominal: e.price <= 100 || undefined })]),
        );
        sales += chunk.length;
      }
      offset += rows.length;
      page++;
      if (seen % 100000 < rows.length) console.log(`[philly] ${upserted} upserted, ${sales} sales`);
      if (opts.maxPages && page >= opts.maxPages) break;
      if (rows.length < PAGE) break;
    }
    if (!opts.maxPages) {
      if (upserted < 500000) throw new Error(`only ${upserted} Philadelphia parcels (expected ~584k) — Carto/paging drift?`);
      const n = await registerParcelSource(FIPS, SQL_API,
        `Philadelphia OPA (opa_properties_public) via keyless Carto SQL API — ${upserted} parcels: owner, situs address, TRUE market value (~100%), priced last-sale (price+date, ~99%) → ${sales} parcel_event sale rows, year built (92%), living-area sqft (92%), beds + baths (first full-universe county with beds/baths), zoning, use. Nominal $1/family transfers kept as legitimate low-consideration sales.`);
      console.log(`[philly] registry ${n}`);
    }
    await closeRun(runId, 'ok', { upserted, notes: `Philadelphia PA: ${upserted} parcels, ${sales} sales${opts.maxPages ? ' (bounded verify)' : ''}` });
    console.log(`[philly] ok: ${upserted} parcels, ${sales} sales${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]}`) {
  ingestPhiladelphia().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}