`src/scripts/run_overnight.ts`

117 lines
/**
 * Overnight orchestrator — runs every importer + enricher in sequence.
 *
 * Sources wired (in order):
 *   1.  OSM Overpass refresh             (LA County office=lawyer / amenity=lawyer)
 *   2.  Wikidata SPARQL                  (notable LA firms — BigLaw filler)
 *   3.  LA City Business Licenses        (Socrata, NAICS 5411)
 *   4.  OSM name-match                   (broader name-based query)
 *   5.  Wikipedia firm articles          (LA County law-firm category)
 *   6.  Wikipedia lawyer biographies     (LA-county lawyers → professionals table)
 *   7.  Aggregation pass                 (firm_size_band recompute)
 *
 * Enrichers (in order, AFTER importers so they have fresh records):
 *   8.  Nominatim geocoder               (fill lat/lng for new addresses)
 *   9.  DNS website discovery            (candidate-domain check + name verify)
 *   10. Firm-website contact pages       (regex phone/email out of /contact pages)
 *
 * Blocked sources (NOT run; documented in docs/blocked_sources.md):
 *   - CA State Bar attorney search   → SPA, requires Playwright
 *   - CA SOS bizfile                 → POST endpoint requires recaptcha token
 *
 * Loop policy: run once, exit 0. Wrap with launchd if continuous mode is desired.
 */
import 'dotenv/config';
import { spawn } from 'node:child_process';
import path from 'node:path';
import fs from 'node:fs';
import { fileURLToPath } from 'node:url';
import { pool, query } from '../db/pool.ts';

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const ROOT = path.resolve(__dirname, '../..');
const LOG_DIR = path.join(ROOT, 'logs');
fs.mkdirSync(LOG_DIR, { recursive: true });

const TS = new Date().toISOString().slice(0, 10);
const LOG = path.join(LOG_DIR, `overnight-${TS}.log`);
const log = fs.createWriteStream(LOG, { flags: 'a' });
function out(s: string) { const line = `[${new Date().toISOString()}] ${s}\n`; process.stdout.write(line); log.write(line); }

// LAUNCHD FIX: was `spawn('npx', ['tsx', ...])` which fails under launchd
// because launchd's minimal PATH doesn't include npm/nvm/homebrew bin dirs
// (manual run from a login shell works because shell PATH inherits them —
// hence "manual returns 0, launchd returns 1, empty stderr"). Resolve tsx
// from local node_modules and run it directly with process.execPath (node).
const TSX_CLI = path.join(ROOT, 'node_modules', 'tsx', 'dist', 'cli.mjs');
function run(name: string, scriptPath: string, env: Record<string, string> = {}): Promise<number> {
  out(`▶ ${name}: tsx ${scriptPath}`);
  return new Promise((resolve) => {
    const child = spawn(process.execPath, [TSX_CLI, scriptPath], {
      cwd: ROOT,
      env: { ...process.env, ...env },
      stdio: ['ignore', 'pipe', 'pipe'],
    });
    child.stdout.on('data', d => log.write(d));
    child.stderr.on('data', d => log.write(d));
    child.on('close', code => { out(`✓ ${name} exited ${code}`); resolve(code ?? 0); });
  });
}

async function aggregationPass() {
  out('▶ aggregation: recompute attorney_count + size band');
  // Future: count professional_locations per organization.
  // For now: leave attorney_count as set by importers; only band by attorney_count if available.
  await query(`
    UPDATE organizations
    SET firm_size_band = CASE
      WHEN attorney_count >= 500 THEN 'biglaw'
      WHEN attorney_count >= 100 THEN 'large'
      WHEN attorney_count >= 25  THEN 'medium'
      WHEN attorney_count >= 5   THEN 'small'
      WHEN attorney_count >= 1   THEN 'solo'
      ELSE firm_size_band
    END
    WHERE type='law_firm' AND attorney_count > 0
  `);
  out('✓ aggregation done');
}

async function summary() {
  const r = await query<{ k: string; v: string }>(`
    SELECT 'firms_total' AS k, COUNT(*)::text AS v FROM organizations WHERE type='law_firm'
    UNION ALL SELECT 'with_geo',  COUNT(*)::text FROM organizations WHERE type='law_firm' AND lat IS NOT NULL
    UNION ALL SELECT 'with_site', COUNT(*)::text FROM organizations WHERE type='law_firm' AND website IS NOT NULL
    UNION ALL SELECT 'biglaw',    COUNT(*)::text FROM organizations WHERE type='law_firm' AND firm_size_band='biglaw'
  `);
  out('── summary ──');
  for (const row of r.rows) out(`  ${row.k.padEnd(14)} ${row.v}`);
}

async function main() {
  out('=== overnight run start ===');
  const codes: Record<string, number> = {};
  codes.osm        = await run('osm-overpass',         'src/ingest/osm_overpass.ts');
  codes.wd         = await run('wikidata',             'src/ingest/wikidata.ts');
  codes.lacity     = await run('la-city-licenses',     'src/ingest/la_city_business_licenses.ts');
  codes.osmName    = await run('osm-name-search',      'src/ingest/osm_name_search.ts');
  codes.wikiFirms  = await run('wikipedia-infobox',    'src/ingest/wikipedia_infobox.ts');
  codes.wikiPros   = await run('wikipedia-lawyers',    'src/ingest/wikipedia_lawyers.ts');
  await aggregationPass();
  codes.geocode    = await run('nominatim-geocoder',   'src/enrich/geocode_nominatim.ts');
  codes.discover   = await run('discover-websites',    'src/enrich/discover_websites.ts');
  codes.contacts   = await run('firm-website-contacts','src/enrich/firm_website_contacts.ts');
  await summary();
  const summary_line = Object.entries(codes).map(([k, v]) => `${k}=${v}`).join(' ');
  out(`=== overnight run end (${summary_line}) ===`);
  await pool.end();
  log.end();
}

main().catch(async (err) => {
  out(`fatal: ${(err as Error).message}`);
  try { await pool.end(); } catch {}
  log.end();
  process.exit(1);
});