diff --git a/config.py b/config.py index 7581fb2..6256b62 100644 --- a/config.py +++ b/config.py @@ -5,13 +5,60 @@ Update DISCORD_WEBHOOK_URL with your actual webhook URL # Discord webhook URL - GET THIS FROM YOUR DISCORD SERVER # Server Settings -> Integrations -> Webhooks -> New Webhook -> Copy Webhook URL -DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/1471927173353963570/8Xh4Y_D8zRDi5MAP6ZlX0rf2Fc5wEtRKCup74kBZltk2qaatxArFp-yrk7ZWh5EOZbHp" +# This is the default/fallback webhook for anything not specifically routed below +DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/1488982276455796756/4HEzQDBF-Hs4a9Iyd7yvj7-SfP0BwrbQGdGBmQ9TL9urWzkTRaHelkvrqX4v9NTf2ZJ_" + +# ============================================================================= +# MULTI-CHANNEL WEBHOOK ROUTING (Optional) +# ============================================================================= +# Route notifications to different Discord channels based on site and/or alert type. +# Leave webhooks as None to use the default DISCORD_WEBHOOK_URL above. +# +# Create webhooks in Discord: Server Settings -> Integrations -> Webhooks -> New Webhook +# +# Lookup priority: +# 1. overrides (specific site + type combo) +# 2. sites (all alerts for a specific store) +# 3. types (all alerts of a specific type) +# 4. default (DISCORD_WEBHOOK_URL) + +DISCORD_WEBHOOKS = { + # Per-site channels - all alerts from a specific store go here first + "sites": { + "target": "https://discord.com/api/webhooks/1488982720901025852/O0KkVFHrQ6-k6FmCqQE4QpuWF6YFJ8YyFgojBPKcQu9g0UHzTpnAU9G1gbhOftARS11R", + "bestbuy": "https://discord.com/api/webhooks/1488982786781216798/AtG6x7FWzJjOsmosq9oPMuwlPrSwNHZJJTUuB3AApjyT7LIOxmEcYeiMFP7HiA_Ov1pn", + "gamestop": "https://discord.com/api/webhooks/1488982854036619355/NWZtydrKEhj-kxKbFi78f1Gpf3h16W_1kAJEcSuIgtjJUoAO8Sk7inOI2T8fb9GXo9xY", + "walmart": "https://discord.com/api/webhooks/1488982929450340375/0g16tJXw8f9UtUrRrLd2jDPWWD__LTDYO_X2VT2WL4F5GBYB8ykVxrjV-bCjzKKjSA5C", + "pokemoncenter": "https://discord.com/api/webhooks/1488983050090975364/DwY9QQvBZa6V-zz8mIg302sukbISSbkXOhDwe1Q6yDG0zAwvaGQcn8iTzNB4D676e3cA", + }, + + # Per-type channels - used when no site-specific webhook is set + "types": { + "restock": "https://discord.com/api/webhooks/1488982403048542269/s1J_p8R6IMGQGypPh_ZYTHHBD0cMnd_BjLGggVWaL2OcyDQAR42KS-CyuHcVr5s21vl3", + "new_drop": "https://discord.com/api/webhooks/1488982511613771898/RLDoHxNjsAx0PfHfT0kZe6-IW3g9PR_ckX_sI-CALcv4Lt0USslvI0rqePqW9gVP7D7v", + "price_change": "https://discord.com/api/webhooks/1488982573723025633/J864UL4vk_q9Rmj4i4wtkGfzB3nKAOc0y7jzBFDgUbhwg4XKm1Z7sWduqRc6x_ThWetc", + "error": "https://discord.com/api/webhooks/1488983128641900657/RP5FRSWwbVaRAf7RdXPs66ETbY8ziI1ekIx7lrHqxp8n200KpzUiX_EylCxXBH5nJD3B", + "captcha": "https://discord.com/api/webhooks/1488983128641900657/RP5FRSWwbVaRAf7RdXPs66ETbY8ziI1ekIx7lrHqxp8n200KpzUiX_EylCxXBH5nJD3B", + }, + + # Override specific site + type combinations (highest priority) + # Use tuples: ("site", "type"): "webhook_url" + "overrides": { + # ("bestbuy", "restock"): "https://discord.com/api/webhooks/xxx/bestbuy-restocks", + # ("pokemoncenter", "new_drop"): "https://discord.com/api/webhooks/xxx/pokemon-drops", + }, +} # Enable/disable Discord notifications NOTIFICATIONS_ENABLED = True # Set to False to disable all Discord alerts NEW_DROP_NOTIFICATIONS = True # Set to False to silence new drop notifications +PRICE_CHANGE_NOTIFICATIONS = True # Set to False to silence price change notifications SKIP_DUPLICATE_SKUS = True # Skip notifications for products we've already notified about +# Restock notification cooldown (prevents duplicate alerts for same product) +# Set to number of hours before a product can trigger another restock notification +RESTOCK_COOLDOWN_HOURS = 6 # Minimum hours between restock alerts for same product + # Discord Bot Token (for interactive bot commands) # Create a bot at https://discord.com/developers/applications # Required for !setlocation, !stores, !stock commands @@ -36,7 +83,7 @@ SORT_BY_NEWEST = True # Which sites to monitor # Note: PokemonCenter has strong bot protection (Imperva) - may need manual workarounds # Note: Walmart has aggressive bot protection (PerimeterX) - may need stealth mode -SITES_ENABLED = ["target", "gamestop", "bestbuy"] # Options: "pokemoncenter", "target", "gamestop", "bestbuy" (disabled - needs rework), "walmart" +SITES_ENABLED = ["target", "bestbuy", "gamestop"] # Options: "pokemoncenter", "target", "gamestop", "bestbuy" (disabled - needs rework), "walmart" # PokemonCenter URLs to monitor POKEMON_CENTER_URLS = [ diff --git a/dashboard/static/app.js b/dashboard/static/app.js index 536ea98..617ac3e 100644 --- a/dashboard/static/app.js +++ b/dashboard/static/app.js @@ -784,6 +784,9 @@ async function loadControlPanel() { // Render scrapers grid renderScrapersGrid(state.scrapers); + // Load Chrome Extension status + loadExtensionStatus(); + // Setup event listeners for control panel buttons setupControlPanelListeners(); @@ -796,10 +799,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/discord_notifier.py b/src/discord_notifier.py index d34db72..3b4e07e 100644 --- a/src/discord_notifier.py +++ b/src/discord_notifier.py @@ -10,10 +10,69 @@ import requests from datetime import datetime from pathlib import Path from typing import Optional, Dict, Set -from config import DISCORD_WEBHOOK_URL, NOTIFICATIONS_ENABLED, SKIP_DUPLICATE_SKUS +from config import DISCORD_WEBHOOK_URL, DISCORD_WEBHOOKS, NOTIFICATIONS_ENABLED, SKIP_DUPLICATE_SKUS logger = logging.getLogger(__name__) + +def _normalize_site_name(site: str) -> str: + """Normalize site name to match config keys (lowercase, no spaces).""" + if not site: + return "" + # Remove spaces and convert to lowercase + # "Best Buy" -> "bestbuy", "Pokemon Center" -> "pokemoncenter" + return site.lower().replace(" ", "").replace("-", "") + + +def get_webhooks(site: str = None, alert_type: str = None) -> list: + """ + Get all applicable webhook URLs based on site and alert type. + Sends to BOTH site-specific AND type-specific channels. + + Args: + site: Site name (target, bestbuy, gamestop, walmart, pokemoncenter) + alert_type: Alert type (restock, new_drop, price_change, error, captcha) + + Returns: + List of webhook URLs to send to (deduplicated) + """ + webhooks = [] + normalized_site = _normalize_site_name(site) + normalized_type = alert_type.lower() if alert_type else None + + # Check for specific override first (highest priority - replaces both site and type) + if normalized_site and normalized_type: + overrides = DISCORD_WEBHOOKS.get("overrides", {}) + override_url = overrides.get((normalized_site, normalized_type)) + if override_url: + return [override_url] # Override replaces all other routing + + # Add site-specific webhook + if normalized_site: + sites = DISCORD_WEBHOOKS.get("sites", {}) + site_url = sites.get(normalized_site) + if site_url: + webhooks.append(site_url) + + # Add type-specific webhook + if normalized_type: + types = DISCORD_WEBHOOKS.get("types", {}) + type_url = types.get(normalized_type) + if type_url and type_url not in webhooks: + webhooks.append(type_url) + + # Fall back to default if no specific webhooks found + if not webhooks: + webhooks.append(DISCORD_WEBHOOK_URL) + + return webhooks + + +def get_webhook(site: str = None, alert_type: str = None) -> str: + """Legacy single-webhook function for backwards compatibility.""" + webhooks = get_webhooks(site, alert_type) + return webhooks[0] if webhooks else DISCORD_WEBHOOK_URL + # Track notified products to avoid duplicates NOTIFIED_FILE = Path(__file__).parent.parent / "data" / "notified_products.json" _notified_urls: Set[str] = set() @@ -56,6 +115,7 @@ def mark_as_notified(product_url: str): COLOR_RESTOCK = 0x00FF00 # Green - item back in stock COLOR_NEW_DROP = 0x0099FF # Blue - new product listing COLOR_PREORDER = 0xFFAA00 # Orange - pre-order available +COLOR_PRICE_CHANGE = 0x9B59B6 # Purple - price changed COLOR_ERROR = 0xFF0000 # Red - error notification COLOR_FAVORITE_HIGH = 0xFF0000 # Red - high priority favorite COLOR_FAVORITE_MEDIUM = 0xFFAA00 # Orange - medium priority favorite @@ -109,12 +169,13 @@ def send_stock_alert( logger.debug(f"Already notified about this product - skipping: {product_name}") return True - # Determine which webhook to use - webhook_url = DISCORD_WEBHOOK_URL + # Determine which webhooks to use (priority: custom > routing > default) if priority_settings and priority_settings.get('custom_webhook'): - webhook_url = priority_settings['custom_webhook'] + webhook_urls = [priority_settings['custom_webhook']] + else: + webhook_urls = get_webhooks(site=site, alert_type=alert_type) - if webhook_url == "YOUR_WEBHOOK_URL_HERE": + if not webhook_urls or webhook_urls[0] == "YOUR_WEBHOOK_URL_HERE": logger.error("Discord webhook URL not configured! Update config.py") return False @@ -189,21 +250,26 @@ def send_stock_alert( "embeds": [embed], } - try: - response = requests.post( - webhook_url, - json=payload, - timeout=10, - ) - response.raise_for_status() - logger.info(f"Discord notification sent for: {product_name} (priority: {priority})") + # Send to all applicable webhooks + success = False + for webhook_url in webhook_urls: + try: + response = requests.post( + webhook_url, + json=payload, + timeout=10, + ) + response.raise_for_status() + success = True + except requests.exceptions.RequestException as e: + logger.error(f"Failed to send Discord notification to {webhook_url}: {e}") + + if success: + logger.info(f"Discord notification sent for: {product_name} to {len(webhook_urls)} channel(s)") # Track that we notified about this product if alert_type == "new_drop": mark_as_notified(product_url) - return True - except requests.exceptions.RequestException as e: - logger.error(f"Failed to send Discord notification: {e}") - return False + return success def send_stock_alert_with_priority(product, alert_type: str = "restock", @@ -231,7 +297,9 @@ def send_error_notification(error_message: str, site: str = "Unknown"): """Send an error notification to Discord""" if not NOTIFICATIONS_ENABLED: return True - if DISCORD_WEBHOOK_URL == "YOUR_WEBHOOK_URL_HERE": + + webhook_urls = get_webhooks(site=site, alert_type="error") + if not webhook_urls or webhook_urls[0] == "YOUR_WEBHOOK_URL_HERE": return False embed = { @@ -245,14 +313,158 @@ def send_error_notification(error_message: str, site: str = "Unknown"): payload = {"embeds": [embed]} - try: - response = requests.post(DISCORD_WEBHOOK_URL, json=payload, timeout=10) - response.raise_for_status() + success = False + for webhook_url in webhook_urls: + try: + response = requests.post(webhook_url, json=payload, timeout=10) + response.raise_for_status() + success = True + except requests.exceptions.RequestException as e: + logger.error(f"Failed to send error notification: {e}") + return success + + +def send_captcha_alert(site: str, url: str, message: str = None): + """ + Send a Discord notification when CAPTCHA is detected and human intervention is needed. + + Args: + site: Site name where CAPTCHA was detected (e.g., "PokemonCenter") + url: URL that triggered the CAPTCHA + message: Optional custom message + """ + if not NOTIFICATIONS_ENABLED: return True - except requests.exceptions.RequestException as e: - logger.error(f"Failed to send error notification: {e}") + + webhook_urls = get_webhooks(site=site, alert_type="captcha") + if not webhook_urls or webhook_urls[0] == "YOUR_WEBHOOK_URL_HERE": return False + default_message = "Human intervention required to solve CAPTCHA. The scraper is paused until resolved." + description = message or default_message + + embed = { + "title": "\U0001F6A7 CAPTCHA DETECTED - Action Required", + "description": description, + "color": 0xFF6600, # Orange color for warning + "fields": [ + {"name": "Site", "value": site, "inline": True}, + {"name": "URL", "value": f"[Open Page]({url})", "inline": True}, + {"name": "Action Needed", "value": "Please solve the CAPTCHA manually to resume monitoring.", "inline": False}, + ], + "footer": {"text": "Pokemon Stock Monitor"}, + "timestamp": datetime.utcnow().isoformat(), + } + + payload = { + "content": "@everyone \U0001F6A7 CAPTCHA requires manual intervention!", + "embeds": [embed] + } + + success = False + for webhook_url in webhook_urls: + try: + response = requests.post(webhook_url, json=payload, timeout=10) + response.raise_for_status() + success = True + except requests.exceptions.RequestException as e: + logger.error(f"Failed to send CAPTCHA alert: {e}") + + if success: + logger.info(f"CAPTCHA alert sent for {site}") + return success + + +def send_price_change_alert( + product_name: str, + product_url: str, + old_price: str, + new_price: str, + site: str, + in_stock: bool = True, + image_url: Optional[str] = None, +): + """ + Send a Discord notification when a product's price changes. + + Args: + product_name: Name of the product + product_url: Direct link to the product + old_price: Previous price string + new_price: New price string + site: Site name (target, bestbuy, etc.) + in_stock: Whether the product is currently in stock + image_url: Optional product image URL + """ + if not NOTIFICATIONS_ENABLED: + logger.debug(f"Notifications disabled - skipping price change alert for: {product_name}") + return True + + webhook_urls = get_webhooks(site=site, alert_type="price_change") + if not webhook_urls or webhook_urls[0] == "YOUR_WEBHOOK_URL_HERE": + logger.error("Discord webhook URL not configured! Update config.py") + return False + + # Determine if price went up or down + try: + old_val = float(old_price.replace("$", "").replace(",", "").strip()) if old_price else 0 + new_val = float(new_price.replace("$", "").replace(",", "").strip()) if new_price else 0 + price_dropped = new_val < old_val + except (ValueError, AttributeError): + price_dropped = False + + # Use green for price drop, purple for price increase + color = 0x00FF00 if price_dropped else COLOR_PRICE_CHANGE + + if price_dropped: + title = "\U0001F4B0 PRICE DROP!" + price_change_text = f"~~{old_price}~~ → **{new_price}**" + else: + title = "\U0001F4C8 Price Changed" + price_change_text = f"{old_price} → {new_price}" + + site_emoji = SITE_EMOJIS.get(site.lower(), "\U0001F6D2") + site_display = site.replace("pokemoncenter", "Pokemon Center").title() + stock_status = "\u2705 In Stock" if in_stock else "\u274C Out of Stock" + + embed = { + "title": title, + "description": f"**{product_name}**", + "url": product_url, + "color": color, + "fields": [ + {"name": "Price Change", "value": price_change_text, "inline": True}, + {"name": "Store", "value": f"{site_emoji} {site_display}", "inline": True}, + {"name": "Stock Status", "value": stock_status, "inline": True}, + {"name": "Link", "value": f"[\U0001F6D2 View Product]({product_url})", "inline": False}, + ], + "footer": {"text": "Pokemon Stock Monitor"}, + "timestamp": datetime.utcnow().isoformat(), + } + + if image_url: + embed["thumbnail"] = {"url": image_url} + + # Only ping everyone for price drops on in-stock items + content = "@everyone" if price_dropped and in_stock else None + + payload = {"embeds": [embed]} + if content: + payload["content"] = content + + success = False + for webhook_url in webhook_urls: + try: + response = requests.post(webhook_url, json=payload, timeout=10) + response.raise_for_status() + success = True + except requests.exceptions.RequestException as e: + logger.error(f"Failed to send price change notification: {e}") + + if success: + logger.info(f"Price change notification sent for: {product_name} ({old_price} -> {new_price}) to {len(webhook_urls)} channel(s)") + return success + def send_startup_notification(): """Send a notification that the monitor has started""" diff --git a/src/product_tracker.py b/src/product_tracker.py index b010286..7f039a8 100644 --- a/src/product_tracker.py +++ b/src/product_tracker.py @@ -9,8 +9,9 @@ import re 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 @@ -60,7 +61,7 @@ class ProductTracker: url = re.sub(r'/sku/\d+$', '', url) return url.rstrip('/') - 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. @@ -69,10 +70,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 = self._normalize_url(product.url) @@ -127,22 +130,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: @@ -154,17 +184,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 @@ -181,7 +212,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"