Coupang APIcoupang.com ↗

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

import os
import sys
import requests
from typing import Optional, List
from dataclasses import dataclass
from datetime import datetime


@dataclass
class Product:
    """Represents a Coupang product"""
    product_id: int
    name: str
    sale_price: int
    original_price: int
    discount_rate: float
    rating: float
    review_count: int
    rocket_delivery: bool
    sold_out: bool
    url: str


@dataclass
class Review:
    """Represents a product review"""
    review_id: int
    rating: int
    title: str
    content: str
    reviewer: str
    review_date: int
    helpful_count: int
    item_name: str
    image_count: int


class ParseClient:
    """Client for interacting with Parse API (Coupang Product & Review endpoints)"""
    
    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.
        
        Raises:
            ValueError: If no API key is provided or available in environment.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "4efec4ec-d019-4a6a-a431-fb9d14f77a19"
        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:
        """
        Make a request to the Parse API.
        
        Args:
            endpoint: The endpoint name (e.g., 'get_product_recommendations')
            method: HTTP method ('GET' or 'POST')
            **params: Query/body parameters
        
        Returns:
            Response JSON as dictionary
        
        Raises:
            requests.exceptions.RequestException: If the API request fails.
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        try:
            if method.upper() == "GET":
                response = requests.get(url, headers=headers, params=params, timeout=30)
            elif method.upper() == "POST":
                response = requests.post(url, headers=headers, json=params, timeout=30)
            else:
                raise ValueError(f"Unsupported HTTP method: {method}")
            
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"API request failed: {e}", file=sys.stderr)
            raise
    
    def get_product_recommendations(
        self,
        product_ids: str = "8499380264",
        max_count: int = 15
    ) -> dict:
        """
        Get recommended products based on one or more product IDs.
        
        Args:
            product_ids: Comma-separated Coupang product IDs (default: "8499380264")
            max_count: Maximum number of products to return, 5-15 (default: 15)
        
        Returns:
            Dictionary with 'title', 'total_products', and 'products' array
        """
        return self._call(
            "get_product_recommendations",
            method="GET",
            product_ids=product_ids,
            max_count=max_count
        )
    
    def get_product_reviews(
        self,
        product_id: str,
        page: int = 1,
        size: int = 10,
        sort_by: str = "ORDER_SCORE_ASC"
    ) -> dict:
        """
        Get paginated reviews for a specific Coupang product.
        
        Args:
            product_id: Coupang product ID (required)
            page: Page number for pagination (default: 1)
            size: Number of reviews per page (default: 10)
            sort_by: Sort order - 'ORDER_SCORE_ASC', 'ORDER_SCORE_DESC', or 'DATE_DESC'
        
        Returns:
            Dictionary with 'rating_summary', 'reviews' array, and pagination info
        """
        return self._call(
            "get_product_reviews",
            method="GET",
            product_id=product_id,
            page=page,
            size=size,
            sort_by=sort_by
        )


def parse_product(product_data: dict) -> Product:
    """Convert API product data to Product dataclass"""
    return Product(
        product_id=product_data['product_id'],
        name=product_data['name'],
        sale_price=product_data['sale_price'],
        original_price=product_data['original_price'],
        discount_rate=product_data['discount_rate'],
        rating=product_data['rating'],
        review_count=product_data['review_count'],
        rocket_delivery=product_data['rocket_delivery'],
        sold_out=product_data['sold_out'],
        url=product_data['url']
    )


def parse_review(review_data: dict) -> Review:
    """Convert API review data to Review dataclass"""
    return Review(
        review_id=review_data['review_id'],
        rating=review_data['rating'],
        title=review_data['title'],
        content=review_data['content'],
        reviewer=review_data['reviewer'],
        review_date=review_data['review_date'],
        helpful_count=review_data['helpful_count'],
        item_name=review_data['item_name'],
        image_count=review_data['image_count']
    )


def format_currency(amount: int) -> str:
    """Format amount as Korean Won"""
    return f"₩{amount:,}"


def print_product_info(product: Product, rank: int = None) -> None:
    """Print formatted product information"""
    status = "🚀" if product.rocket_delivery else "📦"
    sold = "❌ SOLD OUT" if product.sold_out else "✅ Available"
    rank_str = f"#{rank} " if rank else ""
    
    print(f"\n{rank_str}{status} {product.name[:60]}")
    print(f"   Price: {format_currency(product.sale_price)} (was {format_currency(product.original_price)})")
    print(f"   Discount: {product.discount_rate:.1f}% | Rating: ⭐ {product.rating}/5 ({product.review_count:,} reviews)")
    print(f"   Status: {sold}")


def print_review_info(review: Review) -> None:
    """Print formatted review information"""
    star_rating = "⭐" * review.rating + "☆" * (5 - review.rating)
    date_obj = datetime.fromtimestamp(review.review_date / 1000)
    date_str = date_obj.strftime("%Y-%m-%d")
    
    print(f"\n  {star_rating} {review.title}")
    print(f"     By: {review.reviewer} ({date_str})")
    content_preview = review.content[:80] + "..." if len(review.content) > 80 else review.content
    print(f"     \"{content_preview}\"")
    print(f"     👍 {review.helpful_count} helpful | 📷 {review.image_count} images")


def main():
    """
    Practical workflow:
    1. Get product recommendations for a base product
    2. Filter and rank recommendations by value
    3. Get detailed reviews for top 3 candidates
    4. Display comparison and recommendations
    """
    
    # Initialize client
    client = ParseClient()
    print("=" * 80)
    print("🛒 Coupang Smart Product Finder")
    print("=" * 80)
    
    try:
        # Step 1: Get recommendations
        print("\n📊 Step 1: Fetching product recommendations...")
        print("-" * 80)
        
        base_product_id = "8499380264"
        recommendations = client.get_product_recommendations(
            product_ids=base_product_id,
            max_count=10
        )
        
        print(f"✅ Found {recommendations['total_products']} recommended products")
        print(f"   Category: {recommendations['title']}\n")
        
        # Parse all products
        products: List[Product] = []
        for product_data in recommendations['products']:
            product = parse_product(product_data)
            products.append(product)
        
        if not products:
            print("❌ No products found")
            return
        
        # Step 2: Filter and rank products
        print("\n📈 Step 2: Analyzing and ranking products...")
        print("-" * 80)
        
        available_products = [p for p in products if not p.sold_out]
        if not available_products:
            print("❌ No available products")
            return
        
        # Sort by rating first, then by discount rate
        ranked_products = sorted(
            available_products,
            key=lambda p: (-p.rating, -p.discount_rate)
        )
        
        print(f"Available products: {len(available_products)}/{len(products)}")
        print("\nTop 3 candidates for detailed review:")
        
        for idx, product in enumerate(ranked_products[:3], 1):
            print_product_info(product, rank=idx)
        
        # Step 3: Get detailed reviews for top 3 products
        print("\n\n🔍 Step 3: Fetching detailed reviews for top candidates...")
        print("-" * 80)
        
        top_candidates = ranked_products[:3]
        review_cache = {}
        
        for candidate in top_candidates:
            print(f"\n📚 Reviews for: {candidate.name[:50]}")
            print(f"   Product ID: {candidate.product_id}")
            
            try:
                reviews_response = client.get_product_reviews(
                    product_id=str(candidate.product_id),
                    page=1,
                    size=3,
                    sort_by="ORDER_SCORE_DESC"
                )
                
                review_cache[candidate.product_id] = reviews_response
                
                # Display rating summary
                rating_summary = reviews_response['rating_summary']
                print(f"\n   📊 Rating Summary:")
                print(f"      Average: ⭐ {rating_summary['average_rating']}/5")
                print(f"      Total Reviews: {rating_summary['total_reviews']:,}")
                
                # Show rating distribution
                if rating_summary.get('rating_distribution'):
                    print(f"      Distribution:")
                    for dist in rating_summary['rating_distribution']:
                        bar_length = int(dist['percentage'] / 5)
                        bar = "█" * bar_length
                        print(f"      {'⭐' * dist['rating']:5} {bar:<20} {dist['percentage']:3}% ({dist['count']:,})")
                
                # Display top reviews
                reviews = reviews_response.get('reviews', [])
                if reviews:
                    print(f"\n   💬 Recent reviews:")
                    for review in reviews[:3]:
                        print_review_info(parse_review(review))
                
            except Exception as e:
                print(f"      ⚠️  Could not fetch reviews: {e}")
        
        # Step 4: Final recommendation
        print("\n\n" + "=" * 80)
        print("🎯 FINAL RECOMMENDATION")
        print("=" * 80)
        
        best_overall = ranked_products[0]
        best_value = min(
            [p for p in ranked_products if p.rating >= 4.5],
            key=lambda p: p.sale_price,
            default=ranked_products[0]
        )
        
        print(f"\n🏆 Best Overall: {best_overall.name[:55]}")
        print(f"   Rating: ⭐ {best_overall.rating}/5 | Price: {format_currency(best_overall.sale_price)}")
        if best_overall.product_id in review_cache:
            total = review_cache[best_overall.product_id]['rating_summary']['total_reviews']
            print(f"   Based on {total:,} customer reviews")
        
        if best_value.product_id != best_overall.product_id:
            print(f"\n💰 Best Value: {best_value.name[:55]}")
            print(f"   Rating: ⭐ {best_value.rating}/5 | Price: {format_currency(best_value.sale_price)}")
            savings = best_overall.sale_price - best_value.sale_price
            if savings > 0:
                print(f"   Save {format_currency(savings)} vs. best overall!")
        
        # Show complete comparison table
        print("\n📋 Complete Ranking (Top 5):")
        print("-" * 80)
        print(f"{'Rank':<6} {'Name':<40} {'Price':<15} {'Rating':<10} {'Reviews':<10}")
        print("-" * 80)
        
        for idx, product in enumerate(ranked_products[:5], 1):
            name_short = product.name[:38]
            rating_display = f"⭐ {product.rating}/5"
            print(f"{idx:<6} {name_short:<40} {format_currency(product.sale_price):<15} "
                  f"{rating_display:<10} {product.review_count:,}")
        
        print("\n" + "=" * 80)
        print("✅ Analysis completed successfully!")
        print("=" * 80)
        
    except Exception as e:
        print(f"\n❌ Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()