Com APIlotuss.com.my ↗

"""
Lotus's Malaysia Promotional Products API Client

This module provides a Python client for extracting product data from Lotus's Malaysia
promotional and category pages, including prices, discounts, and product information
with pagination support.

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 Parse's Lotus Malaysia promotional products 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 = "7e4ff86c-8b57-45db-969f-17ca1b7035d4"
        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 api_key parameter."
            )

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

        Args:
            endpoint: The endpoint name (e.g., 'get_promotional_products')
            method: HTTP method ('GET' or 'POST')
            **params: Query/body parameters for the request

        Returns:
            JSON response from the API

        Raises:
            requests.RequestException: If the API call fails
        """
        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, 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()

    def get_promotional_products(
        self,
        category_id: str = "60725",
        page: int = 1,
        limit: int = 15,
        sort: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Get products from a Lotus's Malaysia promotional or category page.

        Args:
            category_id: Category ID for the promotional page. Default is 60725 (June Super Weekend Treats).
            page: Page number for pagination (starts at 1).
            limit: Number of products per page (1-100).
            sort: Sort order - one of: best_match, price_low, price_high, newest. None for default ordering.

        Returns:
            Dictionary containing products list and pagination info
        """
        params = {
            "category_id": category_id,
            "page": page,
            "limit": limit,
        }

        if sort:
            params["sort"] = sort

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


def main():
    """Demonstrate practical usage of the Lotus Malaysia promotional products API."""

    try:
        # Initialize the client
        client = ParseClient()
        print("✓ API client initialized successfully\n")

        # Use case: Find discounted coconut products across promotional pages
        print("=" * 70)
        print("PRACTICAL USE CASE: Finding Best Discounted Products")
        print("=" * 70)

        # Step 1: Get initial page of promotional products
        print("\n📦 Step 1: Fetching promotional products (first page)...\n")
        response = client.get_promotional_products(
            category_id="60725", page=1, limit=10, sort="price_low"
        )

        products = response.get("products", [])
        pagination = response.get("pagination", {})

        print(f"Found {pagination.get('total_products')} total products")
        print(f"Displaying page {pagination.get('page')} of {pagination.get('total_pages')}\n")

        # Step 2: Analyze and display products with significant discounts
        print("💰 Best Deals (discount > 10%):")
        print("-" * 70)

        best_deals = []
        for product in products:
            if product.get("discount_percent", 0) > 10:
                best_deals.append(product)

        if best_deals:
            for idx, product in enumerate(best_deals, 1):
                print(f"\n{idx}. {product.get('name')}")
                print(f"   Brand: {product.get('brand')}")
                print(
                    f"   Regular Price: MYR {product.get('regular_price'):.2f} → Promo: MYR {product.get('promotional_price'):.2f}"
                )
                print(f"   Discount: {product.get('discount_percent'):.2f}%")
                print(f"   Category: {product.get('category')}")
                print(f"   Stock: {product.get('stock_status')}")
        else:
            print("No products with discount > 10% on this page")

        # Step 3: Get products sorted by price
        print("\n\n" + "=" * 70)
        print("📊 Step 2: Fetching products sorted by lowest price...")
        print("=" * 70 + "\n")

        price_sorted_response = client.get_promotional_products(
            category_id="60725", page=1, limit=5, sort="price_low"
        )

        price_products = price_sorted_response.get("products", [])

        print("Lowest Priced Products:")
        print("-" * 70)

        for idx, product in enumerate(price_products, 1):
            savings = product.get("regular_price", 0) - product.get(
                "promotional_price", 0
            )
            print(f"\n{idx}. {product.get('name')}")
            print(f"   SKU: {product.get('sku')}")
            print(f"   Promotional Price: MYR {product.get('promotional_price'):.2f}")
            print(f"   You Save: MYR {savings:.2f}")

        # Step 4: Summary statistics
        print("\n\n" + "=" * 70)
        print("📈 Summary Statistics")
        print("=" * 70 + "\n")

        all_products = response.get("products", [])
        if all_products:
            avg_discount = sum(p.get("discount_percent", 0) for p in all_products) / len(
                all_products
            )
            avg_price = sum(p.get("promotional_price", 0) for p in all_products) / len(
                all_products
            )
            in_stock = sum(
                1
                for p in all_products
                if p.get("stock_status") == "IN_STOCK"
            )

            print(f"Average Discount: {avg_discount:.2f}%")
            print(f"Average Promotional Price: MYR {avg_price:.2f}")
            print(f"Products In Stock: {in_stock}/{len(all_products)}")

        print("\n✓ Example completed successfully!")

    except requests.exceptions.RequestException as e:
        print(f"❌ API Error: {e}")
    except ValueError as e:
        print(f"❌ Configuration Error: {e}")


if __name__ == "__main__":
    main()