`scripts/ingest-calaccess-political.mjs`

250 lines
// Ingest CA state campaign-finance CONTRIBUTIONS from the CAL-ACCESS daily bulk dump.
//
// Source: https://campaignfinance.cdn.sos.ca.gov/dbwebexport.zip  (~1.58 GB, daily, free)
// Table of interest: RCPT_CD (receipts / contributions), tab-delimited, header row present.
// RCPT = Schedule A monetary contributions received on Form 460/461/etc.
//
// v1 scope: filter to RCPT_DATE >= 2018-01-01 (RCPT_CD is ~10M+ rows / 3.8 GB uncompressed).
// Maps into political_contributions with jurisdiction='state', source_slug='calaccess',
// form = the row's FORM_TYPE, external_id = FILING_ID:TRAN_ID (idempotent upsert).
//
// Recipient committee/candidate NAME resolved via CVR_CAMPAIGN_DISCLOSURE_CD (FILING_ID ->
// FILER_NAML), which is a direct join every RCPT row supports. We DON'T install the
// django-calaccess-raw-data Django/MySQL stack — we stream the TSVs straight from the zip.
//
// RCPT_CD columns (verified against the dump's own header at runtime, mapped BY NAME):
//   FILING_ID, TRAN_ID, FORM_TYPE, RCPT_DATE, AMOUNT, CMTE_ID,
//   CTRIB_NAML/NAMF/NAMT/NAMS, CTRIB_EMP, CTRIB_OCC, CTRIB_CITY, CTRIB_ST, CTRIB_ZIP4,
//   OFFICE_CD, OFFIC_DSCR
//
// $0 — local bulk parse, no paid API.

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 ZIP = path.join(RAW, 'dbwebexport.zip');
const SLUG = 'calaccess';
const SINCE = process.env.SINCE || '2018-01-01';
const BATCH = 2000;
const MAX_ROWS = process.env.MAX_ROWS ? parseInt(process.env.MAX_ROWS, 10) : 0; // 0 = no cap

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

// Fully-buffered line reader with a per-line async callback that the loop AWAITS,
// so DB flushes apply real backpressure (readline's async iterator pauses the stream
// while we're awaiting inside the loop).
async function forEachLine(entry, onLine) {
  const rl = readline.createInterface({ input: entry.stream(), crlfDelay: Infinity });
  for await (const line of rl) await onLine(line);
}

function headerIndex(line) {
  const idx = {};
  line.split('\t').forEach((name, i) => (idx[name.trim().toUpperCase()] = i));
  return idx;
}

// Strip NUL + other control chars Postgres UTF8 rejects; trim; empty -> null.
function clean(s) {
  if (s == null) return null;
  // Drop C0 control chars (incl. NUL) + DEL.
  const t = String(s).replace(/[\x00-\x1F\x7F]/g, '').trim();
  return t === '' ? null : t;
}

function joinName(...parts) {
  return parts.map((p) => clean(p) || '').filter(Boolean).join(' ') || null;
}

// CAL-ACCESS dates are M/D/YYYY (optionally with a trailing time). Return YYYY-MM-DD or null.
// The raw data has typo'd years (e.g. 8201, 5201, 6830); reject anything outside a sane window
// so a bad year can't slip past the SINCE lexical filter.
function parseDate(s) {
  if (!s) return null;
  const t = s.trim();
  let y, mo, d;
  const m = t.match(/^(\d{1,2})\/(\d{1,2})\/(\d{4})/);
  if (m) { mo = +m[1]; d = +m[2]; y = +m[3]; }
  else {
    const iso = t.match(/^(\d{4})-(\d{1,2})-(\d{1,2})/);
    if (!iso) return null;
    y = +iso[1]; mo = +iso[2]; d = +iso[3];
  }
  if (y < 1974 || y > 2030 || mo < 1 || mo > 12 || d < 1 || d > 31) return null;
  return `${String(y).padStart(4, '0')}-${String(mo).padStart(2, '0')}-${String(d).padStart(2, '0')}`;
}

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;
  // Dedupe within the batch by external_id (RCPT keeps every AMEND_ID, so FILING_ID:TRAN_ID
  // repeats). Postgres rejects a multi-row upsert that touches the same conflict key twice;
  // keep the LAST occurrence (later amendments overwrite earlier ones).
  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,
       office=EXCLUDED.office, form=EXCLUDED.form`,
    vals,
  );
  return rows.length;
}

// Build FILING_ID -> { name, filerId } from CVR_CAMPAIGN_DISCLOSURE_CD (recipient of the filing).
// Keep the AMEND_ID=highest? Simpler: last-writer-wins is fine for a display name.
async function buildRecipientMap() {
  const map = new Map();
  const entry = await openEntry('CVR_CAMPAIGN_DISCLOSURE_CD.TSV');
  if (!entry) {
    console.warn('  CVR_CAMPAIGN_DISCLOSURE_CD.TSV not found — recipient names blank.');
    return map;
  }
  let idx = null;
  await forEachLine(entry, (line) => {
    if (idx === null) { idx = headerIndex(line); return; }
    if (!line) return;
    const c = line.split('\t');
    const filingId = c[idx['FILING_ID']];
    if (!filingId) return;
    const name = joinName(c[idx['FILER_NAMT']], c[idx['FILER_NAMF']], c[idx['FILER_NAML']], c[idx['FILER_NAMS']]);
    const filerId = clean(c[idx['FILER_ID']]);
    if (name || filerId) map.set(filingId, { name, filerId });
  });
  console.log(`  CVR recipient map: ${map.size} filings.`);
  return map;
}

async function main() {
  if (!fs.existsSync(ZIP)) {
    console.error(`Missing ${ZIP}. Download dbwebexport.zip into data/raw/ first.`);
    process.exit(1);
  }
  await upsertSource({
    slug: SLUG,
    name: 'CAL-ACCESS Campaign Finance (RCPT contributions)',
    url: 'https://campaignfinance.cdn.sos.ca.gov/dbwebexport.zip',
    jurisdiction: 'state', family: 'political', granularity: 'donor', access: 'bulk',
    notes: `RCPT_CD Schedule A receipts, contribution_date >= ${SINCE}. Daily CAL-ACCESS dump.`,
  });
  const run = await startRun(SLUG);
  let seen = 0, kept = 0, upserted = 0, skippedOld = 0, skippedNoTran = 0;

  try {
    console.log('Building recipient map from CVR_CAMPAIGN_DISCLOSURE_CD...');
    const recip = await buildRecipientMap();

    const rcpt = await openEntry('RCPT_CD.TSV');
    if (!rcpt) throw new Error('RCPT_CD.TSV not found in dump');

    let idx = null, need = null;
    let batch = [];
    let stop = false;

    await forEachLine(rcpt, async (line) => {
      if (stop) return;
      if (idx === null) {
        idx = headerIndex(line);
        need = {
          filing: idx['FILING_ID'], tran: idx['TRAN_ID'], form: idx['FORM_TYPE'],
          date: idx['RCPT_DATE'], amt: idx['AMOUNT'], cmte: idx['CMTE_ID'],
          naml: idx['CTRIB_NAML'], namf: idx['CTRIB_NAMF'], namt: idx['CTRIB_NAMT'], nams: idx['CTRIB_NAMS'],
          emp: idx['CTRIB_EMP'], occ: idx['CTRIB_OCC'],
          city: idx['CTRIB_CITY'], st: idx['CTRIB_ST'], zip: idx['CTRIB_ZIP4'],
          officeCd: idx['OFFICE_CD'], officeDscr: idx['OFFIC_DSCR'],
        };
        if (need.date === undefined || need.amt === undefined || need.tran === undefined) {
          throw new Error(`RCPT_CD header missing required cols. Header=${line}`);
        }
        return;
      }
      if (!line) return;
      seen++;
      const c = line.split('\t');
      const tran = c[need.tran];
      if (!tran) { skippedNoTran++; return; }
      const date = parseDate(c[need.date]);
      if (!date || date < SINCE) { skippedOld++; return; }

      const filingId = c[need.filing] || null;
      const r = filingId ? recip.get(filingId) : null;
      const amtRaw = c[need.amt];
      const amount = amtRaw != null && amtRaw.trim() !== '' ? Number(amtRaw) : null;
      const office = (need.officeDscr !== undefined ? clean(c[need.officeDscr]) : null)
        || (need.officeCd !== undefined ? clean(c[need.officeCd]) : null) || null;

      batch.push({
        source_slug: SLUG,
        external_id: clean(filingId ? `${filingId}:${tran}` : tran),
        donor_name: joinName(c[need.namt], c[need.namf], c[need.naml], c[need.nams]),
        donor_employer: clean(c[need.emp]),
        donor_occupation: clean(c[need.occ]),
        donor_city: clean(c[need.city]),
        donor_state: clean(c[need.st]),
        donor_zip: clean(c[need.zip]),
        amount: Number.isFinite(amount) ? amount : null,
        contribution_date: date,
        recipient_name: r?.name || null,
        recipient_id: clean(c[need.cmte]) || r?.filerId || filingId || null,
        office,
        jurisdiction: 'state',
        form: clean(c[need.form]) || '460A',
      });
      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',
      `kept ${kept} (>=${SINCE}); skippedOld ${skippedOld}; skippedNoTran ${skippedNoTran}`);
    console.log(`CAL-ACCESS RCPT: seen ${seen}, kept ${kept} (>=${SINCE}), upserted ${upserted}. $0 (local 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();