""" USASpending.gov Contract Search Search federal contract awards to understand company capabilities, contract history, and funding relationships. API Documentation: https://api.usaspending.gov/ Free, no API key required. """ import logging import requests from typing import List, Optional, Dict, Any from datetime import datetime, timedelta from ..search.base import BaseSearcher, SearchResult logger = logging.getLogger(__name__) class ContractSearcher(BaseSearcher): """ Search USASpending.gov for federal contracts. Useful for: - Finding companies with relevant contract experience - Understanding funding levels for technologies - Identifying prime contractors and their focus areas """ BASE_URL = "https://api.usaspending.gov/api/v2" def __init__(self, timeout: int = 30): self.timeout = timeout @property def name(self) -> str: return "USASpending" @property def source_type(self) -> str: return "contract" def search( self, query: str, max_results: int = 20, agency: Optional[str] = None, min_amount: Optional[float] = None, years_back: int = 5 ) -> List[SearchResult]: """ Search federal contracts. Args: query: Keyword search max_results: Maximum results agency: Filter by agency (e.g., "Department of Defense") min_amount: Minimum contract value years_back: How many years back to search Returns: List of SearchResult objects """ # Calculate date range end_date = datetime.now() start_date = end_date - timedelta(days=years_back * 365) payload = { "filters": { "keywords": [query], "time_period": [{ "start_date": start_date.strftime("%Y-%m-%d"), "end_date": end_date.strftime("%Y-%m-%d") }] }, "fields": [ "Award ID", "Recipient Name", "Description", "Award Amount", "Start Date", "Awarding Agency", "Awarding Sub Agency", "Contract Award Type" ], "page": 1, "limit": min(max_results, 100), "sort": "Award Amount", "order": "desc" } if agency: payload["filters"]["agencies"] = [{ "type": "awarding", "tier": "toptier", "name": agency }] if min_amount: payload["filters"]["award_amounts"] = [{ "lower_bound": min_amount }] results = [] try: response = requests.post( f"{self.BASE_URL}/search/spending_by_award/", json=payload, timeout=self.timeout, headers={"Content-Type": "application/json"} ) response.raise_for_status() data = response.json() for rank, award in enumerate(data.get("results", []), 1): if rank > max_results: break award_id = award.get("Award ID", "") # USASpending award detail URL url = f"https://www.usaspending.gov/award/{award_id}" if award_id else "" results.append(SearchResult( title=award.get("Description", "")[:200] if award.get("Description") else "Contract Award", url=url, snippet=award.get("Description", "")[:500] if award.get("Description") else "", source=self.name, source_type=self.source_type, rank=rank, published_date=award.get("Start Date"), organization=award.get("Recipient Name", ""), award_amount=self._parse_amount(award.get("Award Amount")), award_id=award_id, trl_estimate=6, # Contracts typically for more mature tech raw_data=award )) logger.info(f"Contract search for '{query}' returned {len(results)} results") except requests.exceptions.RequestException as e: logger.error(f"USASpending API request failed: {e}") except Exception as e: logger.error(f"Contract search error: {e}") return results def _parse_amount(self, amount) -> Optional[float]: """Parse award amount to float.""" if amount is None: return None try: return float(amount) except (ValueError, TypeError): return None def search_dod(self, query: str, max_results: int = 20) -> List[SearchResult]: """Search specifically DOD contracts.""" return self.search(query, max_results=max_results, agency="Department of Defense") def search_space_force(self, query: str, max_results: int = 20) -> List[SearchResult]: """Search Space Force / space-related DOD contracts.""" space_query = f"{query} (space OR satellite OR orbital OR launch)" return self.search(space_query, max_results=max_results, agency="Department of Defense") def get_recipient_profile(self, recipient_name: str) -> Dict[str, Any]: """ Get profile information about a contract recipient. Returns summary of their contract history. """ payload = { "filters": { "recipient_search_text": [recipient_name], "time_period": [{ "start_date": "2019-01-01", "end_date": datetime.now().strftime("%Y-%m-%d") }] }, "category": "awarding_agency", "limit": 10 } try: response = requests.post( f"{self.BASE_URL}/search/spending_by_category/", json=payload, timeout=self.timeout, headers={"Content-Type": "application/json"} ) response.raise_for_status() data = response.json() return { "recipient_name": recipient_name, "total_contracts": data.get("total", 0), "agencies": data.get("results", []) } except Exception as e: logger.error(f"Failed to get recipient profile: {e}") return {"recipient_name": recipient_name, "error": str(e)} def get_company_contracts( self, company_name: str, max_results: int = 50 ) -> List[SearchResult]: """Get all recent contracts for a specific company.""" payload = { "filters": { "recipient_search_text": [company_name], "time_period": [{ "start_date": "2020-01-01", "end_date": datetime.now().strftime("%Y-%m-%d") }] }, "fields": [ "Award ID", "Recipient Name", "Description", "Award Amount", "Start Date", "Awarding Agency", "Awarding Sub Agency" ], "page": 1, "limit": min(max_results, 100), "sort": "Start Date", "order": "desc" } results = [] try: response = requests.post( f"{self.BASE_URL}/search/spending_by_award/", json=payload, timeout=self.timeout, headers={"Content-Type": "application/json"} ) response.raise_for_status() data = response.json() for rank, award in enumerate(data.get("results", []), 1): award_id = award.get("Award ID", "") url = f"https://www.usaspending.gov/award/{award_id}" if award_id else "" results.append(SearchResult( title=award.get("Description", "")[:200] if award.get("Description") else "Contract", url=url, snippet=f"Agency: {award.get('Awarding Agency', 'Unknown')}", source=self.name, source_type=self.source_type, rank=rank, published_date=award.get("Start Date"), organization=award.get("Recipient Name", ""), award_amount=self._parse_amount(award.get("Award Amount")), award_id=award_id, raw_data=award )) logger.info(f"Found {len(results)} contracts for {company_name}") except Exception as e: logger.error(f"Company contract search failed: {e}") return results