diff --git a/dashboard/static/app.js b/dashboard/static/app.js index a1be244..c1266d7 100644 --- a/dashboard/static/app.js +++ b/dashboard/static/app.js @@ -771,6 +771,9 @@ async function loadControlPanel() { // Render scrapers grid renderScrapersGrid(state.scrapers); + // Load Chrome Extension status + loadExtensionStatus(); + // Setup event listeners for control panel buttons setupControlPanelListeners(); @@ -783,10 +786,36 @@ async function loadControlPanel() { updateMonitorStatus(newState); renderScrapersGrid(newState.scrapers); } + loadExtensionStatus(); } }, 5000); } +async function loadExtensionStatus() { + const badge = document.getElementById('extensionBadge'); + const productCount = document.getElementById('extProductCount'); + const skuCount = document.getElementById('extSkuCount'); + const lastSync = document.getElementById('extLastSync'); + + const stats = await api('/extension/stats'); + + if (stats && stats.last_sync) { + // Extension has synced data + badge.textContent = 'Connected'; + badge.className = 'extension-badge connected'; + productCount.textContent = stats.total_products || 0; + skuCount.textContent = stats.total_skus || 0; + lastSync.textContent = formatTime(new Date(stats.last_sync)); + } else { + // No sync yet + badge.textContent = 'No Data'; + badge.className = 'extension-badge disconnected'; + productCount.textContent = '-'; + skuCount.textContent = '-'; + lastSync.textContent = 'Never'; + } +} + function updateMonitorStatus(state) { const badge = document.getElementById('monitorBadge'); const statusText = document.getElementById('monitorStatusText'); diff --git a/dashboard/static/style.css b/dashboard/static/style.css index 4084576..5a690f7 100644 --- a/dashboard/static/style.css +++ b/dashboard/static/style.css @@ -1103,6 +1103,57 @@ h3 { grid-column: span 2; } +/* Chrome Extension Card */ +.extension-card { + grid-column: span 2; + border-color: #ffcb05; +} + +.extension-badge { + padding: 4px 12px; + border-radius: 12px; + font-size: 12px; + font-weight: 600; +} + +.extension-badge.connected { + background: rgba(0, 255, 0, 0.2); + color: #00ff00; +} + +.extension-badge.disconnected { + background: rgba(255, 100, 100, 0.2); + color: #ff6464; +} + +.extension-stats { + display: grid; + grid-template-columns: repeat(3, 1fr); + gap: 15px; + margin-top: 10px; +} + +.extension-stat { + background: var(--bg-primary); + padding: 12px; + border-radius: 8px; + text-align: center; +} + +.extension-stat .stat-label { + display: block; + font-size: 12px; + color: var(--text-secondary); + margin-bottom: 5px; +} + +.extension-stat .stat-value { + display: block; + font-size: 18px; + font-weight: 600; + color: var(--text-primary); +} + .scrapers-grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); diff --git a/dashboard/templates/index.html b/dashboard/templates/index.html index 2947d87..94f4f88 100644 --- a/dashboard/templates/index.html +++ b/dashboard/templates/index.html @@ -274,6 +274,37 @@ + +
+
+

Pokemon Center (Chrome Extension)

+ Unknown +
+
+

Pokemon Center is monitored via the Chrome Extension, not a Python scraper.

+
+
+
+ Products Tracked: + - +
+
+ SKUs Detected: + - +
+
+ Last Sync: + Never +
+
+
+

+ Open Extension | + Install from chrome-extension/ folder +

+
+
+
diff --git a/main.py b/main.py index d9e0c6c..3bf56a3 100644 --- a/main.py +++ b/main.py @@ -14,7 +14,6 @@ import schedule from config import ( CHECK_INTERVAL_SECONDS, SITES_ENABLED, - POKEMON_CENTER_URLS, TARGET_URLS, GAMESTOP_URLS, BESTBUY_URLS, @@ -23,13 +22,14 @@ from config import ( KEYWORDS, LOG_LEVEL, NEW_DROP_NOTIFICATIONS, + PRICE_CHANGE_NOTIFICATIONS, MAX_PAGES, ) from src.browser import get_browser, shutdown_browser from src.product_tracker import ProductTracker -from src.discord_notifier import send_stock_alert, send_startup_notification, send_error_notification +from src.discord_notifier import send_stock_alert, send_startup_notification, send_error_notification, send_price_change_alert +# Note: Pokemon Center uses Chrome Extension (chrome-extension/), not a Python scraper from scrapers import ( - PokemonCenterScraper, TargetScraper, GameStopScraper, BestBuyScraper, @@ -50,9 +50,8 @@ logger = logging.getLogger(__name__) # Global tracker tracker = ProductTracker() -# Scrapers +# Scrapers (Pokemon Center uses Chrome Extension instead) scrapers = { - "pokemoncenter": PokemonCenterScraper(), "target": TargetScraper(), "gamestop": GameStopScraper(), "bestbuy": BestBuyScraper(), @@ -60,67 +59,14 @@ scrapers = { } -def check_pokemoncenter(): - """Check PokemonCenter for restocks and new drops""" - logger.info("Checking PokemonCenter...") - scraper = scrapers["pokemoncenter"] - - try: - for url in POKEMON_CENTER_URLS: - logger.info(f"Scraping: {url}") - products = scraper.scrape_category_page(url, max_pages=MAX_PAGES.get("pokemoncenter", 1)) - - # Apply keyword filter if enabled - if KEYWORD_FILTER_ENABLED and KEYWORDS: - products = scraper.filter_by_keywords(products, KEYWORDS) - logger.info(f"After keyword filter: {len(products)} products") - - # Process and detect changes - new_products, restocked_products = tracker.process_products(products) - - # Send notifications for new products (if enabled) - if NEW_DROP_NOTIFICATIONS: - for product in new_products: - if product.in_stock: - send_stock_alert( - product_name=product.name, - product_url=product.url, - price=product.price or "See link", - site="pokemoncenter", - alert_type="new_drop", - image_url=product.image_url, - ) - logger.info(f"Sent notification for new product: {product.name}") - - # Send notifications for restocks - for product in restocked_products: - send_stock_alert( - product_name=product.name, - product_url=product.url, - price=product.price or "See link", - site="pokemoncenter", - alert_type="restock", - image_url=product.image_url, - ) - logger.info(f"Sent notification for restock: {product.name}") - - # Log stats - stats = tracker.get_stats() - logger.info( - f"Stats: {stats['total_products']} total, " - f"{stats['in_stock']} in stock, " - f"{stats['out_of_stock']} out of stock" - ) - - except Exception as e: - logger.error(f"Error checking PokemonCenter: {e}") - send_error_notification(f"Error checking PokemonCenter: {str(e)}", "PokemonCenter") - - def check_target(): """Check Target for restocks and new drops""" logger.info("Checking Target...") scraper = scrapers["target"] + start_time = time.time() + total_products = 0 + total_new = 0 + total_restocks = 0 try: for url in TARGET_URLS: @@ -135,8 +81,12 @@ def check_target(): products = scraper.filter_by_keywords(products, KEYWORDS) logger.info(f"After keyword filter: {len(products)} products") + total_products += len(products) + # Process and detect changes - new_products, restocked_products = tracker.process_products(products) + new_products, restocked_products, price_changed_products = tracker.process_products(products) + total_new += len(new_products) + total_restocks += len(restocked_products) # Send notifications for new products (if enabled) if NEW_DROP_NOTIFICATIONS: @@ -164,6 +114,20 @@ def check_target(): ) logger.info(f"Sent notification for restock: {product.name}") + # Send notifications for price changes (if enabled) + if PRICE_CHANGE_NOTIFICATIONS: + for product, old_price, new_price in price_changed_products: + send_price_change_alert( + product_name=product.name, + product_url=product.url, + old_price=old_price, + new_price=new_price, + site="target", + in_stock=product.in_stock, + image_url=product.image_url, + ) + logger.info(f"Sent notification for price change: {product.name} ({old_price} -> {new_price})") + # Log stats stats = tracker.get_stats() logger.info( @@ -172,15 +136,26 @@ def check_target(): f"{stats['out_of_stock']} out of stock" ) + # Log successful check to database + duration_ms = int((time.time() - start_time) * 1000) + tracker.log_check("target", total_products, total_new, total_restocks, duration_ms, success=True) + except Exception as e: logger.error(f"Error checking Target: {e}") send_error_notification(f"Error checking Target: {str(e)}", "Target") + # Log failed check to database + duration_ms = int((time.time() - start_time) * 1000) + tracker.log_check("target", total_products, total_new, total_restocks, duration_ms, success=False, error_message=str(e)) def check_gamestop(): """Check GameStop for restocks and new drops""" logger.info("Checking GameStop...") scraper = scrapers["gamestop"] + start_time = time.time() + total_products = 0 + total_new = 0 + total_restocks = 0 try: for url in GAMESTOP_URLS: @@ -195,8 +170,12 @@ def check_gamestop(): products = scraper.filter_by_keywords(products, KEYWORDS) logger.info(f"After keyword filter: {len(products)} products") + total_products += len(products) + # Process and detect changes - new_products, restocked_products = tracker.process_products(products) + new_products, restocked_products, price_changed_products = tracker.process_products(products) + total_new += len(new_products) + total_restocks += len(restocked_products) # Send notifications for new products (if enabled) if NEW_DROP_NOTIFICATIONS: @@ -224,6 +203,20 @@ def check_gamestop(): ) logger.info(f"Sent notification for restock: {product.name}") + # Send notifications for price changes (if enabled) + if PRICE_CHANGE_NOTIFICATIONS: + for product, old_price, new_price in price_changed_products: + send_price_change_alert( + product_name=product.name, + product_url=product.url, + old_price=old_price, + new_price=new_price, + site="gamestop", + in_stock=product.in_stock, + image_url=product.image_url, + ) + logger.info(f"Sent notification for price change: {product.name} ({old_price} -> {new_price})") + # Log stats stats = tracker.get_stats() logger.info( @@ -232,15 +225,26 @@ def check_gamestop(): f"{stats['out_of_stock']} out of stock" ) + # Log successful check to database + duration_ms = int((time.time() - start_time) * 1000) + tracker.log_check("gamestop", total_products, total_new, total_restocks, duration_ms, success=True) + except Exception as e: logger.error(f"Error checking GameStop: {e}") send_error_notification(f"Error checking GameStop: {str(e)}", "GameStop") + # Log failed check to database + duration_ms = int((time.time() - start_time) * 1000) + tracker.log_check("gamestop", total_products, total_new, total_restocks, duration_ms, success=False, error_message=str(e)) def check_bestbuy(): """Check Best Buy for restocks and new drops""" logger.info("Checking Best Buy...") scraper = scrapers["bestbuy"] + start_time = time.time() + total_products = 0 + total_new = 0 + total_restocks = 0 try: for url in BESTBUY_URLS: @@ -255,8 +259,12 @@ def check_bestbuy(): products = scraper.filter_by_keywords(products, KEYWORDS) logger.info(f"After keyword filter: {len(products)} products") + total_products += len(products) + # Process and detect changes - new_products, restocked_products = tracker.process_products(products) + new_products, restocked_products, price_changed_products = tracker.process_products(products) + total_new += len(new_products) + total_restocks += len(restocked_products) # Send notifications for new products (if enabled) if NEW_DROP_NOTIFICATIONS: @@ -284,6 +292,20 @@ def check_bestbuy(): ) logger.info(f"Sent notification for restock: {product.name}") + # Send notifications for price changes (if enabled) + if PRICE_CHANGE_NOTIFICATIONS: + for product, old_price, new_price in price_changed_products: + send_price_change_alert( + product_name=product.name, + product_url=product.url, + old_price=old_price, + new_price=new_price, + site="bestbuy", + in_stock=product.in_stock, + image_url=product.image_url, + ) + logger.info(f"Sent notification for price change: {product.name} ({old_price} -> {new_price})") + # Log stats stats = tracker.get_stats() logger.info( @@ -292,15 +314,26 @@ def check_bestbuy(): f"{stats['out_of_stock']} out of stock" ) + # Log successful check to database + duration_ms = int((time.time() - start_time) * 1000) + tracker.log_check("bestbuy", total_products, total_new, total_restocks, duration_ms, success=True) + except Exception as e: logger.error(f"Error checking Best Buy: {e}") send_error_notification(f"Error checking Best Buy: {str(e)}", "Best Buy") + # Log failed check to database + duration_ms = int((time.time() - start_time) * 1000) + tracker.log_check("bestbuy", total_products, total_new, total_restocks, duration_ms, success=False, error_message=str(e)) def check_walmart(): """Check Walmart for restocks and new drops""" logger.info("Checking Walmart...") scraper = scrapers["walmart"] + start_time = time.time() + total_products = 0 + total_new = 0 + total_restocks = 0 try: for url in WALMART_URLS: @@ -315,8 +348,12 @@ def check_walmart(): products = scraper.filter_by_keywords(products, KEYWORDS) logger.info(f"After keyword filter: {len(products)} products") + total_products += len(products) + # Process and detect changes - new_products, restocked_products = tracker.process_products(products) + new_products, restocked_products, price_changed_products = tracker.process_products(products) + total_new += len(new_products) + total_restocks += len(restocked_products) # Send notifications for new products (if enabled) if NEW_DROP_NOTIFICATIONS: @@ -344,6 +381,20 @@ def check_walmart(): ) logger.info(f"Sent notification for restock: {product.name}") + # Send notifications for price changes (if enabled) + if PRICE_CHANGE_NOTIFICATIONS: + for product, old_price, new_price in price_changed_products: + send_price_change_alert( + product_name=product.name, + product_url=product.url, + old_price=old_price, + new_price=new_price, + site="walmart", + in_stock=product.in_stock, + image_url=product.image_url, + ) + logger.info(f"Sent notification for price change: {product.name} ({old_price} -> {new_price})") + # Log stats stats = tracker.get_stats() logger.info( @@ -352,17 +403,23 @@ def check_walmart(): f"{stats['out_of_stock']} out of stock" ) + # Log successful check to database + duration_ms = int((time.time() - start_time) * 1000) + tracker.log_check("walmart", total_products, total_new, total_restocks, duration_ms, success=True) + except Exception as e: logger.error(f"Error checking Walmart: {e}") send_error_notification(f"Error checking Walmart: {str(e)}", "Walmart") + # Log failed check to database + duration_ms = int((time.time() - start_time) * 1000) + tracker.log_check("walmart", total_products, total_new, total_restocks, duration_ms, success=False, error_message=str(e)) def run_checks(): """Run all enabled site checks""" logger.info(f"Running checks at {datetime.now().strftime('%H:%M:%S')}") - if "pokemoncenter" in SITES_ENABLED: - check_pokemoncenter() + # Note: Pokemon Center is monitored via Chrome Extension, not here if "target" in SITES_ENABLED: check_target() diff --git a/scrapers/__init__.py b/scrapers/__init__.py index 4414ce3..cdbcc81 100644 --- a/scrapers/__init__.py +++ b/scrapers/__init__.py @@ -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", diff --git a/scrapers/bestbuy.py b/scrapers/bestbuy.py index 6a13a07..298ad30 100644 --- a/scrapers/bestbuy.py +++ b/scrapers/bestbuy.py @@ -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): diff --git a/scrapers/gamestop.py b/scrapers/gamestop.py index 9937530..4154a84 100644 --- a/scrapers/gamestop.py +++ b/scrapers/gamestop.py @@ -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() diff --git a/scrapers/pokemoncenter.py b/scrapers/pokemoncenter.py deleted file mode 100644 index 5dc0316..0000000 --- a/scrapers/pokemoncenter.py +++ /dev/null @@ -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] diff --git a/src/product_tracker.py b/src/product_tracker.py index cf5989b..95be714 100644 --- a/src/product_tracker.py +++ b/src/product_tracker.py @@ -8,8 +8,9 @@ import logging from pathlib import Path from typing import Dict, List, Set, Optional, Tuple from dataclasses import asdict -from datetime import datetime +from datetime import datetime, timedelta +import config from scrapers.base import Product from .database import get_database from .favorites import get_favorites_manager @@ -53,7 +54,7 @@ class ProductTracker: except IOError as e: logger.error(f"Error saving products file: {e}") - def process_products(self, products: List[Product]) -> Tuple[List[Product], List[Product]]: + def process_products(self, products: List[Product]) -> Tuple[List[Product], List[Product], List[Tuple[Product, str, str]]]: """ Process a list of scraped products and detect changes. Also logs events to database for stats tracking. @@ -62,10 +63,12 @@ class ProductTracker: products: List of products from scraper Returns: - Tuple of (new_products, restocked_products) + Tuple of (new_products, restocked_products, price_changed_products) + price_changed_products is a list of tuples: (product, old_price, new_price) """ new_products = [] restocked_products = [] + price_changed_products = [] for product in products: url = product.url @@ -120,22 +123,49 @@ class ProductTracker: existing["image_url"] = product.image_url or existing.get("image_url") existing["db_id"] = db_product_id - # Check for price change - if product.price and product.price != old_price: + # Check for price change (only if both prices exist and are different) + if product.price and old_price and product.price != old_price: + # Record price change for notification + price_changed_products.append((product, old_price, product.price)) + existing["price"] = product.price + self.db.record_price(db_product_id, product.price) + self.db.record_stock_event(db_product_id, "price_change") + logger.info(f"PRICE CHANGE: {product.name}: {old_price} -> {product.price}") + elif product.price and not old_price: + # First time we're seeing a price - just update, don't notify existing["price"] = product.price self.db.record_price(db_product_id, product.price) - logger.debug(f"Price change for {product.name}: {old_price} -> {product.price}") # Check for restock if product.in_stock and not was_in_stock: - # RESTOCK! - restocked_products.append(product) - existing["last_in_stock"] = now - logger.info(f"RESTOCK: {product.name}") + # Check restock cooldown to prevent duplicate notifications + last_restock_notified = existing.get("last_restock_notified") + cooldown_hours = getattr(config, 'RESTOCK_COOLDOWN_HOURS', 6) + cooldown_expired = True - # Log restock event to database - self.db.record_stock_event(db_product_id, "restock") - self.db.record_stock_event(db_product_id, "in_stock") + if last_restock_notified: + try: + last_notified_time = datetime.fromisoformat(last_restock_notified) + cooldown_threshold = datetime.now() - timedelta(hours=cooldown_hours) + cooldown_expired = last_notified_time < cooldown_threshold + except (ValueError, TypeError): + cooldown_expired = True # Invalid timestamp, allow notification + + if cooldown_expired: + # RESTOCK! (outside cooldown period) + restocked_products.append(product) + existing["last_restock_notified"] = now + logger.info(f"RESTOCK: {product.name}") + + # Log restock event to database + self.db.record_stock_event(db_product_id, "restock") + self.db.record_stock_event(db_product_id, "in_stock") + else: + # Within cooldown period - skip notification but still update stock status + logger.debug(f"Restock detected for {product.name} but within {cooldown_hours}h cooldown (last notified: {last_restock_notified})") + self.db.record_stock_event(db_product_id, "in_stock") + + existing["last_in_stock"] = now # Check for out of stock elif not product.in_stock and was_in_stock: @@ -147,17 +177,18 @@ class ProductTracker: self.products[url] = existing self.save() - return new_products, restocked_products + return new_products, restocked_products, price_changed_products - def process_products_with_priority(self, products: List[Product]) -> Tuple[List[Tuple[Product, dict]], List[Tuple[Product, dict]]]: + def process_products_with_priority(self, products: List[Product]) -> Tuple[List[Tuple[Product, dict]], List[Tuple[Product, dict]], List[Tuple[Product, str, str, dict]]]: """ Process products and return with priority/favorite info. Returns: - Tuple of (new_products_with_priority, restocked_products_with_priority) - Each item is a tuple of (Product, notification_settings) + Tuple of (new_products_with_priority, restocked_products_with_priority, price_changed_with_priority) + new/restocked items are tuples of (Product, notification_settings) + price_changed items are tuples of (Product, old_price, new_price, notification_settings) """ - new_products, restocked_products = self.process_products(products) + new_products, restocked_products, price_changed_products = self.process_products(products) def add_priority(product: Product) -> Tuple[Product, dict]: # Get product from DB for category @@ -174,7 +205,13 @@ class ProductTracker: new_with_priority = [add_priority(p) for p in new_products] restocked_with_priority = [add_priority(p) for p in restocked_products] - return new_with_priority, restocked_with_priority + # Price changes include old/new price info + price_changed_with_priority = [] + for product, old_price, new_price in price_changed_products: + _, settings = add_priority(product) + price_changed_with_priority.append((product, old_price, new_price, settings)) + + return new_with_priority, restocked_with_priority, price_changed_with_priority def log_check(self, site: str, products_found: int, new_count: int, restock_count: int, duration_ms: int, success: bool = True, diff --git a/src/scraper_state.py b/src/scraper_state.py index 6fb9c18..7396a51 100644 --- a/src/scraper_state.py +++ b/src/scraper_state.py @@ -15,9 +15,9 @@ from pathlib import Path STATE_FILE = Path(__file__).parent.parent / "data" / "scraper_state.json" # Default state +# Note: Pokemon Center uses Chrome Extension, not a Python scraper DEFAULT_STATE = { "scrapers": { - "pokemoncenter": {"enabled": False, "running": False, "last_run": None, "last_error": None}, "target": {"enabled": True, "running": False, "last_run": None, "last_error": None}, "gamestop": {"enabled": False, "running": False, "last_run": None, "last_error": None}, "walmart": {"enabled": False, "running": False, "last_run": None, "last_error": None}, diff --git a/tools/stealth_browser.py b/tools/stealth_browser.py index 728b683..7c646a7 100644 --- a/tools/stealth_browser.py +++ b/tools/stealth_browser.py @@ -193,6 +193,47 @@ class StealthBrowser: self.driver = None self._setup_complete = False + def is_session_valid(self) -> bool: + """ + Check if the browser session is still valid. + Returns False if the session has crashed or been closed. + """ + if not self.driver: + return False + + try: + # Try to get current URL - this will fail if session is invalid + _ = self.driver.current_url + return True + except Exception as e: + error_msg = str(e).lower() + if "invalid session id" in error_msg or "session deleted" in error_msg or "no such session" in error_msg: + logger.warning(f"Browser session invalid: {e}") + return False + # Other errors might be temporary + return True + + def restart(self): + """ + Restart the browser, cleaning up the old session. + Useful when the session has become invalid. + """ + logger.info("Restarting stealth browser...") + + # Force cleanup of old driver without trying to save cookies (session is dead) + if self.driver: + try: + self.driver.quit() + except Exception: + pass # Ignore errors - session is already dead + finally: + self.driver = None + self._setup_complete = False + + # Start fresh + self.start() + logger.info("Stealth browser restarted successfully") + def _get_cookie_file(self) -> Path: """Get path to cookie file for this session""" return SESSION_DIR / f"{self.session_name}_cookies.pkl"