`scripts/ingest-violations.ts`

127 lines
#!/usr/bin/env tsx
/**
 * Pull LA County DPH Violations dataset (rolling 3-yr window).
 * ArcGIS item: 5eaea9f89b7549ee841da7617d3a9cba
 * ~416K rows, 41 MB CSV, latin-1.
 *
 * Joined to inspection by SERIAL NUMBER. Each violation = one inspector note
 * with code, status, description, point deduction.
 */
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 = '5eaea9f89b7549ee841da7617d3a9cba';
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-violations.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 main() {
  const pulledAt = new Date();
  let counts = { inserted: 0, orphaned: 0, failed: 0 };

  try {
    if (!existsSync(CSV_PATH) || (Date.now() - statSync(CSV_PATH).mtimeMs) > 86400_000) {
      await mkdir(DATA_DIR, { recursive: true });
      console.log(`[violations] downloading…`);
      const res = await fetch(SOURCE_URL, { redirect: 'follow' });
      if (!res.ok || !res.body) throw new Error(`download ${res.status}`);
      await pipeline(Readable.fromWeb(res.body as any), createWriteStream(CSV_PATH));
      console.log(`[violations] downloaded ${(statSync(CSV_PATH).size / 1024 / 1024).toFixed(1)} MB`);
    }

    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(`[violations] parsed ${rows.length.toLocaleString()} rows`);

    // Pre-fetch known serial numbers (FK target)
    const { rows: serials } = await pool.query<{ serial_number: string }>(`SELECT serial_number FROM inspection`);
    const known = new Set(serials.map(s => s.serial_number));
    console.log(`[violations] ${known.size.toLocaleString()} known inspections`);

    // Wipe + bulk insert (cleaner than ON CONFLICT for a child table that
    // doesn't have a stable natural key — same violation can appear across
    // re-pulls with new IDs).
    const client = await pool.connect();
    try {
      await client.query('BEGIN');
      await client.query('TRUNCATE violation');

      // Batch insert via multi-VALUES — much faster than per-row.
      const BATCH = 500;
      for (let i = 0; i < rows.length; i += BATCH) {
        const slice = rows.slice(i, i + BATCH).filter(r => {
          const sn = r['SERIAL NUMBER']?.trim();
          if (!sn || !known.has(sn)) { counts.orphaned++; return false; }
          return true;
        });
        if (slice.length === 0) continue;

        const params: any[] = [];
        const placeholders: string[] = [];
        slice.forEach((r, j) => {
          const base = j * 5;
          placeholders.push(`($${base+1},$${base+2},$${base+3},$${base+4},$${base+5})`);
          params.push(
            r['SERIAL NUMBER'].trim(),
            r['VIOLATION STATUS']?.trim() || null,
            r['VIOLATION CODE']?.trim() || null,
            r['VIOLATION DESCRIPTION']?.trim() || null,
            r['POINTS'] ? parseInt(r['POINTS'], 10) : null,
          );
        });

        try {
          await client.query(
            `INSERT INTO violation (serial_number, violation_status, violation_code, violation_description, points)
             VALUES ${placeholders.join(',')}`,
            params
          );
          counts.inserted += slice.length;
        } catch (e) {
          counts.failed += slice.length;
          if (counts.failed < 5) console.error(`[violations] batch failed at ${i}:`, (e as Error).message);
        }

        if ((i + BATCH) % 50000 === 0 || i + BATCH >= rows.length) {
          console.log(`[violations] progress ${Math.min(i + BATCH, rows.length).toLocaleString()}/${rows.length.toLocaleString()}`);
        }
      }
      await client.query('COMMIT');
    } catch (e) {
      await client.query('ROLLBACK');
      throw e;
    } finally {
      client.release();
    }

    console.log(`[violations] DONE — inserted=${counts.inserted} 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_failed)
       VALUES ('violations', $1, $2, now(), $3, $4, $5)`,
      [SOURCE_URL, pulledAt, rows.length, counts.inserted, counts.failed]
    );
  } finally {
    await pool.end();
  }
}

main();