← back to Restaurant Directory
scripts/ingest-inventory.ts
249 lines
#!/usr/bin/env tsx
/**
* Pull LA County DPH Environmental Health Restaurant & Market Inventory.
*
* Source: ArcGIS Hub (NOT Socrata).
* https://www.arcgis.com/sharing/rest/content/items/4f31c9a99e444a40a3806e3bbe7b5fdd/data
*
* Encoding: Latin-1 (Windows-1252). Confirmed by la-research-agent 2026-05-02.
*
* Strategy: download → parse latin-1 CSV → upsert by FACILITY_ID. Track new vs
* updated vs unchanged. Idempotent — running twice is safe; only changed
* fields trigger UPDATE. New facility_ids get first_seen_at = now() so the
* weekly "what's new" cron can SELECT WHERE first_seen_at > last_run.
*/
import { createWriteStream, existsSync, statSync } from 'node:fs';
import { mkdir, readFile } from 'node:fs/promises';
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';
import path from 'node:path';
import iconv from 'iconv-lite';
import { parse } from 'csv-parse/sync';
import { Pool } from 'pg';
const DATASET_ID = '4f31c9a99e444a40a3806e3bbe7b5fdd';
const SOURCE_URL = `https://www.arcgis.com/sharing/rest/content/items/${DATASET_ID}/data`;
const DATA_DIR = path.resolve(__dirname, '..', 'data');
const CSV_PATH = path.join(DATA_DIR, 'la-eh-inventory.csv');
const pool = new Pool({
host: process.env.PGHOST ?? '/tmp',
database: process.env.PGDATABASE ?? 'restaurant_professional_directory',
user: process.env.PGUSER ?? process.env.USER,
max: 5,
});
function slugify(name: string, address: string | null, city: string | null): string {
const base = [name, address, city].filter(Boolean).join(' ');
return base
.toLowerCase()
.normalize('NFKD')
.replace(/[̀-ͯ]/g, '') // strip combining marks
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-|-$/g, '')
.slice(0, 120); // cap slug length
}
/** PE description → coarse facility_type / seat_tier / risk_tier. */
function classifyPE(desc: string | null): {
facility_type: string | null;
seat_tier: string | null;
risk_tier: string | null;
} {
if (!desc) return { facility_type: null, seat_tier: null, risk_tier: null };
const u = desc.toUpperCase();
let facility_type: string | null = null;
if (u.includes('RESTAURANT')) facility_type = 'restaurant';
else if (u.includes('FOOD MKT') || u.includes('MARKET')) facility_type = 'market';
else if (u.includes('BAKERY')) facility_type = 'bakery';
else if (u.includes('CATERER')) facility_type = 'caterer';
else if (u.includes('MOBILE')) facility_type = 'mobile';
let seat_tier: string | null = null;
if (u.includes('(0-30)')) seat_tier = '0-30';
else if (u.includes('(31-60)')) seat_tier = '31-60';
else if (u.includes('(61-150)')) seat_tier = '61-150';
else if (u.includes('(150 +)') || u.includes('(150+)')) seat_tier = '150+';
let risk_tier: string | null = null;
if (u.includes('HIGH RISK')) risk_tier = 'high';
else if (u.includes('MODERATE RISK')) risk_tier = 'moderate';
else if (u.includes('LOW RISK')) risk_tier = 'low';
return { facility_type, seat_tier, risk_tier };
}
async function downloadCSV(): Promise<void> {
await mkdir(DATA_DIR, { recursive: true });
console.log(`[ingest] downloading ${SOURCE_URL}`);
const res = await fetch(SOURCE_URL, { redirect: 'follow' });
if (!res.ok) throw new Error(`download failed: ${res.status} ${res.statusText}`);
if (!res.body) throw new Error('download returned empty body');
await pipeline(Readable.fromWeb(res.body as any), createWriteStream(CSV_PATH));
const size = statSync(CSV_PATH).size;
console.log(`[ingest] downloaded ${(size / 1024 / 1024).toFixed(1)} MB → ${CSV_PATH}`);
}
async function parseCSV(): Promise<Record<string, string>[]> {
const buf = await readFile(CSV_PATH);
const text = iconv.decode(buf, 'latin1');
const rows = parse(text, {
columns: true,
skip_empty_lines: true,
trim: true,
relax_quotes: true,
relax_column_count: true,
}) as Record<string, string>[];
console.log(`[ingest] parsed ${rows.length.toLocaleString()} rows`);
return rows;
}
type Counts = { inserted: number; updated: number; unchanged: number; failed: number };
async function upsertRows(rows: Record<string, string>[], pulledAt: Date): Promise<Counts> {
const counts: Counts = { inserted: 0, updated: 0, unchanged: 0, failed: 0 };
const client = await pool.connect();
try {
await client.query('BEGIN');
for (const r of rows) {
const facility_id = r['FACILITY ID']?.trim();
const name = r['FACILITY NAME']?.trim();
if (!facility_id || !name) { counts.failed++; continue; }
const address = r['FACILITY ADDRESS']?.trim() || null;
const city = r['FACILITY CITY']?.trim() || null;
const zip = r['FACILITY ZIP']?.trim() || null;
const lat = r['FACILITY LATITUDE'] ? parseFloat(r['FACILITY LATITUDE']) : null;
const lng = r['FACILITY LONGITUDE'] ? parseFloat(r['FACILITY LONGITUDE']) : null;
const pe_code = r['PROGRAM ELEMENT'] ? parseInt(r['PROGRAM ELEMENT'], 10) : null;
const pe_description = r['PE DESCRIPTION']?.trim() || null;
const owner_id = r['OWNER ID']?.trim() || null;
const owner_name = r['OWNER NAME']?.trim() || null;
const owner_address = r['OWNER ADDRESS']?.trim() || null;
const owner_city = r['OWNER CITY']?.trim() || null;
const owner_state = r['OWNER STATE']?.trim() || null;
const owner_zip = r['OWNER ZIP']?.trim() || null;
const { facility_type, seat_tier, risk_tier } = classifyPE(pe_description);
const slug = slugify(name, address, city) + '-' + facility_id.toLowerCase();
try {
const res = await client.query<{ inserted: boolean; updated: boolean }>(
`INSERT INTO facility (
facility_id, slug, name, address, city, state, zip, lat, lng,
pe_code, pe_description, facility_type, seat_tier, risk_tier,
owner_id, owner_name, owner_address, owner_city, owner_state, owner_zip,
source_dataset_id, source_pulled_at, last_updated_at
) VALUES (
$1, $2, $3, $4, $5, COALESCE($6,'CA'), $7, $8, $9,
$10, $11, $12, $13, $14,
$15, $16, $17, $18, $19, $20,
$21, $22, $22
)
ON CONFLICT (facility_id) DO UPDATE SET
name = EXCLUDED.name,
address = EXCLUDED.address,
city = EXCLUDED.city,
zip = EXCLUDED.zip,
lat = EXCLUDED.lat,
lng = EXCLUDED.lng,
pe_code = EXCLUDED.pe_code,
pe_description = EXCLUDED.pe_description,
facility_type = EXCLUDED.facility_type,
seat_tier = EXCLUDED.seat_tier,
risk_tier = EXCLUDED.risk_tier,
owner_id = EXCLUDED.owner_id,
owner_name = EXCLUDED.owner_name,
owner_address = EXCLUDED.owner_address,
owner_city = EXCLUDED.owner_city,
owner_state = EXCLUDED.owner_state,
owner_zip = EXCLUDED.owner_zip,
source_pulled_at = EXCLUDED.source_pulled_at,
last_updated_at = CASE
WHEN facility.name <> EXCLUDED.name
OR facility.address <> EXCLUDED.address
OR facility.lat <> EXCLUDED.lat
OR facility.lng <> EXCLUDED.lng
OR facility.pe_description <> EXCLUDED.pe_description
THEN now()
ELSE facility.last_updated_at
END
RETURNING (xmax = 0) AS inserted,
(xmax <> 0 AND last_updated_at = $22) AS updated`,
[
facility_id, slug, name, address, city, null, zip, lat, lng,
pe_code, pe_description, facility_type, seat_tier, risk_tier,
owner_id, owner_name, owner_address, owner_city, owner_state, owner_zip,
DATASET_ID, pulledAt,
]
);
const row = res.rows[0];
if (row.inserted) counts.inserted++;
else if (row.updated) counts.updated++;
else counts.unchanged++;
} catch (e) {
counts.failed++;
if (counts.failed < 10) console.error(`[ingest] row failed (${facility_id}):`, (e as Error).message);
}
}
await client.query('COMMIT');
} catch (e) {
await client.query('ROLLBACK');
throw e;
} finally {
client.release();
}
return counts;
}
async function logRun(pulledAt: Date, counts: Counts, finishedAt: Date, error: string | null) {
await pool.query(
`INSERT INTO ingest_run (dataset, source_url, started_at, finished_at,
rows_pulled, rows_inserted, rows_updated, rows_unchanged, rows_failed, error)
VALUES ('inventory', $1, $2, $3, $4, $5, $6, $7, $8, $9)`,
[
SOURCE_URL, pulledAt, finishedAt,
counts.inserted + counts.updated + counts.unchanged + counts.failed,
counts.inserted, counts.updated, counts.unchanged, counts.failed,
error,
]
);
}
async function main() {
const pulledAt = new Date();
let counts: Counts = { inserted: 0, updated: 0, unchanged: 0, failed: 0 };
let error: string | null = null;
try {
if (!existsSync(CSV_PATH) || (Date.now() - statSync(CSV_PATH).mtimeMs) > 86400_000) {
await downloadCSV();
} else {
console.log(`[ingest] using cached CSV at ${CSV_PATH} (< 24h old)`);
}
const rows = await parseCSV();
counts = await upsertRows(rows, pulledAt);
console.log(`[ingest] DONE — inserted=${counts.inserted} updated=${counts.updated} unchanged=${counts.unchanged} failed=${counts.failed}`);
// Standing rule: alert via George if failure rate > 10%
const total = counts.inserted + counts.updated + counts.unchanged + counts.failed;
if (total > 0 && counts.failed / total > 0.1) {
console.error(`[ingest] FAILURE RATE ${(counts.failed / total * 100).toFixed(1)}% — should email Steve via George`);
}
} catch (e) {
error = (e as Error).message;
console.error('[ingest] FATAL:', error);
process.exitCode = 1;
} finally {
await logRun(pulledAt, counts, new Date(), error);
await pool.end();
}
}
main();