← back to Stayclaim
scripts/ingest-la-assessor-legacy.ts
211 lines
/**
* ingest-la-assessor-legacy.ts
*
* Downloads the LA County Assessor "Parcels Data 2006 thru 2021" — 17.4 GB CSV
* — and ingests into the existing `parcel` table. Fills coverage gaps in the
* 2021-Present roll (e.g. APN 4349-040-002 which lives in this older dataset
* but was missing from the current snapshot).
*
* Source: ArcGIS Hub item bffc21600e5f408ea6791d1bce7738ae
* download URL: https://www.arcgis.com/sharing/rest/content/items/.../data
*
* Tier: A (LA County Assessor primary record).
*/
import { Pool } from 'pg';
import { spawn } from 'node:child_process';
import { createReadStream, statSync, createWriteStream } from 'node:fs';
import { mkdir, rm } from 'node:fs/promises';
import { createInterface } from 'node:readline';
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,
});
const ITEM_ID = 'bffc21600e5f408ea6791d1bce7738ae';
const ZIP_URL = `https://www.arcgis.com/sharing/rest/content/items/${ITEM_ID}/data`;
const WORKDIR = '/tmp/la-assessor-legacy';
const ZIP_PATH = `${WORKDIR}/data.csv`; // legacy is direct CSV (not zipped per metadata)
const BATCH = 2000;
async function downloadCsv() {
console.log(`Downloading ${ZIP_URL} ...`);
await mkdir(WORKDIR, { recursive: true });
try {
const st = statSync(ZIP_PATH);
if (st.size > 17_000_000_000) {
console.log(` already downloaded (${(st.size/1024/1024/1024).toFixed(2)} GB), skipping`);
return;
}
} catch { /* not present */ }
const res = await fetch(ZIP_URL);
if (!res.ok) throw new Error(`Download failed: ${res.status}`);
const ct = res.headers.get('content-type') ?? '';
console.log(` content-type: ${ct}`);
const total = parseInt(res.headers.get('content-length') ?? '0', 10);
let received = 0;
const w = createWriteStream(ZIP_PATH);
const reader = res.body!.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
w.write(value);
received += value.length;
if (received % (200 * 1024 * 1024) < value.length) {
console.log(` ${(received/1024/1024/1024).toFixed(2)} / ${(total/1024/1024/1024).toFixed(2)} GB`);
}
}
w.end();
await new Promise<void>(r => w.on('close', () => r()));
console.log(` ✓ saved ${(received/1024/1024/1024).toFixed(2)} GB`);
}
function splitCsv(line: string): string[] {
const out: string[] = [];
let cur = '';
let inQ = false;
for (let i = 0; i < line.length; i++) {
const c = line[i];
if (inQ) {
if (c === '"' && line[i+1] === '"') { cur += '"'; i++; }
else if (c === '"') inQ = false;
else cur += c;
} else {
if (c === '"') inQ = true;
else if (c === ',') { out.push(cur); cur = ''; }
else cur += c;
}
}
out.push(cur);
return out;
}
function pi(s?: string): number | null { if (!s) return null; const n = parseInt(s, 10); return isFinite(n) ? n : null; }
function pf(s?: string): number | null { if (!s) return null; const n = parseFloat(s); return isFinite(n) ? n : null; }
async function flush(rows: Record<string, string>[]) {
const tuples: string[] = [];
const values: any[] = [];
let ti = 0;
for (const r of rows) {
const apn = (r.ain ?? r.apn ?? '').replace(/[^0-9]/g, '');
if (!apn) continue;
// Handles BOTH the current roll (snake_case headers from Parcel_Data_0.csv)
// AND the legacy roll (CamelCase headers like RollYear, YearBuilt, TotalValue
// from data.csv 2006-2020). Header normalizer strips spaces+special chars +
// lowercases, so "RollYear" → "rollyear", "TotalValue" → "totalvalue", etc.
const rollYear = pi(r.roll_year ?? r.rollyear);
let situs = (r.property_location ?? r.propertylocation ?? '').trim();
if (!situs) {
const num = (r.address_house_number ?? r.houseno ?? '').trim();
const dir = (r.direction ?? r.streetdirection ?? '').trim();
const street = (r.street ?? r.streetname ?? '').trim();
situs = [num, dir, street].filter(Boolean).join(' ').trim();
}
const city = (r.city ?? '').trim();
const zip = (r.zip_code ?? r.zipcode5 ?? r.zipcode ?? '').trim();
const yearBuilt = pi(r.year_built ?? r.yearbuilt ?? r.effective_year ?? r.effectiveyearbuilt);
const bldg = pi(r.square_footage ?? r.sqftmain);
const beds = pi(r.number_of_bedrooms ?? r.bedrooms);
const baths = pf(r.number_of_bathrooms ?? r.bathrooms);
const av = pf(r.total_value ?? r.totalvalue);
const lv = pf(r.land_value ?? r.landvalue);
const iv = pf(r.improvement_value ?? r.improvementvalue);
const lat = pf(r.location_latitude ?? r.center_lat);
const lon = pf(r.location_longitude ?? r.center_lon);
const base = ti * 18;
tuples.push(`(${Array.from({length: 18}, (_, k) => `$${base+k+1}`).join(',')})`);
values.push(
apn, rollYear, 'Los Angeles County', situs || null, city || null, zip || null,
r.property_use_code ?? r.propertyusecode ?? null,
r.property_use_type ?? r.propertytype ?? r.generalusetype ?? null,
yearBuilt, beds, baths, bldg, av, lv, iv, lat, lon,
`https://portal.assessor.lacounty.gov/parceldetail/${apn}`,
);
ti++;
}
if (tuples.length === 0) return;
await pool.query(
`INSERT INTO parcel
(apn, roll_year, jurisdiction, situs_full, city, zip_code, use_code, use_desc,
year_built, bedrooms, bathrooms, bldg_sqft,
assessed_total, assessed_land, assessed_improvements,
latitude, longitude, source_url)
VALUES ${tuples.join(',')}
ON CONFLICT (apn, roll_year) DO UPDATE SET
situs_full = COALESCE(EXCLUDED.situs_full, parcel.situs_full),
city = COALESCE(EXCLUDED.city, parcel.city),
zip_code = COALESCE(EXCLUDED.zip_code, parcel.zip_code),
year_built = COALESCE(EXCLUDED.year_built, parcel.year_built),
bldg_sqft = COALESCE(EXCLUDED.bldg_sqft, parcel.bldg_sqft),
assessed_total = COALESCE(EXCLUDED.assessed_total, parcel.assessed_total),
retrieved_at = now()`,
values
);
}
async function processCsv(csvPath: string) {
console.log(`\nStreaming ${csvPath} ...`);
const rs = createReadStream(csvPath);
const rl = createInterface({ input: rs, crlfDelay: Infinity });
let headers: string[] | null = null;
const buffer: Record<string, string>[] = [];
let total = 0;
const t0 = Date.now();
for await (const line of rl) {
if (!line.trim()) continue;
const cols = splitCsv(line);
if (!headers) {
headers = cols.map(h => h.toLowerCase().replace(/\s+/g, '_').replace(/[^a-z0-9_]/g, ''));
console.log(` fields (${headers.length}): ${headers.slice(0, 12).join(', ')}…`);
continue;
}
const row: Record<string, string> = {};
headers.forEach((h, i) => { row[h] = cols[i] ?? ''; });
buffer.push(row);
if (buffer.length >= BATCH) {
await flush(buffer);
total += buffer.length;
buffer.length = 0;
const dt = (Date.now() - t0) / 1000;
if (total % 100000 === 0) {
console.log(` ${total.toLocaleString()} parcels (${(total/dt).toFixed(0)}/s, ${(dt/60).toFixed(1)}m elapsed)`);
}
}
}
if (buffer.length) {
await flush(buffer);
total += buffer.length;
}
console.log(` ✓ ${total.toLocaleString()} legacy parcels processed`);
}
async function main() {
await downloadCsv();
await processCsv(ZIP_PATH);
const { rows } = await pool.query<{ n: string; with_value: string; legacy_only: string }>(`
SELECT count(*)::text as n,
count(*) FILTER (WHERE assessed_total IS NOT NULL)::text as with_value,
count(DISTINCT apn) FILTER (WHERE roll_year < 2021)::text as legacy_only
FROM parcel
`);
console.log(`\nparcel: ${parseInt(rows[0].n).toLocaleString()} total, ${parseInt(rows[0].with_value).toLocaleString()} with AV, ${parseInt(rows[0].legacy_only).toLocaleString()} pre-2021 unique APNs`);
// verify Steve's APN now resolves
const { rows: target } = await pool.query<any>(`SELECT roll_year, situs_full, city, year_built, assessed_total::text as av FROM parcel WHERE apn = '4349040002' ORDER BY roll_year DESC LIMIT 5`);
if (target.length) {
console.log(`\n4349-040-002 found: ${target.length} roll years, latest=${target[0].roll_year}, situs="${target[0].situs_full}", AV=${target[0].av}`);
} else {
console.log(`\n4349-040-002 still NOT in parcel table (may be exempt/government parcel)`);
}
await pool.end();
}
main().catch(e => { console.error('FATAL', e); process.exit(1); });