← back to Stayclaim

scripts/ingest-la-lahd.ts

210 lines

/**
 * ingest-la-lahd.ts
 *
 * LA Housing Department (LAHD) Property Look-Up datasets — 7 sources covering
 * tenant/landlord history, evictions, code enforcement, foreclosures.
 *
 *   - 2u8b-eyuu  Eviction Notices                  (99K)
 *   - eagk-wq48  Investigation & Enforcement       (115K)
 *   - ds2y-sb5t  CCRIS Cases                       (316K)
 *   - vpax-89xu  Landlord Declarations             (11K)
 *   - n9x9-zisb  Case Analysis                     (11K)
 *   - ci3m-f23k  Tenant Buyout                     (8K)
 *   - 2qnc-kq4g  Registered Foreclosure Properties (2K)
 *
 * Stored as `place_event` rows linked by APN match to listings (when possible)
 * or to a stub listing keyed by address.
 *
 * Tier: A (LA Housing Department primary record).
 */
import { Pool } from 'pg';

const pool = new Pool({
  host: process.env.PGHOST ?? '/tmp',
  database: process.env.PGDATABASE ?? 'stayclaim',
  user: process.env.PGUSER ?? process.env.USER,
  password: process.env.PGPASSWORD,
  port: parseInt(process.env.PGPORT ?? '5432', 10),
  max: 6,
});

type DS = {
  slug: string;
  kind: string;
  source_label: string;
  expected: number;
  // mapper: row → { apn, addr, date, summary, source_id }
  map: (r: any) => { apn: string | null; addr: string | null; date: string | null; summary: string; source_id: string } | null;
};

function pd(v?: string): string | null {
  if (!v) return null;
  // Sometimes "MM-DD-YYYY", sometimes ISO
  const iso = v.match(/^(\d{4})-(\d{2})-(\d{2})/);
  if (iso) return `${iso[1]}-${iso[2]}-${iso[3]}`;
  const us = v.match(/^(\d{2})-(\d{2})-(\d{4})/);
  if (us) return `${us[3]}-${us[1]}-${us[2]}`;
  return null;
}

const DATASETS: DS[] = [
  {
    slug: '2u8b-eyuu', kind: 'eviction', source_label: 'LAHD Eviction Notice', expected: 99832,
    map: (r) => r.officialaddress || r.apn ? {
      apn: r.apn ?? null,
      addr: r.officialaddress ?? null,
      date: pd(r.notice_date ?? r.received),
      summary: `Eviction notice (${r.notice_type ?? '?'}) — ${r.eviction_category ?? 'category n/a'}`,
      source_id: `lahd_eviction:${r.apn ?? ''}:${r.notice_date ?? r.received ?? ''}:${r.notice_type ?? ''}`.slice(0, 200),
    } : null,
  },
  {
    slug: 'eagk-wq48', kind: 'investigation', source_label: 'LAHD Investigation & Enforcement', expected: 115111,
    map: (r) => r.apn || r.officialaddress ? {
      apn: r.apn ?? null,
      addr: r.officialaddress ?? r.address ?? null,
      date: pd(r.case_open_date ?? r.case_opened ?? r.received ?? r.date_opened),
      summary: `LAHD investigation/enforcement case ${r.case_number ?? ''}`.trim(),
      source_id: `lahd_investigation:${r.case_number ?? r.apn ?? ''}`.slice(0, 200),
    } : null,
  },
  {
    slug: 'ds2y-sb5t', kind: 'ccris_case', source_label: 'LAHD CCRIS Case', expected: 316832,
    map: (r) => r.apn || r.address ? {
      apn: r.apn ?? null,
      addr: r.address ?? r.officialaddress ?? null,
      date: pd(r.case_open_date ?? r.received ?? r.date),
      summary: `CCRIS case ${r.case_number ?? r.apn ?? ''}`.trim(),
      source_id: `lahd_ccris:${r.case_number ?? r.apn ?? ''}`.slice(0, 200),
    } : null,
  },
  {
    slug: 'vpax-89xu', kind: 'landlord_decl', source_label: 'LAHD Landlord Declaration', expected: 11382,
    map: (r) => r.apn || r.address ? {
      apn: r.apn ?? null,
      addr: r.address ?? r.officialaddress ?? null,
      date: pd(r.received ?? r.date),
      summary: `Landlord declaration filed (${r.declaration_type ?? 'type n/a'})`,
      source_id: `lahd_landlord:${r.case_number ?? r.apn ?? ''}:${r.received ?? ''}`.slice(0, 200),
    } : null,
  },
  {
    slug: 'n9x9-zisb', kind: 'case_analysis', source_label: 'LAHD Case Analysis', expected: 11245,
    map: (r) => r.apn || r.address ? {
      apn: r.apn ?? null,
      addr: r.address ?? r.officialaddress ?? null,
      date: pd(r.received ?? r.date_opened ?? r.date),
      summary: `LAHD case analysis ${r.case_number ?? r.apn ?? ''}`.trim(),
      source_id: `lahd_analysis:${r.case_number ?? r.apn ?? ''}`.slice(0, 200),
    } : null,
  },
  {
    slug: 'ci3m-f23k', kind: 'tenant_buyout', source_label: 'LAHD Tenant Buyout', expected: 8792,
    map: (r) => r.apn || r.address ? {
      apn: r.apn ?? null,
      addr: r.address ?? r.officialaddress ?? null,
      date: pd(r.date_filed ?? r.received ?? r.date),
      summary: `Tenant buyout filing (${r.amount ? `$${r.amount}` : 'amount n/a'})`,
      source_id: `lahd_buyout:${r.case_number ?? r.apn ?? ''}:${r.date_filed ?? ''}`.slice(0, 200),
    } : null,
  },
  {
    slug: '2qnc-kq4g', kind: 'foreclosure', source_label: 'LAHD Foreclosure Registration', expected: 2143,
    map: (r) => r.apn || r.propertyaddress ? {
      apn: r.apn ?? null,
      addr: r.propertyaddress ?? null,
      date: pd(r.registered_date),
      summary: `Foreclosure registered with lender ${r.lender ?? '?'}`,
      source_id: `lahd_foreclosure:${r.apn ?? r.propertyaddress ?? ''}`.slice(0, 200),
    } : null,
  },
];

const PAGE = 50000;
const BATCH = 1000;

async function fetchPage(slug: string, offset: number): Promise<any[]> {
  const u = `https://data.lacity.org/resource/${slug}.json?$limit=${PAGE}&$offset=${offset}`;
  for (let attempt = 1; attempt <= 4; attempt++) {
    try {
      const res = await fetch(u);
      if (res.status === 429 || res.status === 503) { await new Promise(r => setTimeout(r, 2000 * attempt)); continue; }
      if (!res.ok) throw new Error(`HTTP ${res.status}`);
      return await res.json();
    } catch (e) {
      if (attempt >= 4) throw e;
      await new Promise(r => setTimeout(r, 1500 * attempt));
    }
  }
  return [];
}

async function processDataset(ds: DS) {
  console.log(`\n=== ${ds.slug} ${ds.source_label} (${ds.expected.toLocaleString()}) ===`);
  let offset = 0, total = 0, inserted = 0;
  const t0 = Date.now();
  while (true) {
    const rows = await fetchPage(ds.slug, offset);
    if (rows.length === 0) break;
    // map each row, dedupe by source_id
    const seen = new Set<string>();
    const events = rows.map(r => ds.map(r)).filter((e): e is NonNullable<ReturnType<typeof ds.map>> => !!e && !seen.has(e.source_id) && (seen.add(e.source_id), true));
    if (events.length === 0) {
      total += rows.length;
      offset += rows.length;
      continue;
    }
    for (let i = 0; i < events.length; i += BATCH) {
      const slice = events.slice(i, i + BATCH);
      const tuples: string[] = [];
      const values: any[] = [];
      slice.forEach((e, idx) => {
        const base = idx * 5;
        tuples.push(`($${base+1},$${base+2}::date,$${base+3},$${base+4},$${base+5})`);
        values.push(e.summary, e.date, e.source_id, e.apn, e.addr);
      });
      // Match listing by APN first (fast indexed), then by address fallback
      const r = await pool.query(
        `INSERT INTO place_event
           (listing_id, kind, summary, valid_from, source_tier, source_label, source_url, source_id, public_visible)
         SELECT
           COALESCE(
             (SELECT lp.listing_id FROM permit lp WHERE lp.apn IS NOT NULL AND lp.apn = t.apn LIMIT 1),
             (SELECT id FROM listing WHERE source = 'ladbs_permit'
               AND source_id = 'ladbs:' || lower(regexp_replace(regexp_replace(coalesce(t.addr,''), '[^\\w\\s-]', '', 'g'), '\\s+', '-', 'g'))
               LIMIT 1)
           ),
           '${ds.kind}', t.summary, t.dt::date, 'A', '${ds.source_label}',
           'https://data.lacity.org/resource/${ds.slug}.json',
           t.sid, true
         FROM (VALUES ${tuples.map((_, k) => {
           const b = k * 5;
           return `($${b+1}::text,$${b+2},$${b+3}::text,$${b+4}::text,$${b+5}::text)`;
         }).join(',')}) AS t(summary, dt, sid, apn, addr)
         WHERE COALESCE(
           (SELECT lp.listing_id FROM permit lp WHERE lp.apn IS NOT NULL AND lp.apn = t.apn LIMIT 1),
           (SELECT id FROM listing WHERE source = 'ladbs_permit'
             AND source_id = 'ladbs:' || lower(regexp_replace(regexp_replace(coalesce(t.addr,''), '[^\\w\\s-]', '', 'g'), '\\s+', '-', 'g'))
             LIMIT 1)
         ) IS NOT NULL
         ON CONFLICT DO NOTHING`,
        values
      );
      inserted += r.rowCount ?? 0;
    }
    total += rows.length;
    const dt = (Date.now() - t0) / 1000;
    console.log(`  ${ds.slug}: ${total.toLocaleString()}/${ds.expected.toLocaleString()} pulled, ${inserted.toLocaleString()} events inserted (${(total/dt).toFixed(0)}/s)`);
    if (rows.length < PAGE) break;
    offset += PAGE;
  }
  console.log(`  ✓ ${ds.slug}: ${total.toLocaleString()} pulled, ${inserted.toLocaleString()} events`);
}

async function main() {
  for (const ds of DATASETS) await processDataset(ds);
  await pool.end();
}

main().catch(e => { console.error('FATAL', e); process.exit(1); });