Amazon APIamazon.in ↗

"""
Amazon India API Parser Client
Get your API key from: https://parse.bot/settings
"""

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


class ParseClient:
    """Client for Amazon India API via Parse Bot."""
    
    def __init__(self, api_key: Optional[str] = None):
        """Initialize the Parse client with API credentials."""
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "db3857ae-e8f1-442e-896c-1c15b19a5ea9"
        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 API call to Parse Bot endpoint.
        
        Args:
            endpoint: The API endpoint name
            method: HTTP method (GET or POST)
            **params: Query/body parameters
            
        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)
        elif method == "POST":
            payload = params
            response = requests.post(url, headers=headers, json=payload)
        else:
            raise ValueError(f"Unsupported HTTP method: {method}")
        
        response.raise_for_status()
        return response.json()
    
    def get_search_suggestions(self, query: str) -> dict[str, Any]:
        """Get search autocomplete suggestions for a query prefix.
        
        Args:
            query: Search query prefix to get suggestions for
            
        Returns:
            Dictionary with search suggestions
        """
        return self._call("get_search_suggestions", method="GET", query=query)
    
    def search_products(self, query: str, page: int = 1) -> dict[str, Any]:
        """Search for products by keyword on Amazon.in.
        
        Args:
            query: Search keyword
            page: Page number for pagination (default: 1)
            
        Returns:
            Dictionary with product search results
        """
        return self._call("search_products", method="GET", query=query, page=page)
    
    def get_product_details(self, asin: str) -> dict[str, Any]:
        """Get detailed product information from Amazon.in product page.
        
        Args:
            asin: Amazon Standard Identification Number of the product
            
        Returns:
            Dictionary with product details
        """
        return self._call("get_product_details", method="GET", asin=asin)
    
    def get_bestsellers(self, category: str = "") -> dict[str, Any]:
        """Get Amazon Bestsellers list for a category.
        
        Args:
            category: Category path slug (e.g. 'electronics', 'books'). 
                     Empty string for overall bestsellers. (default: "")
            
        Returns:
            Dictionary with bestseller products
        """
        return self._call("get_bestsellers", method="GET", category=category)
    
    def get_product_reviews(self, asin: str, sort_by: str = "recent", star_filter: str = "") -> dict[str, Any]:
        """Extract customer reviews for an Amazon.in product.
        
        Args:
            asin: Amazon Standard Identification Number of the product
            sort_by: Sort order for reviews ('recent' or 'top'). (default: 'recent')
            star_filter: Filter reviews by star rating ('1'-'5' or ''). (default: '')
            
        Returns:
            Dictionary with product reviews
        """
        return self._call("get_product_reviews", method="GET", asin=asin, sort_by=sort_by, star_filter=star_filter)


def main():
    """Demonstrate practical usage of Amazon India API with real workflow."""
    
    # Initialize client
    client = ParseClient()
    
    print("=" * 80)
    print("Amazon India Product Research Tool")
    print("=" * 80)
    
    # Step 1: Get search suggestions for user query
    search_query = "wireless headphones"
    print(f"\n[Step 1] Getting autocomplete suggestions for '{search_query}'...")
    
    suggestions_response = client.get_search_suggestions(search_query)
    
    if suggestions_response.get("status") != "success":
        print("   ✗ Failed to get suggestions")
        return
    
    suggestions_data = suggestions_response.get("data", {})
    suggestions = suggestions_data.get("suggestions", [])
    
    if not suggestions:
        print("   ✗ No suggestions found")
        return
    
    print(f"   ✓ Found {len(suggestions)} suggestions")
    for i, sugg in enumerate(suggestions[:3], 1):
        print(f"      {i}. {sugg.get('value')} ({sugg.get('type')})")
    
    # Use first suggestion for refined search
    refined_query = suggestions[0].get("value", search_query)
    print(f"   → Refining search with: '{refined_query}'")
    
    # Step 2: Search for products with refined query
    print(f"\n[Step 2] Searching for products...")
    
    search_response = client.search_products(refined_query, page=1)
    
    if search_response.get("status") != "success":
        print("   ✗ Failed to search products")
        return
    
    search_data = search_response.get("data", {})
    products = search_data.get("products", [])
    results_count = search_data.get("results_count", 0)
    
    if not products:
        print("   ✗ No products found")
        return
    
    print(f"   ✓ Found {results_count} total results ({len(products)} on page 1)")
    
    # Step 3: Analyze top products with detailed information and reviews
    print(f"\n[Step 3] Analyzing top products in detail...\n")
    
    top_products = []
    
    for idx, product in enumerate(products[:3], 1):
        asin = product.get("asin")
        title = product.get("title", "N/A")
        price = product.get("price", "N/A")
        rating = product.get("rating", "N/A")
        
        print(f"   Product #{idx}")
        print(f"   ├─ Title: {title}")
        print(f"   ├─ Price: ₹{price}")
        print(f"   ├─ Rating: {rating}/5 (from search listing)")
        
        if not asin:
            print(f"   └─ No ASIN available\n")
            continue
        
        # Get detailed product information
        print(f"   ├─ Fetching detailed information (ASIN: {asin})...")
        details_response = client.get_product_details(asin)
        
        if details_response.get("status") != "success":
            print(f"   │  ✗ Failed to retrieve details\n")
            continue
        
        details_data = details_response.get("data", {})
        brand = details_data.get("brand", "N/A")
        availability = details_data.get("availability", "N/A")
        review_count = details_data.get("review_count", "N/A")
        features = details_data.get("features", [])
        images_count = len(details_data.get("images", []))
        
        print(f"   ├─ Brand: {brand}")
        print(f"   ├─ Availability: {availability}")
        print(f"   ├─ Review Count: {review_count}")
        print(f"   ├─ Product Images: {images_count} available")
        
        if features:
            print(f"   ├─ Top Features:")
            for feat_idx, feature in enumerate(features[:2], 1):
                feat_text = feature.replace("\n", " ").strip()[:65]
                marker = "├─" if feat_idx < len(features[:2]) else "└─"
                print(f"   │  {marker} {feat_text}...")
        
        # Get customer reviews
        print(f"   ├─ Fetching customer reviews...")
        reviews_response = client.get_product_reviews(asin, sort_by="recent", star_filter="")
        
        if reviews_response.get("status") != "success":
            print(f"   │  ✗ Failed to retrieve reviews\n")
            continue
        
        reviews_data = reviews_response.get("data", {})
        reviews = reviews_data.get("reviews", [])
        overall_rating = reviews_data.get("overall_rating", "N/A")
        total_ratings = reviews_data.get("total_ratings", "N/A")
        
        print(f"   ├─ Overall Rating: {overall_rating}")
        print(f"   ├─ Total Ratings: {total_ratings}")
        
        if reviews:
            print(f"   ├─ Recent Reviews ({len(reviews)} available):")
            for review_idx, review in enumerate(reviews[:2], 1):
                author = review.get("author", "Anonymous")
                rev_rating = review.get("rating", "N/A")
                rev_title = review.get("title", "No title")[:45]
                verified = "✓ Verified" if review.get("verified_purchase") else "✗ Not verified"
                date = review.get("date", "").replace("Reviewed in India on ", "")
                
                marker = "├─" if review_idx < len(reviews[:2]) else "└─"
                print(f"   │  {marker} {rev_title}")
                print(f"   │     • {author} - {rev_rating}★ ({verified})")
                print(f"   │     • {date}")
        
        top_products.append({
            "asin": asin,
            "title": title,
            "price": price,
            "rating": rating,
            "brand": brand,
            "availability": availability
        })
        
        print()
    
    # Step 4: Compare with bestsellers
    print("[Step 4] Comparing with Electronics Bestsellers...\n")
    
    bestsellers_response = client.get_bestsellers("electronics")
    
    if bestsellers_response.get("status") == "success":
        bestsellers_data = bestsellers_response.get("data", {})
        bestseller_products = bestsellers_data.get("products", [])
        
        print(f"   ✓ Top 3 Electronics Bestsellers:\n")
        
        for rank_idx, bestseller in enumerate(bestseller_products[:3], 1):
            rank = bestseller.get("rank", "N/A")
            title = bestseller.get("title", "N/A")[:60]
            price = bestseller.get("price", "N/A")
            rating = bestseller.get("rating", "N/A")
            
            print(f"   {rank_idx}. {rank}")
            print(f"      Title: {title}")
            print(f"      Price: {price} | Rating: {rating}★\n")
    
    # Final summary
    print("=" * 80)
    print("Research Summary")
    print("=" * 80)
    print(f"✓ Retrieved {len(top_products)} products with complete analysis")
    print(f"✓ Gathered customer reviews and detailed ratings")
    print(f"✓ Compared against current bestsellers")
    print(f"✓ Analysis complete!")
    
    if top_products:
        print(f"\nTop Product Found:")
        top = top_products[0]
        print(f"  • {top['title'][:60]}")
        print(f"  • Price: ₹{top['price']} | Rating: {top['rating']}/5")
        print(f"  • Brand: {top['brand']} | Stock: {top['availability']}")
    
    print("=" * 80)


if __name__ == "__main__":
    main()