Browser session recovery, price tracking, and dashboard improvements
- Add browser session validation and auto-restart for BestBuy/GameStop scrapers - Add price change detection and notifications - Remove PokemonCenter scraper (now uses Chrome Extension) - Dashboard UI improvements Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,12 +1,11 @@
|
||||
# Scrapers package
|
||||
from .pokemoncenter import PokemonCenterScraper
|
||||
# Note: Pokemon Center uses Chrome Extension (chrome-extension/), not a Python scraper
|
||||
from .target import TargetScraper
|
||||
from .gamestop import GameStopScraper, warmup_gamestop
|
||||
from .bestbuy import BestBuyScraper
|
||||
from .walmart import WalmartScraper
|
||||
|
||||
__all__ = [
|
||||
"PokemonCenterScraper",
|
||||
"TargetScraper",
|
||||
"GameStopScraper",
|
||||
"BestBuyScraper",
|
||||
|
||||
+128
-36
@@ -27,15 +27,55 @@ class BestBuyScraper(BaseScraper):
|
||||
|
||||
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"""
|
||||
"""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:
|
||||
from tools.stealth_browser import StealthBrowser
|
||||
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
|
||||
@@ -65,39 +105,62 @@ class BestBuyScraper(BaseScraper):
|
||||
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:
|
||||
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")
|
||||
|
||||
@@ -272,11 +335,17 @@ class BestBuyScraper(BaseScraper):
|
||||
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=True,
|
||||
in_stock=in_stock,
|
||||
image_url=None,
|
||||
site=self.site_name,
|
||||
product_id=sku_id,
|
||||
@@ -359,12 +428,28 @@ class BestBuyScraper(BaseScraper):
|
||||
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=True, # Assume in stock if listed
|
||||
in_stock=in_stock,
|
||||
image_url=image_url,
|
||||
site=self.site_name,
|
||||
product_id=sku_id,
|
||||
@@ -512,7 +597,14 @@ class BestBuyScraper(BaseScraper):
|
||||
if "availability" in data:
|
||||
availability = data["availability"]
|
||||
if isinstance(availability, dict):
|
||||
in_stock = availability.get("isAvailable", True) or availability.get("available", True)
|
||||
# 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):
|
||||
|
||||
+67
-8
@@ -107,15 +107,55 @@ class GameStopScraper(BaseScraper):
|
||||
|
||||
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 GameStop"""
|
||||
"""Get or create stealth browser for GameStop, 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:
|
||||
from tools.stealth_browser import StealthBrowser
|
||||
logger.info("Creating new GameStop stealth browser...")
|
||||
self._stealth_browser = StealthBrowser(headless=False, session_name="gamestop")
|
||||
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"GameStop 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("GameStop 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 GameStop search/category page for all products
|
||||
@@ -145,14 +185,33 @@ class GameStopScraper(BaseScraper):
|
||||
for page_num in range(1, max_pages + 1):
|
||||
products = []
|
||||
page_url = url if page_num == 1 else f"{url}&start={24 * (page_num - 1)}"
|
||||
retry_count = 0
|
||||
max_retries = 2
|
||||
|
||||
while retry_count <= max_retries:
|
||||
try:
|
||||
logger.info(f"Navigating to: {page_url}")
|
||||
browser.driver.get(page_url)
|
||||
time.sleep(5) # Wait for page load
|
||||
|
||||
html = browser.driver.page_source
|
||||
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)
|
||||
logger.error(f"Failed to load page after {max_retries} retries")
|
||||
return all_products
|
||||
|
||||
try:
|
||||
logger.info(f"Navigating to: {page_url}")
|
||||
browser.driver.get(page_url)
|
||||
time.sleep(5) # Wait for page load
|
||||
|
||||
html = browser.driver.page_source
|
||||
|
||||
# Check for Cloudflare challenge
|
||||
html_lower = html.lower()
|
||||
title = browser.driver.title.lower()
|
||||
|
||||
@@ -1,334 +0,0 @@
|
||||
"""
|
||||
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]
|
||||
Reference in New Issue
Block a user