Flipkart APIflipkart.com ↗

"""
Flipkart Product Reviews API Client
Get your API key from: https://parse.bot/settings
"""

import os
import requests
from typing import Optional, Dict, Any
from urllib.parse import urlencode


class ParseClient:
    """Client for interacting with the Parse API for Flipkart product reviews."""
    
    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 = "dfeb72c1-9b76-4102-a752-70e10f3a0c50"
        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 environment variable 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., 'get_rating_summary')
            method: HTTP method - 'GET' or 'POST'
            **params: Parameters to send with the request
            
        Returns:
            The JSON response from the API
        """
        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)
        else:  # POST
            response = requests.post(url, headers=headers, json=params)
        
        response.raise_for_status()
        return response.json()
    
    def get_rating_summary(
        self, 
        product_url: str, 
        sort_order: str = "MOST_RECENT"
    ) -> Dict[str, Any]:
        """
        Get the rating summary for a Flipkart product.
        
        Args:
            product_url: Flipkart product reviews URL path (e.g., /product-name/product-reviews/itemid?pid=...)
            sort_order: Sort order - MOST_RECENT, MOST_HELPFUL, POSITIVE_FIRST, NEGATIVE_FIRST
            
        Returns:
            Rating summary data including average rating, total ratings, and breakdown by stars
        """
        return self._call(
            "get_rating_summary",
            method="GET",
            product_url=product_url,
            sort_order=sort_order
        )
    
    def get_reviews(
        self,
        product_url: str,
        page: str = "1",
        sort_order: str = "MOST_RECENT",
        certified_buyer: str = "false"
    ) -> Dict[str, Any]:
        """
        Get individual product reviews with full details.
        
        Args:
            product_url: Flipkart product reviews URL path
            page: Page number for pagination (1-based)
            sort_order: Sort order - MOST_RECENT, MOST_HELPFUL, POSITIVE_FIRST, NEGATIVE_FIRST
            certified_buyer: Filter to certified buyer reviews - 'true' or 'false'
            
        Returns:
            Paginated reviews data with individual review details
        """
        return self._call(
            "get_reviews",
            method="GET",
            product_url=product_url,
            page=page,
            sort_order=sort_order,
            certified_buyer=certified_buyer
        )


def analyze_product_reviews():
    """
    Practical workflow: Analyze a Flipkart product's reviews comprehensively.
    
    This example demonstrates:
    1. Getting rating summary for overview statistics
    2. Fetching reviews across multiple pages
    3. Analyzing patterns in the review data
    """
    
    client = ParseClient()
    
    # Example product URL - this would be a real Flipkart product reviews URL
    product_url = "/nivea-body-milk-nourishing-lotion-400ml-120-ml-pack-2/product-reviews/itmf6y86zahhzntz?pid=MSCF6Y86XFKRFZDG&lid=LSTMSCF6Y86XFKRFZDG42HEUC"
    
    print("=" * 80)
    print("FLIPKART PRODUCT REVIEWS ANALYSIS")
    print("=" * 80)
    
    # Step 1: Get rating summary
    print("\n[1] Fetching rating summary...")
    rating_response = client.get_rating_summary(product_url, sort_order="MOST_HELPFUL")
    
    if rating_response.get("status") == "success":
        rating_data = rating_response.get("data", {})
        
        print(f"\n📊 RATING SUMMARY:")
        print(f"   Average Rating: {rating_data.get('average_rating', 'N/A')}/5.0")
        print(f"   Total Ratings: {rating_data.get('total_ratings', 'N/A'):,}")
        print(f"   Total Reviews: {rating_data.get('total_reviews', 'N/A'):,}")
        
        # Display rating breakdown
        breakdown = rating_data.get("rating_breakdown", {})
        print(f"\n⭐ RATING BREAKDOWN:")
        for star_count in [5, 4, 3, 2, 1]:
            key = f"{star_count}_star"
            count = breakdown.get(key, 0)
            percentage = (count / rating_data.get('total_ratings', 1)) * 100
            bar = "█" * int(percentage / 2)
            print(f"   {star_count}★ {bar} {count:,} ({percentage:.1f}%)")
        
        # Step 2: Fetch reviews from multiple pages
        print(f"\n[2] Fetching reviews (most helpful first)...")
        
        all_reviews = []
        pages_to_fetch = min(3, (rating_data.get('total_reviews', 10) // 10) + 1)  # Fetch up to 3 pages
        
        for page_num in range(1, pages_to_fetch + 1):
            print(f"   Loading page {page_num}...")
            reviews_response = client.get_reviews(
                product_url,
                page=str(page_num),
                sort_order="MOST_HELPFUL",
                certified_buyer="false"
            )
            
            if reviews_response.get("status") == "success":
                reviews_data = reviews_response.get("data", {})
                reviews = reviews_data.get("reviews", [])
                all_reviews.extend(reviews)
            else:
                break
        
        # Step 3: Analyze review patterns
        print(f"\n[3] ANALYZING {len(all_reviews)} REVIEWS:")
        
        rating_distribution = {5: 0, 4: 0, 3: 0, 2: 0, 1: 0}
        certified_count = 0
        avg_helpful = 0
        locations_seen = set()
        
        for review in all_reviews:
            rating = review.get("rating", 0)
            rating_distribution[rating] = rating_distribution.get(rating, 0) + 1
            
            if review.get("certified_buyer"):
                certified_count += 1
            
            avg_helpful += review.get("helpful_count", 0)
            
            location = review.get("location", {})
            if location.get("city"):
                locations_seen.add(f"{location.get('city')}, {location.get('state', 'N/A')}")
        
        avg_helpful = avg_helpful / len(all_reviews) if all_reviews else 0
        
        print(f"   Certified Buyers: {certified_count}/{len(all_reviews)} ({(certified_count/len(all_reviews)*100):.1f}%)")
        print(f"   Avg Helpful Count: {avg_helpful:.1f}")
        print(f"   Unique Locations: {len(locations_seen)}")
        
        # Sample reviews
        print(f"\n[4] SAMPLE REVIEWS (Top {min(3, len(all_reviews))} Most Helpful):")
        for i, review in enumerate(all_reviews[:3], 1):
            print(f"\n   Review #{i}:")
            print(f"   👤 {review.get('author', 'Anonymous')}")
            print(f"   ⭐ Rating: {review.get('rating')}/5")
            print(f"   ✍️  Title: {review.get('title', 'No title')}")
            print(f"   📝 {review.get('text', 'No text')[:100]}...")
            print(f"   📅 Date: {review.get('created', 'N/A')}")
            print(f"   👍 Helpful: {review.get('helpful_count', 0)}")
            if review.get("product_attributes"):
                attrs = ", ".join([f"{a.get('name')}: {a.get('value')}" for a in review.get("product_attributes", [])])
                print(f"   📦 Variant: {attrs}")
    
    print("\n" + "=" * 80)
    print("Analysis complete!")
    print("=" * 80)


if __name__ == "__main__":
    analyze_product_reviews()