MegaMarket APImegamarket.ru ↗

"""
MegaMarket.ru Product API Parse Client

Search and browse products on MegaMarket.ru (Sber's Russian marketplace).
Includes product search, details, reviews, catalog categories, and search suggestions.

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

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


class ParseClient:
    """Client for MegaMarket.ru Product API via Parse.bot"""

    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the Parse API client.

        Args:
            api_key: API key from parse.bot. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "c32d9ff8-1dc2-4c8d-920b-a69bb80b66f7"
        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 env var or pass api_key parameter."
            )

    def _call(
        self, endpoint: str, method: str = "POST", **params
    ) -> Dict[str, Any]:
        """
        Make a request to the Parse API.

        Args:
            endpoint: API endpoint name
            method: HTTP method (GET or POST)
            **params: Endpoint 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.upper() == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method.upper() == "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,
        limit: int = 20,
        offset: int = 0,
        sorting: int = 0,
        location_id: str = "50",
    ) -> Dict[str, Any]:
        """
        Search products by text query with pagination and sorting.

        Args:
            query: Search text query
            limit: Number of results per page (max 44)
            offset: Pagination offset
            sorting: Sort order (0=relevance, 1=price_asc, 2=price_desc, 3=rating, 4=discount)
            location_id: Region ID (50=Moscow)

        Returns:
            Search results with total count and product items
        """
        return self._call(
            "search_products",
            method="GET",
            query=query,
            limit=limit,
            offset=offset,
            sorting=sorting,
            location_id=location_id,
        )

    def get_product_details(
        self,
        goods_id: str,
        merchant_id: Optional[str] = None,
        location_id: str = "50",
    ) -> Dict[str, Any]:
        """
        Get detailed product information including all attributes/specs.

        Args:
            goods_id: Product goods ID (e.g. '700001132179_254730')
            merchant_id: Merchant ID (auto-extracted from goods_id if contains underscore)
            location_id: Region ID

        Returns:
            Detailed product information with attributes, images, and description
        """
        params = {"goods_id": goods_id, "location_id": location_id}
        if merchant_id:
            params["merchant_id"] = merchant_id

        return self._call("get_product_details", method="GET", **params)

    def get_product_reviews(
        self,
        goods_id: str,
        limit: int = 20,
        offset: int = 0,
        sort_type: str = "HELPFULNESS",
        sort_direction: str = "DESC",
        location_id: str = "50",
    ) -> Dict[str, Any]:
        """
        Get product reviews with pagination and sorting.

        Args:
            goods_id: Product goods ID
            limit: Number of reviews per page (max 50)
            offset: Pagination offset
            sort_type: Sort by HELPFULNESS or DATE
            sort_direction: Sort direction (ASC or DESC)
            location_id: Region ID

        Returns:
            Reviews with ratings, text, and helpfulness metrics
        """
        return self._call(
            "get_product_reviews",
            method="GET",
            goods_id=goods_id,
            limit=limit,
            offset=offset,
            sort_type=sort_type,
            sort_direction=sort_direction,
            location_id=location_id,
        )

    def get_catalog_menu(
        self,
        depth_level: int = 4,
        depth_level_from: int = 0,
        parent_id: str = "0",
        location_id: str = "50",
    ) -> Dict[str, Any]:
        """
        Get the catalog category tree/menu hierarchy.

        Args:
            depth_level: Max depth level of category tree
            depth_level_from: Starting depth level
            parent_id: Parent category ID (0 for root)
            location_id: Region ID

        Returns:
            Category tree with hierarchical structure
        """
        return self._call(
            "get_catalog_menu",
            method="GET",
            depth_level=depth_level,
            depth_level_from=depth_level_from,
            parent_id=parent_id,
            location_id=location_id,
        )

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

        Args:
            query: Search text for suggestions
            location_id: Region ID

        Returns:
            Suggested search terms and their parameters
        """
        return self._call(
            "search_suggest", method="GET", query=query, location_id=location_id
        )


def main():
    """Practical usage example: Search for iPhones and analyze top results."""

    # Initialize client
    client = ParseClient()

    print("=" * 70)
    print("MegaMarket.ru Product Search & Analysis")
    print("=" * 70)

    # Step 1: Get search suggestions
    print("\n[1] Getting search suggestions for 'iphone'...")
    suggestions = client.search_suggest(query="iphone")
    if suggestions.get("suggests"):
        top_suggestion = suggestions["suggests"][0]
        print(f"    Top suggestion: {top_suggestion['text']}")

    # Step 2: Search for products
    print("\n[2] Searching for iPhones...")
    search_results = client.search_products(
        query="iphone", limit=5, sorting=3  # Sort by rating
    )
    print(f"    Found {search_results['total']} products total")
    print(f"    Showing {len(search_results['items'])} results (sorted by rating)")

    # Step 3: Process each product
    print("\n[3] Analyzing top products...")
    for idx, product in enumerate(search_results["items"], 1):
        goods_id = product["goods_id"]
        title = product["title"]
        price = product["price"]
        final_price = product["final_price"]
        discount = int((1 - final_price / price) * 100) if price > 0 else 0
        rating = product.get("rating", "N/A")
        review_count = product.get("review_count", 0)
        available = product.get("is_available", False)

        print(f"\n    Product #{idx}: {title}")
        print(f"      Price: {price} ₽ → {final_price} ₽ ({discount}% off)")
        print(f"      Rating: {rating}/5 ({review_count} reviews)")
        print(f"      Available: {'Yes' if available else 'No'}")
        print(f"      Merchant: {product.get('merchant_name', 'N/A')}")

        # Step 4: Get detailed information for top 2 products
        if idx <= 2:
            print(f"      → Fetching detailed specs...")
            try:
                details = client.get_product_details(goods_id=goods_id)
                attributes = details.get("attributes", {})
                if attributes:
                    print(f"        Key specs:")
                    for attr_name, attr_value in list(attributes.items())[:3]:
                        print(f"          • {attr_name}: {attr_value}")
            except Exception as e:
                print(f"        (Could not fetch details: {e})")

            # Step 5: Get reviews for top product
            if idx == 1:
                print(f"      → Fetching recent reviews...")
                try:
                    reviews_data = client.get_product_reviews(
                        goods_id=goods_id, limit=3, sort_type="DATE", sort_direction="DESC"
                    )
                    reviews = reviews_data.get("reviews", [])
                    print(f"        Found {reviews_data.get('total', 0)} reviews total")
                    for review in reviews[:2]:
                        rating_review = review.get("rating", "N/A")
                        body = review.get("body", "No text")[:60]
                        likes = review.get("likes", 0)
                        print(f"          ⭐ {rating_review}: {body}... ({likes} likes)")
                except Exception as e:
                    print(f"        (Could not fetch reviews: {e})")

    # Step 6: Browse catalog
    print("\n[4] Browsing catalog categories...")
    try:
        catalog = client.get_catalog_menu(depth_level=2)
        categories = catalog.get("categories", [])
        print(f"    Top categories:")
        for category in categories[:5]:
            title = category.get("title", "Unknown")
            print(f"      • {title}")
    except Exception as e:
        print(f"    (Could not fetch catalog: {e})")

    print("\n" + "=" * 70)
    print("Analysis complete!")
    print("=" * 70)


if __name__ == "__main__":
    main()