Trustpilot APIuk.trustpilot.com ↗

"""
Trustpilot Data API Client

Comprehensive API for Trustpilot review data, company insights, categories, and blog content.
Get your API key from: https://parse.bot/settings
"""

import os
import requests
from typing import Any, Optional


class ParseClient:
    """Client for interacting with the Trustpilot Data API through Parse."""

    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 env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "bb2e1c0f-26df-4ad6-b6a0-874d5bc47c6f"
        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: API endpoint name
            method: HTTP method (GET or POST)
            **params: Query parameters or request body
            
        Returns:
            API response 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 == "GET":
            response = requests.get(url, headers=headers, params=params)
        else:  # POST
            response = requests.post(url, headers=headers, json=params)
        
        response.raise_for_status()
        return response.json()

    def search_companies(self, query: str, limit: int = 20, page: int = 1) -> dict[str, Any]:
        """Search for companies by name or keyword.
        
        Args:
            query: Search keyword (e.g., 'John Lewis')
            limit: Max results per page (default: 20)
            page: Page number (default: 1)
            
        Returns:
            Dictionary containing matching companies with basic info
        """
        return self._call(
            "search_companies",
            method="GET",
            query=query,
            limit=limit,
            page=page
        )

    def get_company_overview(self, domain: str) -> dict[str, Any]:
        """Get the summary/overview for a company by its domain.
        
        Args:
            domain: Company domain (e.g., 'www.johnlewis.com')
            
        Returns:
            Dictionary containing trust score, star rating, review count, and AI summary
        """
        return self._call(
            "get_company_overview",
            method="GET",
            domain=domain
        )

    def get_company_reviews(
        self,
        domain: str,
        language: Optional[str] = None,
        stars: Optional[str] = None,
        page: int = 1,
        date: Optional[str] = None,
        sort: str = "recency"
    ) -> dict[str, Any]:
        """Get paginated reviews for a company with optional filtering.
        
        Args:
            domain: Company domain
            language: ISO language code (e.g., 'en')
            stars: Filter by star rating (1-5)
            page: Page number (default: 1)
            date: Date filter (last30days, last3months, last6months, last12months)
            sort: Sort order (recency or relevance)
            
        Returns:
            Dictionary containing reviews and pagination info
        """
        params = {
            "domain": domain,
            "page": page,
            "sort": sort
        }
        if language:
            params["language"] = language
        if stars:
            params["stars"] = stars
        if date:
            params["date"] = date
            
        return self._call("get_company_reviews", method="GET", **params)

    def get_reviews_for_month(
        self,
        domain: str,
        month: int = 7,
        year: int = 2025,
        max_pages: int = 50
    ) -> dict[str, Any]:
        """Get all reviews for a specific month and year.
        
        Args:
            domain: Company domain
            month: Target month (1-12)
            year: Target year
            max_pages: Max pages to scan
            
        Returns:
            Dictionary containing reviews from the specified month
        """
        return self._call(
            "get_reviews_for_month",
            method="GET",
            domain=domain,
            month=month,
            year=year,
            max_pages=max_pages
        )

    def get_categories(self) -> dict[str, Any]:
        """Get the hierarchical category structure of Trustpilot.
        
        Returns:
            Dictionary containing category hierarchy
        """
        return self._call("get_categories", method="GET")

    def get_category_companies(self, category_id: str, page: int = 1) -> dict[str, Any]:
        """Get companies listed under a specific category.
        
        Args:
            category_id: Category ID (slug) e.g., 'bank'
            page: Page number (default: 1)
            
        Returns:
            Dictionary containing companies in the category
        """
        return self._call(
            "get_category_companies",
            method="GET",
            category_id=category_id,
            page=page
        )

    def get_company_top_mentions(
        self,
        domain: Optional[str] = None,
        business_unit_id: Optional[str] = None
    ) -> dict[str, Any]:
        """Get top mention tags/topics extracted from reviews for a company.
        
        Args:
            domain: Company domain (will be resolved to ID if provided)
            business_unit_id: Business Unit ID (hash)
            
        Returns:
            Dictionary containing top topics/mentions
        """
        params = {}
        if domain:
            params["domain"] = domain
        if business_unit_id:
            params["business_unit_id"] = business_unit_id
            
        return self._call("get_company_top_mentions", method="GET", **params)

    def get_similar_companies(self, domain: str) -> dict[str, Any]:
        """Get similar companies users also viewed.
        
        Args:
            domain: Company domain
            
        Returns:
            Dictionary containing similar business units
        """
        return self._call(
            "get_similar_companies",
            method="GET",
            domain=domain
        )

    def get_blog_posts(self) -> dict[str, Any]:
        """Get recent blog posts from the Trustpilot blog.
        
        Returns:
            Dictionary containing featured article and pillar articles
        """
        return self._call("get_blog_posts", method="GET")


def main():
    """Practical workflow: Compare companies within a category and analyze their reviews."""
    client = ParseClient()
    
    print("=" * 70)
    print("Trustpilot Data API - Competitive Analysis Workflow")
    print("=" * 70)
    
    # Step 1: Get all available categories
    print("\n[Step 1] Fetching Trustpilot categories...")
    categories = client.get_categories()
    
    # Find a specific category (e.g., banks or department stores)
    target_category = None
    for category in categories.get("categories", []):
        if category["categoryId"] == "department_store":
            target_category = category
            break
    
    if target_category:
        print(f"✓ Found category: {target_category['displayName']}")
    
    # Step 2: Get top companies in the category
    print(f"\n[Step 2] Getting top companies in '{target_category['displayName']}'...")
    category_companies = client.get_category_companies(
        category_id=target_category["categoryId"],
        page=1
    )
    
    companies_list = category_companies["businessUnits"]["businesses"][:3]
    print(f"✓ Retrieved {len(companies_list)} top companies")
    
    # Store company data for comparison
    company_analysis = []
    
    # Step 3: For each company, gather detailed insights
    for idx, company in enumerate(companies_list, 1):
        print(f"\n[Step 3.{idx}] Analyzing {company['displayName']}...")
        
        domain = company["identifyingName"]
        company_id = company["businessUnitId"]
        
        # Get company overview
        overview = client.get_company_overview(domain=domain)
        
        # Get recent reviews
        reviews_data = client.get_company_reviews(
            domain=domain,
            page=1,
            sort="recency"
        )
        
        # Get top mentions
        mentions = client.get_company_top_mentions(domain=domain)
        
        # Compile analysis
        analysis = {
            "name": company["displayName"],
            "domain": domain,
            "unit_id": company_id,
            "reviews_count": company["numberOfReviews"],
            "stars": company.get("stars"),
            "trust_score": company.get("trustScore"),
            "recent_reviews": reviews_data.get("reviews", [])[:3],
            "top_topics": mentions.get("topics", [])[:5]
        }
        
        company_analysis.append(analysis)
        
        print(f"  ✓ Rating: {analysis['stars']}⭐ | Trust Score: {analysis['trust_score']}")
        print(f"  ✓ Total Reviews: {analysis['reviews_count']:,}")
        print(f"  ✓ Retrieved {len(analysis['recent_reviews'])} recent reviews")
        print(f"  ✓ Identified {len(analysis['top_topics'])} top discussion topics")
    
    # Step 4: Comparative analysis
    print("\n" + "=" * 70)
    print("COMPARATIVE ANALYSIS")
    print("=" * 70)
    
    print("\n[Ratings Comparison]")
    for company in company_analysis:
        print(f"  {company['name']:30} {company['stars']}⭐ ({company['trust_score']} trust score)")
    
    print("\n[Review Volume]")
    for company in company_analysis:
        print(f"  {company['name']:30} {company['reviews_count']:,} reviews")
    
    print("\n[Recent Review Sentiment]")
    for company in company_analysis:
        if company["recent_reviews"]:
            ratings = [r.get("rating", 0) for r in company["recent_reviews"]]
            avg_rating = sum(ratings) / len(ratings) if ratings else 0
            print(f"  {company['name']:30} Last 3 reviews avg: {avg_rating:.1f}⭐")
    
    print("\n[Top Discussion Topics]")
    for company in company_analysis:
        topics = ", ".join([t.get("displayName", "Unknown") for t in company["top_topics"][:3]])
        print(f"  {company['name']:30} {topics}")
    
    # Step 5: Deep dive into best-performing company
    best_company = max(company_analysis, key=lambda x: x["stars"] if x["stars"] else 0)
    print("\n" + "=" * 70)
    print(f"DEEP DIVE: {best_company['name']}")
    print("=" * 70)
    
    print(f"\n[Filtering reviews by rating...]")
    positive_reviews = client.get_company_reviews(
        domain=best_company["domain"],
        stars="5",
        page=1
    )
    
    print(f"  5-star reviews found: {len(positive_reviews.get('reviews', []))}")
    
    # Check for similar companies
    print(f"\n[Finding competitors...]")
    similar = client.get_similar_companies(domain=best_company["domain"])
    
    if similar.get("similarBusinessUnits"):
        print(f"  Users also compared with:")
        for comp in similar["similarBusinessUnits"][:3]:
            comp_name = comp.get("businessUnitDisplayName", "Unknown")
            comp_rating = comp.get("stars", "N/A")
            print(f"    - {comp_name} ({comp_rating}⭐)")
    
    # Step 6: Blog insights
    print(f"\n[Latest Trustpilot Insights]")
    blog = client.get_blog_posts()
    
    if blog.get("featuredArticle"):
        featured = blog["featuredArticle"]
        print(f"  Featured: {featured.get('title', 'N/A')}")
        print(f"  Published: {featured.get('date', 'N/A')}")
    
    print("\n" + "=" * 70)
    print("✓ Analysis complete! Use this data for competitive benchmarking.")
    print("=" * 70)


if __name__ == "__main__":
    main()