Ozon APIozon.kz ↗

"""
Ozon.kz Catalog API Client

A Python client for searching products, retrieving details, browsing categories,
and more on Ozon.kz.

Get your API key from: https://parse.bot/settings
"""

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


class ParseClient:
    """Client for interacting with the Ozon.kz Catalog API via Parse Bot."""

    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 = "3181e6fd-238a-49a5-986f-96065147b60f"
        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 API endpoint name
            method: HTTP method (GET or POST)
            **params: Parameters to pass to the endpoint

        Returns:
            Response data as dictionary

        Raises:
            requests.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"
        }

        if method == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method == "POST":
            response = requests.post(url, headers=headers, json=params)
        else:
            raise ValueError(f"Unsupported HTTP method: {method}")

        response.raise_for_status()
        return response.json()

    def search_products(
        self,
        query: str,
        sort: str = "score",
        page: int = 1
    ) -> Dict[str, Any]:
        """
        Search for products by keyword with optional sorting.

        Args:
            query: Search query string
            sort: Sort order (score, price_asc, price_desc, rating, discount)
            page: Page number for pagination

        Returns:
            Dictionary containing products list and page number
        """
        return self._call(
            "search_products",
            method="GET",
            query=query,
            sort=sort,
            page=page
        )

    def get_product_details(self, product_id: str) -> Dict[str, Any]:
        """
        Get detailed information for a specific product.

        Args:
            product_id: Product ID (SKU) or URL slug

        Returns:
            Dictionary containing detailed product information
        """
        return self._call(
            "get_product_details",
            method="GET",
            product_id=product_id
        )

    def get_category_listing(self, category_id: str, page: int = 1) -> Dict[str, Any]:
        """
        Browse products in a specific category.

        Args:
            category_id: Category ID or path slug
            page: Page number for pagination

        Returns:
            Dictionary containing products in the category and page number
        """
        return self._call(
            "get_category_listing",
            method="GET",
            category_id=category_id,
            page=page
        )

    def get_category_tree(self) -> Dict[str, Any]:
        """
        Retrieve the category tree for the catalog menu.

        Returns:
            Dictionary containing nested category structure
        """
        return self._call("get_category_tree", method="GET")

    def get_product_reviews(self, product_id: str, page: int = 1) -> Dict[str, Any]:
        """
        Get customer reviews for a specific product.

        Args:
            product_id: Product SKU ID
            page: Page number for pagination

        Returns:
            Dictionary containing reviews list and page number
        """
        return self._call(
            "get_product_reviews",
            method="GET",
            product_id=product_id,
            page=page
        )

    def get_seller_info(self, seller_id: str) -> Dict[str, Any]:
        """
        Get information about a specific seller.

        Args:
            seller_id: Seller ID or slug

        Returns:
            Dictionary containing seller name and rating
        """
        return self._call(
            "get_seller_info",
            method="GET",
            seller_id=seller_id
        )

    def get_search_suggestions(self, query: str) -> Dict[str, Any]:
        """
        Get autocomplete suggestions for a search query.

        Args:
            query: Partial search term

        Returns:
            Dictionary containing suggestion list
        """
        return self._call(
            "get_search_suggestions",
            method="GET",
            query=query
        )


if __name__ == "__main__":
    # Initialize the client
    client = ParseClient()

    print("=" * 80)
    print("OZON.KZ CATALOG API - PRACTICAL USAGE EXAMPLE")
    print("=" * 80)

    # Step 1: Get search suggestions
    print("\n📝 Step 1: Getting search suggestions for 'iphone'...")
    suggestions_response = client.get_search_suggestions("iphone")
    if suggestions_response.get("suggestions"):
        suggestions = [s["text"] for s in suggestions_response["suggestions"]]
        print(f"   Found suggestions: {suggestions[:3]}")  # Show first 3
    
    # Step 2: Search for products
    search_query = "iPhone 15"
    print(f"\n🔍 Step 2: Searching for '{search_query}'...")
    search_results = client.search_products(query=search_query, sort="rating", page=1)
    
    if search_results.get("products"):
        products = search_results["products"]
        print(f"   Found {len(products)} products on page {search_results['page']}")
        
        # Step 3: Get details for the top 2 products and their reviews
        print(f"\n📦 Step 3: Getting details and reviews for top products...")
        for idx, product in enumerate(products[:2], 1):
            product_id = product["id"]
            product_name = product["name"]
            
            print(f"\n   Product {idx}: {product_name}")
            print(f"   Price: {product['price']} (Original: {product.get('original_price', 'N/A')})")
            print(f"   Discount: {product.get('discount', 'None')}")
            print(f"   Rating: {product['rating']} ⭐ ({product['reviews_count']} reviews)")
            
            # Get detailed information
            try:
                details = client.get_product_details(product_id)
                print(f"   Description: {details.get('description', 'N/A')[:100]}...")
                
                # Get seller information
                if details.get("seller"):
                    seller = details["seller"]
                    seller_info = client.get_seller_info(seller["id"])
                    print(f"   Seller: {seller['name']} (Rating: {seller_info.get('rating', 'N/A')})")
                
                # Get product reviews
                reviews_response = client.get_product_reviews(product_id, page=1)
                if reviews_response.get("reviews"):
                    reviews = reviews_response["reviews"]
                    print(f"   Recent reviews ({len(reviews)} shown):")
                    for review in reviews[:2]:
                        print(f"      - {review['author']}: {review['content'][:60]}... ({review['score']}⭐)")
            except Exception as e:
                print(f"   Error getting details: {e}")
    
    # Step 4: Browse a category
    print(f"\n🏷️  Step 4: Browsing Electronics category...")
    try:
        category_tree = client.get_category_tree()
        if category_tree.get("data"):
            # Find Electronics category
            for category in category_tree["data"][:2]:
                print(f"   Found category: {category['name']} (ID: {category['id']})")
                if category.get("subcategories"):
                    for subcat in category["subcategories"][:2]:
                        print(f"      └─ {subcat['name']}")
    except Exception as e:
        print(f"   Error getting category tree: {e}")
    
    print("\n" + "=" * 80)
    print("✅ Example workflow completed!")
    print("=" * 80)