← back to Small Business Builder

scripts/ingest-bbcb-salons.js

235 lines

#!/usr/bin/env node
// Ingest CA Board of Barbering & Cosmetology (BBCB) Establishment licenses
// for LA + 4 adjacent counties. Source: CA DCA Public Information bulk export
// (free, no CPRA, refreshes monthly).
//
// Per la-research-agent (2026-05-04):
//   GET https://www.dca.ca.gov/boxToken/getToken           → Bearer token
//   GET https://api.box.com/2.0/files/2216160745307/content → Data00 ~77MB
//
// File is tab-delimited UTF-8 with `.xls` extension (NOT binary Excel).
// 20 columns:
//   0  Agency Code
//   1  Agency Name
//   2  LicTypeCode
//   3  License Type
//   4  License Number
//   5  Indiv/Org             ← filter to 'O'
//   6  Org/Last Name
//   7  First Name
//   8  Middle Name
//   9  Suffix
//   10 Address Line 1
//   11 Address Line 2
//   12 City
//   13 County                ← filter to 5 counties
//   14 State
//   15 Zip
//   16 Country
//   17 Original Issue Date
//   18 Expiration Date
//   19 License Status        ← filter to 'Current'
//
// Expected: 23,370 active shops across LA + Orange + Riverside + SB + Ventura.
//
//   node scripts/ingest-bbcb-salons.js              # full ingest, upsert mode
//   DRY=1 node scripts/ingest-bbcb-salons.js        # parse + filter, no DB writes
//   LIMIT=200 node scripts/ingest-bbcb-salons.js    # cap row count for testing

import 'dotenv/config';
import fs from 'node:fs';
import path from 'node:path';
import readline from 'node:readline';
import { query, one } from '../src/lib/db.js';
import { slugify } from '../src/lib/slug.js';

const TARGET_COUNTIES = new Set(['LOS ANGELES','ORANGE','VENTURA','SAN BERNARDINO','RIVERSIDE']);
const SHOP_TYPES      = new Set(['Establishment','Barber Shop','Chain Establishment']);
const TOKEN_URL       = 'https://www.dca.ca.gov/boxToken/getToken';
const FILE_IDS        = ['2216160745307']; // Data00 only — Data01 is individuals (no orgs)
const TMP_DIR         = '/tmp/bbcb-ingest';
const SOURCE          = 'dca_bbcb';
const SOURCE_URL      = 'https://www.dca.ca.gov/consumers/public_info/index.html';

const DRY   = process.env.DRY === '1';
const LIMIT = process.env.LIMIT ? parseInt(process.env.LIMIT, 10) : null;

function ts() { return new Date().toISOString().replace('T', ' ').slice(0, 19); }
function log(msg) { console.log(`[${ts()}] ${msg}`); }

// Map BBCB license type to our category enum.
function categoryFromLicType(licType) {
  return licType === 'Barber Shop' ? 'barbershop' : 'salon';
}

// "01/15/2014" or "1/15/2014" or "" → years (int) or null.
function yearsFromIssueDate(s) {
  if (!s) return null;
  const m = String(s).match(/(\d{4})\s*$/);
  if (!m) return null;
  const y = parseInt(m[1], 10);
  if (!y || y < 1900 || y > 2100) return null;
  return new Date().getFullYear() - y;
}

// Robust slug: name + zip is the agent's recipe; on collision the unique-index
// catches it and we fall back to appending the license number.
function makeSlug(name, zip, licNum) {
  const base = slugify(name) || 'salon';
  const zipPart = (zip || '').slice(0, 5);
  return zipPart ? `${base}-${zipPart}` : `${base}-${licNum}`;
}

// === Box token + download ==========================================
async function getBoxToken() {
  const res = await fetch(TOKEN_URL);
  if (!res.ok) throw new Error(`token endpoint HTTP ${res.status}`);
  const tok = (await res.text()).trim();
  if (!tok || tok.length < 30) throw new Error(`unexpected token: "${tok.slice(0, 80)}"`);
  return tok;
}

async function downloadFile(fileId, token, outPath) {
  const url = `https://api.box.com/2.0/files/${fileId}/content`;
  const res = await fetch(url, { headers: { Authorization: `Bearer ${token}` }, redirect: 'follow' });
  if (!res.ok) throw new Error(`Box file ${fileId} HTTP ${res.status}: ${(await res.text().catch(() => '')).slice(0, 200)}`);
  // Node's native fetch returns a Web ReadableStream — not a Node stream — so
  // res.body.pipe() doesn't exist. arrayBuffer() is fine here (77MB max).
  const buf = Buffer.from(await res.arrayBuffer());
  fs.writeFileSync(outPath, buf);
  log(`  downloaded ${fileId} → ${outPath} (${(buf.length / 1e6).toFixed(1)} MB)`);
}

// === Row processing =================================================
async function processFile(filePath) {
  const rl = readline.createInterface({ input: fs.createReadStream(filePath) });
  let lineNum = 0;
  let matched = 0, inserted = 0, updated = 0, skipped = 0, errors = 0;
  let countyHits = {};

  for await (const raw of rl) {
    lineNum++;
    if (lineNum === 1) continue; // header

    const f = raw.split('\t');
    if (f.length < 20) { skipped++; continue; }

    const [, , , licTypeRaw, licNum, indivOrg, orgName, , , ,
           addr1, addr2, cityRaw, countyRaw, stateRaw, zipRaw, ,
           issueDate, , statusRaw] = f;

    if (indivOrg !== 'O') continue;
    const licType = (licTypeRaw || '').trim();
    if (!SHOP_TYPES.has(licType)) continue;
    const county = (countyRaw || '').trim().toUpperCase();
    if (!TARGET_COUNTIES.has(county)) continue;
    const status = (statusRaw || '').trim();
    if (status !== 'Current') continue;

    matched++;
    countyHits[county] = (countyHits[county] || 0) + 1;
    if (LIMIT && matched > LIMIT) break;

    const name = (orgName || '').trim();
    if (!name) { skipped++; continue; }
    const address = [addr1, addr2].map(s => (s || '').trim()).filter(Boolean).join(', ');
    const city = (cityRaw || '').trim();
    const state = (stateRaw || '').trim() || 'CA';
    const zip = (zipRaw || '').trim().slice(0, 10);
    const slug = makeSlug(name, zip, (licNum || '').trim());
    const category = categoryFromLicType(licType);
    const years = yearsFromIssueDate(issueDate);

    const sourceData = {
      source: SOURCE,
      source_id: (licNum || '').trim(),
      source_url: SOURCE_URL,
      tier: 'A',
      retrieved_at: new Date().toISOString(),
      license_type: licType,
      license_number: (licNum || '').trim(),
      issue_date: (issueDate || '').trim(),
      license_status: status,
      county,
    };

    if (DRY) {
      if (matched <= 12) console.log(`  ${slug.padEnd(60)} ${name} · ${city}, ${county}`);
      continue;
    }

    try {
      // Try INSERT; on slug collision, UPDATE only the source-bearing fields.
      const r = await query(
        `INSERT INTO businesses
           (slug, name, category, address, city, state, zip,
            neighborhood, years_in_business, source_data_json, tier)
         VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,'free')
         ON CONFLICT (slug) DO UPDATE SET
           name = EXCLUDED.name,
           address = EXCLUDED.address,
           city = EXCLUDED.city,
           zip = EXCLUDED.zip,
           neighborhood = COALESCE(businesses.neighborhood, EXCLUDED.neighborhood),
           source_data_json = EXCLUDED.source_data_json,
           updated_at = now()
         RETURNING (xmax = 0) AS was_insert`,
        [slug, name, category, address, city, state, zip,
         county, years, JSON.stringify(sourceData)]
      );
      if (r.rows[0]?.was_insert) inserted++; else updated++;
    } catch (e) {
      errors++;
      if (errors <= 5) log(`  ERR ${slug}: ${e.message.slice(0, 200)}`);
    }

    if ((inserted + updated) > 0 && (inserted + updated) % 500 === 0) {
      log(`  …progress: ins=${inserted} upd=${updated} err=${errors}`);
    }
  }

  log(`COUNTY HITS: ${Object.entries(countyHits).map(([c, n]) => `${c}=${n}`).join(' ')}`);
  log(`DONE ${path.basename(filePath)}: matched=${matched} inserted=${inserted} updated=${updated} skipped=${skipped} errors=${errors}`);
  return { matched, inserted, updated, skipped, errors };
}

// === Main ==========================================================
async function main() {
  fs.mkdirSync(TMP_DIR, { recursive: true });
  log(`mode=${DRY ? 'DRY' : 'WRITE'}${LIMIT ? ` limit=${LIMIT}` : ''}`);

  log('fetching DCA Box token…');
  const token = await getBoxToken();
  log(`token len=${token.length} (head: ${token.slice(0, 12)}…)`);

  const totals = { matched: 0, inserted: 0, updated: 0, skipped: 0, errors: 0 };
  for (const fileId of FILE_IDS) {
    const outPath = path.join(TMP_DIR, `bbcb_${fileId}.tsv`);
    if (!fs.existsSync(outPath) || fs.statSync(outPath).size < 1e6) {
      log(`downloading file ${fileId}…`);
      await downloadFile(fileId, token, outPath);
    } else {
      log(`reusing cached ${outPath} (${(fs.statSync(outPath).size / 1e6).toFixed(1)} MB)`);
    }
    const r = await processFile(outPath);
    for (const k of Object.keys(totals)) totals[k] += r[k];
  }

  log(`TOTAL: matched=${totals.matched} inserted=${totals.inserted} updated=${totals.updated} skipped=${totals.skipped} errors=${totals.errors}`);

  if (!DRY) {
    const tally = await one(`
      SELECT
        COUNT(*) AS total,
        COUNT(*) FILTER (WHERE source_data_json->>'source' = 'dca_bbcb') AS bbcb,
        COUNT(*) FILTER (WHERE category = 'barbershop') AS barbershops,
        COUNT(*) FILTER (WHERE category = 'salon') AS salons
      FROM businesses
    `);
    log(`DB now: total=${tally.total}  bbcb=${tally.bbcb}  barbershops=${tally.barbershops}  salons=${tally.salons}`);
  }
  process.exit(totals.errors > totals.inserted + totals.updated ? 1 : 0);
}

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