`src/cli.js`

91 lines
#!/usr/bin/env node
import { SOURCES, GROUPS } from './sources.js';
import { ingestSource, closePool } from './ingest.js';
import { classifyRunResults } from './run-policy.js';

const argv = process.argv.slice(2);
const flags = new Set(argv.filter((a) => a.startsWith('--') && !a.includes('=')));
const kv = Object.fromEntries(
  argv.filter((a) => a.startsWith('--') && a.includes('=')).map((a) => a.slice(2).split('='))
);
const positional = argv.filter((a) => !a.startsWith('--'));

function printList() {
  console.log('\nLA Socrata/ArcGIS ingester — sources:\n');
  for (const [group, names] of Object.entries(GROUPS)) {
    console.log(`  ${group}`);
    for (const n of names) {
      const s = SOURCES[n];
      const id = s.datasetId || (s.endpoint ? 'arcgis' : '');
      console.log(`    • ${n.padEnd(30)} ${s.platform.padEnd(8)} ${id}${s.static ? '  [static]' : ''}`);
    }
  }
  console.log(`\nUsage:
  node src/cli.js --list
  node src/cli.js <source|group|all> [--full] [--max=N] [--page=N]

Examples:
  node src/cli.js building_permits --max=25        # smoke test: 25 rows
  node src/cli.js code_enforcement_open            # incremental refresh
  node src/cli.js "building permits"               # whole group
  node src/cli.js assessor_parcels --full          # 12.1M-row backfill (all roll years)
  node src/cli.js all                              # every non-static source, incremental
`);
}

function resolveTargets(token) {
  if (!token || token === 'all') {
    // everything except static historical bulk (run those explicitly)
    return Object.keys(SOURCES).filter((n) => !SOURCES[n].static);
  }
  if (GROUPS[token]) return GROUPS[token];
  if (SOURCES[token]) return [token];
  return null;
}

async function main() {
  if (flags.has('--list') || argv.length === 0) return printList();

  const token = positional[0];
  const targets = resolveTargets(token);
  if (!targets) {
    console.error(`Unknown source/group: "${token}". Run --list to see options.`);
    process.exitCode = 1;
    return;
  }

  const opts = {
    full: flags.has('--full'),
    maxRows: kv.max ? Number(kv.max) : undefined,
    pageSize: kv.page ? Number(kv.page) : undefined,
    year: kv.year || undefined,
  };

  const results = [];
  for (const name of targets) {
    try {
      results.push(await ingestSource(name, opts));
    } catch (err) {
      results.push({ name, error: err.message });
    }
  }

  const policy = classifyRunResults(results, SOURCES, token);
  const toleratedNames = new Set(policy.tolerated.map((r) => r.name));

  console.log('\n=== summary ===');
  for (const r of results) {
    if (!r.error) console.log(`  ✔ ${r.name}: ${r.total} rows`);
    else if (toleratedNames.has(r.name)) {
      const meta = SOURCES[r.name].allowFailure;
      console.log(`  ⚠ ${r.name}: known degraded (${meta.since}; ${meta.reason}): ${r.error}`);
    } else console.log(`  ✖ ${r.name}: ${r.error}`);
  }
  if (policy.tolerated.length) {
    console.log(`  ⚠ tolerated ${policy.tolerated.length} audited failure(s) in aggregate all run`);
  }
  if (policy.exitCode) process.exitCode = policy.exitCode;
}

main().finally(closePool);