975 lines
42 KiB
Python
975 lines
42 KiB
Python
"""
|
|
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
|
|
from config import BESTBUY_HEADLESS
|
|
|
|
# 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=BESTBUY_HEADLESS, 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",
|
|
"newconnectionerror", "connection refused", "winerror 10061",
|
|
"failed to establish a new connection", "max retries exceeded",
|
|
]
|
|
|
|
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 = []
|
|
|
|
# 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(15) # Wait for React hydration and GraphQL fetches to complete
|
|
|
|
# Scroll to trigger any remaining lazy-loaded products
|
|
scroll_pause = 1.5
|
|
scroll_step = 800
|
|
current_pos = 0
|
|
for _ in range(20):
|
|
current_pos += scroll_step
|
|
browser.driver.execute_script(f"window.scrollTo(0, {current_pos})")
|
|
time.sleep(scroll_pause)
|
|
page_height = browser.driver.execute_script("return document.body.scrollHeight")
|
|
if current_pos >= page_height:
|
|
break
|
|
browser.driver.execute_script("window.scrollTo(0, 0)")
|
|
time.sleep(2)
|
|
|
|
# Wait for products to load (check for content)
|
|
for _ in range(10):
|
|
html = browser.driver.page_source
|
|
if any(kw in html for kw in ["sku-item", "sku-title", "priceView", "product-list-item", "ApolloSSRDataTransport", '"pdp":']):
|
|
logger.info("Product content detected")
|
|
break
|
|
time.sleep(1)
|
|
|
|
# Get HTML after waiting
|
|
html = browser.driver.page_source
|
|
|
|
# Save debug screenshot and HTML for analysis
|
|
try:
|
|
browser.driver.save_screenshot("debug_bestbuy.png")
|
|
with open("debug_bestbuy.html", "w", encoding="utf-8") as f:
|
|
f.write(html)
|
|
logger.info("Saved debug screenshot and HTML to debug_bestbuy.*")
|
|
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_graphql_response(self, data: dict) -> List[Product]:
|
|
"""Extract products from BestBuy's intercepted GraphQL search response"""
|
|
products = []
|
|
try:
|
|
search = data.get('searchResult') or data.get('searchResults') or {}
|
|
items = search.get('products') or search.get('items') or []
|
|
logger.info(f"GraphQL response has {len(items)} products")
|
|
for item in items:
|
|
product = item.get('product') or item
|
|
name = (product.get('name') or {}).get('short') or (product.get('name') or {}).get('title') or product.get('displayName') or ''
|
|
if not name or len(name) < 5:
|
|
continue
|
|
sku_id = str(product.get('skuId') or '')
|
|
url_obj = product.get('url') or {}
|
|
url = url_obj.get('skuSpecificUrl') or url_obj.get('pdp') or (f"{self.base_url}/site/{sku_id}.p" if sku_id else None)
|
|
if not url:
|
|
continue
|
|
image = (product.get('primaryImage') or {}).get('piscesHref')
|
|
price = None
|
|
price_info = product.get('priceInfo') or product.get('price') or {}
|
|
if isinstance(price_info, dict):
|
|
raw = price_info.get('currentPrice') or price_info.get('regularPrice')
|
|
if isinstance(raw, (int, float)):
|
|
price = f"${raw:.2f}"
|
|
# Check stock via button state
|
|
in_stock = True
|
|
fulfillment = product.get('fulfillmentOptions') or {}
|
|
btn_states = fulfillment.get('buttonStates') or []
|
|
if btn_states:
|
|
state = btn_states[0].get('buttonState', '')
|
|
in_stock = state in ('ADD_TO_CART', 'PRE_ORDER', 'CHECK_STORES')
|
|
products.append(Product(
|
|
name=name, url=url, price=price, in_stock=in_stock,
|
|
image_url=image, site=self.site_name, product_id=sku_id,
|
|
))
|
|
except Exception as e:
|
|
logger.error(f"Error parsing GraphQL response: {e}")
|
|
return 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: Extract from live Apollo client cache (captures client-side loaded products)
|
|
# BestBuy SSRs only ~4 featured products; the rest load via client-side GraphQL.
|
|
# The Apollo client cache holds all of them after hydration.
|
|
apollo_cache_script = """
|
|
try {
|
|
var client = window.__APOLLO_CLIENT__;
|
|
if (!client) return null;
|
|
var cache = client.cache.extract();
|
|
var products = [];
|
|
for (var key in cache) {
|
|
var obj = cache[key];
|
|
if (obj && obj.__typename === 'Product' && obj.skuId && obj.name && obj.name.short) {
|
|
products.push({
|
|
skuId: obj.skuId,
|
|
name: obj.name.short || obj.name.title || '',
|
|
pdp: obj.url ? obj.url.pdp || obj.url.skuSpecificUrl : null,
|
|
price: obj.priceAndAvailabilityInfo ? obj.priceAndAvailabilityInfo.currentPrice : null,
|
|
inStock: obj.fulfillmentOptions ? !!(obj.fulfillmentOptions.buttonStates || []).find(function(b){ return b.buttonState === 'ADD_TO_CART'; }) : true,
|
|
image: obj.primaryImage ? obj.primaryImage.piscesHref : null
|
|
});
|
|
}
|
|
}
|
|
return products.length > 0 ? JSON.stringify(products) : null;
|
|
} catch(e) { return null; }
|
|
"""
|
|
try:
|
|
apollo_json = browser.driver.execute_script(apollo_cache_script)
|
|
if apollo_json:
|
|
apollo_products = json.loads(apollo_json)
|
|
logger.info(f"Extracted {len(apollo_products)} products from Apollo client cache")
|
|
for item in apollo_products:
|
|
if not item.get('name') or len(item.get('name', '')) < 5:
|
|
continue
|
|
sku_id = str(item.get('skuId', ''))
|
|
pdp = item.get('pdp')
|
|
url = pdp if pdp else (f"{self.base_url}/site/{sku_id}.p" if sku_id else None)
|
|
if not url:
|
|
continue
|
|
price = item.get('price')
|
|
products.append(Product(
|
|
name=item['name'],
|
|
url=url,
|
|
price=f"${price:.2f}" if isinstance(price, (int, float)) else str(price) if price else None,
|
|
in_stock=item.get('inStock', True),
|
|
image_url=item.get('image'),
|
|
site=self.site_name,
|
|
product_id=sku_id,
|
|
))
|
|
except Exception as e:
|
|
logger.debug(f"Could not extract from Apollo client cache: {e}")
|
|
|
|
# Method 4: 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]
|
|
|
|
# Determine stock status by anchoring to the SKU ID in the Apollo data,
|
|
# then searching for buttonState within 500 chars after it.
|
|
# This is more reliable than looking near the URL (which bleeds across products).
|
|
in_stock = True
|
|
if sku_id:
|
|
sku_anchor = f'"skuId":"{sku_id}"'
|
|
sku_pos = script_content.find(sku_anchor)
|
|
if sku_pos >= 0:
|
|
sku_context = script_content[sku_pos:sku_pos + 500]
|
|
btn_match = re.search(r'"buttonState":"([^"]+)"', sku_context)
|
|
if btn_match:
|
|
in_stock = btn_match.group(1) in ("ADD_TO_CART", "PRE_ORDER", "CHECK_STORES")
|
|
|
|
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 — "add to cart" takes priority over "unavailable"
|
|
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 ["add to cart", "add to bag", "pre-order"]):
|
|
break # purchasable — in_stock stays True
|
|
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()
|
|
|
|
# "Add to Cart" takes priority — a card can show "Unavailable" for in-store/delivery
|
|
# while still being purchasable online, so check purchasable state first.
|
|
if "add to cart" in card_text or "add to bag" in card_text:
|
|
return True
|
|
if "pre-order" in card_text:
|
|
return True
|
|
|
|
# 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
|
|
|
|
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
|