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