`scripts/ingest-la-business-licenses.ts`

180 lines
/**
 * ingest-la-business-licenses.ts
 *
 * LA City "Listing of Active Businesses" (619K rows) — historical commercial
 * tenants by address. Same Socrata pattern as LADBS permits.
 *
 * Source: data.lacity.org dataset 6rrh-rzua
 *
 * Each row creates a business_license record linked to a listing matched by
 * primary_address (LADBS-style canonicalization).
 *
 * Tier: A (LA City Office of Finance 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 = '6rrh-rzua';
const PAGE = 50000;
const BATCH = 1000;

type Row = {
  location_account?: string;
  business_name?: string;
  dba_name?: string;
  street_address?: string;
  city?: string;
  zip_code?: string;
  location_description?: string;
  mailing_address?: string;
  mailing_city?: string;
  naics?: string;
  primary_naics_description?: string;
  council_district?: string;
  location_start_date?: string;
  location_end_date?: string;
  location_1?: { latitude?: string; longitude?: 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 ensureSchema() {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS business_license (
      id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
      source TEXT NOT NULL DEFAULT 'la_city',
      source_dataset TEXT NOT NULL DEFAULT '6rrh-rzua',
      account_number TEXT NOT NULL,
      listing_id UUID REFERENCES listing(id),
      business_name TEXT,
      dba_name TEXT,
      primary_address TEXT,
      city TEXT,
      zip_code TEXT,
      naics TEXT,
      naics_description TEXT,
      council_district TEXT,
      location_start_date DATE,
      location_end_date DATE,
      latitude NUMERIC(9,6),
      longitude NUMERIC(9,6),
      source_url TEXT,
      source_label TEXT NOT NULL DEFAULT 'LA City Office of Finance — Active Businesses',
      source_tier CHAR(1) NOT NULL DEFAULT 'A',
      retrieved_at TIMESTAMPTZ DEFAULT now(),
      UNIQUE (account_number)
    );
    CREATE INDEX IF NOT EXISTS idx_bl_listing ON business_license(listing_id);
    CREATE INDEX IF NOT EXISTS idx_bl_naics ON business_license(naics);
    CREATE INDEX IF NOT EXISTS idx_bl_address ON business_license(primary_address);
    CREATE INDEX IF NOT EXISTS idx_bl_dates ON business_license(location_start_date, location_end_date);
  `);
}

async function fetchPage(offset: number): Promise<Row[]> {
  const u = `https://data.lacity.org/resource/${SLUG}.json?$limit=${PAGE}&$offset=${offset}&$order=location_account`;
  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[]) {
  // Match by primary_address against existing LADBS listings
  const seen = new Set<string>();
  const valid = rows.filter(r => r.location_account && !seen.has(r.location_account) && (seen.add(r.location_account), true));
  if (valid.length === 0) return 0;
  const tuples: string[] = [];
  const values: any[] = [];
  valid.forEach((r, idx) => {
    const addr = (r.street_address ?? '').trim();
    const canon = addr ? canonicalize(addr) : '';
    const lat = r.location_1?.latitude ? parseFloat(r.location_1.latitude) : null;
    const lon = r.location_1?.longitude ? parseFloat(r.location_1.longitude) : null;
    const base = idx * 13;
    tuples.push(`(${Array.from({length: 13}, (_, k) => `$${base+k+1}`).join(',')})`);
    values.push(
      r.location_account,
      r.business_name ?? null,
      r.dba_name ?? null,
      addr || null,
      r.city ?? null,
      r.zip_code ?? null,
      r.naics ?? null,
      r.primary_naics_description ?? null,
      r.council_district ?? null,
      pd(r.location_start_date),
      pd(r.location_end_date),
      lat,
      lon,
    );
  });
  await pool.query(
    `INSERT INTO business_license
       (account_number, business_name, dba_name, primary_address, city, zip_code,
        naics, naics_description, council_district,
        location_start_date, location_end_date, latitude, longitude,
        listing_id, source_url)
     SELECT t.acct, t.bn, t.dba, t.addr, t.city, t.zip, t.naics, t.naicsd, t.cd,
            t.startd, t.endd, t.lat, t.lon, l.id,
            'https://data.lacity.org/resource/6rrh-rzua.json?location_account=' || t.acct
     FROM (VALUES ${tuples.map((_, k) => {
       const b = k * 13;
       return `($${b+1}::text,$${b+2}::text,$${b+3}::text,$${b+4}::text,$${b+5}::text,$${b+6}::text,$${b+7}::text,$${b+8}::text,$${b+9}::text,$${b+10}::date,$${b+11}::date,$${b+12}::numeric,$${b+13}::numeric)`;
     }).join(',')}) AS t(acct, bn, dba, addr, city, zip, naics, naicsd, cd, startd, endd, lat, lon)
     LEFT JOIN listing l ON l.source = 'ladbs_permit'
                       AND l.source_id = 'ladbs:' || lower(regexp_replace(regexp_replace(coalesce(t.addr,''), '[^\\w\\s-]', '', 'g'), '\\s+', '-', 'g'))
     ON CONFLICT (account_number) DO UPDATE SET
       business_name = COALESCE(EXCLUDED.business_name, business_license.business_name),
       location_end_date = COALESCE(EXCLUDED.location_end_date, business_license.location_end_date),
       listing_id = COALESCE(business_license.listing_id, EXCLUDED.listing_id),
       retrieved_at = now()`,
    values
  );
  return valid.length;
}

async function main() {
  await ensureSchema();
  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(`  business_license: ${total.toLocaleString()} (${(total/dt).toFixed(0)}/s)`);
    if (rows.length < PAGE) break;
    offset += PAGE;
  }
  const { rows: stats } = await pool.query<{ n: string; matched: string }>(
    `SELECT count(*)::text as n, count(*) FILTER (WHERE listing_id IS NOT NULL)::text as matched FROM business_license`
  );
  console.log(`✓ ${parseInt(stats[0]!.n).toLocaleString()} businesses (${parseInt(stats[0]!.matched).toLocaleString()} matched to a listing)`);
  await pool.end();
}

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