`scripts/ingest-990-grants.mjs`

299 lines
// 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();