Ozon APIozon.ru ↗

"""
Ozon.ru API Parser Client
Get your API key from: https://parse.bot/settings
"""

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


class ParseClient:
    """Client for accessing Ozon.ru product data via Parse API."""

    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 environment variable.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "38bf3079-48f3-41e3-926d-7c6ea7fa1fe9"
        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 an API call to the Parse endpoint.
        
        Args:
            endpoint: The endpoint name (e.g., 'search_products')
            method: HTTP method ('GET' or 'POST')
            **params: Parameters to pass to the endpoint
            
        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"
        }
        
        try:
            if method == "GET":
                response = requests.get(url, headers=headers, params=params, timeout=30)
            else:  # POST
                response = requests.post(url, headers=headers, json=params, timeout=30)
            
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"API call failed: {e}")

    def search_products(
        self,
        query: str,
        page: int = 1,
        sort: str = "popular"
    ) -> Dict[str, Any]:
        """Search for products by keyword.
        
        Args:
            query: Search keyword
            page: Page number (default: 1)
            sort: Sort order - popular, new, price_asc, price_desc, rating, discount
            
        Returns:
            Dictionary containing products list and total count
        """
        return self._call("search_products", method="GET", query=query, page=page, sort=sort)

    def get_category_products(
        self,
        category_url: str,
        page: int = 1,
        sort: str = "popular"
    ) -> Dict[str, Any]:
        """Browse products in a specific category.
        
        Args:
            category_url: Full URL of the category page
            page: Page number (default: 1)
            sort: Sort order (default: 'popular')
            
        Returns:
            Dictionary containing products list and total count
        """
        return self._call("get_category_products", method="GET", category_url=category_url, page=page, sort=sort)

    def get_product_details(self, url: str) -> Dict[str, Any]:
        """Get full product detail page data.
        
        Args:
            url: Product URL
            
        Returns:
            Dictionary containing product details
        """
        return self._call("get_product_details", method="GET", url=url)

    def get_product_reviews(self, url: str) -> Dict[str, Any]:
        """Retrieve customer reviews for a specific product.
        
        Args:
            url: Product URL
            
        Returns:
            Dictionary containing reviews list
        """
        return self._call("get_product_reviews", method="GET", url=url)

    def get_product_questions(self, url: str) -> Dict[str, Any]:
        """Retrieve Q&A for a specific product.
        
        Args:
            url: Product URL
            
        Returns:
            Dictionary containing questions and answers
        """
        return self._call("get_product_questions", method="GET", url=url)

    def get_category_tree(self) -> Dict[str, Any]:
        """Get the full hierarchical category tree.
        
        Returns:
            Dictionary containing category hierarchy
        """
        return self._call("get_category_tree", method="GET")

    def get_search_filters(self, query: str) -> Dict[str, Any]:
        """Get available filter options for a search query.
        
        Args:
            query: Search keyword
            
        Returns:
            Dictionary containing available filters and their values
        """
        return self._call("get_search_filters", method="GET", query=query)

    def get_seller_info(self, seller_slug: str) -> Dict[str, Any]:
        """Get information about a seller by slug.
        
        Args:
            seller_slug: Seller slug from URL
            
        Returns:
            Dictionary containing seller information
        """
        return self._call("get_seller_info", method="GET", seller_slug=seller_slug)


if __name__ == "__main__":
    # Initialize the client
    client = ParseClient()
    
    print("=" * 80)
    print("OZON.RU PRODUCT SEARCH AND ANALYSIS EXAMPLE")
    print("=" * 80)
    
    # Step 1: Search for products
    search_query = "iPhone"
    print(f"\n1. Searching for products: '{search_query}'")
    print("-" * 80)
    
    search_results = client.search_products(query=search_query, sort="rating", page=1)
    products = search_results.get("products", [])
    total_products = search_results.get("total", 0)
    
    print(f"Found {total_products} products matching '{search_query}'")
    print(f"Displaying {len(products)} products on this page:\n")
    
    # Step 2: Display search results summary
    for idx, product in enumerate(products[:3], 1):  # Show first 3 products
        title = product.get("title", "N/A")
        url = product.get("url", "N/A")
        price = product.get("price", {})
        current_price = price.get("current", "N/A")
        discount = price.get("discount", "N/A")
        rating_info = product.get("rating", {})
        rating = rating_info.get("rating", "N/A")
        reviews = rating_info.get("reviews", "N/A")
        product_id = product.get("id", "N/A")
        
        print(f"Product {idx}: {title[:60]}")
        print(f"  Price: {current_price} (Discount: {discount})")
        print(f"  Rating: {rating}/5.0 ({reviews})")
        print(f"  ID: {product_id}")
        print()
    
    # Step 3: Get detailed information for the first product
    if products:
        first_product = products[0]
        product_url = first_product.get("url")
        product_title = first_product.get("title")
        
        if product_url:
            print("\n2. Getting detailed information for first product")
            print("-" * 80)
            print(f"Product: {product_title[:70]}")
            
            details = client.get_product_details(url=product_url)
            
            # Display key details
            detail_price = details.get("price", "N/A")
            detail_rating = details.get("rating", "N/A")
            review_count = details.get("review_count", "N/A")
            description = details.get("description", "N/A")
            sku = details.get("sku", "N/A")
            images_count = len(details.get("images", []))
            chars_count = len(details.get("characteristics", []))
            
            print(f"  SKU: {sku}")
            print(f"  Price: {detail_price}")
            print(f"  Rating: {detail_rating}/5.0")
            print(f"  Reviews: {review_count}")
            print(f"  Images: {images_count}")
            print(f"  Characteristics: {chars_count}")
            if description and description != "N/A":
                print(f"  Description: {description[:150]}...")
            
            # Step 4: Get reviews for the product
            print("\n3. Fetching customer reviews")
            print("-" * 80)
            reviews_data = client.get_product_reviews(url=product_url)
            reviews = reviews_data.get("reviews", [])
            
            print(f"Found {len(reviews)} reviews. Showing first 2:\n")
            for review_idx, review in enumerate(reviews[:2], 1):
                author = review.get("author", "Anonymous")
                rating_val = review.get("rating", "N/A")
                text = review.get("text", "N/A")
                date = review.get("date", "N/A")
                
                print(f"Review {review_idx}:")
                print(f"  Author: {author}")
                print(f"  Rating: {rating_val}/5")
                print(f"  Date: {date}")
                print(f"  Text: {text[:100]}...")
                print()
            
            # Step 5: Get Q&A for the product
            print("4. Fetching Q&A section")
            print("-" * 80)
            qa_data = client.get_product_questions(url=product_url)
            questions = qa_data.get("questions", [])
            
            print(f"Found {len(questions)} Q&A entries. Showing first 2:\n")
            for qa_idx, qa in enumerate(questions[:2], 1):
                question_text = qa.get("text", "N/A")
                question_date = qa.get("date", "N/A")
                answers = qa.get("answers", [])
                
                print(f"Q&A {qa_idx}:")
                print(f"  Question: {question_text}")
                print(f"  Asked: {question_date}")
                print(f"  Answers: {len(answers)}")
                if answers:
                    answer = answers[0]
                    answer_author = answer.get("author", "N/A")
                    answer_text = answer.get("text", "N/A")
                    print(f"    - {answer_author}: {answer_text[:80]}...")
                print()
    
    # Step 6: Get search filters for advanced filtering
    print("5. Available filters for search")
    print("-" * 80)
    filters_data = client.get_search_filters(query=search_query)
    filters = filters_data.get("filters", [])
    
    if filters:
        for filter_item in filters[:2]:  # Show first 2 filters
            filter_name = filter_item.get("name", "N/A")
            filter_id = filter_item.get("id", "N/A")
            values = filter_item.get("values", [])
            
            print(f"\nFilter: {filter_name} (ID: {filter_id})")
            print(f"  Available values:")
            for value in values[:3]:  # Show first 3 values
                value_name = value.get("name", "N/A")
                value_count = value.get("count", "0")
                print(f"    - {value_name} ({value_count} products)")
    
    # Step 7: Get category tree
    print("\n\n6. Category tree structure")
    print("-" * 80)
    category_tree = client.get_category_tree()
    categories = category_tree.get("data", [])
    
    if categories:
        for cat in categories[:2]:  # Show first 2 main categories
            cat_title = cat.get("title", "N/A")
            cat_id = cat.get("id", "N/A")
            items = cat.get("items", [])
            
            print(f"\nCategory: {cat_title} (ID: {cat_id})")
            print(f"  Subcategories:")
            for item in items[:3]:  # Show first 3 subcategories
                item_title = item.get("title", "N/A")
                item_link = item.get("link", "N/A")
                print(f"    - {item_title}")
    
    print("\n" + "=" * 80)
    print("SEARCH AND ANALYSIS COMPLETE")
    print("=" * 80)