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.
335 lines
12 KiB
Python
335 lines
12 KiB
Python
"""
|
|
PokemonCenter.com scraper
|
|
Handles bot protection with undetected-chromedriver stealth browser
|
|
"""
|
|
|
|
import re
|
|
import logging
|
|
from typing import List, Optional
|
|
from bs4 import BeautifulSoup
|
|
|
|
from .base import BaseScraper, Product
|
|
import config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_stealth_or_regular_browser():
|
|
"""Get appropriate browser based on config"""
|
|
if config.USE_STEALTH_BROWSER:
|
|
from tools.stealth_browser import get_stealth_browser
|
|
return get_stealth_browser(
|
|
headless=config.STEALTH_HEADLESS,
|
|
session_name=config.STEALTH_SESSION_NAME
|
|
), True
|
|
else:
|
|
from src.browser import get_browser
|
|
return get_browser(), False
|
|
|
|
|
|
class PokemonCenterScraper(BaseScraper):
|
|
"""Scraper for PokemonCenter.com"""
|
|
|
|
site_name = "pokemoncenter"
|
|
base_url = "https://www.pokemoncenter.com"
|
|
|
|
def scrape_category_page(self, url: str, max_pages: int = 1) -> List[Product]:
|
|
"""
|
|
Scrape a PokemonCenter category page for all products
|
|
|
|
Args:
|
|
url: Category page URL
|
|
max_pages: Maximum pages to scrape (not yet implemented for PokemonCenter)
|
|
|
|
Returns:
|
|
List of Product objects
|
|
"""
|
|
# TODO: Implement pagination for PokemonCenter when needed
|
|
browser, is_stealth = get_stealth_or_regular_browser()
|
|
products = []
|
|
|
|
try:
|
|
if is_stealth:
|
|
# Use stealth browser (undetected-chromedriver)
|
|
html = browser.get_page(url, wait_time=config.HUMAN_DELAY_MAX)
|
|
|
|
# Check for CAPTCHA
|
|
if browser.check_for_captcha():
|
|
logger.warning("CAPTCHA detected on PokemonCenter!")
|
|
if config.PAUSE_ON_CAPTCHA and not config.STEALTH_HEADLESS:
|
|
browser.wait_for_captcha_solve(config.CAPTCHA_WAIT_TIMEOUT)
|
|
# Re-fetch page after CAPTCHA solve
|
|
html = browser.get_page(url, wait_time=config.HUMAN_DELAY_MAX)
|
|
else:
|
|
logger.error("Cannot solve CAPTCHA in headless mode")
|
|
return []
|
|
|
|
# Save screenshot for debugging
|
|
try:
|
|
browser.screenshot("debug_screenshot.png")
|
|
logger.info("Saved debug screenshot to debug_screenshot.png")
|
|
except:
|
|
pass
|
|
|
|
page = None # Stealth browser doesn't return page object
|
|
else:
|
|
# Use regular Playwright browser
|
|
page, html = browser.get_page_content(
|
|
url,
|
|
wait_for_selector=None, # Let it use networkidle instead
|
|
timeout=60000,
|
|
)
|
|
|
|
# Save screenshot for debugging if needed
|
|
try:
|
|
page.screenshot(path="debug_screenshot.png")
|
|
logger.info("Saved debug screenshot to debug_screenshot.png")
|
|
except:
|
|
pass
|
|
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
|
|
# Try multiple selectors for product cards
|
|
# PokemonCenter may use different structures
|
|
product_cards = (
|
|
soup.select("[data-testid='product-card']")
|
|
or soup.select(".product-card")
|
|
or soup.select(".product-tile")
|
|
or soup.select("article[data-product-id]")
|
|
or soup.select(".product-grid-item")
|
|
or soup.select("[class*='ProductCard']")
|
|
)
|
|
|
|
logger.info(f"Found {len(product_cards)} product cards")
|
|
|
|
# If no product cards found, try a more generic approach
|
|
if not product_cards:
|
|
# Look for any links that look like product pages
|
|
product_links = soup.select("a[href*='/product/']")
|
|
logger.info(f"Fallback: Found {len(product_links)} product links")
|
|
|
|
for link in product_links:
|
|
href = link.get("href", "")
|
|
if href and "/product/" in href:
|
|
full_url = href if href.startswith("http") else f"{self.base_url}{href}"
|
|
|
|
# Try to get product name from link text or nearby elements
|
|
name = link.get_text(strip=True)
|
|
if not name or len(name) < 3:
|
|
name = link.get("title", "") or link.get("aria-label", "") or "Unknown Product"
|
|
|
|
product = Product(
|
|
name=name,
|
|
url=full_url,
|
|
price=None,
|
|
in_stock=True, # Assume in stock if listed, will verify later
|
|
image_url=None,
|
|
site=self.site_name,
|
|
product_id=self._extract_product_id(full_url),
|
|
)
|
|
products.append(product)
|
|
|
|
else:
|
|
for card in product_cards:
|
|
product = self._parse_product_card(card)
|
|
if product:
|
|
products.append(product)
|
|
|
|
# Close page only for Playwright browser
|
|
if page is not None:
|
|
page.close()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error scraping PokemonCenter 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 PokemonCenter")
|
|
return unique_products
|
|
|
|
def _parse_product_card(self, card) -> Optional[Product]:
|
|
"""Parse a product card element into a Product object"""
|
|
try:
|
|
# Try to find product link
|
|
link = card.select_one("a[href*='/product/']") 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 product name
|
|
name_elem = (
|
|
card.select_one("[data-testid='product-name']")
|
|
or card.select_one(".product-name")
|
|
or card.select_one("h2")
|
|
or card.select_one("h3")
|
|
or card.select_one("[class*='title']")
|
|
or card.select_one("[class*='name']")
|
|
)
|
|
name = name_elem.get_text(strip=True) if name_elem else link.get_text(strip=True)
|
|
|
|
if not name or len(name) < 3:
|
|
name = link.get("title", "") or link.get("aria-label", "") or "Unknown Product"
|
|
|
|
# Get price
|
|
price_elem = (
|
|
card.select_one("[data-testid='product-price']")
|
|
or card.select_one(".product-price")
|
|
or card.select_one("[class*='price']")
|
|
or card.select_one("span:contains('$')")
|
|
)
|
|
price = None
|
|
if price_elem:
|
|
price_text = price_elem.get_text(strip=True)
|
|
# Extract price with regex
|
|
price_match = re.search(r"\$[\d,]+\.?\d*", price_text)
|
|
if price_match:
|
|
price = price_match.group()
|
|
|
|
# Check stock status
|
|
in_stock = self._check_card_stock_status(card)
|
|
|
|
# Get image URL
|
|
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"{self.base_url}{image_url}"
|
|
|
|
return Product(
|
|
name=name,
|
|
url=url,
|
|
price=price,
|
|
in_stock=in_stock,
|
|
image_url=image_url,
|
|
site=self.site_name,
|
|
product_id=self._extract_product_id(url),
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Error parsing product card: {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=True) if hasattr(card, "get_text") else str(card).lower()
|
|
|
|
# Out of stock indicators
|
|
out_of_stock_phrases = [
|
|
"sold out",
|
|
"out of stock",
|
|
"unavailable",
|
|
"coming soon",
|
|
"notify me",
|
|
]
|
|
|
|
for phrase in out_of_stock_phrases:
|
|
if phrase in card_text:
|
|
return False
|
|
|
|
# In stock indicators
|
|
in_stock_phrases = [
|
|
"add to cart",
|
|
"add to bag",
|
|
"buy now",
|
|
"in stock",
|
|
"available",
|
|
]
|
|
|
|
for phrase in in_stock_phrases:
|
|
if phrase in card_text:
|
|
return True
|
|
|
|
# If we can't determine, assume it might be in stock (will verify on product page)
|
|
return True
|
|
|
|
def check_product_stock(self, product_url: str) -> tuple[bool, Optional[str]]:
|
|
"""
|
|
Check if a specific product is in stock by visiting its page
|
|
|
|
Args:
|
|
product_url: URL of the product page
|
|
|
|
Returns:
|
|
Tuple of (is_in_stock, price)
|
|
"""
|
|
browser, is_stealth = get_stealth_or_regular_browser()
|
|
|
|
try:
|
|
if is_stealth:
|
|
# Use stealth browser
|
|
html = browser.get_page(product_url, wait_time=config.HUMAN_DELAY_MAX)
|
|
|
|
# Check for CAPTCHA
|
|
if browser.check_for_captcha():
|
|
logger.warning("CAPTCHA detected on product page!")
|
|
if config.PAUSE_ON_CAPTCHA and not config.STEALTH_HEADLESS:
|
|
browser.wait_for_captcha_solve(config.CAPTCHA_WAIT_TIMEOUT)
|
|
html = browser.get_page(product_url, wait_time=config.HUMAN_DELAY_MAX)
|
|
else:
|
|
return False, None
|
|
|
|
page = None
|
|
else:
|
|
# Use regular Playwright browser
|
|
page, html = browser.get_page_content(
|
|
product_url,
|
|
wait_for_selector="button, [data-testid]",
|
|
timeout=30000,
|
|
)
|
|
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
page_text = soup.get_text().lower()
|
|
|
|
# Check for out of stock indicators
|
|
out_of_stock = any(
|
|
phrase in page_text
|
|
for phrase in ["sold out", "out of stock", "unavailable", "notify me when available"]
|
|
)
|
|
|
|
# Check for add to cart button
|
|
add_to_cart = soup.select_one(
|
|
"button:contains('Add to Cart'), button:contains('Add to Bag'), [data-testid='add-to-cart']"
|
|
)
|
|
|
|
in_stock = not out_of_stock and add_to_cart is not None
|
|
|
|
# Get price
|
|
price = None
|
|
price_elem = soup.select_one("[data-testid='product-price'], .product-price, [class*='price']")
|
|
if price_elem:
|
|
price_match = re.search(r"\$[\d,]+\.?\d*", price_elem.get_text())
|
|
if price_match:
|
|
price = price_match.group()
|
|
|
|
if page is not None:
|
|
page.close()
|
|
return in_stock, price
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking product stock: {e}")
|
|
return False, None
|
|
|
|
def _extract_product_id(self, url: str) -> str:
|
|
"""Extract product ID from URL"""
|
|
# PokemonCenter URLs typically look like:
|
|
# https://www.pokemoncenter.com/product/123456/product-name
|
|
match = re.search(r"/product/(\d+)", url)
|
|
if match:
|
|
return match.group(1)
|
|
|
|
# Fallback: use URL path as ID
|
|
return url.split("/")[-1]
|