""" Walmart.com scraper Note: Walmart has aggressive bot protection (PerimeterX). May need stealth browser mode for sustained use. """ import re import json import logging from typing import List, Optional from bs4 import BeautifulSoup from .base import BaseScraper, Product from src.browser import get_browser logger = logging.getLogger(__name__) class WalmartScraper(BaseScraper): """Scraper for Walmart.com""" site_name = "walmart" base_url = "https://www.walmart.com" def scrape_category_page(self, url: str, max_pages: int = 1) -> List[Product]: """ Scrape a Walmart search/category page for all products Args: url: Search/category page URL max_pages: Maximum pages to scrape (not yet implemented for Walmart) Returns: List of Product objects """ # TODO: Implement pagination for Walmart when needed browser = get_browser() products = [] try: page, html = browser.get_page_content( url, wait_for_selector=None, timeout=60000, ) # Save debug screenshot try: page.screenshot(path="debug_walmart.png") logger.info("Saved debug screenshot to debug_walmart.png") except: pass soup = BeautifulSoup(html, "html.parser") # Try to extract from __NEXT_DATA__ JSON (Walmart uses Next.js) next_data = soup.select_one("script#__NEXT_DATA__") if next_data: try: data = json.loads(next_data.string) products.extend(self._extract_products_from_next_data(data)) except (json.JSONDecodeError, TypeError) as e: logger.debug(f"Error parsing __NEXT_DATA__: {e}") # Also try application/json scripts scripts = soup.find_all("script", type="application/json") for script in scripts: try: data = json.loads(script.string) products.extend(self._extract_products_from_json(data)) except (json.JSONDecodeError, TypeError): continue # Try multiple selectors for product cards product_cards = ( soup.select("[data-item-id]") or soup.select(".search-result-gridview-item") or soup.select("[data-testid='list-view']") or soup.select("[class*='product-item']") or soup.select("[class*='ProductCard']") ) logger.info(f"Found {len(product_cards)} product cards") for card in product_cards: product = self._parse_product_card(card) if product: products.append(product) # Fallback: parse product links if not product_cards and not products: product_links = soup.select("a[href*='/ip/']") logger.info(f"Fallback: Found {len(product_links)} product links") href_to_links = {} for link in product_links: href = link.get("href", "") if not href or "/ip/" not in href: continue if href not in href_to_links: href_to_links[href] = [] href_to_links[href].append(link) for href, links in href_to_links.items(): best_link = links[0] for link in links: if link.get("aria-label") or len(link.get_text(strip=True)) > 10: best_link = link break product = self._parse_product_link(best_link, soup) if product: products.append(product) page.close() except Exception as e: logger.error(f"Error scraping Walmart category page: {e}") raise # Remove duplicates seen_urls = set() unique_products = [] for p in products: if p.url not in seen_urls: seen_urls.add(p.url) unique_products.append(p) logger.info(f"Scraped {len(unique_products)} unique products from Walmart") return unique_products def _extract_products_from_next_data(self, data: dict) -> List[Product]: """Extract products from Next.js __NEXT_DATA__ JSON""" products = [] try: # Navigate to search results in Next.js data structure props = data.get("props", {}) page_props = props.get("pageProps", {}) initial_data = page_props.get("initialData", {}) search_result = initial_data.get("searchResult", {}) item_stacks = search_result.get("itemStacks", []) for stack in item_stacks: items = stack.get("items", []) for item in items: product = self._parse_walmart_item(item) if product: products.append(product) except Exception as e: logger.debug(f"Error extracting from __NEXT_DATA__: {e}") return products def _extract_products_from_json(self, data, depth=0) -> List[Product]: """Recursively search JSON for product data""" products = [] if depth > 10: return products if isinstance(data, dict): # Check if this looks like a Walmart product if "usItemId" in data or ("name" in data and "priceInfo" in data): product = self._parse_walmart_item(data) if product: products.append(product) for value in data.values(): products.extend(self._extract_products_from_json(value, depth + 1)) elif isinstance(data, list): for item in data: products.extend(self._extract_products_from_json(item, depth + 1)) return products def _parse_walmart_item(self, data: dict) -> Optional[Product]: """Parse a product from Walmart's JSON data""" try: name = data.get("name") or data.get("title", "") if not name: return None item_id = data.get("usItemId") or data.get("id", "") canonical_url = data.get("canonicalUrl") or data.get("productPageUrl", "") if canonical_url: url = canonical_url if canonical_url.startswith("http") else f"{self.base_url}{canonical_url}" elif item_id: url = f"{self.base_url}/ip/{item_id}" else: return None # Get price price = None price_info = data.get("priceInfo", {}) if isinstance(price_info, dict): current_price = price_info.get("currentPrice", {}) if isinstance(current_price, dict): price = current_price.get("priceString") elif price_info.get("priceString"): price = price_info.get("priceString") if not price and "price" in data: price_val = data.get("price") if isinstance(price_val, (int, float)): price = f"${price_val:.2f}" # Check availability in_stock = True availability = data.get("availabilityStatusV2", {}) if isinstance(availability, dict): status = availability.get("value", "").upper() in_stock = status not in ["OUT_OF_STOCK", "NOT_AVAILABLE"] elif data.get("availabilityStatus"): in_stock = data.get("availabilityStatus") != "OUT_OF_STOCK" # Get image image_url = None image_info = data.get("imageInfo", {}) if isinstance(image_info, dict): image_url = image_info.get("thumbnailUrl") or image_info.get("url") elif data.get("image"): image_url = data.get("image") return Product( name=name, url=url, price=price, in_stock=in_stock, image_url=image_url, site=self.site_name, product_id=str(item_id), ) except Exception as e: logger.debug(f"Error parsing Walmart product JSON: {e}") return None def _parse_product_card(self, card) -> Optional[Product]: """Parse a product card element""" try: # Find link link = card.select_one("a[href*='/ip/']") or card.select_one("a") if not link: return None href = link.get("href", "") if not href: return None url = href if href.startswith("http") else f"{self.base_url}{href}" # Get name name_elem = ( card.select_one("[data-automation-id='product-title']") or card.select_one(".product-title-link span") or card.select_one("[class*='ProductTitle']") or card.select_one("[class*='product-title']") or link ) name = name_elem.get_text(strip=True) if name_elem else "Unknown" # Skip if name too short if len(name) < 5: return None # Get price price_elem = ( card.select_one("[data-automation-id='product-price']") or card.select_one(".price-current") or card.select_one("[class*='ProductPrice']") or card.select_one("[class*='price']") ) price = None if price_elem: price_match = re.search(r"\$[\d,]+\.?\d*", price_elem.get_text()) if price_match: price = price_match.group() # Check stock in_stock = self._check_card_stock_status(card) # Get image img = card.select_one("img") image_url = None if img: image_url = img.get("src") or img.get("data-src") if image_url and not image_url.startswith("http"): image_url = f"https:{image_url}" if image_url.startswith("//") else f"{self.base_url}{image_url}" # Extract item ID from URL product_id = "" match = re.search(r"/ip/[^/]+/(\d+)", url) or re.search(r"/ip/(\d+)", url) if match: product_id = match.group(1) return Product( name=name, url=url, price=price, in_stock=in_stock, image_url=image_url, site=self.site_name, product_id=product_id, ) except Exception as e: logger.debug(f"Error parsing Walmart product card: {e}") return None def _parse_product_link(self, link, soup) -> Optional[Product]: """Parse a product from a product link element""" try: href = link.get("href", "") if not href: return None url = href if href.startswith("http") else f"{self.base_url}{href}" name = link.get("aria-label", "") or link.get_text(strip=True) if not name or len(name) < 5: return None # Extract item ID product_id = "" match = re.search(r"/ip/[^/]+/(\d+)", url) or re.search(r"/ip/(\d+)", url) if match: product_id = match.group(1) # Find price near link parent = link.find_parent() price = None for _ in range(5): if parent: price_text = parent.get_text() price_match = re.search(r"\$[\d,]+\.?\d*", price_text) if price_match: price = price_match.group() break parent = parent.find_parent() # Find image image_url = None parent = link.find_parent() for _ in range(5): if parent: img = parent.select_one("img") if img: image_url = img.get("src") or img.get("data-src") break parent = parent.find_parent() # Check stock in_stock = True parent = link.find_parent() for _ in range(5): if parent: text = parent.get_text().lower() if any(phrase in text for phrase in ["out of stock", "unavailable", "not available"]): in_stock = False break parent = parent.find_parent() return Product( name=name, url=url, price=price, in_stock=in_stock, image_url=image_url, site=self.site_name, product_id=product_id, ) except Exception as e: logger.debug(f"Error parsing Walmart product link: {e}") return None def _check_card_stock_status(self, card) -> bool: """Check if a product card indicates in-stock status""" card_text = card.get_text().lower() if hasattr(card, "get_text") else str(card).lower() out_of_stock_phrases = [ "out of stock", "unavailable", "not available", "sold out", "pickup not available", ] for phrase in out_of_stock_phrases: if phrase in card_text: return False in_stock_phrases = [ "add to cart", "available", "in stock", "pickup available", "delivery available", ] for phrase in in_stock_phrases: if phrase in card_text: return True return True def check_product_stock(self, product_url: str) -> tuple[bool, Optional[str]]: """Check if a specific product is in stock""" browser = get_browser() try: page, html = browser.get_page_content(product_url, timeout=30000) soup = BeautifulSoup(html, "html.parser") page_text = soup.get_text().lower() out_of_stock = any( phrase in page_text for phrase in ["out of stock", "unavailable", "not available", "sold out"] ) in_stock = not out_of_stock # Get price price = None price_elem = ( soup.select_one("[data-automation-id='product-price']") or soup.select_one("[itemprop='price']") or soup.select_one("[class*='price']") ) if price_elem: price_match = re.search(r"\$[\d,]+\.?\d*", price_elem.get_text()) if price_match: price = price_match.group() page.close() return in_stock, price except Exception as e: logger.error(f"Error checking Walmart product stock: {e}") return False, None