Drimble APIdrimble.nl ↗

"""
Drimble.nl Business Listings API Client
Extract business listings and full company details from drimble.nl

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

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


class ParseClient:
    """Client for interacting with the Parse API for Drimble.nl business listings."""

    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the Parse API client.

        Args:
            api_key: API key for authentication. Defaults to PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "9681bf2d-0c8b-4bb8-b56b-ab6902bea290"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError("API key not provided and PARSE_API_KEY env var not set")

    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: Query or body parameters depending on method

        Returns:
            Response JSON as a 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: Optional[str] = None,
        sbi: Optional[str] = None,
        prid: Optional[int] = None,
        gid: Optional[int] = None,
        pid: Optional[int] = None,
        wid: Optional[int] = None,
        bid: Optional[int] = None,
        page: Optional[int] = None
    ) -> Dict[str, Any]:
        """
        Search for companies on Drimble.nl using keywords and filters.

        Args:
            query: Keyword search for company name or related terms
            sbi: SBI code filter (e.g., 'F' for construction)
            prid: Province ID filter
            gid: Gemeente (Municipality) ID filter
            pid: Plaats (City) ID filter
            wid: Wijk (District) ID filter
            bid: Buurt (Neighborhood) ID filter
            page: Page number for pagination (default: 1)

        Returns:
            Dictionary containing companies list and available filters
        """
        params = {}
        if query:
            params["query"] = query
        if sbi:
            params["sbi"] = sbi
        if prid:
            params["prid"] = prid
        if gid:
            params["gid"] = gid
        if pid:
            params["pid"] = pid
        if wid:
            params["wid"] = wid
        if bid:
            params["bid"] = bid
        if page:
            params["page"] = page

        return self._call("search_companies", method="GET", **params)

    def get_company_details(self, url: str) -> Dict[str, Any]:
        """
        Get full details for a specific business listing.

        Args:
            url: The company detail page URL or relative path from search results

        Returns:
            Dictionary containing company details including address, SBI codes, and description
        """
        return self._call("get_company_details", method="GET", url=url)


def main():
    """
    Demonstrate practical workflow: search for companies and retrieve their details.
    """
    # Initialize the client
    client = ParseClient()

    print("=" * 70)
    print("Drimble.nl Business Listings API - Practical Business Intelligence")
    print("=" * 70)

    # Step 1: Search for IT/Software companies (SBI code 'J' for Information and Communication)
    print("\n📊 Step 1: Searching for IT/Software companies...")
    search_params = {
        "query": "software",
        "sbi": "J",
        "page": 1
    }
    print(f"   Search parameters: {search_params}")

    search_results = client.search_companies(**search_params)

    # Display search results summary
    if "data" in search_results:
        data = search_results["data"]
        companies = data.get("companies", [])
        print(f"\n✓ Found {len(companies)} software companies in results")

        # Display available SBI filters to understand the data structure
        if "sbi_filter" in data:
            sbi_options = data["sbi_filter"].get("options", [])
            print(f"\n📋 Available Industry Categories (SBI Codes): {len(sbi_options)} total")
            for sbi_option in sbi_options[:5]:
                print(f"   • {sbi_option.get('code')}: {sbi_option.get('description')}")

        # Display geographic filter options
        if "geolocation_filter" in data:
            geoloc = data["geolocation_filter"]
            print(f"\n🗺️  Geographic Filter Level: {geoloc.get('level')}")
            options = geoloc.get("options", [])
            if options:
                print(f"   Available regions: {len(options)}")
                for option in options[:5]:
                    print(f"   • {option.get('name')} (ID: {option.get('id')})")

        # Step 2: Process each company and get detailed information
        if companies:
            print(f"\n📝 Step 2: Retrieving detailed information for companies...")
            print("-" * 70)

            for idx, company in enumerate(companies[:3], 1):  # Process first 3 companies
                print(f"\nCompany #{idx}: {company.get('company_name', 'Unknown')}")
                print(f"   KVK Number: {company.get('kvk_number')}")
                print(f"   Branch Number: {company.get('branch_number')}")
                print(f"   Registration Date: {company.get('registration_date', 'N/A')}")

                # Fetch detailed information for this company
                company_url = company.get("url")
                if company_url:
                    try:
                        print(f"   → Fetching complete profile...")
                        details = client.get_company_details(url=company_url)

                        if "data" in details:
                            detail_data = details["data"]

                            # Display address information
                            if "address" in detail_data:
                                addr = detail_data["address"]
                                print(f"\n   📍 Address Information:")
                                print(f"      Street: {addr.get('street')} {addr.get('house_number')}")
                                if addr.get('house_number_extension'):
                                    print(f"      Extension: {addr.get('house_number_extension')}")
                                print(f"      Postcode: {addr.get('postcode')}")
                                print(f"      City: {addr.get('plaats')}")
                                print(f"      Municipality: {addr.get('municipality')}")
                                print(f"      Province: {addr.get('province')}")

                            # Display business details
                            print(f"\n   💼 Business Details:")
                            print(f"      Legal Form: {detail_data.get('legal_form', 'N/A')}")
                            print(f"      Employees: {detail_data.get('employees', 'N/A')}")

                            # Display SBI codes
                            if "sbi_codes" in detail_data:
                                sbi_codes = detail_data["sbi_codes"]
                                if sbi_codes:
                                    print(f"\n   🏭 Industry Classifications:")
                                    for sbi in sbi_codes:
                                        print(f"      • {sbi.get('code')}: {sbi.get('description', 'N/A')}")
                                else:
                                    print(f"\n   🏭 Industry Classifications: Not specified")

                            # Display business description
                            if "description" in detail_data:
                                description = detail_data["description"]
                                print(f"\n   📄 Business Description:")
                                # Truncate long descriptions
                                if len(description) > 150:
                                    print(f"      {description[:150]}...")
                                else:
                                    print(f"      {description}")

                    except requests.exceptions.RequestException as e:
                        print(f"      ✗ Error fetching details: {e}")
                    except Exception as e:
                        print(f"      ✗ Unexpected error: {e}")

        print("\n" + "-" * 70)
        print("✓ Analysis Complete!")
        print("\nKey Insights:")
        print(f"   • Total companies found: {len(companies)}")
        print(f"   • Companies processed: {min(3, len(companies))}")
        print(f"   • Available industry categories: {len(search_results['data'].get('sbi_filter', {}).get('options', []))}")
        if "geolocation_filter" in search_results["data"]:
            print(f"   • Geographic regions available: {len(search_results['data']['geolocation_filter'].get('options', []))}")

    print("\n" + "=" * 70)
    print("Workflow completed successfully!")
    print("=" * 70)


if __name__ == "__main__":
    main()