Registro Imprese APIregistroimprese.it ↗

"""
Italian Business Registry API (Registro Imprese) - Parse Client
Search and extract official company data from the Italian Business Registry.
Get your API key from: https://parse.bot/settings
"""

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


class ParseClient:
    """Client for the Italian Business Registry API via Parse."""

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

        Args:
            api_key: API key for Parse. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "718193c2-9b35-4cca-a2a7-60c1e7c12f98"
        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: The API endpoint name
            method: HTTP method (GET or POST)
            **params: Parameters to send with the request

        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 == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method == "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, province: Optional[str] = None) -> Dict[str, Any]:
        """
        Search for Italian companies by name and province.

        Args:
            query: Company name or keyword to search for
            province: Two-letter province code (e.g., 'VI' for Vicenza), optional

        Returns:
            Dictionary with total count and items list containing company info
        """
        params = {"query": query}
        if province:
            params["province"] = province

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

    def get_company_details(self, p_auth: str, instance_id: str, company_id: str) -> Dict[str, Any]:
        """
        Get full details for a specific company.

        Args:
            p_auth: Session auth token from search results
            instance_id: Portlet instance ID from search results
            company_id: Encrypted company ID from search results

        Returns:
            Dictionary with complete company details
        """
        params = {
            "p_auth": p_auth,
            "instance_id": instance_id,
            "company_id": company_id
        }
        return self._call("get_company_details", method="GET", **params)


def main():
    """Practical workflow example: search companies and get their details."""

    # Initialize the client
    client = ParseClient()

    print("=" * 60)
    print("Italian Business Registry API - Company Search")
    print("=" * 60)

    # Search for companies
    search_query = "ARROWELD"
    province_code = "VI"  # Vicenza

    print(f"\nSearching for companies: '{search_query}' in province {province_code}...")
    search_results = client.search_companies(query=search_query, province=province_code)

    print(f"\nFound {search_results['total']} company/companies:")
    print("-" * 60)

    # Get p_auth token for use in subsequent calls
    p_auth = search_results.get('p_auth')

    # Process each company found
    for idx, company in enumerate(search_results.get('items', []), 1):
        print(f"\nResult {idx}: {company['name']}")
        print(f"  Location: {company['municipality']}, {company['province']}")
        print(f"  Address: {company['address']}")
        print(f"  Instance ID: {company['instance_id']}")

        # Get detailed information for this company
        print("\n  Fetching detailed information...")
        try:
            details = client.get_company_details(
                p_auth=p_auth,
                instance_id=company['instance_id'],
                company_id=company['company_id']
            )

            print(f"  Full Details:")
            print(f"    Legal Form: {details.get('legal_form', 'N/A')}")
            print(f"    VAT Number: {details.get('vat_number', 'N/A')}")
            print(f"    Fiscal Code: {details.get('fiscal_code', 'N/A')}")
            print(f"    REA Number: {details.get('rea_number', 'N/A')}")
            print(f"    Full Address: {details.get('address', 'N/A')}")
            print(f"    PEC Email: {details.get('pec', 'N/A')}")

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

    print("\n" + "=" * 60)
    print("Search completed successfully!")
    print("=" * 60)


if __name__ == "__main__":
    main()