""" Best Buy scraper - Uses undetected-chromedriver to bypass bot protection """ import re import json import logging import time import sys import os from typing import List, Optional from bs4 import BeautifulSoup from .base import BaseScraper, Product # Add parent directory to path for tools import sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) logger = logging.getLogger(__name__) class BestBuyScraper(BaseScraper): """Scraper for BestBuy.com - Uses undetected-chromedriver""" site_name = "bestbuy" base_url = "https://www.bestbuy.com" def __init__(self): self._stealth_browser = None self._restart_attempts = 0 self._max_restart_attempts = 3 def _get_stealth_browser(self): """Get or create stealth browser for Best Buy, with auto-restart on session errors""" from tools.stealth_browser import StealthBrowser # Check if we need to create a new browser if self._stealth_browser is None: logger.info("Creating new Best Buy stealth browser...") self._stealth_browser = StealthBrowser(headless=False, session_name="bestbuy") self._stealth_browser.start() self._restart_attempts = 0 return self._stealth_browser # Check if existing session is still valid if not self._stealth_browser.is_session_valid(): if self._restart_attempts >= self._max_restart_attempts: logger.error(f"Best Buy browser failed after {self._max_restart_attempts} restart attempts") # Reset counter and try one more time after clearing self._restart_attempts = 0 self._stealth_browser = None return self._get_stealth_browser() logger.warning("Best Buy browser session invalid, restarting...") self._restart_attempts += 1 self._stealth_browser.restart() return self._stealth_browser def _handle_session_error(self, error: Exception) -> bool: """ Check if error is a session error and handle it. Returns True if browser was restarted and operation should be retried. """ error_msg = str(error).lower() session_errors = ["invalid session id", "session deleted", "no such session", "browser has closed"] if any(err in error_msg for err in session_errors): logger.warning(f"Session error detected: {error}") if self._stealth_browser and self._restart_attempts < self._max_restart_attempts: self._restart_attempts += 1 try: self._stealth_browser.restart() return True # Retry operation except Exception as e: logger.error(f"Failed to restart browser: {e}") return False def scrape_category_page(self, url: str, max_pages: int = 1) -> List[Product]: """ Scrape a Best Buy search/category page for all products Uses undetected-chromedriver to bypass bot protection Args: url: Search/category page URL max_pages: Maximum number of pages to scrape (default 1) Returns: List of Product objects """ all_products = [] # Add sort by newest if not already in URL if "sort=" not in url: separator = "&" if "?" in url else "?" url = f"{url}{separator}sp=-releasedate" # Use stealth browser try: browser = self._get_stealth_browser() except Exception as e: logger.error(f"Failed to start stealth browser: {e}") return all_products for page_num in range(1, max_pages + 1): products = [] page_url = url if page_num == 1 else f"{url}&cp={page_num}" retry_count = 0 max_retries = 2 html = None while retry_count <= max_retries: try: logger.info(f"Navigating to: {page_url}") browser.driver.get(page_url) time.sleep(8) # Wait longer for initial page load # Scroll for lazy loading - wait longer between scrolls browser.driver.execute_script("window.scrollTo(0, 500)") time.sleep(2) browser.driver.execute_script("window.scrollTo(0, 1500)") time.sleep(2) browser.driver.execute_script("window.scrollTo(0, 3000)") time.sleep(2) browser.driver.execute_script("window.scrollTo(0, 0)") # Scroll back to top time.sleep(3) # Wait for products to load (check for content) for _ in range(10): html = browser.driver.page_source if "sku-item" in html or "sku-title" in html or "priceView" in html: logger.info("Product content detected") break time.sleep(1) # Get HTML after waiting html = browser.driver.page_source # Save debug screenshot try: browser.driver.save_screenshot("debug_bestbuy.png") logger.info("Saved debug screenshot to debug_bestbuy.png") except: pass break # Success, exit retry loop except Exception as e: if self._handle_session_error(e) and retry_count < max_retries: retry_count += 1 logger.info(f"Retrying after browser restart (attempt {retry_count}/{max_retries})") browser = self._get_stealth_browser() # Get restarted browser continue else: logger.error(f"Error navigating to {page_url}: {e}") return all_products else: # while loop completed without break (all retries exhausted) if html is None: logger.error(f"Failed to load page after {max_retries} retries") return all_products try: soup = BeautifulSoup(html, "html.parser") # Try to find product data in 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 - Best Buy updates these frequently product_cards = ( soup.select("li.product-list-item") # Current Best Buy format (2026) or soup.select("[data-testid='list-item']") # Modern React testid or soup.select("[data-testid='product-card']") # Alternate testid or soup.select("[class*='ProductCard']") # React component class or soup.select("li.sku-item") # Legacy or soup.select("[data-sku-id]") or soup.select(".sku-item") or soup.select("[class*='sku-item']") or soup.select("div.shop-sku-list-item") or soup.select("[class*='productCard']") or soup.select("[class*='product-card']") or soup.select(".list-item") or soup.select("[class*='listItem']") # camelCase variant or soup.select("article[class*='product']") # Semantic HTML ) logger.info(f"Found {len(product_cards)} product cards on page {page_num}") for card in product_cards: product = self._parse_product_card(card) if product: products.append(product) # Fallback 1: Try to extract from JS state / __NEXT_DATA__ (most reliable) if not product_cards or not products: logger.info("Trying JS state extraction...") js_products = self._extract_from_page_data(soup, browser) if js_products: logger.info(f"Extracted {len(js_products)} products from JS state") products.extend(js_products) # Fallback 2: Parse product links (resilient to DOM changes) if not products: # Try multiple link patterns - Best Buy uses different URL formats # New format: /product/pokemon-card-name/ABC123/sku/12345 # Old format: /site/product-name/12345.p product_links = ( soup.select("a[href*='/product/'][href*='/sku/']") # New format with /sku/ or soup.select("a[href*='/site/'][href*='.p']") # Legacy format or soup.select("a[href*='skuId=']") # URL param format ) logger.info(f"Fallback links: Found {len(product_links)} product links") href_to_links = {} for link in product_links: href = link.get("href", "") if not href: continue # Accept links with /sku/, .p suffix, or skuId parameter if "/sku/" not in href and ".p" not in href and "skuId=" 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) # Debug: Save HTML if no products found for analysis if not products and not product_cards: try: debug_path = "debug_bestbuy.html" with open(debug_path, "w", encoding="utf-8") as f: f.write(html) logger.warning(f"No products found - saved HTML to {debug_path} for debugging") except Exception as e: logger.debug(f"Could not save debug HTML: {e}") # Don't close - reuse browser for next page except Exception as e: logger.error(f"Error scraping Best Buy page {page_num}: {e}") break # Don't raise - just return what we have all_products.extend(products) # Stop if no products found on this page (no more pages) if not products: break # Remove duplicates seen_urls = set() unique_products = [] for p in all_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 Best Buy") return unique_products def _extract_from_page_data(self, soup, browser) -> List[Product]: """Extract products from page data attributes and evaluate JS if needed""" products = [] # Method 1: Parse Apollo SSR data from script tags (Best Buy's current format) # Best Buy uses window[Symbol.for("ApolloSSRDataTransport")] format for script in soup.find_all("script"): if script.string and "ApolloSSRDataTransport" in script.string: logger.info("Found Apollo SSR data in script tag") products.extend(self._extract_from_apollo_data(script.string)) if products: break # Method 2: Try to parse __NEXT_DATA__ script tag (older format) if not products: next_data_script = soup.select_one("script#__NEXT_DATA__") if next_data_script and next_data_script.string: try: data = json.loads(next_data_script.string) logger.info("Found __NEXT_DATA__ script tag in HTML") products.extend(self._extract_products_from_json(data)) except (json.JSONDecodeError, TypeError) as e: logger.debug(f"Could not parse __NEXT_DATA__ from HTML: {e}") # Method 3: Try multiple JS state sources via browser execution if not products: state_scripts = [ "return window.__NEXT_DATA__ ? JSON.stringify(window.__NEXT_DATA__) : null", "return window.__INITIAL_STATE__ ? JSON.stringify(window.__INITIAL_STATE__) : null", "return window.__PRELOADED_STATE__ ? JSON.stringify(window.__PRELOADED_STATE__) : null", "return window.__APP_STATE__ ? JSON.stringify(window.__APP_STATE__) : null", ] for script in state_scripts: try: state_json = browser.driver.execute_script(script) if state_json: data = json.loads(state_json) logger.info(f"Extracted state from JS: {script[:50]}...") extracted = self._extract_products_from_json(data) if extracted: products.extend(extracted) break # Stop if we found products except Exception as e: logger.debug(f"Could not extract from JS state ({script[:30]}): {e}") # Method 3: Try to get product data from data attributes if not products: items_with_data = soup.select("[data-testid][data-sku-id]") or soup.select("[data-sku-id]") for item in items_with_data: sku_id = item.get("data-sku-id", "") if sku_id: # Find name and price within this element name_elem = item.select_one("h4") or item.select_one("[class*='title']") or item.select_one("a") name = name_elem.get_text(strip=True) if name_elem else "" if name and len(name) > 5: price_text = item.get_text() price_match = re.search(r"\$[\d,]+\.?\d*", price_text) price = price_match.group() if price_match else None # Check stock status from element text item_text = item.get_text().lower() in_stock = True if any(phrase in item_text for phrase in ["sold out", "out of stock", "unavailable", "coming soon"]): in_stock = False products.append(Product( name=name, url=f"{self.base_url}/site/{sku_id}.p", price=price, in_stock=in_stock, image_url=None, site=self.site_name, product_id=sku_id, )) return products def _extract_from_apollo_data(self, script_content: str) -> List[Product]: """Extract products from Best Buy's Apollo SSR data format""" products = [] try: # Extract product URLs - new format: /product/name/ID/sku/skuId url_pattern = r'"pdp":"(https://www\.bestbuy\.com/product/[^"]+)"' url_matches = re.findall(url_pattern, script_content) logger.info(f"Found {len(url_matches)} product URLs in Apollo data") # Extract product names (short format) name_pattern = r'"short":"([^"]+)"' name_matches = re.findall(name_pattern, script_content) # Extract SKU IDs sku_pattern = r'"skuId":"(\d+)"' sku_matches = re.findall(sku_pattern, script_content) # Extract prices - look for priceEventPrice or similar # Prices in Apollo format: "priceEventPrice":29.99 or "regularPrice":39.99 price_pattern = r'"(?:priceEventPrice|regularPrice|currentPrice)":(\d+\.?\d*)' price_matches = re.findall(price_pattern, script_content) # Extract images image_pattern = r'"piscesHref":"(https://pisces\.bbystatic\.com/[^"]+)"' image_matches = re.findall(image_pattern, script_content) logger.info(f"Apollo extraction: {len(url_matches)} URLs, {len(name_matches)} names, {len(sku_matches)} SKUs, {len(price_matches)} prices") # Create products from URLs (most reliable source) seen_urls = set() for url in url_matches: if url in seen_urls: continue seen_urls.add(url) # Extract SKU from URL: /product/.../sku/12345 sku_match = re.search(r'/sku/(\d+)', url) sku_id = sku_match.group(1) if sku_match else "" # Try to find name for this product # Look for name near the URL in the data name = None url_pos = script_content.find(url) if url_pos > 0: # Look for "short":"..." within 2000 chars before the URL context = script_content[max(0, url_pos-2000):url_pos] name_match = re.search(r'"short":"([^"]+)"[^}]*$', context) if name_match: name = name_match.group(1) if not name and name_matches: # Use any name that contains pokemon (fallback) for n in name_matches: if 'pok' in n.lower(): name = n break if not name: # Extract from URL url_parts = url.split('/') if len(url_parts) > 4: name = url_parts[4].replace('-', ' ').title() # Find price price = None if price_matches: # Use first available price as default price = f"${float(price_matches[0]):.2f}" # Find image image_url = None if image_matches: image_url = image_matches[0] # Try to determine stock status from Apollo data context in_stock = True # Default to True if no stock info found if url_pos > 0: # Look for availability info near this product's URL context_start = max(0, url_pos - 3000) context_end = min(len(script_content), url_pos + 500) product_context = script_content[context_start:context_end] # Check for explicit out of stock indicators if '"isAvailable":false' in product_context or '"available":false' in product_context: in_stock = False elif '"soldOut":true' in product_context or '"outOfStock":true' in product_context: in_stock = False elif 'sold out' in product_context.lower() or 'out of stock' in product_context.lower(): in_stock = False if name and len(name) > 5: products.append(Product( name=name, url=url, price=price, in_stock=in_stock, image_url=image_url, site=self.site_name, product_id=sku_id, )) # Deduplicate by SKU seen_skus = set() unique_products = [] for p in products: if p.product_id and p.product_id not in seen_skus: seen_skus.add(p.product_id) unique_products.append(p) elif not p.product_id: unique_products.append(p) logger.info(f"Extracted {len(unique_products)} unique products from Apollo data") return unique_products except Exception as e: logger.error(f"Error parsing Apollo data: {e}") return [] def _extract_products_from_json(self, data, depth=0) -> List[Product]: """Recursively search JSON for product data""" products = [] if depth > 15: # Increased depth for deeply nested structures return products if isinstance(data, dict): # First check common Best Buy JSON paths (Next.js structure) if depth == 0: # Try common paths in __NEXT_DATA__ common_paths = [ ("props", "pageProps", "products"), ("props", "pageProps", "initialData", "products"), ("props", "pageProps", "searchResults", "products"), ("props", "pageProps", "items"), ("props", "pageProps", "initialData", "searchResult", "products"), ("props", "initialState", "products"), ("pageProps", "products"), ("pageProps", "items"), ] for path in common_paths: obj = data for key in path: if isinstance(obj, dict) and key in obj: obj = obj[key] else: obj = None break if obj and isinstance(obj, list): logger.info(f"Found products at path: {'.'.join(path)}") for item in obj: if isinstance(item, dict): product = self._parse_product_json(item) if product: products.append(product) # Check if this looks like a Best Buy product if "skuId" in data or "sku" in data: product = self._parse_product_json(data) if product: products.append(product) elif "name" in data and ("regularPrice" in data or "salePrice" in data or "price" in data): product = self._parse_product_json(data) if product: products.append(product) # Continue recursive search 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_product_json(self, data: dict) -> Optional[Product]: """Parse a product from Best Buy's JSON data""" try: # Try multiple name fields name = ( data.get("name") or data.get("displayName") or data.get("title") or data.get("productName") or "" ) if not name or len(name) < 5: return None # Try multiple SKU fields sku_id = ( data.get("skuId") or data.get("sku") or data.get("productId") or data.get("id") or "" ) # Try multiple URL fields url_slug = ( data.get("url") or data.get("pdpUrl") or data.get("productUrl") or data.get("link") or "" ) if url_slug: url = url_slug if url_slug.startswith("http") else f"{self.base_url}{url_slug}" elif sku_id: url = f"{self.base_url}/site/{sku_id}.p" else: return None # Get price - try multiple price structures price = None if "regularPrice" in data: price = f"${data['regularPrice']:.2f}" if isinstance(data['regularPrice'], (int, float)) else data['regularPrice'] elif "salePrice" in data: price = f"${data['salePrice']:.2f}" if isinstance(data['salePrice'], (int, float)) else data['salePrice'] elif "currentPrice" in data: price = f"${data['currentPrice']:.2f}" if isinstance(data['currentPrice'], (int, float)) else data['currentPrice'] elif "price" in data: p = data['price'] if isinstance(p, dict): price = p.get("currentPrice") or p.get("regularPrice") or p.get("salePrice") if isinstance(price, (int, float)): price = f"${price:.2f}" elif isinstance(p, (int, float)): price = f"${p:.2f}" else: price = str(p) if p else None elif "priceInfo" in data: price_info = data["priceInfo"] if isinstance(price_info, dict): price = price_info.get("currentPrice") or price_info.get("price") if isinstance(price, (int, float)): price = f"${price:.2f}" # Check availability - handle multiple formats in_stock = True if "availability" in data: availability = data["availability"] if isinstance(availability, dict): # Check isAvailable first, then available - don't use 'or' which short-circuits incorrectly is_available = availability.get("isAvailable") available = availability.get("available") if is_available is not None: in_stock = bool(is_available) elif available is not None: in_stock = bool(available) # else keep default True elif isinstance(availability, bool): in_stock = availability elif isinstance(availability, str): in_stock = availability.lower() not in ["unavailable", "sold out", "out of stock"] if data.get("orderable") is False: in_stock = False if data.get("inStock") is False: in_stock = False # Get image - try multiple fields image_url = ( data.get("image") or data.get("thumbnailImage") or data.get("imageUrl") or data.get("thumbnail") ) if isinstance(image_url, dict): image_url = image_url.get("src") or image_url.get("url") return Product( name=name, url=url, price=price, in_stock=in_stock, image_url=image_url, site=self.site_name, product_id=str(sku_id) if sku_id else "", ) except Exception as e: logger.debug(f"Error parsing Best Buy product JSON: {e}") return None def _parse_product_card(self, card) -> Optional[Product]: """Parse a product card element""" try: # Find link - try multiple patterns link = ( card.select_one("a[href*='/product/']") # New Best Buy format or card.select_one("a[href*='/site/'][href*='.p']") # Legacy format or card.select_one("a[href*='skuId=']") or card.select_one("a.image-link") or card.select_one("[data-testid='product-link']") or card.select_one("a[data-track]") 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 - try multiple modern selectors name_elem = ( card.select_one("h4") # Current Best Buy format or card.select_one("h3") or card.select_one("[data-testid='product-title']") or card.select_one("[data-testid='product-name']") or card.select_one("[class*='productTitle']") or card.select_one("[class*='ProductTitle']") or card.select_one(".sku-title a") or card.select_one("h4.sku-header a") or card.select_one(".sku-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 - try multiple modern selectors price_elem = ( card.select_one("div.pricing") # Current Best Buy format or card.select_one("[class*='pricing']") or card.select_one("[data-testid='customer-price']") or card.select_one("[data-testid='current-price']") or card.select_one("[class*='customerPrice']") or card.select_one("[class*='CurrentPrice']") or card.select_one(".priceView-customer-price span") or card.select_one(".pricing-price__regular-price") 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 product ID from URL - handle multiple formats # New: /product/product-name/ABC123 or /product/.../sku/12345 # Old: /site/.../12345.p product_id = "" # Try /sku/12345 format first match = re.search(r"/sku/(\d+)", url) if match: product_id = match.group(1) else: # Try old .p format match = re.search(r"/(\d+)\.p", url) if match: product_id = match.group(1) else: # New format: last path segment is the ID url_parts = url.rstrip('/').split('/') if url_parts: product_id = url_parts[-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 Best Buy 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 SKU ID - handle both old and new formats # New: /product/.../sku/12345 # Old: /site/.../12345.p product_id = "" match = re.search(r"/sku/(\d+)", url) # New format first if match: product_id = match.group(1) else: match = re.search(r"/(\d+)\.p", url) # Old format 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 ["sold out", "out of stock", "unavailable"]): 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 Best Buy 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() # Check for disabled add to cart button add_btn = card.select_one(".add-to-cart-button") if add_btn and "btn-disabled" in add_btn.get("class", []): return False out_of_stock_phrases = [ "sold out", "out of stock", "unavailable", "coming soon", "not available", ] for phrase in out_of_stock_phrases: if phrase in card_text: return False in_stock_phrases = [ "add to cart", "add to bag", "available", "in stock", ] 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""" try: browser = self._get_stealth_browser() browser.driver.get(product_url) time.sleep(3) html = browser.driver.page_source soup = BeautifulSoup(html, "html.parser") page_text = soup.get_text().lower() out_of_stock = any( phrase in page_text for phrase in ["sold out", "out of stock", "unavailable", "coming soon"] ) in_stock = not out_of_stock # Get price price = None price_elem = ( soup.select_one(".priceView-customer-price span") or soup.select_one("[data-testid='customer-price']") ) if price_elem: price_match = re.search(r"\$[\d,]+\.?\d*", price_elem.get_text()) if price_match: price = price_match.group() return in_stock, price except Exception as e: logger.error(f"Error checking Best Buy product stock: {e}") return False, None