`scripts/ingest-bh-permits.ts`

159 lines
#!/usr/bin/env -S npx tsx
/**
 * Beverly Hills nightly permit feed ingest.
 *
 * Drop-in slot for the real BH issued-permits dataset. Behaviour:
 *   1. Reads from a CSV path (--file) or HTTP URL (--url) or the bundled sample.
 *   2. Normalizes the address string and matches it to an existing listing.
 *      If no match, creates a stub listing in DRAFT state (is_public=false)
 *      so the permit isn't lost — editorial can promote later.
 *   3. Appends a row to `place_event` per permit, idempotent on permit_number
 *      (extracted into source_id so repeat runs don't double-insert).
 *
 * Usage:
 *   npx tsx scripts/ingest-bh-permits.ts                       # bundled sample
 *   npx tsx scripts/ingest-bh-permits.ts --file path/to.csv
 *   npx tsx scripts/ingest-bh-permits.ts --url https://data.beverlyhills.org/.../permits.csv
 *
 * Wire-up for the live feed (when the URL lands):
 *   - Add ~/Library/LaunchAgents/com.pastdoor.bh-permits.plist (cron 03:00 daily)
 *   - Or pm2: `pm2 start "npx tsx scripts/ingest-bh-permits.ts --url $URL" --cron "0 3 * * *"`
 */
import fs from 'node:fs';
import path from 'node:path';
import { Pool } from 'pg';
import slugify from 'slugify';

type Row = {
  permit_number: string;
  issued_date: string;
  address: string;
  scope: string;
  permit_type: string;
  valuation: string;
  applicant: string;
};

function parseCsv(text: string): Row[] {
  // Lightweight CSV parser — sample file uses clean quoting.
  const lines = text.split(/\r?\n/).filter(Boolean);
  const headers = lines[0]!.split(',');
  return lines.slice(1).map(line => {
    // naive comma-split — sample data has no embedded commas inside fields
    const cols = line.split(',');
    const row: Record<string, string> = {};
    headers.forEach((h, i) => (row[h.trim()] = (cols[i] ?? '').trim()));
    return row as unknown as Row;
  });
}

function normalizeAddress(raw: string): string {
  return raw
    .toLowerCase()
    .replace(/[.,]/g, '')
    .replace(/\bnorth\b/g, 'n')
    .replace(/\bsouth\b/g, 's')
    .replace(/\beast\b/g, 'e')
    .replace(/\bwest\b/g, 'w')
    .replace(/\bdrive\b/g, 'dr')
    .replace(/\bavenue\b/g, 'ave')
    .replace(/\bboulevard\b/g, 'blvd')
    .replace(/\bstreet\b/g, 'st')
    .replace(/\bway\b/g, 'wy')
    .replace(/\broad\b/g, 'rd')
    .replace(/\splace\b/g, ' pl')
    .replace(/\bcanyon\b/g, 'canyon')
    .replace(/\s+/g, ' ')
    .trim();
}

const pool = new Pool({
  host: process.env.PGHOST ?? '/tmp',
  database: process.env.PGDATABASE ?? 'stayclaim',
  user: process.env.PGUSER ?? process.env.USER,
  max: 4,
});

async function findOrCreateListingByAddress(address: string): Promise<string> {
  const norm = normalizeAddress(address);
  const { rows: hit } = await pool.query<{ id: string }>(
    `SELECT id FROM listing
     WHERE city = 'Beverly Hills'
       AND lower(regexp_replace(address_line1, '[.,]', '', 'g')) ILIKE $1
     LIMIT 1`,
    [`%${norm}%`]
  );
  if (hit[0]) return hit[0].id;

  // Create a stub. is_public=false, so the page won't index until enriched.
  const slug = slugify(`${address}-bh`, { lower: true, strict: true });
  const { rows } = await pool.query<{ id: string }>(
    `INSERT INTO listing (slug, source, source_id, title, address_line1, city, state, country, is_public, privacy_class)
     VALUES ($1, 'bh-permits', $2, $3, $4, 'Beverly Hills', 'CA', 'US', false, 'public_history_only')
     ON CONFLICT (slug) DO UPDATE SET updated_at = now()
     RETURNING id`,
    [slug, slug, address, address]
  );
  return rows[0]!.id;
}

async function appendPermitEvent(listingId: string, row: Row) {
  const summary = row.scope ? `${row.scope}` : `Permit issued — ${row.permit_type}`;
  const sourceLabel = `BH Planning · permit ${row.permit_number}`;
  const { rowCount } = await pool.query(
    `INSERT INTO place_event (listing_id, kind, summary, valid_from, source_tier, source_label, source_id, confidence)
     SELECT $1, 'permit', $2, $3::date, 'A', $4, $5, 0.94
     WHERE NOT EXISTS (SELECT 1 FROM place_event WHERE source_id = $5 AND kind = 'permit')`,
    [listingId, summary, row.issued_date, sourceLabel, row.permit_number]
  );
  return rowCount === 1;
}

async function main() {
  const args = process.argv.slice(2);
  const fileArg = args.find((a, i) => args[i - 1] === '--file');
  const urlArg = args.find((a, i) => args[i - 1] === '--url');
  let csv: string;
  if (urlArg) {
    const r = await fetch(urlArg);
    if (!r.ok) throw new Error(`Fetch ${urlArg} ${r.status}`);
    csv = await r.text();
  } else {
    const filePath = fileArg ?? path.resolve(__dirname, '..', 'data', 'bh-permits-sample.csv');
    csv = fs.readFileSync(filePath, 'utf8');
  }

  const rows = parseCsv(csv);
  let inserted = 0;
  let matched = 0;
  let stubbed = 0;
  for (const row of rows) {
    const before = await pool.query<{ id: string }>(
      `SELECT id FROM listing WHERE source = 'bh-permits' AND source_id = $1`,
      [slugify(`${row.address}-bh`, { lower: true, strict: true })]
    );
    const wasStub = before.rowCount! > 0;
    const listingId = await findOrCreateListingByAddress(row.address);
    if (!wasStub) {
      const after = await pool.query<{ id: string }>(
        `SELECT id FROM listing WHERE source = 'bh-permits' AND source_id = $1`,
        [slugify(`${row.address}-bh`, { lower: true, strict: true })]
      );
      if (after.rowCount! > 0) stubbed++;
      else matched++;
    }
    const did = await appendPermitEvent(listingId, row);
    if (did) inserted++;
  }
  console.log(JSON.stringify(
    { processed: rows.length, events_inserted: inserted, matched_existing: matched, stubs_created: stubbed },
    null, 2
  ));
  await pool.end();
}

main().catch(e => {
  console.error(e);
  process.exit(1);
});