"""Full FEC individual contributions ingest from bulk files. Streams indiv24.zip and indiv26.zip, filters to a ZIP prefix, parses the 21-column pipe-delimited layout, and loads into fec_contributions via COPY. Also loads the small committee master file (cm[cy].zip) so we can resolve committee_id → name/type/party. Rerunnable: truncates fec_contributions before loading. """ from __future__ import annotations import csv import gzip import io import subprocess import sys import time import zipfile from pathlib import Path import requests from sqlalchemy import create_engine, text DATA_DIR = Path("/home/m3ac/genoa-entwuerfe.com") CACHE = DATA_DIR / "gis_scripts" / ".cache" CACHE.mkdir(exist_ok=True) ZIP_PREFIX = "43082" # ZIP_CODE column starts_with CYCLES = [26, 24] # 2-digit cycle labels used in FEC filenames # Field order in indiv*.txt (pipe-delimited, 21 cols) INDIV_COLS = [ "cmte_id", "amndt_ind", "rpt_tp", "transaction_pgi", "image_num", "transaction_tp", "entity_tp", "name", "city", "state", "zip_code", "employer", "occupation", "transaction_dt", "transaction_amt", "other_id", "tran_id", "file_num", "memo_cd", "memo_text", "sub_id", ] # Field order in cm.txt (committee master, 15 cols) CM_COLS = [ "cmte_id", "cmte_nm", "tres_nm", "cmte_st1", "cmte_st2", "cmte_city", "cmte_st", "cmte_zip", "cmte_dsgn", "cmte_tp", "cmte_pty_affiliation", "cmte_filing_freq", "org_tp", "connected_org_nm", "cand_id", ] def _env(var: str) -> str: for line in (DATA_DIR / ".env.gis").read_text().splitlines(): if line.startswith(f"{var}="): return line.split("=", 1)[1].strip() raise SystemExit(f"{var} missing") def _download(url: str, dest: Path) -> Path: if dest.exists() and dest.stat().st_size > 100_000: print(f"[cache] {dest.name} ({dest.stat().st_size/1e6:.1f} MB)") return dest print(f"[fetch] {url}") t0 = time.time() with requests.get(url, stream=True, timeout=600) as r: r.raise_for_status() total = int(r.headers.get("content-length") or 0) with open(dest, "wb") as f: got = 0 for chunk in r.iter_content(chunk_size=1024*1024): f.write(chunk); got += len(chunk) if total and got % (50*1024*1024) < 1024*1024: print(f" {got/1e6:.0f}/{total/1e6:.0f} MB") print(f" done in {time.time()-t0:.0f}s") return dest def _filter_indiv(zip_path: Path, cycle: int, out: list[dict]) -> int: """Stream unzip the file inside the archive and filter by ZIP prefix.""" n_kept = 0 n_seen = 0 with zipfile.ZipFile(zip_path) as z: # the txt file is usually named itcont.txt member = next(n for n in z.namelist() if n.endswith(".txt")) print(f"[stream] {zip_path.name} ({member})") with z.open(member) as fh: reader = csv.reader(io.TextIOWrapper(fh, encoding="latin-1", newline=""), delimiter="|", quoting=csv.QUOTE_NONE) for row in reader: n_seen += 1 if len(row) < 21: continue if not row[10].startswith(ZIP_PREFIX): continue rec = dict(zip(INDIV_COLS, row)) # split "LAST, FIRST MIDDLE" name = rec["name"] or "" last = first = middle = None if "," in name: left, right = [x.strip() for x in name.split(",", 1)] last = left or None parts = right.split() if parts: first = parts[0] if len(parts) > 1: middle = " ".join(parts[1:]) # transaction_dt is MMDDYYYY dt = rec["transaction_dt"] or "" iso = None if len(dt) == 8 and dt.isdigit(): iso = f"{dt[4:8]}-{dt[0:2]}-{dt[2:4]}" try: amt = float(rec["transaction_amt"] or 0) except ValueError: amt = 0 try: sub_id = int(rec["sub_id"]) except (ValueError, TypeError): continue out.append({ "sub_id": sub_id, "cycle": 2000 + cycle, "contribution_date": iso, "amount": amt, "contributor_name": name or None, "contributor_first_name": first, "contributor_last_name": last, "contributor_middle_name": middle, "contributor_street_1": None, # bulk file omits street "contributor_city": rec["city"] or None, "contributor_state": rec["state"] or None, "contributor_zip": (rec["zip_code"] or "")[:5] or None, "contributor_employer": rec["employer"] or None, "contributor_occupation": rec["occupation"] or None, "committee_id": rec["cmte_id"] or None, "committee_name": None, "committee_type": None, "candidate_id": rec["other_id"] or None, # populated when target is a candidate "candidate_name": None, "candidate_office": None, }) n_kept += 1 if n_kept % 5000 == 0: print(f" kept {n_kept} of {n_seen} seen") print(f"[filter] cycle=20{cycle:02d}: kept {n_kept} of {n_seen} rows") return n_kept def _copy_indiv(engine, rows: list[dict]) -> None: if not rows: return # write CSV to temp file and use raw psycopg2 COPY for speed import psycopg2, os dsn = f"host=127.0.0.1 dbname=westerville user=gisapp password={_env('GISPASS')}" tmp = CACHE / "fec_indiv_load.csv" cols = list(rows[0].keys()) with open(tmp, "w", newline="") as f: w = csv.writer(f) for r in rows: w.writerow([r[c] if r[c] is not None else "" for c in cols]) with psycopg2.connect(dsn) as conn, conn.cursor() as cur: cur.execute("CREATE TEMP TABLE fec_stage (LIKE fec_contributions INCLUDING DEFAULTS)") with open(tmp) as f: cur.copy_expert( f"COPY fec_stage ({','.join(cols)}) FROM STDIN WITH (FORMAT CSV, NULL '')", f, ) cur.execute( "INSERT INTO fec_contributions (" f"{','.join(cols)}) SELECT {','.join(cols)} FROM fec_stage " "ON CONFLICT (sub_id) DO NOTHING" ) conn.commit() tmp.unlink() def _load_committees(engine, cycle: int) -> int: url = f"https://www.fec.gov/files/bulk-downloads/20{cycle:02d}/cm{cycle:02d}.zip" dest = CACHE / f"cm{cycle:02d}.zip" _download(url, dest) committees = [] with zipfile.ZipFile(dest) as z: member = next(n for n in z.namelist() if n.endswith(".txt")) with z.open(member) as fh: reader = csv.reader(io.TextIOWrapper(fh, encoding="latin-1", newline=""), delimiter="|", quoting=csv.QUOTE_NONE) for row in reader: if len(row) < 15: continue r = dict(zip(CM_COLS, row)) committees.append({ "committee_id": r["cmte_id"], "name": r["cmte_nm"] or None, "type": r["cmte_tp"] or None, "party": r["cmte_pty_affiliation"] or None, "candidate_id": r["cand_id"] or None, "cycle": 2000 + cycle, }) with engine.begin() as conn: conn.execute(text(""" CREATE TABLE IF NOT EXISTS fec_committees ( committee_id TEXT, cycle INT, name TEXT, type TEXT, party TEXT, candidate_id TEXT, PRIMARY KEY (committee_id, cycle) ); """)) if committees: cols = list(committees[0].keys()) vals_ph = ",".join( "(" + ",".join(f":{k}_{i}" for k in cols) + ")" for i, _ in enumerate(committees) ) params = {} for i, row in enumerate(committees): for k, v in row.items(): params[f"{k}_{i}"] = v conn.execute(text( f"INSERT INTO fec_committees ({','.join(cols)}) VALUES {vals_ph} " "ON CONFLICT (committee_id, cycle) DO UPDATE SET " "name = EXCLUDED.name, type = EXCLUDED.type, party = EXCLUDED.party" ), params) return len(committees) def main() -> None: engine = create_engine( f"postgresql+psycopg2://gisapp:{_env('GISPASS')}@127.0.0.1:5432/westerville" ) with engine.begin() as conn: conn.execute(text("TRUNCATE fec_contributions")) print("[fec] truncated fec_contributions") grand = 0 for cycle in CYCLES: url = f"https://www.fec.gov/files/bulk-downloads/20{cycle:02d}/indiv{cycle:02d}.zip" dest = CACHE / f"indiv{cycle:02d}.zip" _download(url, dest) rows: list[dict] = [] n_kept = _filter_indiv(dest, cycle, rows) _copy_indiv(engine, rows) grand += n_kept for cycle in CYCLES: n = _load_committees(engine, cycle) print(f"[cm] cycle=20{cycle:02d}: loaded {n} committees") # attach committee names/type/party to contributions with engine.begin() as conn: conn.execute(text(""" UPDATE fec_contributions f SET committee_name = c.name, committee_type = c.type FROM fec_committees c WHERE c.committee_id = f.committee_id AND c.cycle = f.cycle AND f.committee_name IS NULL; """)) (n,) = conn.execute(text("SELECT count(*) FROM fec_contributions")).one() (nn,) = conn.execute(text( "SELECT count(*) FROM fec_contributions WHERE committee_name IS NOT NULL" )).one() print(f"[done] fec_contributions: {n} rows, {nn} enriched with committee names") if __name__ == "__main__": main()