`src/ingest/census_acs.ts`

117 lines
/**
 * Census ACS 5-year (2023 vintage) — one call for all counties.
 * Metrics -> metric_series source 'acs', period 2023-07-01 (mid-vintage anchor).
 * Also updates region.population from B01003.
 * Keyless was probed 2026-07-21 and REDIRECTS to missing_key.html — a key is required;
 * we read CENSUS_API_KEY from ./.env then ~/Projects/secrets-manager/.env.
 */
import { readFileSync, existsSync } from 'node:fs';
import { join } from 'node:path';
import { homedir } from 'node:os';
import { query, pool } from '../../db/pool.ts';
import { openRun, closeRun } from './run.ts';
import { loadRegionMaps, upsertMetricRows, UnmatchedCollector, type MetricRow } from './crosswalk.ts';

const ACS_PERIOD = '2023-07-01';
const SOURCE = 'acs';
const VARS = 'NAME,B19013_001E,B25077_001E,B25064_001E,B01003_001E,B25002_001E,B25002_003E,B25003_001E,B25003_003E';
const BASE = `https://api.census.gov/data/2023/acs/acs5?get=${VARS}&for=county:*`;

function findCensusKey(): string | null {
  if (process.env.CENSUS_API_KEY) return process.env.CENSUS_API_KEY;
  for (const p of [join(process.cwd(), '.env'), join(homedir(), 'Projects', 'secrets-manager', '.env')]) {
    if (!existsSync(p)) continue;
    const m = readFileSync(p, 'utf8').match(/^CENSUS_API_KEY=(.+)$/m);
    if (m) return m[1].trim().replace(/^["']|["']$/g, '');
  }
  return null;
}

/** Census sentinel negatives (-666666666 etc.) and null -> undefined. */
function num(v: string | null): number | undefined {
  if (v == null || v === '') return undefined;
  const n = Number(v);
  if (!Number.isFinite(n) || n < 0) return undefined;
  return n;
}

async function main() {
  const key = findCensusKey();
  const url = key ? `${BASE}&key=${key}` : BASE;
  const runId = await openRun(SOURCE, BASE);
  try {
    const res = await fetch(url, { redirect: 'follow' });
    if (!res.ok) throw new Error(`ACS fetch failed: ${res.status} ${res.statusText}`);
    if (res.url.includes('missing_key')) throw new Error('ACS redirected to missing_key.html — CENSUS_API_KEY required/invalid');
    const data: string[][] = await res.json();
    if (!Array.isArray(data) || data.length < 2) throw new Error(`ACS returned ${Array.isArray(data) ? data.length : 0} rows`);

    const header = data[0];
    const col = (name: string) => {
      const i = header.indexOf(name);
      if (i < 0) throw new Error(`ACS header missing ${name}: ${header.join(',')}`);
      return i;
    };
    const iIncome = col('B19013_001E'), iValue = col('B25077_001E'), iRent = col('B25064_001E'),
      iPop = col('B01003_001E'), iUnits = col('B25002_001E'), iVacant = col('B25002_003E'),
      iOccTot = col('B25003_001E'), iRenter = col('B25003_003E'),
      iState = col('state'), iCounty = col('county'), iName = col('NAME');

    const maps = await loadRegionMaps();
    const unmatched = new UnmatchedCollector();
    const rows: MetricRow[] = [];
    const popUpdates: { regionId: number; population: number }[] = [];
    let skipped = 0;

    for (const r of data.slice(1)) {
      const fips = r[iState].padStart(2, '0') + r[iCounty].padStart(3, '0');
      const region = maps.byFips.get(fips);
      if (!region) {
        unmatched.add(SOURCE, { fips, name: r[iName] });
        skipped++;
        continue;
      }
      const push = (metric: string, value: number | undefined) => {
        if (value !== undefined) rows.push({ regionId: region.id, source: SOURCE, metric, period: ACS_PERIOD, value });
      };
      push('median_hh_income', num(r[iIncome]));
      push('median_home_value_acs', num(r[iValue]));
      push('median_gross_rent', num(r[iRent]));
      const pop = num(r[iPop]);
      push('population', pop);
      if (pop !== undefined) popUpdates.push({ regionId: region.id, population: pop });
      const units = num(r[iUnits]), vacant = num(r[iVacant]);
      if (units !== undefined && units > 0 && vacant !== undefined) push('vacancy_rate', vacant / units);
      const occ = num(r[iOccTot]), renter = num(r[iRenter]);
      if (occ !== undefined && occ > 0 && renter !== undefined) push('renter_share', renter / occ);
    }

    if (rows.length === 0) throw new Error('ACS parse produced 0 metric rows');

    const upserted = await upsertMetricRows(rows, runId);
    for (let i = 0; i < popUpdates.length; i += 500) {
      const batch = popUpdates.slice(i, i + 500);
      const params: unknown[] = [];
      const values = batch.map((u, j) => { params.push(u.regionId, u.population); return `($${j * 2 + 1}::int,$${j * 2 + 2}::int)`; });
      await query(
        `UPDATE region rg SET population = v.pop FROM (VALUES ${values.join(',')}) AS v(id, pop) WHERE rg.id = v.id`,
        params,
      );
    }
    unmatched.flush();
    await closeRun(runId, 'ok', {
      upserted, skipped,
      notes: `counties=${data.length - 1} key=${key ? 'used' : 'keyless'} pop_updates=${popUpdates.length} unmatched=${unmatched.count(SOURCE)}`,
    });
    console.log(`[acs] ok: ${upserted} metric rows, ${popUpdates.length} region.population updates, ${skipped} skipped`);
  } catch (e) {
    await closeRun(runId, 'failed', { notes: String(e) });
    console.error('[acs] FAILED:', e);
    process.exitCode = 1;
  } finally {
    await pool.end();
  }
}

main();