Walmart APIwalmart.com.mx ↗

"""
Walmart Mexico API Client - Parse Bot Integration
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 the Walmart Mexico API via Parse Bot."""
    
    def __init__(self, api_key: Optional[str] = None):
        """Initialize the Parse API client.
        
        Args:
            api_key: API key for Parse Bot. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "ea0bc2df-b753-4a65-9458-80d8167fb00d"
        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 a request to the Parse API.
        
        Args:
            endpoint: The endpoint name (e.g., 'search_products')
            method: HTTP method (GET or POST)
            **params: Parameters to pass to the endpoint
            
        Returns:
            API response 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.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()
        except requests.exceptions.RequestException as e:
            print(f"API request failed: {e}")
            raise
    
    def search_products(
        self,
        query: str,
        page: int = 1,
        sort: str = "best_match",
        cat_id: Optional[str] = None,
        min_price: Optional[str] = None,
        max_price: Optional[str] = None,
        facet: Optional[str] = None
    ) -> Dict[str, Any]:
        """Search for products on walmart.com.mx.
        
        Args:
            query: Search keyword (required)
            page: Page number, 1-indexed (default: 1)
            sort: Sort order - best_match, price_low, price_high (default: best_match)
            cat_id: Category ID to filter results
            min_price: Minimum price filter
            max_price: Maximum price filter
            facet: Additional filter facets
            
        Returns:
            Search results with item stacks
        """
        params = {
            "query": query,
            "page": page,
            "sort": sort,
        }
        
        if cat_id:
            params["cat_id"] = cat_id
        if min_price:
            params["min_price"] = min_price
        if max_price:
            params["max_price"] = max_price
        if facet:
            params["facet"] = facet
        
        return self._call("search_products", method="GET", **params)
    
    def get_product_details(self, product_id: str) -> Dict[str, Any]:
        """Get full product details including pricing and availability.
        
        Args:
            product_id: The Walmart product ID
            
        Returns:
            Product details including name, brand, price, and availability
        """
        return self._call("get_product_details", method="GET", product_id=product_id)
    
    def list_categories(self) -> Dict[str, Any]:
        """List top-level departments and navigation links.
        
        Returns:
            List of categories with display names and URLs
        """
        return self._call("list_categories", method="GET")
    
    def get_similar_products(self, product_id: str) -> Dict[str, Any]:
        """Get similar and recommended products for a given item.
        
        Args:
            product_id: The Walmart product ID
            
        Returns:
            Carousels with similar products
        """
        return self._call("get_similar_products", method="GET", product_id=product_id)


def extract_products_from_search(search_result: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Extract product items from search response.
    
    Args:
        search_result: The searchResult object from search response
        
    Returns:
        List of product items
    """
    products = []
    
    item_stacks = search_result.get("itemStacks", [])
    for stack in item_stacks:
        items = stack.get("items", [])
        products.extend(items)
    
    return products


def format_price(price: float) -> str:
    """Format price for display.
    
    Args:
        price: Price value
        
    Returns:
        Formatted price string
    """
    return f"${price:.2f}" if isinstance(price, (int, float)) else "N/A"


def print_product_summary(product: Dict[str, Any]) -> None:
    """Print a summary of a product.
    
    Args:
        product: Product dictionary from search results
    """
    name = product.get("name", "Unknown")
    brand = product.get("brand", "Unknown")
    price = product.get("price", "N/A")
    rating = product.get("rating", {})
    avg_rating = rating.get("averageRating", "N/A")
    num_reviews = rating.get("numberOfReviews", 0)
    availability = product.get("availabilityStatusV2", {}).get("display", "Unknown")
    
    print(f"  • {name}")
    print(f"    Brand: {brand} | Price: {format_price(price)}")
    print(f"    Rating: {avg_rating} ⭐ ({num_reviews} reviews) | {availability}")


def main():
    """Practical workflow demonstrating the Walmart Mexico API."""
    
    # Initialize the client
    client = ParseClient()
    
    print("=" * 80)
    print("Walmart Mexico Product Search & Analysis Workflow")
    print("=" * 80)
    
    # Step 1: Explore categories
    print("\n[Step 1] Discovering available product categories...")
    try:
        categories_response = client.list_categories()
        categories = categories_response.get("data", {}).get("categories", [])
        print(f"✓ Found {len(categories)} main categories")
        
        if categories:
            print("\n  Available categories:")
            for i, cat in enumerate(categories[:5], 1):
                cat_name = cat.get("name", {}).get("title", "Unknown")
                subcats = cat.get("subcategories", [])
                print(f"    {i}. {cat_name} ({len(subcats)} subcategories)")
    except Exception as e:
        print(f"✗ Error fetching categories: {e}")
        return
    
    # Step 2: Search for products
    search_query = "leche"
    print(f"\n[Step 2] Searching for '{search_query}' products...")
    
    try:
        search_response = client.search_products(
            query=search_query,
            page=1,
            sort="best_match"
        )
        
        search_result = search_response.get("data", {}).get("searchResult", {})
        total_count = search_result.get("aggregatedCount", 0)
        products = extract_products_from_search(search_result)
        max_pages = search_result.get("paginationV2", {}).get("maxPage", 1)
        
        print(f"✓ Search found {total_count} total results ({max_pages} pages)")
        print(f"  Showing {len(products)} products on page 1:\n")
        
        for product in products[:3]:
            print_product_summary(product)
        
        if not products:
            print("  No products found.")
            return
        
        # Step 3: Get detailed information for top products
        print(f"\n[Step 3] Fetching detailed information for top {min(3, len(products))} products...")
        
        selected_products = []
        for idx, product in enumerate(products[:3], 1):
            product_id = product.get("id")
            product_name = product.get("name", "Unknown")
            
            if not product_id:
                continue
            
            print(f"\n  Fetching details for: {product_name}...")
            
            try:
                details = client.get_product_details(product_id)
                product_data = details.get("data", {})
                
                selected_products.append({
                    "id": product_id,
                    "name": product_name,
                    "details": product_data
                })
                
                brand = product_data.get("brand", "Unknown")
                price_info = product_data.get("priceInfo", {})
                current_price = price_info.get("currentPrice", {})
                price_value = current_price.get("price", "N/A")
                availability = product_data.get("availabilityStatus", "Unknown")
                short_desc = product_data.get("shortDescription", "")
                
                print(f"    ✓ Brand: {brand}")
                print(f"    ✓ Price: {format_price(price_value)}")
                print(f"    ✓ Availability: {availability}")
                
                if short_desc:
                    desc_preview = short_desc[:80] + "..." if len(short_desc) > 80 else short_desc
                    print(f"    ✓ Description: {desc_preview}")
            
            except Exception as e:
                print(f"    ✗ Error fetching details: {e}")
        
        # Step 4: Compare prices across similar products
        print(f"\n[Step 4] Comparing prices for similar products...")
        
        if selected_products:
            first_product = selected_products[0]
            print(f"\n  Looking for alternatives to: {first_product['name']}")
            
            try:
                similar_response = client.get_similar_products(first_product["id"])
                
                # Parse similar products data
                similar_data = similar_response.get("data", {})
                if isinstance(similar_data, str):
                    print("    (Similar products data available but in complex format)")
                else:
                    carousels = similar_data.get("carousels", []) if isinstance(similar_data, dict) else []
                    
                    if carousels:
                        carousel = carousels[0]
                        carousel_title = carousel.get("title", "Similar Products")
                        similar_products = carousel.get("products", [])
                        
                        print(f"    ✓ {carousel_title}: {len(similar_products)} items")
                        
                        price_list = []
                        for similar in similar_products[:5]:
                            similar_name = similar.get("name", "Unknown")
                            similar_price_info = similar.get("priceInfo", {})
                            similar_price = similar_price_info.get("currentPrice", {})
                            price_val = similar_price.get("price", "N/A")
                            
                            if isinstance(price_val, (int, float)):
                                price_list.append((similar_name, price_val))
                        
                        if price_list:
                            price_list.sort(key=lambda x: x[1])
                            print("\n      Cheapest alternatives:")
                            for name, price in price_list[:3]:
                                print(f"        • {name}: {format_price(price)}")
                    else:
                        print("    No similar products carousel found")
            
            except Exception as e:
                print(f"    ✗ Error finding similar products: {e}")
        
        # Step 5: Search with price filters
        print(f"\n[Step 5] Searching for budget options (max $50)...")
        
        try:
            budget_response = client.search_products(
                query=search_query,
                page=1,
                sort="price_low",
                max_price="50"
            )
            
            budget_result = budget_response.get("data", {}).get("searchResult", {})
            budget_products = extract_products_from_search(budget_result)
            
            print(f"✓ Found {len(budget_products)} products under $50 (sorted by price):\n")
            
            for product in budget_products[:5]:
                price = product.get("price", "N/A")
                name = product.get("name", "Unknown")
                availability = product.get("availabilityStatusV2", {}).get("display", "Unknown")
                print(f"  • {name}")
                print(f"    Price: {format_price(price)} | {availability}\n")
        
        except Exception as e:
            print(f"✗ Error with budget search: {e}")
    
    except Exception as e:
        print(f"✗ Error during workflow: {e}")
    
    print("=" * 80)
    print("Workflow completed successfully!")
    print("=" * 80)


if __name__ == "__main__":
    main()