Scrape Government Contract Data from SAM.gov with Python (Green List #4)

python -m venv .venv
source .venv/bin/activate
pip install requests tenacity pandas beautifulsoup4 lxml

from __future__ import annotations

import random
import time
from urllib.parse import quote

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

TIMEOUT = (10, 60)
UA = "Mozilla/5.0 (compatible; ProxiesAPI-Guides/1.0)"

session = requests.Session()
session.headers.update({
    "User-Agent": UA,
    "Accept": "text/html,application/json;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
})


class FetchError(RuntimeError):
    pass


@retry(
    reraise=True,
    stop=stop_after_attempt(6),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=retry_if_exception_type((requests.RequestException, FetchError)),
)
def fetch(url: str) -> requests.Response:
    r = session.get(url, timeout=TIMEOUT)
    if r.status_code in (403, 429):
        raise FetchError(f"blocked status={r.status_code}")
    r.raise_for_status()

    # small jitter helps reduce bursts
    time.sleep(0.3 + random.random() * 0.5)
    return r


def proxiesapi_url(target_url: str, api_key: str) -> str:
    return "http://api.proxiesapi.com/?key=" + quote(api_key) + "&url=" + quote(target_url, safe="")

import json


def fetch_json(url: str) -> dict:
    r = fetch(url)
    ct = r.headers.get("content-type", "")
    if "json" not in ct.lower():
        raise FetchError(f"expected json, got content-type={ct}")
    return r.json()


def parse_opportunities_from_json(payload: dict) -> list[dict]:
    """Normalize a list of opportunities from a JSON payload.

    Update field paths based on the payload you observe.
    """
    items = payload.get("items") or payload.get("results") or []

    out = []
    for it in items:
        out.append({
            "notice_id": it.get("noticeId") or it.get("id"),
            "title": it.get("title"),
            "agency": it.get("agency") or it.get("department"),
            "posted_date": it.get("postedDate") or it.get("posted"),
            "response_deadline": it.get("responseDeadline") or it.get("dueDate"),
            "url": it.get("url"),
        })

    return out

from bs4 import BeautifulSoup
from urllib.parse import urljoin

BASE = "https://sam.gov"


def parse_opportunity_cards(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "lxml")

    out: list[dict] = []

    # Many sites use headings/anchors for card titles.
    # We collect plausible listing anchors and then enrich by nearby text.
    for a in soup.select("a[href]"):
        href = a.get("href") or ""
        text = a.get_text(" ", strip=True)
        if not text:
            continue

        # Heuristic: opportunity detail URLs often contain "opportunity" or "notice".
        if "opportunity" in href or "notice" in href:
            url = urljoin(BASE, href)

            # attempt to find the parent card container and pull nearby metadata
            card = a.find_parent(["div", "article", "li"]) or a
            card_text = card.get_text("\n", strip=True)

            out.append({
                "title": text,
                "url": url,
                "raw": card_text[:800],
            })

    # Deduplicate by URL
    seen = set()
    uniq = []
    for row in out:
        if row["url"] in seen:
            continue
        seen.add(row["url"])
        uniq.append(row)

    return uniq

import re
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse


def bump_page(url: str, page: int) -> str:
    """Generic helper for URLs with a page query parameter."""
    parts = urlparse(url)
    q = parse_qs(parts.query)
    q["page"] = [str(page)]
    new_query = urlencode(q, doseq=True)
    return urlunparse(parts._replace(query=new_query))


def parse_next_link(html: str, current_url: str) -> str | None:
    soup = BeautifulSoup(html, "lxml")
    for a in soup.select("a[href]"):
        txt = a.get_text(" ", strip=True).lower()
        if txt in {"next", "next page", "›"} or re.fullmatch(r"next\s*›?", txt):
            return urljoin(current_url, a.get("href"))

    rel = soup.select_one("a[rel='next']")
    if rel and rel.get("href"):
        return urljoin(current_url, rel.get("href"))

    return None

import json


def crawl_html(start_url: str, max_pages: int = 5, api_key: str | None = None) -> list[dict]:
    rows: list[dict] = []
    seen = set()

    current = start_url
    page = 0

    while current and page < max_pages:
        page += 1

        url = proxiesapi_url(current, api_key) if api_key else current
        r = fetch(url)
        html = r.text

        batch = parse_opportunity_cards(html)
        for row in batch:
            if row["url"] in seen:
                continue
            seen.add(row["url"])
            rows.append(row)

        next_url = parse_next_link(html, current)
        print(f"page={page} found={len(batch)} total={len(rows)} next={bool(next_url)}")
        current = next_url

    return rows


if __name__ == "__main__":
    START_URL = "PASTE_A_SAM_GOV_SEARCH_RESULTS_URL_HERE"

    data = crawl_html(START_URL, max_pages=3, api_key=None)  # set api_key to use ProxiesAPI

    with open("sam_opportunities.json", "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

    try:
        import pandas as pd
        pd.DataFrame(data).to_csv("sam_opportunities.csv", index=False)
        print("wrote sam_opportunities.csv")
    except Exception as e:
        print("CSV export skipped:", e)

    print("wrote sam_opportunities.json", len(data))