osint-investigation

1#!/usr/bin/env python32"""Fetch federal contracts/awards from USAspending.gov API v2.3 4No auth required. POST to /api/v2/search/spending_by_award/ with filters.5"""6from __future__ import annotations7 8import argparse9import csv10import json11import sys12import time13import urllib.request14from pathlib import Path15 16ENDPOINT = "https://api.usaspending.gov/api/v2/search/spending_by_award/"17COLUMNS = [18    "award_id",19    "recipient_name",20    "recipient_uei",21    "recipient_duns",22    "recipient_parent_name",23    "recipient_state",24    "awarding_agency",25    "awarding_sub_agency",26    "award_type",27    "award_amount",28    "award_date",29    "period_of_performance_start",30    "period_of_performance_end",31    "naics_code",32    "psc_code",33    "competition_extent",34    "description",35]36 37# USAspending result column "code" → human label mapping for output.38_FIELDS = [39    "Award ID",40    "Recipient Name",41    "Recipient UEI",42    "Recipient DUNS Number",43    "Recipient Parent Name",44    "Recipient State Code",45    "Awarding Agency",46    "Awarding Sub Agency",47    "Award Type",48    "Award Amount",49    "Start Date",50    "End Date",51    "NAICS Code",52    "PSC Code",53    "Type of Set Aside",54    "Description",55]56 57 58def _post(body: dict) -> dict:59    req = urllib.request.Request(60        ENDPOINT,61        data=json.dumps(body).encode("utf-8"),62        headers={"Content-Type": "application/json", "User-Agent": "hermes-agent osint-investigation"},63        method="POST",64    )65    with urllib.request.urlopen(req, timeout=60) as resp:66        return json.loads(resp.read().decode("utf-8"))67 68 69def fetch(70    recipient: str | None,71    agency: str | None,72    fy: int,73    sole_source_only: bool,74    out_path: str,75    page_size: int = 100,76    max_pages: int = 20,77) -> int:78    filters: dict = {79        "time_period": [{"start_date": f"{fy - 1}-10-01", "end_date": f"{fy}-09-30"}],80        # Contracts only by default; adjust award_type_codes for grants/loans.81        "award_type_codes": ["A", "B", "C", "D"],82    }83    if recipient:84        filters["recipient_search_text"] = [recipient]85    if agency:86        filters["agencies"] = [{"type": "awarding", "tier": "toptier", "name": agency}]87 88    rows: list[dict[str, str]] = []89    page = 190    while page <= max_pages:91        body = {92            "filters": filters,93            "fields": _FIELDS,94            "page": page,95            "limit": page_size,96            "sort": "Award Amount",97            "order": "desc",98        }99        try:100            payload = _post(body)101        except Exception as e:  # noqa: BLE001102            print(f"USAspending error on page {page}: {e}", file=sys.stderr)103            break104        results = payload.get("results", [])105        if not results:106            break107        for r in results:108            set_aside = r.get("Type of Set Aside", "") or ""109            if sole_source_only and "sole" not in set_aside.lower():110                continue111            rows.append(112                {113                    "award_id": r.get("Award ID", "") or "",114                    "recipient_name": r.get("Recipient Name", "") or "",115                    "recipient_uei": r.get("Recipient UEI", "") or "",116                    "recipient_duns": r.get("Recipient DUNS Number", "") or "",117                    "recipient_parent_name": r.get("Recipient Parent Name", "") or "",118                    "recipient_state": r.get("Recipient State Code", "") or "",119                    "awarding_agency": r.get("Awarding Agency", "") or "",120                    "awarding_sub_agency": r.get("Awarding Sub Agency", "") or "",121                    "award_type": r.get("Award Type", "") or "",122                    "award_amount": str(r.get("Award Amount", "") or ""),123                    "award_date": r.get("Start Date", "") or "",124                    "period_of_performance_start": r.get("Start Date", "") or "",125                    "period_of_performance_end": r.get("End Date", "") or "",126                    "naics_code": str(r.get("NAICS Code", "") or ""),127                    "psc_code": str(r.get("PSC Code", "") or ""),128                    "competition_extent": set_aside,129                    "description": r.get("Description", "") or "",130                }131            )132        meta = payload.get("page_metadata", {})133        if not meta.get("hasNext"):134            break135        page += 1136        time.sleep(0.5)137 138    Path(out_path).parent.mkdir(parents=True, exist_ok=True)139    with open(out_path, "w", newline="", encoding="utf-8") as fh:140        w = csv.DictWriter(fh, fieldnames=COLUMNS)141        w.writeheader()142        w.writerows(rows)143    return len(rows)144 145 146def main() -> int:147    p = argparse.ArgumentParser(description=__doc__)148    p.add_argument("--recipient", help="Recipient name search")149    p.add_argument("--agency", help="Awarding agency (top-tier)")150    p.add_argument("--fy", type=int, default=2024, help="Federal fiscal year")151    p.add_argument("--sole-source-only", action="store_true")152    p.add_argument("--max-pages", type=int, default=20)153    p.add_argument("--out", required=True)154    a = p.parse_args()155    if not (a.recipient or a.agency):156        p.error("must supply at least one of --recipient / --agency")157    n = fetch(158        recipient=a.recipient,159        agency=a.agency,160        fy=a.fy,161        sole_source_only=a.sole_source_only,162        out_path=a.out,163        max_pages=a.max_pages,164    )165    print(f"Wrote {n} USAspending rows to {a.out}")166    return 0167 168 169if __name__ == "__main__":170    raise SystemExit(main())171