osint-investigation

1#!/usr/bin/env python32"""Fetch Senate Lobbying Disclosure (LD-1 / LD-2) filings.3 4Anonymous: 120 req/hour. Token (SENATE_LDA_TOKEN): 1200 req/hour.5"""6from __future__ import annotations7 8import argparse9import csv10import os11import sys12import time13from pathlib import Path14 15sys.path.insert(0, str(Path(__file__).parent))16from _http import get_json  # noqa: E40217 18ENDPOINT = "https://lda.senate.gov/api/v1/filings/"19COLUMNS = [20    "filing_uuid",21    "filing_type",22    "filing_year",23    "filing_period",24    "registrant_name",25    "registrant_id",26    "client_name",27    "client_id",28    "client_general_description",29    "income",30    "expenses",31    "lobbyists",32    "issues",33    "government_entities",34    "filing_date",35]36 37 38def fetch(39    client: str | None,40    registrant: str | None,41    year: int,42    token: str | None,43    out_path: str,44    page_size: int = 100,45    max_pages: int = 25,46) -> int:47    params: dict = {"filing_year": year, "page_size": page_size}48    if client:49        params["client_name"] = client50    if registrant:51        params["registrant_name"] = registrant52 53    headers = {"Authorization": f"Token {token}"} if token else None54    rows: list[dict[str, str]] = []55    url = ENDPOINT56    page = 057    while page < max_pages:58        try:59            payload = get_json(url, params=params if page == 0 else None, headers=headers)60        except Exception as e:  # noqa: BLE00161            print(f"Senate LDA error on page {page + 1}: {e}", file=sys.stderr)62            break63        if not isinstance(payload, dict):64            break65        results = payload.get("results", [])66        for r in results:67            client_obj = r.get("client") or {}68            registrant_obj = r.get("registrant") or {}69            lobbying_activities = r.get("lobbying_activities") or []70            lobbyists = []71            issues = []72            entities = []73            for la in lobbying_activities:74                for lob in la.get("lobbyists") or []:75                    lob_obj = lob.get("lobbyist") or {}76                    name = " ".join(77                        x for x in (lob_obj.get("first_name", ""), lob_obj.get("last_name", "")) if x78                    )79                    if name:80                        lobbyists.append(name)81                desc = la.get("description") or ""82                if desc:83                    issues.append(desc)84                for ge in la.get("government_entities") or []:85                    nm = ge.get("name") or ""86                    if nm:87                        entities.append(nm)88            rows.append(89                {90                    "filing_uuid": r.get("filing_uuid", "") or "",91                    "filing_type": r.get("filing_type", "") or "",92                    "filing_year": str(r.get("filing_year", "") or year),93                    "filing_period": r.get("filing_period", "") or "",94                    "registrant_name": registrant_obj.get("name", "") or "",95                    "registrant_id": str(registrant_obj.get("id", "") or ""),96                    "client_name": client_obj.get("name", "") or "",97                    "client_id": str(client_obj.get("id", "") or ""),98                    "client_general_description": client_obj.get("general_description", "") or "",99                    "income": str(r.get("income", "") or ""),100                    "expenses": str(r.get("expenses", "") or ""),101                    "lobbyists": "; ".join(sorted(set(lobbyists))),102                    "issues": "; ".join(issues),103                    "government_entities": "; ".join(sorted(set(entities))),104                    "filing_date": (r.get("dt_posted") or "")[:10],105                }106            )107        next_url = payload.get("next")108        if not next_url:109            break110        url = next_url111        page += 1112        time.sleep(1.0 if not token else 0.3)113 114    Path(out_path).parent.mkdir(parents=True, exist_ok=True)115    with open(out_path, "w", newline="", encoding="utf-8") as fh:116        w = csv.DictWriter(fh, fieldnames=COLUMNS)117        w.writeheader()118        w.writerows(rows)119    return len(rows)120 121 122def main() -> int:123    p = argparse.ArgumentParser(description=__doc__)124    p.add_argument("--client", help="Client name filter")125    p.add_argument("--registrant", help="Registrant (lobbying firm) name filter")126    p.add_argument("--year", type=int, default=2024)127    p.add_argument("--token", default=os.environ.get("SENATE_LDA_TOKEN"))128    p.add_argument("--max-pages", type=int, default=25)129    p.add_argument("--out", required=True)130    a = p.parse_args()131    if not (a.client or a.registrant):132        p.error("must supply at least one of --client / --registrant")133    n = fetch(134        client=a.client,135        registrant=a.registrant,136        year=a.year,137        token=a.token,138        out_path=a.out,139        max_pages=a.max_pages,140    )141    print(f"Wrote {n} Senate LDA rows to {a.out}")142    return 0143 144 145if __name__ == "__main__":146    raise SystemExit(main())147