mmcghen
8d382e723f
- Added Walmart scraper to scrape product data from Walmart.com, including category pages and product details. - Introduced a stealth browser module to handle bot protection and improve scraping reliability. - Created a SQLite database for tracking product history, price changes, stock events, and user favorites. - Developed a Discord bot for user interaction, allowing location setting and stock checking at local stores. - Implemented a favorites system to manage priority products and categories with custom notification settings. - Added news aggregation module to fetch and analyze Pokemon TCG news from various sources. - Created tools for API discovery and monitoring, including a backend monitor for detecting new products. - Added unit tests for database operations, product filtering, and API endpoints to ensure functionality. - Enhanced existing modules with improved error handling and logging for better maintainability.
506 lines
18 KiB
Python
506 lines
18 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
|
|
|
|
# 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
|
|
|
|
def _get_stealth_browser(self):
|
|
"""Get or create stealth browser for Best Buy"""
|
|
if self._stealth_browser is None:
|
|
from tools.stealth_browser import StealthBrowser
|
|
self._stealth_browser = StealthBrowser(headless=False, session_name="bestbuy")
|
|
self._stealth_browser.start()
|
|
return self._stealth_browser
|
|
|
|
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}"
|
|
|
|
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
|
|
|
|
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.sku-item")
|
|
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")
|
|
)
|
|
|
|
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: parse product links
|
|
if not product_cards:
|
|
product_links = soup.select("a[href*='/site/'][href*='.p']")
|
|
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 ".p" 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)
|
|
|
|
# Also try parsing from data attributes and script tags
|
|
if not products:
|
|
products.extend(self._extract_from_page_data(soup, browser))
|
|
|
|
# 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 = []
|
|
|
|
# Try to get product data from data attributes
|
|
items_with_data = soup.select("[data-testid][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
|
|
|
|
products.append(Product(
|
|
name=name,
|
|
url=f"{self.base_url}/site/{sku_id}.p",
|
|
price=price,
|
|
in_stock=True,
|
|
image_url=None,
|
|
site=self.site_name,
|
|
product_id=sku_id,
|
|
))
|
|
|
|
# Try to extract from window.__INITIAL_STATE__ or similar JS objects
|
|
try:
|
|
initial_state = browser.driver.execute_script("""
|
|
if (window.__INITIAL_STATE__) return JSON.stringify(window.__INITIAL_STATE__);
|
|
if (window.__NEXT_DATA__) return JSON.stringify(window.__NEXT_DATA__);
|
|
return null;
|
|
""")
|
|
if initial_state:
|
|
data = json.loads(initial_state)
|
|
products.extend(self._extract_products_from_json(data))
|
|
except Exception as e:
|
|
logger.debug(f"Could not extract from JS state: {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 Best Buy product
|
|
if "skuId" in data or ("name" in data and "regularPrice" in data):
|
|
product = self._parse_product_json(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_product_json(self, data: dict) -> Optional[Product]:
|
|
"""Parse a product from Best Buy's JSON data"""
|
|
try:
|
|
name = data.get("name") or data.get("displayName", "")
|
|
if not name:
|
|
return None
|
|
|
|
sku_id = data.get("skuId") or data.get("sku", "")
|
|
url_slug = data.get("url") 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
|
|
price = None
|
|
if "regularPrice" in data:
|
|
price = f"${data['regularPrice']:.2f}"
|
|
elif "salePrice" in data:
|
|
price = f"${data['salePrice']:.2f}"
|
|
|
|
# Check availability
|
|
in_stock = True
|
|
availability = data.get("availability", {})
|
|
if isinstance(availability, dict):
|
|
in_stock = availability.get("isAvailable", True)
|
|
elif data.get("orderable") is False:
|
|
in_stock = False
|
|
|
|
# Get image
|
|
image_url = data.get("image") or data.get("thumbnailImage")
|
|
|
|
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),
|
|
)
|
|
|
|
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
|
|
link = card.select_one("a[href*='/site/'][href*='.p']") or card.select_one("a.image-link")
|
|
if not link:
|
|
link = 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(".sku-title a")
|
|
or card.select_one("h4.sku-header a")
|
|
or card.select_one("[data-testid='product-title']")
|
|
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
|
|
price_elem = (
|
|
card.select_one(".priceView-customer-price span")
|
|
or card.select_one("[data-testid='customer-price']")
|
|
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 SKU ID from URL
|
|
product_id = ""
|
|
match = re.search(r"/(\d+)\.p", 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 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
|
|
product_id = ""
|
|
match = re.search(r"/(\d+)\.p", 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 ["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
|