`scripts/ingest-inspections.ts`

132 lines
#!/usr/bin/env tsx
/**
 * Pull LA County DPH Restaurant & Market Inspections (rolling 3-yr window).
 *
 * Source: ArcGIS Hub item 19b6607ac82c4512b10811870975dbdc
 *   https://www.arcgis.com/sharing/rest/content/items/19b6607ac82c4512b10811870975dbdc/data
 *
 * Encoding: Latin-1. Idempotent upsert keyed on SERIAL_NUMBER.
 * Skips inspections whose facility_id is not in the facility table.
 */
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 = '19b6607ac82c4512b10811870975dbdc';
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-inspections.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,
});

async function downloadCSV(): Promise<void> {
  await mkdir(DATA_DIR, { recursive: true });
  console.log(`[inspections] downloading ${SOURCE_URL}`);
  const res = await fetch(SOURCE_URL, { redirect: 'follow' });
  if (!res.ok) throw new Error(`download failed: ${res.status}`);
  if (!res.body) throw new Error('empty body');
  await pipeline(Readable.fromWeb(res.body as any), createWriteStream(CSV_PATH));
  console.log(`[inspections] downloaded ${(statSync(CSV_PATH).size / 1024 / 1024).toFixed(1)} MB`);
}

function parseDate(s: string | undefined): Date | null {
  if (!s) return null;
  const d = new Date(s);
  return isNaN(d.getTime()) ? null : d;
}

async function main() {
  const pulledAt = new Date();
  let counts = { inserted: 0, updated: 0, unchanged: 0, failed: 0, orphaned: 0 };

  try {
    if (!existsSync(CSV_PATH) || (Date.now() - statSync(CSV_PATH).mtimeMs) > 86400_000) {
      await downloadCSV();
    }

    const buf = await readFile(CSV_PATH);
    const rows = parse(iconv.decode(buf, 'latin1'), {
      columns: true, skip_empty_lines: true, trim: true,
      relax_quotes: true, relax_column_count: true,
    }) as Record<string, string>[];
    console.log(`[inspections] parsed ${rows.length.toLocaleString()} rows`);

    // Pre-fetch all known facility_ids to skip orphans cheaply
    const { rows: facRows } = await pool.query<{ facility_id: string }>(`SELECT facility_id FROM facility`);
    const known = new Set(facRows.map(r => r.facility_id));
    console.log(`[inspections] ${known.size.toLocaleString()} known facilities`);

    const client = await pool.connect();
    try {
      await client.query('BEGIN');
      for (const r of rows) {
        const facility_id = r['FACILITY ID']?.trim();
        const serial_number = r['SERIAL NUMBER']?.trim();
        const activity_date = parseDate(r['ACTIVITY DATE']);
        if (!facility_id || !serial_number || !activity_date) { counts.failed++; continue; }
        if (!known.has(facility_id)) { counts.orphaned++; continue; }

        const score = r['SCORE'] ? parseInt(r['SCORE'], 10) : null;
        const grade = r['GRADE']?.trim().charAt(0).toUpperCase() || null;
        const program_status = r['PROGRAM STATUS']?.trim() || null;

        try {
          const res = await client.query<{ inserted: boolean }>(
            `INSERT INTO inspection (
              facility_id, serial_number, activity_date, service_code, service_description,
              score, grade, program_status, employee_id, source_pulled_at
            ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
            ON CONFLICT (serial_number) DO UPDATE SET
              activity_date    = EXCLUDED.activity_date,
              score            = EXCLUDED.score,
              grade            = EXCLUDED.grade,
              program_status   = EXCLUDED.program_status,
              source_pulled_at = EXCLUDED.source_pulled_at
            RETURNING (xmax = 0) AS inserted`,
            [
              facility_id, serial_number, activity_date,
              r['SERVICE CODE']?.trim() || null,
              r['SERVICE DESCRIPTION']?.trim() || null,
              score, grade && /^[ABC]$/.test(grade) ? grade : null, program_status,
              r['EMPLOYEE ID']?.trim() || null, pulledAt,
            ]
          );
          res.rows[0].inserted ? counts.inserted++ : counts.updated++;
        } catch (e) {
          counts.failed++;
          if (counts.failed < 5) console.error(`[inspections] row failed (${serial_number}):`, (e as Error).message);
        }
      }
      await client.query('COMMIT');
    } catch (e) {
      await client.query('ROLLBACK');
      throw e;
    } finally {
      client.release();
    }

    console.log(`[inspections] DONE — inserted=${counts.inserted} updated=${counts.updated} orphaned=${counts.orphaned} failed=${counts.failed}`);

    await pool.query(
      `INSERT INTO ingest_run (dataset, source_url, started_at, finished_at,
                               rows_pulled, rows_inserted, rows_updated, rows_failed)
       VALUES ('inspections', $1, $2, now(), $3, $4, $5, $6)`,
      [SOURCE_URL, pulledAt, rows.length, counts.inserted, counts.updated, counts.failed]
    );
  } finally {
    await pool.end();
  }
}

main();