OpenCorporates APIopencorporates.com ↗

"""
OpenCorporates API & Scraper Client

Get your API key from: https://parse.bot/settings
"""

import os
import requests
from typing import Optional, Dict, Any, List


class ParseClient:
    """Client for interacting with OpenCorporates data through Parse API."""

    def __init__(self, api_key: Optional[str] = None):
        """Initialize the Parse client.
        
        Args:
            api_key: OpenCorporates API token. If not provided, will use PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "e95ef62d-606c-418a-ae4e-3133289dc899"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError("API key must be provided or set in PARSE_API_KEY environment variable")

    def _call(self, endpoint: str, method: str = "POST", **params) -> Dict[str, Any]:
        """Make a request to the Parse API.
        
        Args:
            endpoint: The endpoint name (e.g., 'search_companies')
            method: HTTP method ('GET' or 'POST')
            **params: Parameters to pass to the endpoint
            
        Returns:
            Response JSON as dictionary
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        if method.upper() == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method.upper() == "POST":
            response = requests.post(url, headers=headers, json=params)
        else:
            raise ValueError(f"Unsupported HTTP method: {method}")
        
        response.raise_for_status()
        return response.json()

    def search_companies(
        self,
        query: str,
        api_token: Optional[str] = None,
        country_code: Optional[str] = None,
        jurisdiction_code: Optional[str] = None,
        company_type: Optional[str] = None,
        current_status: Optional[str] = None,
        page: int = 1
    ) -> Dict[str, Any]:
        """Search for companies by name across all or a specific jurisdiction.
        
        Args:
            query: Search keyword (company name)
            api_token: OpenCorporates API token
            country_code: Filter by country code
            jurisdiction_code: Filter by jurisdiction code (e.g., us_de)
            company_type: Filter by company type
            current_status: Filter by current status (e.g., Active)
            page: Page number for pagination
            
        Returns:
            Dictionary with 'companies' array and 'total_count'
        """
        params = {"query": query, "page": page}
        if api_token:
            params["api_token"] = api_token
        if country_code:
            params["country_code"] = country_code
        if jurisdiction_code:
            params["jurisdiction_code"] = jurisdiction_code
        if company_type:
            params["company_type"] = company_type
        if current_status:
            params["current_status"] = current_status
        
        return self._call("search_companies", method="GET", **params)

    def get_company(
        self,
        jurisdiction_code: str,
        company_number: str,
        api_token: Optional[str] = None,
        sparse: bool = False
    ) -> Dict[str, Any]:
        """Retrieve full details for a specific company.
        
        Args:
            jurisdiction_code: Jurisdiction code (e.g., us_de)
            company_number: Company registration number
            api_token: OpenCorporates API token
            sparse: Return lightweight response if True
            
        Returns:
            Dictionary with company details
        """
        params = {
            "jurisdiction_code": jurisdiction_code,
            "company_number": company_number,
            "sparse": str(sparse).lower()
        }
        if api_token:
            params["api_token"] = api_token
        
        return self._call("get_company", method="GET", **params)

    def scrape_company_page(
        self,
        jurisdiction_code: str,
        company_number: str
    ) -> Dict[str, Any]:
        """Scrape OpenCorporates HTML company page for registration details, officers, and filings.
        
        Args:
            jurisdiction_code: Jurisdiction code (e.g., us_de)
            company_number: Company registration number
            
        Returns:
            Dictionary with company_name, officers, filings, attributes, and url
        """
        params = {
            "jurisdiction_code": jurisdiction_code,
            "company_number": company_number
        }
        
        return self._call("scrape_company_page", method="GET", **params)

    def scrape_officer_page(self, officer_id: str) -> Dict[str, Any]:
        """Scrape OpenCorporates HTML officer page for details and associated companies.
        
        Args:
            officer_id: OpenCorporates officer ID
            
        Returns:
            Dictionary with officer_name, attributes, companies, and url
        """
        params = {"id": officer_id}
        
        return self._call("scrape_officer_page", method="GET", **params)

    def search_officers(
        self,
        query: str,
        api_token: Optional[str] = None,
        page: int = 1
    ) -> Dict[str, Any]:
        """Search for company officers by name.
        
        Args:
            query: Officer name search keyword
            api_token: OpenCorporates API token
            page: Page number for pagination
            
        Returns:
            Dictionary with 'officers' array and 'total_count'
        """
        params = {"query": query, "page": page}
        if api_token:
            params["api_token"] = api_token
        
        return self._call("search_officers", method="GET", **params)

    def list_jurisdictions(self, api_token: Optional[str] = None) -> Dict[str, Any]:
        """List all jurisdictions supported by OpenCorporates.
        
        Args:
            api_token: OpenCorporates API token
            
        Returns:
            Dictionary with 'jurisdictions' array
        """
        params = {}
        if api_token:
            params["api_token"] = api_token
        
        return self._call("list_jurisdictions", method="GET", **params)


def main():
    """Practical workflow example: Research companies and their officers."""
    
    # Initialize the client
    client = ParseClient()
    
    # Step 1: Search for companies by name
    print("=" * 60)
    print("STEP 1: Searching for companies named 'Cribl'...")
    print("=" * 60)
    
    search_results = client.search_companies(
        query="Cribl",
        jurisdiction_code="us_de",
        current_status="Active"
    )
    
    print(f"Found {search_results['total_count']} company(ies)")
    
    if search_results['companies']:
        # Step 2: Get details for the first company found
        first_company = search_results['companies'][0]['company']
        jurisdiction = first_company['jurisdiction_code']
        company_number = first_company['company_number']
        company_name = first_company['name']
        
        print(f"\nSelected company: {company_name} ({jurisdiction}/{company_number})")
        
        # Step 3: Scrape the company page for detailed information
        print("\n" + "=" * 60)
        print("STEP 2: Scraping company page for officers and filings...")
        print("=" * 60)
        
        company_page = client.scrape_company_page(
            jurisdiction_code=jurisdiction,
            company_number=company_number
        )
        
        print(f"Company: {company_page.get('company_name')}")
        print(f"URL: {company_page.get('url')}")
        
        # Display company attributes
        if company_page.get('attributes'):
            print("\nCompany Attributes:")
            for key, value in company_page['attributes'].items():
                print(f"  {key}: {value}")
        
        # Step 4: Process officers
        if company_page.get('officers'):
            print(f"\nOfficers ({len(company_page['officers'])}):")
            for officer in company_page['officers']:
                print(f"  - {officer.get('name', 'Unknown')} ({officer.get('role', 'Unknown role')})")
        
        # Step 5: Display filings
        if company_page.get('filings'):
            print(f"\nRecent Filings ({len(company_page['filings'])}):")
            for filing in company_page['filings'][:5]:  # Show first 5
                print(f"  - {filing.get('title', 'Unknown')} ({filing.get('date', 'Unknown date')})")
        
        # Step 6: Search for officers with the same name
        print("\n" + "=" * 60)
        print("STEP 3: Searching for officers with similar names...")
        print("=" * 60)
        
        if company_page.get('officers'):
            first_officer_name = company_page['officers'][0].get('name')
            if first_officer_name:
                officer_search = client.search_officers(query=first_officer_name)
                print(f"Found {officer_search.get('total_count', 0)} officer(s) named '{first_officer_name}'")
                
                if officer_search.get('officers'):
                    print(f"\nFirst officer result: {officer_search['officers'][0]['officer']['name']}")
    
    print("\n" + "=" * 60)
    print("Workflow completed successfully!")
    print("=" * 60)


if __name__ == "__main__":
    main()