`src/ingest/la_city_business_licenses.ts`

245 lines
/**
 * LA City Active Business Licenses importer.
 *
 * Source: https://data.lacity.org/resource/6rrh-rzua.json (Socrata API, free, no key).
 *         "Listing of Active Businesses" — every business currently licensed by LA City.
 * Filter: NAICS 541100 (Legal services), 541110 (Offices of Lawyers), 541199 (Other legal).
 *         Excludes 541120 notaries and 541191 title offices (not lawyers).
 *
 * Yields ~8K rows for LA City — vastly bigger than OSM (190) or Wikidata (29).
 * Includes solo practitioners working from home addresses, which is realistic
 * coverage of "where lawyers actually practice in LA."
 */
import 'dotenv/config';
import crypto from 'node:crypto';
import { fetch } from 'undici';
import { pool, query, withTx } from '../db/pool.ts';

const SOURCE_NAME = 'LA City Active Business Licenses';
const ENDPOINT = 'https://data.lacity.org/resource/6rrh-rzua.json';
const USER_AGENT = process.env.USER_AGENT
  || 'LawyerDirectoryBuilder/0.1 (research; contact: steveabramsdesigns@gmail.com)';

// NAICS codes worth importing as law firms.
const KEEP_NAICS = new Set(['541100', '541110', '541199']);

const PAGE_SIZE = 1000;

interface Row {
  location_account?: string;
  business_name?: string;
  street_address?: string;
  city?: string;
  zip_code?: string;
  location_description?: string;
  naics?: string;
  primary_naics_description?: string;
  council_district?: string;
  location_start_date?: string;
  location_1?: { latitude?: string; longitude?: string };
}

function clean(s: string | undefined | null) {
  if (!s) return null;
  const t = String(s).trim();
  return t.length === 0 ? null : t;
}

function titleCase(s: string | null) {
  if (!s) return null;
  return s.toLowerCase().replace(/\b([a-z])([a-z]*)/g, (_, a, b) => a.toUpperCase() + b);
}

function normAddress(s: string | null) {
  if (!s) return null;
  return s.toLowerCase().replace(/[.,]/g, ' ').replace(/\b(suite|ste|unit|apt|#)\b/g, '').replace(/\s+/g, ' ').trim();
}

function zip5(z: string | undefined | null) {
  if (!z) return null;
  const m = String(z).match(/\d{5}/);
  return m ? m[0] : null;
}

function neighborhoodOf(city: string | null) {
  return city ? titleCase(city) : null;
}

function firmSizeBand(name: string | null): string | null {
  if (!name) return null;
  const n = name.toLowerCase();
  if (/\b(llp|pllc)\b/.test(n)) return 'medium';     // LLPs lean medium-large
  if (/\b(pc|p\.c\.|aplc|apc)\b/.test(n)) return 'small'; // small CA prof corps
  if (/&|and associates|group|partners/.test(n)) return 'small';
  return 'solo';
}

async function ensureSource(): Promise<number> {
  await query(`
    INSERT INTO sources (source_name, source_type, base_url, terms_notes, allowed_method, rate_limit_rps)
    VALUES ($1, 'api',
      'https://data.lacity.org/resource/6rrh-rzua.json',
      'Socrata Open Data — public domain. NAICS 5411 (legal services). Polite-use, no key required.',
      'api', 5.0)
    ON CONFLICT (source_name) DO NOTHING
  `, [SOURCE_NAME]);
  const r = await query<{ id: number }>(`SELECT id FROM sources WHERE source_name = $1`, [SOURCE_NAME]);
  return r.rows[0].id;
}

async function startJob(sourceId: number, label: string) {
  const r = await query<{ id: number }>(`
    INSERT INTO scrape_jobs (source_id, job_label, status, started_at)
    VALUES ($1, $2, 'running', NOW()) RETURNING id
  `, [sourceId, label]);
  return r.rows[0].id;
}

async function finishJob(jobId: number, fields: Record<string, unknown>) {
  const sets: string[] = [];
  const params: unknown[] = [];
  let i = 1;
  for (const [k, v] of Object.entries(fields)) { sets.push(`${k} = $${i++}`); params.push(v); }
  sets.push(`finished_at = NOW()`);
  params.push(jobId);
  await query(`UPDATE scrape_jobs SET ${sets.join(', ')} WHERE id = $${i}`, params);
}

async function fetchPage(offset: number): Promise<Row[]> {
  const where = `naics in ('541100','541110','541199')`;
  const url = `${ENDPOINT}?$where=${encodeURIComponent(where)}&$limit=${PAGE_SIZE}&$offset=${offset}&$order=location_account`;
  const r = await fetch(url, { headers: { 'User-Agent': USER_AGENT, Accept: 'application/json' }, signal: AbortSignal.timeout(60000) });
  if (!r.ok) throw new Error(`Socrata ${r.status}: ${(await r.text()).slice(0, 300)}`);
  return await r.json() as Row[];
}

async function upsert(row: Row, sourceId: number) {
  if (!KEEP_NAICS.has(String(row.naics || ''))) return null;

  const rawName = clean(row.business_name);
  if (!rawName) return null;
  const name = titleCase(rawName);                                   // "LAW OFFICES OF X" → "Law Offices Of X"
  const street = titleCase(clean(row.street_address));
  const city = titleCase(clean(row.city));
  const zip = zip5(row.zip_code);
  const fullAddress = [street, city, 'CA', zip].filter(Boolean).join(', ');
  const lat = row.location_1?.latitude ? parseFloat(row.location_1.latitude) : null;
  const lng = row.location_1?.longitude ? parseFloat(row.location_1.longitude) : null;
  const sourceUrl = `https://data.lacity.org/resource/6rrh-rzua.json?$where=location_account='${row.location_account}'`;
  const addressNorm = normAddress(fullAddress);

  return await withTx(async (client) => {
    // Dedup: name + address_norm match an existing org (most likely from OSM with full address).
    let orgId: number | null = null;
    if (addressNorm) {
      const r = await client.query<{ id: number }>(
        `SELECT id FROM organizations WHERE address_norm = $1 AND LOWER(name) = LOWER($2) LIMIT 1`,
        [addressNorm, name!]);
      if (r.rowCount && r.rowCount > 0) orgId = r.rows[0].id;
    }
    // Fallback: name match in same city
    if (!orgId && city) {
      const r = await client.query<{ id: number }>(
        `SELECT id FROM organizations WHERE LOWER(name) = LOWER($1) AND LOWER(city) = LOWER($2) LIMIT 1`,
        [name!, city]);
      if (r.rowCount && r.rowCount > 0) orgId = r.rows[0].id;
    }

    if (orgId) {
      // Enrich with anything we have that's missing.
      await client.query(`
        UPDATE organizations
        SET address = COALESCE(address, $2),
            address_norm = COALESCE(address_norm, $3),
            city = COALESCE(city, $4),
            neighborhood = COALESCE(neighborhood, $4),
            zip = COALESCE(zip, $5),
            lat = COALESCE(lat, $6::double precision),
            lng = COALESCE(lng, $7::double precision),
            geocoded_at = COALESCE(geocoded_at, CASE WHEN $6::double precision IS NOT NULL THEN NOW() END),
            firm_size_band = COALESCE(firm_size_band, $8),
            source_url = COALESCE(source_url, $9),
            updated_at = NOW()
        WHERE id = $1
      `, [orgId, fullAddress, addressNorm, city, zip, lat, lng, firmSizeBand(name), sourceUrl]);
    } else {
      const r = await client.query<{ id: number }>(`
        INSERT INTO organizations (
          name, type, address, address_norm, city, neighborhood, state, county, zip,
          lat, lng, geocoded_at, firm_size_band, source_url
        ) VALUES (
          $1, 'law_firm', $2, $3, $4, $4, 'CA', 'Los Angeles', $5,
          $6::double precision, $7::double precision,
          CASE WHEN $6::double precision IS NOT NULL THEN NOW() END,
          $8, $9
        )
        RETURNING id
      `, [name, fullAddress, addressNorm, city, zip, lat, lng, firmSizeBand(name), sourceUrl]);
      orgId = r.rows[0].id;
    }

    const rawJson = JSON.stringify(row);
    const hash = crypto.createHash('sha256').update(rawJson + '|la-city|' + (row.location_account || '') + '|' + orgId).digest('hex');
    await client.query(`
      INSERT INTO raw_records (source_id, source_url, entity_type, entity_id, raw_json, fetched_at, hash)
      VALUES ($1, $2, 'organization', $3, $4::jsonb, NOW(), $5)
      ON CONFLICT (source_id, hash) DO NOTHING
    `, [sourceId, sourceUrl, orgId, rawJson, hash]);

    return orgId;
  });
}

async function main() {
  console.log('[la-city] importing legal-services business licenses (NAICS 5411)…');
  const sourceId = await ensureSource();
  const jobId = await startJob(sourceId, 'la-city:business-licenses');

  let offset = 0;
  let inserted = 0;
  let skipped = 0;
  let total = 0;

  try {
    while (true) {
      const page = await fetchPage(offset);
      if (page.length === 0) break;
      total += page.length;

      for (const row of page) {
        try {
          const id = await upsert(row, sourceId);
          if (id) inserted++; else skipped++;
        } catch (e) {
          console.error(`[la-city] upsert err ${row.location_account}: ${(e as Error).message}`);
          skipped++;
        }
      }

      console.log(`[la-city] page offset=${offset} (+${page.length}) → kept=${inserted} skipped=${skipped}`);
      if (page.length < PAGE_SIZE) break;
      offset += PAGE_SIZE;
    }
  } catch (e) {
    await finishJob(jobId, { status: 'failed', error_message: (e as Error).message,
                              records_found: total, records_inserted: inserted, records_skipped: skipped });
    throw e;
  }

  await finishJob(jobId, {
    status: 'completed',
    records_found: total,
    records_inserted: inserted,
    records_skipped: skipped,
  });

  console.log(`[la-city] done. seen=${total} kept=${inserted} skipped=${skipped}`);
  await pool.end();
}

main().catch(async (err) => {
  console.error('[la-city] fatal:', err);
  try { await pool.end(); } catch {}
  process.exit(1);
});