`scripts/ingest-fec-ca.mjs`

195 lines
// Ingest FEDERAL individual contributions from CA donors, from the FEC bulk download.
//
// Source: https://www.fec.gov/files/bulk-downloads/2024/indiv24.zip  (~4.2 GB, 302 -> S3)
// File inside: itcont.txt — PIPE-delimited, NO header row. Fixed 21-column layout:
//   0 CMTE_ID 1 AMNDT_IND 2 RPT_TP 3 TRANSACTION_PGI 4 IMAGE_NUM 5 TRANSACTION_TP
//   6 ENTITY_TP 7 NAME 8 CITY 9 STATE 10 ZIP_CODE 11 EMPLOYER 12 OCCUPATION
//   13 TRANSACTION_DT (MMDDYYYY) 14 TRANSACTION_AMT 15 OTHER_ID 16 TRAN_ID
//   17 FILE_NUM 18 MEMO_CD 19 MEMO_TEXT 20 SUB_ID
// Layout doc: https://www.fec.gov/campaign-finance-data/contributions-individuals-file-description/
//
// Filter STATE=CA, map into political_contributions with jurisdiction='federal',
// form='FEC_A', source_slug='fec_bulk', external_id=SUB_ID (idempotent upsert).
// Recipient committee NAME resolved from cm<cycle>.zip (CMTE_ID -> CMTE_NM).
//
// Stream-parse + filter; the file never enters memory whole. $0 — free FEC bulk.

import fs from 'node:fs';
import path from 'node:path';
import readline from 'node:readline';
import { fileURLToPath } from 'node:url';
import unzipper from 'unzipper';
import { q, startRun, upsertSource, pool } from '../lib/db.js';
import { refreshPoliticalAgg } from './refresh-political-agg.mjs';

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const RAW = path.join(__dirname, '..', 'data', 'raw');
const CYCLE = process.env.CYCLE || '24';
const INDIV_ZIP = path.join(RAW, `indiv${CYCLE}.zip`);
const CM_ZIP = path.join(RAW, `cm${CYCLE}.zip`);
const SLUG = 'fec_bulk';
const STATE = 'CA';
const BATCH = 2000;
const MAX_ROWS = process.env.MAX_ROWS ? parseInt(process.env.MAX_ROWS, 10) : 0;

// FEC indiv fixed column indices.
const F = {
  CMTE_ID: 0, NAME: 7, CITY: 8, STATE: 9, ZIP: 10, EMPLOYER: 11, OCCUPATION: 12,
  DT: 13, AMT: 14, SUB_ID: 20,
};

function clean(s) {
  if (s == null) return null;
  const t = String(s).replace(/[\x00-\x1F\x7F]/g, '').trim();
  return t === '' ? null : t;
}

// FEC dates are MMDDYYYY.
function parseFecDate(s) {
  if (!s) return null;
  const t = s.trim();
  const m = t.match(/^(\d{2})(\d{2})(\d{4})$/);
  if (!m) return null;
  const mo = +m[1], d = +m[2], y = +m[3];
  if (y < 1974 || y > 2030 || mo < 1 || mo > 12 || d < 1 || d > 31) return null;
  return `${y}-${m[1]}-${m[2]}`;
}

async function openTxt(zipPath, wantBase) {
  const dir = await unzipper.Open.file(zipPath);
  return dir.files.find((f) => path.basename(f.path).toLowerCase() === wantBase.toLowerCase()) || null;
}

async function forEachLine(entry, onLine) {
  const rl = readline.createInterface({ input: entry.stream(), crlfDelay: Infinity });
  for await (const line of rl) await onLine(line);
}

// Build CMTE_ID -> committee name from cm<cycle>.zip / cm.txt (pipe-delimited, no header).
async function buildCommitteeMap() {
  const map = new Map();
  if (!fs.existsSync(CM_ZIP)) {
    console.warn(`  ${CM_ZIP} not present — recipient names will be committee ids.`);
    return map;
  }
  const entry = await openTxt(CM_ZIP, 'cm.txt');
  if (!entry) { console.warn('  cm.txt not in cm zip.'); return map; }
  await forEachLine(entry, (line) => {
    if (!line) return;
    const c = line.split('|');
    const id = clean(c[0]);
    const nm = clean(c[1]);
    if (id && nm) map.set(id, nm);
  });
  console.log(`  FEC committee map: ${map.size} committees.`);
  return map;
}

const COLS = [
  'source_slug', 'external_id', 'donor_name', 'donor_employer', 'donor_occupation',
  'donor_city', 'donor_state', 'donor_zip', 'amount', 'contribution_date',
  'recipient_name', 'recipient_id', 'office', 'jurisdiction', 'form',
];

async function flush(rows) {
  if (!rows.length) return 0;
  const byKey = new Map();
  for (const r of rows) byKey.set(r.external_id, r);
  rows = [...byKey.values()];
  const ph = rows.map((_, i) => {
    const b = i * COLS.length;
    return '(' + COLS.map((_, j) => `$${b + j + 1}`).join(',') + ')';
  });
  const vals = [];
  for (const r of rows) vals.push(...COLS.map((c) => r[c]));
  await q(
    `INSERT INTO political_contributions (${COLS.join(',')}) VALUES ${ph.join(',')}
     ON CONFLICT (source_slug, external_id) DO UPDATE SET
       donor_name=EXCLUDED.donor_name, donor_employer=EXCLUDED.donor_employer,
       donor_occupation=EXCLUDED.donor_occupation, donor_city=EXCLUDED.donor_city,
       donor_state=EXCLUDED.donor_state, donor_zip=EXCLUDED.donor_zip,
       amount=EXCLUDED.amount, contribution_date=EXCLUDED.contribution_date,
       recipient_name=EXCLUDED.recipient_name, recipient_id=EXCLUDED.recipient_id,
       form=EXCLUDED.form`,
    vals,
  );
  return rows.length;
}

async function main() {
  if (!fs.existsSync(INDIV_ZIP)) {
    console.error(`Missing ${INDIV_ZIP}. Download indiv${CYCLE}.zip into data/raw/ first.`);
    process.exit(1);
  }
  await upsertSource({
    slug: SLUG,
    name: `FEC bulk individual contributions (cycle 20${CYCLE})`,
    url: `https://www.fec.gov/files/bulk-downloads/20${CYCLE}/indiv${CYCLE}.zip`,
    jurisdiction: 'federal', family: 'political', granularity: 'donor', access: 'bulk',
    notes: `itcont.txt filtered STATE=CA. SUB_ID = external_id. Committee names from cm${CYCLE}.`,
  });
  const run = await startRun(SLUG);
  let seen = 0, kept = 0, upserted = 0, skippedNoSub = 0;

  try {
    const cmtes = await buildCommitteeMap();
    const entry = await openTxt(INDIV_ZIP, 'itcont.txt');
    if (!entry) throw new Error('itcont.txt not found in indiv zip');

    let batch = [];
    let stop = false;

    await forEachLine(entry, async (line) => {
      if (stop || !line) return;
      seen++;
      const c = line.split('|');
      if (clean(c[F.STATE]) !== STATE) return; // filter to CA donors
      const subId = clean(c[F.SUB_ID]);
      if (!subId) { skippedNoSub++; return; }

      const amtRaw = c[F.AMT];
      const amount = amtRaw != null && amtRaw.trim() !== '' ? Number(amtRaw) : null;
      const cmteId = clean(c[F.CMTE_ID]);

      batch.push({
        source_slug: SLUG,
        external_id: subId,
        donor_name: clean(c[F.NAME]),
        donor_employer: clean(c[F.EMPLOYER]),
        donor_occupation: clean(c[F.OCCUPATION]),
        donor_city: clean(c[F.CITY]),
        donor_state: 'CA',
        donor_zip: clean(c[F.ZIP]),
        amount: Number.isFinite(amount) ? amount : null,
        contribution_date: parseFecDate(c[F.DT]),
        recipient_name: (cmteId && cmtes.get(cmteId)) || null,
        recipient_id: cmteId,
        office: null,
        jurisdiction: 'federal',
        form: 'FEC_A',
      });
      kept++;

      if (batch.length >= BATCH) { upserted += await flush(batch); batch = []; }
      if (MAX_ROWS && kept >= MAX_ROWS) stop = true;
    });

    if (batch.length) upserted += await flush(batch);

    console.log('Refreshing political_agg rollup...');
    const aggCounts = await refreshPoliticalAgg();
    for (const r of aggCounts) console.log(`  agg ${r.dimension}: ${r.groups} groups (min_donors=${r.min_donors})`);

    await run.done(seen, upserted, 'ok', `CA kept ${kept} of ${seen}; skippedNoSub ${skippedNoSub}`);
    console.log(`FEC indiv${CYCLE}: seen ${seen}, CA kept ${kept}, upserted ${upserted}. $0 (free bulk).`);
  } catch (e) {
    await run.done(seen, upserted, 'error', String(e.message));
    console.error('FAILED:', e.stack || e.message);
    process.exitCode = 1;
  } finally {
    await pool.end();
  }
}

main();