← 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
A scripts/ingest-990-grants.mjsA scripts/ingest-calaccess-political.mjsA scripts/ingest-fec-ca.mjs
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 →