`scripts/ingest-la-boe-permits.ts`

119 lines
/**
 * ingest-la-boe-permits.ts
 *
 * LA Bureau of Engineering permits (j7mw-thyc, ~626K rows).
 * ROW/sidewalk/sewer/curb-cut work permits — historical infrastructure changes
 * per address.
 *
 * Different schema than building permits: id, location, permitname, permitno,
 * permittype, permitissuedate, permiturl, refno.
 *
 * Tier: A (LA Bureau of Engineering primary record).
 */
import { Pool } from 'pg';

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 SLUG = 'j7mw-thyc';
const PAGE = 50000;
const BATCH = 1000;

type Row = {
  id?: string;
  location?: string;
  permitname?: string;
  permitno?: string;
  permittype?: string;
  permitissuedate?: string;
  permiturl?: string;
  refno?: string;
};

function canonicalize(addr: string): string {
  return addr.toLowerCase().replace(/[^\w\s-]/g, '').replace(/\s+/g, '-').replace(/-+/g, '-').replace(/^-|-$/g, '').slice(0, 90);
}
function pd(v?: string): string | null { if (!v) return null; const m = v.match(/^(\d{4}-\d{2}-\d{2})/); return m ? m[1] : null; }

async function fetchPage(offset: number): Promise<Row[]> {
  const u = `https://data.lacity.org/resource/${SLUG}.json?$limit=${PAGE}&$offset=${offset}&$order=permitno`;
  for (let attempt = 1; attempt <= 4; attempt++) {
    try {
      const res = await fetch(u, { headers: { 'User-Agent': 'stayclaim-ingest/1.0' } });
      if (res.status === 429 || res.status === 503) { await new Promise(r => setTimeout(r, 2000 * attempt)); continue; }
      if (!res.ok) throw new Error(`HTTP ${res.status}`);
      return await res.json() as Row[];
    } catch (e) {
      if (attempt >= 4) throw e;
      await new Promise(r => setTimeout(r, 1500 * attempt));
    }
  }
  return [];
}

async function processBatch(rows: Row[]) {
  const seen = new Set<string>();
  const valid = rows.filter(r => r.permitno && !seen.has(r.permitno) && (seen.add(r.permitno), true));
  if (valid.length === 0) return 0;
  const tuples: string[] = [];
  const values: any[] = [];
  valid.forEach((r, idx) => {
    const addr = (r.location ?? '').trim();
    const base = idx * 9;
    tuples.push(`(${Array.from({length: 9}, (_, k) => `$${base+k+1}`).join(',')})`);
    values.push(
      r.permitno!.trim(),                           // 1 permit_number
      addr || null,                                 // 2 primary_address
      r.permitname ?? null,                         // 3 work_desc
      r.permittype ?? null,                         // 4 permit_type
      pd(r.permitissuedate),                        // 5 issue_date
      r.permiturl ?? `https://data.lacity.org/resource/${SLUG}.json?permitno=${encodeURIComponent(r.permitno!)}`, // 6 source_url
      r.refno ?? null,                              // 7 zone (re-purposed for refno display)
      r.permitname ?? null,                         // 8 use_desc
      addr,                                         // 9 — address for join
    );
  });
  await pool.query(
    `INSERT INTO permit (source, source_dataset, permit_number, primary_address,
       work_desc, permit_type, issue_date, source_url, zone, use_desc, listing_id, source_label)
     SELECT 'la_city_boe', '${SLUG}', t.pn, t.addr, t.wd, t.pt, t.id_, t.url, t.zo, t.ud, l.id,
            'LA Bureau of Engineering Permits'
     FROM (VALUES ${tuples.map((_, k) => {
       const b = k * 9;
       return `($${b+1},$${b+2},$${b+3},$${b+4},$${b+5}::date,$${b+6},$${b+7},$${b+8},$${b+9})`;
     }).join(',')}) AS t(pn, addr, wd, pt, id_, url, zo, ud, joinaddr)
     LEFT JOIN listing l ON l.source = 'ladbs_permit'
                       AND l.source_id = 'ladbs:' || lower(regexp_replace(regexp_replace(coalesce(t.joinaddr,''), '[^\\w\\s-]', '', 'g'), '\\s+', '-', 'g'))
     ON CONFLICT (source_dataset, permit_number) DO UPDATE SET
       listing_id = COALESCE(permit.listing_id, EXCLUDED.listing_id),
       retrieved_at = now()`,
    values
  );
  return valid.length;
}

async function main() {
  let offset = 0, total = 0;
  const t0 = Date.now();
  while (true) {
    const rows = await fetchPage(offset);
    if (rows.length === 0) break;
    for (let i = 0; i < rows.length; i += BATCH) await processBatch(rows.slice(i, i + BATCH));
    total += rows.length;
    const dt = (Date.now() - t0) / 1000;
    console.log(`  boe: ${total.toLocaleString()} (${(total/dt).toFixed(0)}/s)`);
    if (rows.length < PAGE) break;
    offset += PAGE;
  }
  console.log(`✓ BOE permits: ${total.toLocaleString()} total`);
  await pool.end();
}

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