ConnectLI APIconnectli.com ↗

"""
ConnectLI Business Directory API Client

Access contact information for businesses listed on ConnectLI.com,
a Long Island business directory covering Nassau & Suffolk County.

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 the ConnectLI Business Directory API."""

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

        Args:
            api_key: API key for authentication. If not provided,
                    reads from PARSE_API_KEY environment variable.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "d9b117f0-b797-47a5-be3b-7bd2f0fe8dd4"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError(
                "API key not provided. Set PARSE_API_KEY environment variable "
                "or pass api_key parameter."
            )

    def _call(
        self, endpoint: str, method: str = "POST", **params
    ) -> Dict[str, Any]:
        """
        Make a request to the Parse API.

        Args:
            endpoint: API endpoint name
            method: HTTP method (GET or POST)
            **params: Query/body parameters

        Returns:
            Parsed JSON response

        Raises:
            requests.exceptions.RequestException: If the request fails
        """
        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 list_businesses(
        self,
        category: str,
        page: int = 1,
        per_page: int = 50,
    ) -> Dict[str, Any]:
        """
        List all businesses in a specified category with contact information.

        Args:
            category: Business category. Accepted values: architects, insurance,
                     law_firms, it_companies
            page: Page number for pagination (default: 1)
            per_page: Number of results per page (default: 50)

        Returns:
            Dictionary containing category, total_found, page, per_page,
            max_pages, and listings array with business objects
        """
        return self._call(
            "list_businesses",
            method="POST",
            category=category,
            page=page,
            per_page=per_page,
        )


def main():
    """
    Practical workflow: Search for businesses in multiple categories
    and display contact information for easy outreach.
    """
    # Initialize client
    client = ParseClient()

    # Categories we want to search
    categories = ["architects", "it_companies"]

    all_businesses = []

    print("=" * 70)
    print("ConnectLI Business Directory Search")
    print("=" * 70)

    # Iterate through categories
    for category in categories:
        print(f"\n📂 Searching category: {category.replace('_', ' ').title()}")
        print("-" * 70)

        try:
            # Get first page of results for this category
            response = client.list_businesses(
                category=category,
                page=1,
                per_page=50,
            )

            total_found = response.get("total_found", 0)
            max_pages = response.get("max_pages", 1)
            listings = response.get("listings", [])

            print(f"Found {total_found} businesses across {max_pages} page(s)")

            # Process listings from this page
            for idx, business in enumerate(listings, 1):
                business_info = {
                    "category": category,
                    "id": business.get("id"),
                    "name": business.get("name"),
                    "phone": business.get("phone"),
                    "address": business.get("address"),
                    "website": business.get("website"),
                    "description": business.get("description", "N/A")[:80] + "...",
                }
                all_businesses.append(business_info)

                # Display first 3 businesses per category as preview
                if idx <= 3:
                    print(f"\n  Business #{idx}: {business.get('name')}")
                    print(f"    📞 Phone: {business.get('phone')}")
                    print(f"    📍 Address: {business.get('address')}")
                    print(f"    🌐 Website: {business.get('website', 'N/A')}")

            if len(listings) > 3:
                print(f"\n  ... and {len(listings) - 3} more businesses")

            # If there are multiple pages, fetch additional pages
            if max_pages > 1:
                print(f"\n  📄 Pagination available: {max_pages} total pages")

        except requests.exceptions.RequestException as e:
            print(f"  ❌ Error fetching {category}: {e}")

    # Summary statistics
    print("\n" + "=" * 70)
    print("SUMMARY")
    print("=" * 70)
    print(f"Total businesses found: {len(all_businesses)}")

    if all_businesses:
        print("\n📋 Contact Summary:")
        print("-" * 70)

        # Group by category
        by_category = {}
        for biz in all_businesses:
            cat = biz["category"]
            if cat not in by_category:
                by_category[cat] = []
            by_category[cat].append(biz)

        for category, businesses in sorted(by_category.items()):
            print(f"\n{category.replace('_', ' ').title()} ({len(businesses)})")
            for biz in businesses[:2]:  # Show first 2 per category
                print(f"  • {biz['name']:40} | {biz['phone']}")


if __name__ == "__main__":
    main()