[object Object]

← back to Ca Donations

ca-donations: political + foundation-grant ingest landed (both families live)

37ac67804142a6b06826e7c953461d0cee7b0904 · 2026-08-22 10:43:40 -0700 · Steve

- CAL-ACCESS 2018+ state contributions + FEC CA federal -> 6.95M political_contributions
- 990-PF Part XV grants (top 200 CA foundations) -> 40.5K charitable_grants
- ingest scripts: calaccess-political, fec-ca, 990-grants (adm-zip/unzipper/fast-xml-parser)
- verified: /api/political + /api/grants serve real rows; both grid tabs now populate

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Files touched

Diff

commit 37ac67804142a6b06826e7c953461d0cee7b0904
Author: Steve <steve@designerwallcoverings.com>
Date:   Sat Aug 22 10:43:40 2026 -0700

    ca-donations: political + foundation-grant ingest landed (both families live)
    
    - CAL-ACCESS 2018+ state contributions + FEC CA federal -> 6.95M political_contributions
    - 990-PF Part XV grants (top 200 CA foundations) -> 40.5K charitable_grants
    - ingest scripts: calaccess-political, fec-ca, 990-grants (adm-zip/unzipper/fast-xml-parser)
    - verified: /api/political + /api/grants serve real rows; both grid tabs now populate
    
    Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 scripts/ingest-990-grants.mjs          | 298 +++++++++++++++++++++++++++++++++
 scripts/ingest-calaccess-political.mjs | 244 +++++++++++++++++++++++++++
 scripts/ingest-fec-ca.mjs              | 189 +++++++++++++++++++++
 3 files changed, 731 insertions(+)

diff --git a/scripts/ingest-990-grants.mjs b/scripts/ingest-990-grants.mjs
new file mode 100644
index 0000000..3b01137
--- /dev/null
+++ b/scripts/ingest-990-grants.mjs
@@ -0,0 +1,298 @@
+// Ingest CA private-foundation GRANT records (990-PF Part XV outbound grants) into
+// charitable_grants. Also captures 990 Schedule I / Schedule F grant tables when present.
+//
+// SOURCE PATH (verified 2026-08-21):
+//   - The AWS s3://irs-form-990 HTTPS mirror is DEAD (404) and ProPublica blocks bulk XML.
+//   - The working free path is the IRS canonical e-file XML: the annual TEOS index CSV
+//     (https://apps.irs.gov/pub/epostcard/990/xml/2024/index_2024.csv) maps every filing to
+//     an OBJECT_ID + XML_BATCH_ID (the zip that contains <OBJECT_ID>_public.xml). We download
+//     only the batch zips that hold the most CA foundations, and extract ONLY the target XMLs.
+//
+// v1 BOUND (tractable): CA foundations (subsection 990-PF filers, joined from charitable_orgs)
+//   present in the 2024 index, restricted to the batch zips already downloaded into data/raw/,
+//   then capped to the TOP N (default 200) foundations by total grant dollars observed. The
+//   script logs covered vs skipped counts — no silent truncation.
+//
+// Grant parse:
+//   990-PF Part XV  -> SupplementaryInformationGrp/GrantOrContributionPdDurYrGrp
+//                      { RecipientPersonNm | RecipientBusinessName/BusinessNameLine1Txt,
+//                        RecipientUSAddress/{CityNm,StateAbbreviationCd},
+//                        GrantOrContributionPurposeTxt, Amt }         -> grant_type 990pf_partxv
+//   990 Schedule I  -> IRS990ScheduleI/RecipientTable                 -> grant_type 990_sched_i
+//   990 Schedule F  -> IRS990ScheduleF/GrantsToOrgOutsideUSGrp / ...  -> grant_type 990_sched_f
+//
+// external_id = <object_id>:<grant_type>:<index>  (stable, idempotent).
+// $0 — IRS bulk + already-loaded ProPublica org list; no paid API.
+
+import fs from 'node:fs';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+import unzipper from 'unzipper';
+import { XMLParser } from 'fast-xml-parser';
+import { q, startRun, upsertSource, pool } from '../lib/db.js';
+
+const __dirname = path.dirname(fileURLToPath(import.meta.url));
+const RAW = path.join(__dirname, '..', 'data', 'raw');
+const YEAR = process.env.YEAR || '2024';
+const INDEX_CSV = path.join(RAW, `index_${YEAR}.csv`);
+const SLUG = 'irs_990';
+const TOP_N = process.env.TOP_N ? parseInt(process.env.TOP_N, 10) : 200;
+
+const parser = new XMLParser({
+  ignoreAttributes: true,
+  parseTagValue: false, // keep amounts as strings; we Number() ourselves
+  trimValues: true,
+});
+
+function clean(s) {
+  if (s == null) return null;
+  const t = String(s).replace(/[\x00-\x1F\x7F]/g, '').trim();
+  return t === '' ? null : t;
+}
+function num(s) {
+  if (s == null || s === '') return null;
+  const n = Number(String(s).replace(/[$,]/g, ''));
+  return Number.isFinite(n) ? n : null;
+}
+function arr(x) { return x == null ? [] : Array.isArray(x) ? x : [x]; }
+
+// Read CA foundation EINs (dashless) from the DB.
+async function caFoundationEins() {
+  const rows = await q(`SELECT replace(ein,'-','') AS ein FROM charitable_orgs WHERE ein IS NOT NULL`);
+  return new Set(rows.map((r) => r.ein));
+}
+
+// Parse the 2024 index CSV -> per-EIN newest 990PF filing {object_id, batch, tax_period, name}.
+function readIndexTargets(caEins) {
+  const text = fs.readFileSync(INDEX_CSV, 'utf8');
+  const lines = text.split(/\r?\n/); // handle CRLF
+  const header = lines[0].split(',');
+  const col = Object.fromEntries(header.map((h, i) => [h.trim(), i]));
+  const hasBatch = col['XML_BATCH_ID'] !== undefined;
+  const best = new Map();
+  for (let i = 1; i < lines.length; i++) {
+    const line = lines[i];
+    if (!line) continue;
+    const c = line.split(',');
+    if (c[col['RETURN_TYPE']] !== '990PF') continue;
+    const ein = c[col['EIN']];
+    if (!caEins.has(ein)) continue;
+    const tp = c[col['TAX_PERIOD']];
+    const cur = best.get(ein);
+    if (!cur || tp > cur.tax_period) {
+      best.set(ein, {
+        ein,
+        object_id: c[col['OBJECT_ID']],
+        batch: hasBatch ? c[col['XML_BATCH_ID']].toLowerCase() : null,
+        tax_period: tp,
+        name: c[col['TAXPAYER_NAME']],
+      });
+    }
+  }
+  return best;
+}
+
+// Which batch zips are actually present in data/raw/?
+function presentBatches() {
+  const set = new Set();
+  for (const f of fs.readdirSync(RAW)) {
+    const m = f.match(/^(\d{4}_TEOS_XML_\d{2}[A-Za-z])\.zip$/);
+    if (m) set.add(m[1].toLowerCase());
+  }
+  return set;
+}
+
+// Extract grant rows from one parsed 990-PF/990 return object.
+function extractGrants(obj, objectId) {
+  const rd = obj?.Return?.ReturnData || {};
+  const hdr = obj?.Return?.ReturnHeader || {};
+  const filer = hdr.Filer || {};
+  const grantorEin = clean(filer.EIN);
+  const grantorName = clean(filer.BusinessName?.BusinessNameLine1Txt || filer.BusinessName?.BusinessNameLine1Txt);
+  const taxYear = clean(hdr.TaxYr) ? parseInt(clean(hdr.TaxYr), 10) : null;
+  const out = [];
+
+  // --- 990-PF Part XV outbound grants ---
+  const pf = rd.IRS990PF || {};
+  const supp = pf.SupplementaryInformationGrp || {};
+  const partxv = arr(supp.GrantOrContributionPdDurYrGrp);
+  partxv.forEach((g, i) => {
+    const grantee = clean(g.RecipientPersonNm)
+      || clean(g.RecipientBusinessName?.BusinessNameLine1Txt);
+    const addr = g.RecipientUSAddress || g.RecipientForeignAddress || {};
+    out.push({
+      external_id: `${objectId}:990pf_partxv:${i}`,
+      grantor_ein: grantorEin, grantor_name: grantorName,
+      grantee_name: grantee,
+      grantee_city: clean(addr.CityNm),
+      grantee_state: clean(addr.StateAbbreviationCd || addr.CountryCd),
+      amount: num(g.Amt),
+      purpose: clean(g.GrantOrContributionPurposeTxt),
+      tax_year: taxYear,
+      grant_type: '990pf_partxv',
+    });
+  });
+
+  // --- 990 Schedule I domestic grants (public charities / DAF sponsors) ---
+  const schI = rd.IRS990ScheduleI || {};
+  arr(schI.RecipientTable).forEach((g, i) => {
+    const grantee = clean(g.RecipientBusinessName?.BusinessNameLine1Txt) || clean(g.RecipientPersonNm);
+    const addr = g.USAddress || {};
+    out.push({
+      external_id: `${objectId}:990_sched_i:${i}`,
+      grantor_ein: grantorEin, grantor_name: grantorName,
+      grantee_name: grantee,
+      grantee_city: clean(addr.CityNm),
+      grantee_state: clean(addr.StateAbbreviationCd),
+      amount: num(g.CashGrantAmt) ?? num(g.NonCashAssistanceAmt),
+      purpose: clean(g.PurposeOfGrantTxt),
+      tax_year: taxYear,
+      grant_type: '990_sched_i',
+    });
+  });
+
+  // --- 990 Schedule F foreign grants ---
+  const schF = rd.IRS990ScheduleF || {};
+  arr(schF.GrantsToOrgOutsideUSGrp).forEach((g, i) => {
+    out.push({
+      external_id: `${objectId}:990_sched_f:${i}`,
+      grantor_ein: grantorEin, grantor_name: grantorName,
+      grantee_name: clean(g.NameOfOrganizationTxt) || null,
+      grantee_city: clean(g.RegionTxt),
+      grantee_state: clean(g.RegionTxt),
+      amount: num(g.CashGrantAmt),
+      purpose: clean(g.PurposeOfGrantTxt),
+      tax_year: taxYear,
+      grant_type: '990_sched_f',
+    });
+  });
+
+  // keep only rows with a grantee name or an amount (drop empty artifacts)
+  return out.filter((g) => g.grantee_name || g.amount);
+}
+
+const COLS = [
+  'source_slug', 'external_id', 'grantor_ein', 'grantor_name', 'grantee_name',
+  'grantee_city', 'grantee_state', 'amount', 'purpose', 'tax_year', 'grant_type',
+];
+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) => (c === 'source_slug' ? SLUG : r[c])));
+  await q(
+    `INSERT INTO charitable_grants (${COLS.join(',')}) VALUES ${ph.join(',')}
+     ON CONFLICT (source_slug, external_id) DO UPDATE SET
+       grantor_ein=EXCLUDED.grantor_ein, grantor_name=EXCLUDED.grantor_name,
+       grantee_name=EXCLUDED.grantee_name, grantee_city=EXCLUDED.grantee_city,
+       grantee_state=EXCLUDED.grantee_state, amount=EXCLUDED.amount,
+       purpose=EXCLUDED.purpose, tax_year=EXCLUDED.tax_year, grant_type=EXCLUDED.grant_type`,
+    vals,
+  );
+  return rows.length;
+}
+
+async function main() {
+  if (!fs.existsSync(INDEX_CSV)) {
+    console.error(`Missing ${INDEX_CSV}. Download index_${YEAR}.csv into data/raw/ first.`);
+    process.exit(1);
+  }
+  await upsertSource({
+    slug: SLUG,
+    name: 'IRS Form 990 e-file XML (990-PF Part XV + 990 Sched I/F grants)',
+    url: 'https://www.irs.gov/charities-non-profits/form-990-series-downloads',
+    jurisdiction: 'federal', family: 'charitable', granularity: 'grant', access: 'bulk',
+    notes: `CA private-foundation outbound grants, filing year ${YEAR}. Object XML from TEOS batch zips.`,
+  });
+  const run = await startRun(SLUG);
+
+  let foundationsCovered = 0, foundationsSkippedNoBatch = 0, grantsUpserted = 0;
+  try {
+    const caEins = await caFoundationEins();
+    const targets = readIndexTargets(caEins); // ein -> filing
+    const present = presentBatches();
+    console.log(`CA 990PF foundations in ${YEAR} index: ${targets.size}`);
+    console.log(`Batch zips present in data/raw/: ${[...present].join(', ') || '(none)'}`);
+
+    // Restrict to foundations whose batch zip we actually have.
+    const reachable = [];
+    for (const t of targets.values()) {
+      if (t.batch && present.has(t.batch)) reachable.push(t);
+      else foundationsSkippedNoBatch++;
+    }
+    console.log(`Reachable (batch present): ${reachable.length}; unreachable (batch not downloaded): ${foundationsSkippedNoBatch}`);
+
+    // Group reachable targets by batch so each zip is opened once.
+    const byBatch = new Map();
+    for (const t of reachable) {
+      if (!byBatch.has(t.batch)) byBatch.set(t.batch, new Map());
+      byBatch.get(t.batch).set(t.object_id, t);
+    }
+
+    // First pass: parse every reachable foundation's grants into memory, keyed by ein,
+    // so we can rank by total grant dollars and cap at TOP_N.
+    const perFoundation = []; // { ein, name, total, grants[] }
+    for (const [batch, oidMap] of byBatch) {
+      // Find the actual zip file on disk case-insensitively (batch id is lowercased).
+      const diskName = fs.readdirSync(RAW).find((f) => f.toLowerCase() === `${batch}.zip`);
+      const dir = await unzipper.Open.file(path.join(RAW, diskName));
+      // index zip entries by object id substring for quick lookup
+      const wanted = new Set(oidMap.keys());
+      let hit = 0;
+      for (const f of dir.files) {
+        const base = path.basename(f.path);
+        const m = base.match(/^(\d+)_public\.xml$/);
+        if (!m) continue;
+        const oid = m[1];
+        if (!wanted.has(oid)) continue;
+        hit++;
+        const t = oidMap.get(oid);
+        const xml = (await f.buffer()).toString('utf8');
+        let obj;
+        try { obj = parser.parse(xml); } catch { continue; }
+        const grants = extractGrants(obj, oid);
+        const total = grants.reduce((s, g) => s + (g.amount || 0), 0);
+        perFoundation.push({ ein: t.ein, name: t.name, total, grants });
+      }
+      console.log(`  ${batch}: matched ${hit}/${wanted.size} target XMLs.`);
+    }
+
+    // Rank by total grant dollars, cap at TOP_N.
+    perFoundation.sort((a, b) => b.total - a.total);
+    const withGrants = perFoundation.filter((f) => f.grants.length);
+    const chosen = withGrants.slice(0, TOP_N);
+    const skippedByCap = Math.max(0, withGrants.length - chosen.length);
+    console.log(`Foundations with >=1 grant: ${withGrants.length}; taking top ${chosen.length} by grant $, skipping ${skippedByCap} beyond cap.`);
+
+    // Load chosen foundations' grants.
+    let batch = [];
+    for (const f of chosen) {
+      foundationsCovered++;
+      for (const g of f.grants) {
+        batch.push(g);
+        if (batch.length >= 1000) { grantsUpserted += await flush(batch); batch = []; }
+      }
+    }
+    if (batch.length) grantsUpserted += await flush(batch);
+
+    const detail = `covered ${foundationsCovered} foundations (top ${TOP_N} by grant $); `
+      + `${grantsUpserted} grant rows; skippedByCap ${skippedByCap}; `
+      + `unreachable(no batch dl) ${foundationsSkippedNoBatch} of ${targets.size} indexed.`;
+    await run.done(chosen.length, grantsUpserted, 'ok', detail);
+    console.log(`IRS 990-PF grants: ${detail} $0 (IRS bulk).`);
+  } catch (e) {
+    await run.done(foundationsCovered, grantsUpserted, 'error', String(e.message));
+    console.error('FAILED:', e.stack || e.message);
+    process.exitCode = 1;
+  } finally {
+    await pool.end();
+  }
+}
+
+main();
diff --git a/scripts/ingest-calaccess-political.mjs b/scripts/ingest-calaccess-political.mjs
new file mode 100644
index 0000000..2fa0db0
--- /dev/null
+++ b/scripts/ingest-calaccess-political.mjs
@@ -0,0 +1,244 @@
+// 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';
+
+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);
+
+    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();
diff --git a/scripts/ingest-fec-ca.mjs b/scripts/ingest-fec-ca.mjs
new file mode 100644
index 0000000..aa486df
--- /dev/null
+++ b/scripts/ingest-fec-ca.mjs
@@ -0,0 +1,189 @@
+// 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';
+
+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);
+
+    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();

← 5197daa auto-data-snapshot: 2026-08-21T08:45:49 (2 data files) — pac  ·  back to Ca Donations  ·  ca-donations: wire server-side sort (mandatory sort rule) ac 9e7508b →