"""Ingest FEC individual contributions for a set of ZIPs into PostGIS. Uses openFEC schedule_a endpoint. Paginates via last_indexes cursor. API key = api.data.gov key stored in .env.gis (APIDATAGOV_KEY). """ from __future__ import annotations import json import sys import time from pathlib import Path import pandas as pd import requests from sqlalchemy import create_engine, text DATA_DIR = Path("/home/m3ac/genoa-entwuerfe.com") BASE = "https://api.open.fec.gov/v1/schedules/schedule_a/" ZIPS = ["43082"] CYCLES = [2026, 2024] PER_PAGE = 100 MAX_PAGES_PER_JOB = 5000 # safety 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 _ensure_table(engine) -> None: with engine.begin() as conn: conn.execute(text(""" CREATE TABLE IF NOT EXISTS fec_contributions ( sub_id BIGINT PRIMARY KEY, cycle INT, contribution_date DATE, amount NUMERIC(14,2), contributor_name TEXT, contributor_first_name TEXT, contributor_last_name TEXT, contributor_middle_name TEXT, contributor_street_1 TEXT, contributor_city TEXT, contributor_state TEXT, contributor_zip TEXT, contributor_employer TEXT, contributor_occupation TEXT, committee_id TEXT, committee_name TEXT, committee_type TEXT, candidate_id TEXT, candidate_name TEXT, candidate_office TEXT, raw JSONB, loaded_at TIMESTAMPTZ DEFAULT now() ); """)) # (safe re-run) add indexes if missing conn.execute(text("CREATE INDEX IF NOT EXISTS fec_contrib_zip_idx ON fec_contributions (contributor_zip);")) conn.execute(text("CREATE INDEX IF NOT EXISTS fec_contrib_name_idx ON fec_contributions (upper(contributor_last_name), upper(contributor_first_name));")) conn.execute(text("CREATE INDEX IF NOT EXISTS fec_contrib_committee_idx ON fec_contributions (committee_id);")) conn.execute(text("CREATE INDEX IF NOT EXISTS fec_contrib_date_idx ON fec_contributions (contribution_date);")) conn.execute(text("CREATE INDEX IF NOT EXISTS fec_contrib_cycle_idx ON fec_contributions (cycle);")) def _flatten(rec: dict, cycle: int) -> dict: cm = rec.get("committee") or {} return { "sub_id": int(rec["sub_id"]) if rec.get("sub_id") else None, "cycle": cycle, "contribution_date": rec.get("contribution_receipt_date"), "amount": rec.get("contribution_receipt_amount"), "contributor_name": (rec.get("contributor_name") or "").strip() or None, "contributor_first_name": (rec.get("contributor_first_name") or "").strip() or None, "contributor_last_name": (rec.get("contributor_last_name") or "").strip() or None, "contributor_middle_name":(rec.get("contributor_middle_name") or "").strip() or None, "contributor_street_1": (rec.get("contributor_street_1") or "").strip() or None, "contributor_city": (rec.get("contributor_city") or "").strip() or None, "contributor_state": (rec.get("contributor_state") or "").strip() or None, "contributor_zip": (rec.get("contributor_zip") or "").strip()[:5] or None, "contributor_employer": (rec.get("contributor_employer") or "").strip() or None, "contributor_occupation": (rec.get("contributor_occupation") or "").strip() or None, "committee_id": cm.get("committee_id"), "committee_name": cm.get("name"), "committee_type": cm.get("committee_type_full"), "candidate_id": rec.get("candidate_id"), "candidate_name": (rec.get("candidate_name") or "").strip() or None, "candidate_office": rec.get("candidate_office_full"), "raw": json.dumps(rec, default=str), } def _fetch_zip_cycle(api_key: str, engine, zip5: str, cycle: int) -> int: print(f"[fec] fetching zip={zip5} cycle={cycle}") params = { "api_key": api_key, "contributor_zip": zip5, "two_year_transaction_period": cycle, "per_page": PER_PAGE, "sort": "contribution_receipt_date", "sort_hide_null": "false", } total = 0 page = 0 last_index = None last_sort = None while page < MAX_PAGES_PER_JOB: p = dict(params) if last_index is not None: p["last_index"] = last_index if last_sort is not None: p["last_contribution_receipt_date"] = last_sort r = requests.get(BASE, params=p, timeout=45) if r.status_code == 429: wait = int(r.headers.get("Retry-After") or 60) print(f" [429] rate-limited, sleeping {wait}s") time.sleep(wait) continue r.raise_for_status() body = r.json() results = body.get("results", []) if not results: break rows = [_flatten(rec, cycle) for rec in results if rec.get("sub_id")] if rows: with engine.begin() as conn: cols = ",".join(rows[0].keys()) # build VALUES placeholders per row vals_ph = ",".join( "(" + ",".join(f":{k}_{i}" for k in row.keys()) + ")" for i, row in enumerate(rows) ) params_dict = {} for i, row in enumerate(rows): for k, v in row.items(): params_dict[f"{k}_{i}"] = v sql = ( f"INSERT INTO fec_contributions ({cols}) VALUES {vals_ph} " "ON CONFLICT (sub_id) DO NOTHING" ) conn.execute(text(sql), params_dict) total += len(rows) page += 1 page_info = body.get("pagination", {}) li = page_info.get("last_indexes") or {} last_index = li.get("last_index") last_sort = li.get("last_contribution_receipt_date") if not last_index: break remaining_hdr = r.headers.get("X-RateLimit-Remaining") if page % 25 == 0: print(f" page {page}: {total} rows inserted so far (remaining/hour: {remaining_hdr})") # pace: 4/sec is fine well under 1000/hr in bursts, but keep it easy time.sleep(0.15) print(f"[fec] zip={zip5} cycle={cycle} → {total} rows, {page} pages") return total def main() -> None: api_key = _env("APIDATAGOV_KEY") engine = create_engine( f"postgresql+psycopg2://gisapp:{_env('GISPASS')}@127.0.0.1:5432/westerville" ) _ensure_table(engine) grand = 0 for cycle in CYCLES: for z in ZIPS: grand += _fetch_zip_cycle(api_key, engine, z, cycle) with engine.begin() as conn: (n,) = conn.execute(text("SELECT count(*) FROM fec_contributions")).one() print(f"[done] total in table: {n} (added ~{grand} this run)") if __name__ == "__main__": main()