5ka API5ka.ru ↗

"""
Pyaterochka (5ka.ru) API Client

This module provides a Python client for the Parse API to interact with
Pyaterochka (5ka.ru) product catalog, categories, pricing, and store information.

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

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


class ParseClient:
    """Client for interacting with Pyaterochka 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 env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "ae4a3593-8723-4991-8373-5de87af48eed"
        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 it directly."
            )

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

        Args:
            endpoint: The API endpoint name.
            method: HTTP method (GET or POST).
            **params: Additional parameters for the request.

        Returns:
            Response data as a 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 get_categories(self) -> Dict[str, Any]:
        """
        Fetch all top-level product categories.

        Returns:
            Dictionary containing categories with 'count' and 'results' fields.
        """
        return self._call("get_categories", method="GET")

    def get_products(
        self,
        query: Optional[str] = None,
        store_id: Optional[str] = None,
        page: int = 1,
        limit: int = 20,
        category_code: Optional[str] = None,
        order_by: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Fetch a paginated list of products with optional filters.

        Args:
            query: Search keyword.
            store_id: Filter by store ID (SAP code).
            page: Page number (default: 1).
            limit: Number of items per page (default: 20).
            category_code: Filter by category ID/code.
            order_by: Field to sort by (e.g., 'price', '-price').

        Returns:
            Dictionary containing products with pagination info and 'results' field.
        """
        params = {
            "page": page,
            "limit": limit,
        }
        if query:
            params["query"] = query
        if store_id:
            params["store_id"] = store_id
        if category_code:
            params["category_code"] = category_code
        if order_by:
            params["order_by"] = order_by

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

    def get_product_detail(self, product_id: str) -> Dict[str, Any]:
        """
        Fetch full detail for a single product, including nutritional values.

        Args:
            product_id: Unique product ID (PLU).

        Returns:
            Dictionary containing product details including nutritional information.
        """
        return self._call("get_product_detail", method="GET", product_id=product_id)

    def get_special_offers(
        self, store_id: Optional[str] = None, page: int = 1, limit: int = 20
    ) -> Dict[str, Any]:
        """
        Fetch products currently on sale or promotion.

        Args:
            store_id: Store ID (SAP code).
            page: Page number (default: 1).
            limit: Number of items per page (default: 20).

        Returns:
            Dictionary containing promotional products with 'results' field.
        """
        params = {
            "page": page,
            "limit": limit,
        }
        if store_id:
            params["store_id"] = store_id

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

    def get_stores(self) -> Dict[str, Any]:
        """
        Fetch store locations and details.

        Returns:
            Dictionary containing store information with 'results' field.
        """
        return self._call("get_stores", method="GET")


def main():
    """Demonstrate practical usage of the ParseClient."""
    # Initialize the client
    client = ParseClient()

    print("=" * 60)
    print("Pyaterochka (5ka.ru) API Client - Practical Usage Example")
    print("=" * 60)

    # Step 1: Get all available stores
    print("\n1. Fetching available stores...")
    stores_data = client.get_stores()
    stores = stores_data.get("results", [])
    print(f"   Found {len(stores)} stores")

    if stores:
        first_store = stores[0]
        print(f"   Example store: {first_store.get('address')} (ID: {first_store.get('id')})")
        store_id = str(first_store.get("id"))
    else:
        store_id = None
        print("   No stores found, proceeding without store filter")

    # Step 2: Get product categories
    print("\n2. Fetching product categories...")
    categories_data = client.get_categories()
    categories = categories_data.get("results", [])
    print(f"   Found {categories_data.get('count', 0)} categories")

    if categories:
        # Show first few categories
        for i, category in enumerate(categories[:3]):
            print(f"   - {category.get('name')} (ID: {category.get('id')})")

    # Step 3: Get special offers/promotions
    print("\n3. Fetching special offers and promotions...")
    params = {"limit": 5}
    if store_id:
        params["store_id"] = store_id
    offers_data = client.get_special_offers(**params)
    offers = offers_data.get("results", [])
    print(f"   Found {len(offers)} promotional items")

    if offers:
        print("\n   Promotional Products:")
        for offer in offers:
            regular_price = offer.get("price_regular")
            promo_price = offer.get("price_promo")
            discount = (
                ((regular_price - promo_price) / regular_price * 100)
                if regular_price
                else 0
            )
            print(
                f"   - {offer.get('name')}: {promo_price} ₽ (was {regular_price} ₽, -{discount:.0f}%)"
            )

    # Step 4: Search for products (e.g., dairy products)
    print("\n4. Searching for 'milk' products...")
    search_params = {"query": "milk", "limit": 5}
    if store_id:
        search_params["store_id"] = store_id
    search_results = client.get_products(**search_params)
    products = search_results.get("results", [])
    print(f"   Found {search_results.get('count', 0)} products matching 'milk'")

    # Step 5: Get details for each product from search
    if products:
        print("\n5. Fetching detailed information for search results...")
        for i, product in enumerate(products[:3]):
            product_id = str(product.get("id"))
            print(f"\n   Product {i + 1}: {product.get('name')}")
            print(f"   - Price: {product.get('price')} ₽")

            # Get detailed information including nutritional data
            try:
                detail = client.get_product_detail(product_id)
                nutritional = detail.get("nutritional_value", {})

                if nutritional:
                    print(
                        f"   - Nutritional Info (per 100g):"
                    )
                    print(
                        f"     • Calories: {nutritional.get('calories')} kcal"
                    )
                    print(
                        f"     • Proteins: {nutritional.get('proteins')} g"
                    )
                    print(
                        f"     • Fats: {nutritional.get('fats')} g"
                    )
                    print(
                        f"     • Carbohydrates: {nutritional.get('carbohydrates')} g"
                    )
            except Exception as e:
                print(f"   - Could not fetch details: {e}")

    # Step 6: Browse products by category with sorting
    if categories:
        sample_category = categories[0]
        category_id = str(sample_category.get("id"))
        print(f"\n6. Browsing products in category '{sample_category.get('name')}'...")

        params = {"category_code": category_id, "limit": 5, "order_by": "price"}
        if store_id:
            params["store_id"] = store_id

        category_products = client.get_products(**params)
        category_results = category_products.get("results", [])

        if category_results:
            print(f"   Found {category_products.get('count', 0)} products in this category")
            print("   Top 5 cheapest products in category:")
            for i, product in enumerate(category_results[:5], 1):
                print(
                    f"   {i}. {product.get('name')} - {product.get('price')} ₽"
                )
        else:
            print("   No products found in this category")

    print("\n" + "=" * 60)
    print("API Usage Example Complete!")
    print("=" * 60)


if __name__ == "__main__":
    main()