commit 9d49a99916d1b40f7e8d12d865dfe6460b025696 Author: Mike McGhen Date: Tue Mar 24 12:39:09 2026 -0400 Pokemon Stock Monitor - Initial commit Chrome extension for PokemonCenter monitoring with Discord notifications. Includes Python scripts for Target monitoring. Co-Authored-By: Claude Opus 4.5 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6f0ecc3 --- /dev/null +++ b/.gitignore @@ -0,0 +1,20 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +.env +venv/ +.venv/ + +# Logs and data +*.log +products.json +debug_*.png + +# IDE +.vscode/ +.idea/ + +# OS +.DS_Store +Thumbs.db diff --git a/SETUP_GUIDE.md b/SETUP_GUIDE.md new file mode 100644 index 0000000..9dd1837 --- /dev/null +++ b/SETUP_GUIDE.md @@ -0,0 +1,286 @@ +# Pokemon Stock Monitor - Setup Guide + +## Overview +This monitor watches retail sites for Pokemon TCG restocks and new drops, sending Discord notifications when products become available. + +**Supported Sites:** +- Target (working) +- PokemonCenter (requires special setup - see below) +- Walmart (planned) +- BestBuy (planned) + +--- + +## Prerequisites + +- Python 3.10+ +- Chrome/Chromium browser +- Discord server with webhook access +- ~500MB disk space (for browser) + +--- + +## Quick Start (5 minutes) + +### 1. Clone/Copy the Project +```bash +# Copy the pokemon-stock-monitor folder to your server +cd pokemon-stock-monitor +``` + +### 2. Install Dependencies +```bash +pip install -r requirements.txt +playwright install chromium +``` + +### 3. Configure Discord Webhook + +1. Open Discord → Your Server → Server Settings → Integrations → Webhooks +2. Click "New Webhook" +3. Name it "Pokemon Stock Monitor" +4. Copy the Webhook URL +5. Edit `config.py`: +```python +DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/YOUR_WEBHOOK_URL_HERE" +``` + +### 4. Configure Sites to Monitor + +Edit `config.py`: +```python +# Enable the sites you want +SITES_ENABLED = ["target"] # Add "pokemoncenter" after special setup + +# Customize check interval (seconds) +CHECK_INTERVAL_SECONDS = 60 + +# For headless server, set to True +HEADLESS = True +``` + +### 5. Run the Monitor +```bash +python main.py +``` + +You should see: +``` +================================================== +Pokemon Stock Monitor +================================================== +Monitoring: target +Check interval: 60 seconds +================================================== + +Running initial check... +``` + +--- + +## Running as a Background Service + +### Option A: Using Screen (Linux) +```bash +# Start a screen session +screen -S pokemon-monitor + +# Run the monitor +python main.py + +# Detach: Press Ctrl+A, then D +# Reattach later: screen -r pokemon-monitor +``` + +### Option B: Using systemd (Linux) + +Create `/etc/systemd/system/pokemon-monitor.service`: +```ini +[Unit] +Description=Pokemon Stock Monitor +After=network.target + +[Service] +Type=simple +User=YOUR_USERNAME +WorkingDirectory=/path/to/pokemon-stock-monitor +ExecStart=/usr/bin/python3 main.py +Restart=always +RestartSec=10 + +[Install] +WantedBy=multi-user.target +``` + +Then: +```bash +sudo systemctl daemon-reload +sudo systemctl enable pokemon-monitor +sudo systemctl start pokemon-monitor + +# Check status +sudo systemctl status pokemon-monitor + +# View logs +journalctl -u pokemon-monitor -f +``` + +### Option C: Using Task Scheduler (Windows) + +1. Open Task Scheduler +2. Create Basic Task → Name: "Pokemon Monitor" +3. Trigger: "When the computer starts" +4. Action: Start a program + - Program: `python` + - Arguments: `main.py` + - Start in: `C:\path\to\pokemon-stock-monitor` +5. Check "Run whether user is logged on or not" + +--- + +## PokemonCenter Setup (Special - Requires GUI) + +PokemonCenter has strong bot protection (Imperva). To monitor it: + +### Method 1: Use Real Chrome Profile (Recommended) + +This uses your actual Chrome browser with all its cookies/history, making it appear human. + +1. **Find your Chrome profile path:** + - Windows: `C:\Users\USERNAME\AppData\Local\Google\Chrome\User Data` + - Linux: `~/.config/google-chrome` + - Mac: `~/Library/Application Support/Google/Chrome` + +2. **Edit `config.py`:** +```python +# Add this line +CHROME_USER_DATA_DIR = "C:\\Users\\USERNAME\\AppData\\Local\\Google\\Chrome\\User Data" + +# Enable PokemonCenter +SITES_ENABLED = ["target", "pokemoncenter"] + +# Must be False to use Chrome profile +HEADLESS = False +``` + +3. **Important:** Close Chrome before running the monitor (can't use same profile twice) + +4. **First run:** Manually solve any CAPTCHA that appears, then the session should stay valid + +### Method 2: VM with Desktop Environment + +If running on a headless server, set up a VM with a desktop: + +1. Install a lightweight desktop (XFCE, LXDE) +2. Install Chrome and browse PokemonCenter manually once +3. Run the monitor with `HEADLESS = False` +4. Use VNC/RDP to check on it occasionally + +--- + +## Configuration Reference + +### config.py Options + +```python +# Discord webhook URL (required) +DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/..." + +# Check interval in seconds (60 = 1 minute) +CHECK_INTERVAL_SECONDS = 60 + +# Sites to monitor +SITES_ENABLED = ["target"] # Options: "target", "pokemoncenter" + +# URLs to monitor per site +TARGET_URLS = [ + "https://www.target.com/s?searchTerm=pokemon+tcg", +] + +POKEMON_CENTER_URLS = [ + "https://www.pokemoncenter.com/category/tcg-cards?sort=relevance", +] + +# Keyword filtering (optional) +KEYWORD_FILTER_ENABLED = False +KEYWORDS = ["chaos rising", "booster", "etb", "elite trainer"] + +# Browser settings +HEADLESS = True # True for servers, False for desktop/VM +SLOW_MO = 0 # Slow down browser (ms) for debugging + +# Logging level +LOG_LEVEL = "INFO" # DEBUG, INFO, WARNING, ERROR +``` + +--- + +## Troubleshooting + +### "No products found" +- Check `debug_screenshot.png` or `debug_target.png` in the project folder +- The site may have changed its HTML structure +- Try increasing wait times in `browser.py` + +### "Discord notification not sending" +- Verify webhook URL is correct in `config.py` +- Test webhook: `python -c "from discord_notifier import test_webhook; test_webhook()"` + +### "Browser failed to start" +```bash +# Reinstall Playwright browsers +playwright install chromium --force + +# On Linux, install dependencies +playwright install-deps chromium +``` + +### "Access denied" / Bot blocked +- PokemonCenter: Use the Chrome profile method above +- Target: Should work, try increasing `CHECK_INTERVAL_SECONDS` to 120+ +- All sites: Don't run too frequently (rate limiting) + +### High CPU/Memory Usage +- Set `HEADLESS = True` (uses less resources) +- Increase `CHECK_INTERVAL_SECONDS` +- The browser stays open between checks (by design, for session persistence) + +--- + +## File Structure + +``` +pokemon-stock-monitor/ +├── main.py # Entry point - run this +├── config.py # Configuration - edit this +├── browser.py # Browser automation +├── discord_notifier.py # Discord webhook +├── product_tracker.py # Tracks products for restock detection +├── products.json # Auto-generated product database +├── monitor.log # Log file +├── scrapers/ +│ ├── base.py # Base scraper class +│ ├── pokemoncenter.py # PokemonCenter scraper +│ └── target.py # Target scraper +└── requirements.txt # Python dependencies +``` + +--- + +## Adding More Sites + +The scraper architecture is modular. To add a new site: + +1. Create `scrapers/newsite.py` based on `target.py` +2. Add to `scrapers/__init__.py` +3. Add URL config to `config.py` +4. Add check function to `main.py` + +--- + +## Support + +If you encounter issues: +1. Check `monitor.log` for errors +2. Check debug screenshots (`debug_*.png`) +3. Try running with `LOG_LEVEL = "DEBUG"` for more info diff --git a/browser.py b/browser.py new file mode 100644 index 0000000..3c8af36 --- /dev/null +++ b/browser.py @@ -0,0 +1,183 @@ +""" +Browser automation with Playwright and stealth patches +Handles bot protection bypass for retail sites +""" + +import logging +from playwright.sync_api import sync_playwright, Browser, Page, BrowserContext +from playwright_stealth import Stealth +from config import HEADLESS, SLOW_MO, CHROME_USER_DATA_DIR, USE_REAL_CHROME + +logger = logging.getLogger(__name__) + +# Create stealth instance +stealth = Stealth() + + +class StealthBrowser: + """Manages a stealth browser instance that bypasses bot detection""" + + def __init__(self): + self.playwright = None + self.browser: Browser = None + self.context: BrowserContext = None + self.using_persistent = False + + def start(self): + """Initialize the browser""" + logger.info("Starting stealth browser...") + self.playwright = sync_playwright().start() + + # Check if we should use a persistent Chrome profile + if CHROME_USER_DATA_DIR: + logger.info(f"Using Chrome profile from: {CHROME_USER_DATA_DIR}") + self._start_with_profile() + else: + self._start_fresh() + + logger.info("Browser started successfully") + + def _start_fresh(self): + """Start with a fresh browser profile""" + # Launch browser - use real Chrome if configured + launch_options = { + "headless": HEADLESS, + "slow_mo": SLOW_MO, + "args": [ + "--disable-blink-features=AutomationControlled", + "--disable-dev-shm-usage", + "--no-sandbox", + ] + } + + if USE_REAL_CHROME: + launch_options["channel"] = "chrome" + logger.info("Using real Chrome browser") + + self.browser = self.playwright.chromium.launch(**launch_options) + + # Create context with realistic viewport and user agent + self.context = self.browser.new_context( + viewport={"width": 1920, "height": 1080}, + user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", + locale="en-US", + timezone_id="America/New_York", + ) + + # Apply stealth to context (affects all pages created from it) + stealth.apply_stealth_sync(self.context) + + def _start_with_profile(self): + """Start using an existing Chrome profile (better for bot detection)""" + self.using_persistent = True + + # Launch persistent context with user's REAL Chrome (not Playwright's Chromium) + # This is needed because Chrome encrypts credentials with its own binary + self.context = self.playwright.chromium.launch_persistent_context( + user_data_dir=CHROME_USER_DATA_DIR, + channel="chrome", # Use real Chrome, not Playwright's Chromium + headless=HEADLESS, + slow_mo=SLOW_MO, + viewport={"width": 1920, "height": 1080}, + locale="en-US", + timezone_id="America/New_York", + args=[ + "--disable-blink-features=AutomationControlled", + "--disable-dev-shm-usage", + ] + ) + + # Apply stealth + stealth.apply_stealth_sync(self.context) + + def new_page(self) -> Page: + """Create a new page with stealth patches applied""" + if not self.context: + raise RuntimeError("Browser not started. Call start() first.") + + page = self.context.new_page() + return page + + def get_page_content(self, url: str, wait_for_selector: str = None, timeout: int = 30000) -> tuple[Page, str]: + """ + Navigate to URL and return page + HTML content + + Args: + url: URL to navigate to + wait_for_selector: CSS selector to wait for before returning + timeout: Max time to wait in milliseconds + + Returns: + Tuple of (page, html_content) + """ + page = self.new_page() + + try: + logger.info(f"Navigating to: {url}") + page.goto(url, wait_until="domcontentloaded", timeout=timeout) + + # Wait for specific element if provided + if wait_for_selector: + logger.debug(f"Waiting for selector: {wait_for_selector}") + page.wait_for_selector(wait_for_selector, timeout=timeout) + else: + # Wait for DOM to be ready, then give extra time for JS rendering + try: + page.wait_for_load_state("networkidle", timeout=15000) + except: + # networkidle can timeout on dynamic sites, that's OK + logger.debug("networkidle timeout, continuing anyway") + + # Let dynamic content render + page.wait_for_timeout(5000) + + # Scroll down to trigger lazy loading + page.evaluate("window.scrollTo(0, document.body.scrollHeight / 2)") + page.wait_for_timeout(2000) + page.evaluate("window.scrollTo(0, document.body.scrollHeight)") + page.wait_for_timeout(2000) + page.evaluate("window.scrollTo(0, 0)") + page.wait_for_timeout(1000) + + content = page.content() + logger.info(f"Page loaded, content length: {len(content)}") + + return page, content + + except Exception as e: + logger.error(f"Error loading page {url}: {e}") + page.close() + raise + + def stop(self): + """Clean up browser resources""" + logger.info("Stopping browser...") + if self.context: + self.context.close() + if self.browser and not self.using_persistent: + # Only close browser if not using persistent context + self.browser.close() + if self.playwright: + self.playwright.stop() + logger.info("Browser stopped") + + +# Global browser instance +_browser: StealthBrowser = None + + +def get_browser() -> StealthBrowser: + """Get or create the global browser instance""" + global _browser + if _browser is None: + _browser = StealthBrowser() + _browser.start() + return _browser + + +def shutdown_browser(): + """Shutdown the global browser instance""" + global _browser + if _browser: + _browser.stop() + _browser = None diff --git a/chrome-extension/README.md b/chrome-extension/README.md new file mode 100644 index 0000000..5a865e9 --- /dev/null +++ b/chrome-extension/README.md @@ -0,0 +1,68 @@ +# Pokemon Stock Monitor - Chrome Extension + +A Chrome extension that monitors PokemonCenter for stock changes and sends Discord notifications. + +## Why a Chrome Extension? + +PokemonCenter uses Imperva bot protection that blocks automated browsers (Playwright, Selenium). This extension runs **inside your real Chrome browser**, making it undetectable - to the website, it looks exactly like you browsing manually. + +## Installation + +1. **Add Icons** (optional but recommended): + - Create or download three PNG icons: `icon16.png`, `icon48.png`, `icon128.png` + - Place them in this folder + - You can use any Pokemon-themed icons or simple colored squares + +2. **Load the Extension**: + - Open Chrome and go to `chrome://extensions` + - Enable "Developer mode" (toggle in top right) + - Click "Load unpacked" + - Select this `chrome-extension` folder + - The extension icon should appear in your toolbar + +3. **Configure**: + - Click the extension icon + - Paste your Discord webhook URL + - Add URLs to monitor (default is PokemonCenter TCG) + - Set check interval (1-60 minutes) + - Click "Save Settings" + +## How It Works + +1. **Background Service Worker**: Runs continuously in Chrome +2. **Periodic Checks**: Uses Chrome's alarm API to check at your interval +3. **Product Parsing**: Fetches pages and parses product data from HTML +4. **Change Detection**: Compares against known products to detect: + - New product listings + - Restocks (was out of stock, now in stock) +5. **Discord Notifications**: Sends rich embeds with product info and direct links + +## Features + +- **Keyword Filtering**: Only track products matching specific keywords +- **New Product Alerts**: Get notified when new items appear +- **Restock Alerts**: Get notified when out-of-stock items return +- **Browser Notifications**: Local notifications in addition to Discord +- **Persistent Storage**: Remembers products across browser restarts + +## Tips + +- **Keep Chrome Running**: The extension only works while Chrome is open +- **First Run**: After installing, click "Check Now" to do an initial scan +- **Clear History**: Use this if you want to re-detect all products as "new" +- **Multiple URLs**: Add one URL per line in the URLs field + +## Troubleshooting + +**No notifications sending:** +- Check that your Discord webhook URL is correct +- Make sure "Monitor Enabled" is toggled on +- Check Chrome's console for errors (right-click extension → Inspect popup) + +**Extension not loading:** +- Make sure manifest.json is valid JSON +- Check for any icon files referenced but missing + +**Products not detected:** +- PokemonCenter may have changed their HTML structure +- Check the browser console for parsing errors diff --git a/chrome-extension/background.js b/chrome-extension/background.js new file mode 100644 index 0000000..3410066 --- /dev/null +++ b/chrome-extension/background.js @@ -0,0 +1,317 @@ +// Pokemon Stock Monitor - Background Service Worker + +const DEFAULT_CONFIG = { + discordWebhook: "", + checkIntervalMinutes: 1, + enabled: true, + urls: [ + "https://www.pokemoncenter.com/category/tcg-cards?sort=relevance" + ], + keywords: [], // Empty = all products, or ["chaos rising", "booster"] etc + notifyNewProducts: true, + notifyRestocks: true +}; + +// Store known products +let knownProducts = {}; +let config = DEFAULT_CONFIG; + +// Initialize +chrome.runtime.onInstalled.addListener(() => { + console.log("Pokemon Stock Monitor installed"); + loadConfig(); + loadProducts(); + setupAlarm(); +}); + +// Load on startup +chrome.runtime.onStartup.addListener(() => { + loadConfig(); + loadProducts(); + setupAlarm(); +}); + +// Handle alarm +chrome.alarms.onAlarm.addListener((alarm) => { + if (alarm.name === "stockCheck") { + runStockCheck(); + } +}); + +// Setup periodic alarm +function setupAlarm() { + chrome.alarms.create("stockCheck", { + periodInMinutes: config.checkIntervalMinutes + }); + console.log(`Alarm set for every ${config.checkIntervalMinutes} minute(s)`); +} + +// Load config from storage +async function loadConfig() { + const stored = await chrome.storage.local.get("config"); + if (stored.config) { + config = { ...DEFAULT_CONFIG, ...stored.config }; + } + console.log("Config loaded:", config); +} + +// Save config to storage +async function saveConfig() { + await chrome.storage.local.set({ config }); + setupAlarm(); // Reset alarm with new interval +} + +// Load known products from storage +async function loadProducts() { + const stored = await chrome.storage.local.get("knownProducts"); + if (stored.knownProducts) { + knownProducts = stored.knownProducts; + } + console.log(`Loaded ${Object.keys(knownProducts).length} known products`); +} + +// Save known products to storage +async function saveProducts() { + await chrome.storage.local.set({ knownProducts }); +} + +// Main stock check function +async function runStockCheck() { + if (!config.enabled) { + console.log("Stock check disabled"); + return; + } + + console.log("Running stock check..."); + + for (const url of config.urls) { + try { + const products = await fetchAndParseProducts(url); + const { newProducts, restockedProducts } = processProducts(products); + + // Send notifications + if (config.notifyNewProducts) { + for (const product of newProducts) { + if (product.inStock) { + await sendDiscordNotification(product, "new_drop"); + } + } + } + + if (config.notifyRestocks) { + for (const product of restockedProducts) { + await sendDiscordNotification(product, "restock"); + } + } + + console.log(`Check complete: ${products.length} products, ${newProducts.length} new, ${restockedProducts.length} restocks`); + + } catch (error) { + console.error(`Error checking ${url}:`, error); + } + } + + await saveProducts(); +} + +// Fetch and parse products from a URL using content script +async function fetchAndParseProducts(url) { + return new Promise(async (resolve, reject) => { + try { + // Create a tab to load the page + const tab = await chrome.tabs.create({ url, active: false }); + + console.log(`Created tab ${tab.id} for ${url}`); + + // Wait for tab to finish loading + const waitForLoad = () => { + return new Promise((res) => { + const listener = (tabId, changeInfo) => { + if (tabId === tab.id && changeInfo.status === "complete") { + chrome.tabs.onUpdated.removeListener(listener); + res(); + } + }; + chrome.tabs.onUpdated.addListener(listener); + + // Timeout after 30 seconds + setTimeout(() => { + chrome.tabs.onUpdated.removeListener(listener); + res(); + }, 30000); + }); + }; + + await waitForLoad(); + console.log(`Tab ${tab.id} loaded`); + + // Give extra time for dynamic content to load + console.log("Waiting for dynamic content..."); + await new Promise(r => setTimeout(r, 5000)); + + // Send message to content script to extract products + let products = []; + try { + const response = await chrome.tabs.sendMessage(tab.id, { type: "extractProducts" }); + products = response || []; + console.log(`Content script returned ${products.length} products`); + } catch (e) { + console.log("Content script not ready, injecting manually..."); + // Inject content script if not already there + await chrome.scripting.executeScript({ + target: { tabId: tab.id }, + files: ["content.js"] + }); + await new Promise(r => setTimeout(r, 500)); + const response = await chrome.tabs.sendMessage(tab.id, { type: "extractProducts" }); + products = response || []; + } + + // Apply keyword filter + if (config.keywords && config.keywords.length > 0) { + products = products.filter(p => { + const nameLower = p.name.toLowerCase(); + return config.keywords.some(kw => nameLower.includes(kw.toLowerCase())); + }); + console.log(`After keyword filter: ${products.length} products`); + } + + // Close the tab + await chrome.tabs.remove(tab.id); + console.log(`Closed tab ${tab.id}`); + + resolve(products); + + } catch (error) { + reject(error); + } + }); +} + +// Note: Product parsing is now done by the content script (content.js) + +// Process products - detect new and restocked +function processProducts(products) { + const newProducts = []; + const restockedProducts = []; + const now = new Date().toISOString(); + + for (const product of products) { + const existing = knownProducts[product.url]; + + if (!existing) { + // New product + newProducts.push(product); + knownProducts[product.url] = { + ...product, + firstSeen: now, + lastSeen: now, + lastInStock: product.inStock ? now : null + }; + } else { + // Existing product - check for restock + if (product.inStock && !existing.inStock) { + restockedProducts.push(product); + existing.lastInStock = now; + } + + // Update + existing.inStock = product.inStock; + existing.price = product.price || existing.price; + existing.lastSeen = now; + } + } + + return { newProducts, restockedProducts }; +} + +// Send Discord notification +async function sendDiscordNotification(product, alertType) { + if (!config.discordWebhook) { + console.log("No Discord webhook configured"); + return; + } + + const color = alertType === "restock" ? 0x00FF00 : 0x0099FF; + const title = alertType === "restock" ? "RESTOCK ALERT" : "NEW DROP"; + + const embed = { + title: title, + description: `**${product.name}**`, + url: product.url, + color: color, + fields: [ + { name: "Price", value: product.price || "See link", inline: true }, + { name: "Store", value: "Pokemon Center", inline: true }, + { name: "Link", value: `[BUY NOW](${product.url})`, inline: false } + ], + footer: { text: "Pokemon Stock Monitor (Chrome Extension)" }, + timestamp: new Date().toISOString() + }; + + if (product.imageUrl) { + embed.thumbnail = { url: product.imageUrl }; + } + + try { + const response = await fetch(config.discordWebhook, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + content: "@everyone", + embeds: [embed] + }) + }); + + if (response.ok) { + console.log(`Discord notification sent for: ${product.name}`); + + // Also show browser notification + chrome.notifications.create({ + type: "basic", + iconUrl: "icon128.png", + title: title, + message: product.name + }); + } else { + console.error("Discord notification failed:", response.status); + } + } catch (error) { + console.error("Error sending Discord notification:", error); + } +} + +// Listen for messages from popup +chrome.runtime.onMessage.addListener((message, sender, sendResponse) => { + if (message.type === "getConfig") { + sendResponse(config); + } else if (message.type === "saveConfig") { + config = { ...config, ...message.config }; + saveConfig(); + sendResponse({ success: true }); + } else if (message.type === "runCheck") { + runStockCheck(); + sendResponse({ success: true }); + } else if (message.type === "getStats") { + sendResponse({ + totalProducts: Object.keys(knownProducts).length, + enabled: config.enabled + }); + } else if (message.type === "clearProducts") { + knownProducts = {}; + saveProducts(); + sendResponse({ success: true }); + } + return true; +}); + +// Run initial check after a short delay +setTimeout(() => { + loadConfig().then(() => { + loadProducts().then(() => { + if (config.enabled && config.discordWebhook) { + runStockCheck(); + } + }); + }); +}, 5000); diff --git a/chrome-extension/content.js b/chrome-extension/content.js new file mode 100644 index 0000000..276f4ee --- /dev/null +++ b/chrome-extension/content.js @@ -0,0 +1,109 @@ +// Pokemon Stock Monitor - Content Script +// Runs on PokemonCenter pages to extract product data + +(function() { + console.log("[Pokemon Monitor] Content script loaded on:", window.location.href); + + // Listen for messages from background script + chrome.runtime.onMessage.addListener((message, sender, sendResponse) => { + if (message.type === "extractProducts") { + console.log("[Pokemon Monitor] Extracting products..."); + const products = extractProducts(); + console.log("[Pokemon Monitor] Found", products.length, "products"); + sendResponse(products); + } + return true; + }); + + function extractProducts() { + const products = []; + + // Group all product links by URL + const urlToLinks = {}; + document.querySelectorAll('a[href*="/product/"]').forEach(link => { + const url = link.href; + if (!urlToLinks[url]) { + urlToLinks[url] = []; + } + urlToLinks[url].push(link); + }); + + console.log("[Pokemon Monitor] Found", Object.keys(urlToLinks).length, "unique product URLs"); + + // Process each unique product URL + for (const [url, links] of Object.entries(urlToLinks)) { + // Find the best name from all links to this product + let bestName = ""; + let isSoldOut = false; + + for (const link of links) { + const text = link.textContent.trim(); + + // Check if any link says "SOLD OUT" + if (text.toUpperCase().includes("SOLD OUT")) { + isSoldOut = true; + } + + // Pick the longest non-"SOLD OUT" text as the name + if (text.length > bestName.length && + !text.toUpperCase().startsWith("SOLD") && + text.length > 10) { + bestName = text; + } + } + + // Skip if we couldn't find a good name + if (!bestName || bestName.length < 10) continue; + + // Clean up name - remove price if embedded + bestName = bestName.replace(/\$[\d,.]+/g, "").trim(); + // Remove "Add to Cart" etc + bestName = bestName.replace(/Add to Cart/gi, "").trim(); + // Clean whitespace + bestName = bestName.replace(/\s+/g, " ").trim(); + + // Find price from any of the links' containers + let price = null; + for (const link of links) { + let parent = link.parentElement; + for (let i = 0; i < 6 && parent && !price; i++) { + const priceMatch = parent.textContent.match(/\$[\d,]+\.?\d*/); + if (priceMatch) { + price = priceMatch[0]; + break; + } + parent = parent.parentElement; + } + if (price) break; + } + + // Find image from any of the links + let imageUrl = null; + for (const link of links) { + const img = link.querySelector("img") || + link.closest("[class*='product']")?.querySelector("img"); + if (img) { + imageUrl = img.src || img.dataset.src; + if (imageUrl && !imageUrl.startsWith("data:")) break; + } + } + + // Extract product ID from URL + const idMatch = url.match(/\/product\/([^\/]+)/); + const productId = idMatch ? idMatch[1] : url; + + products.push({ + name: bestName, + url: url, + price: price, + inStock: !isSoldOut, + imageUrl: imageUrl, + productId: productId, + site: "pokemoncenter" + }); + } + + console.log("[Pokemon Monitor] Extracted products:", products.map(p => ({name: p.name.slice(0,40), inStock: p.inStock}))); + return products; + } +})(); diff --git a/chrome-extension/manifest.json b/chrome-extension/manifest.json new file mode 100644 index 0000000..e0f03e8 --- /dev/null +++ b/chrome-extension/manifest.json @@ -0,0 +1,30 @@ +{ + "manifest_version": 3, + "name": "Pokemon Stock Monitor", + "version": "1.0.1", + "description": "Monitors PokemonCenter for restocks and new drops, sends Discord notifications", + "permissions": [ + "alarms", + "storage", + "notifications", + "tabs", + "scripting" + ], + "host_permissions": [ + "https://www.pokemoncenter.com/*", + "https://discord.com/api/*" + ], + "background": { + "service_worker": "background.js" + }, + "action": { + "default_popup": "popup.html" + }, + "content_scripts": [ + { + "matches": ["https://www.pokemoncenter.com/*"], + "js": ["content.js"], + "run_at": "document_idle" + } + ] +} diff --git a/chrome-extension/popup.html b/chrome-extension/popup.html new file mode 100644 index 0000000..21ee715 --- /dev/null +++ b/chrome-extension/popup.html @@ -0,0 +1,254 @@ + + + + + + + +

⚡ Pokemon Stock Monitor

+ +
+
+ Status: + Active +
+
+ Products tracked: + 0 +
+
+ +
+ + +
+ +
+ + +
+ +
+ + +
+ +
+
+ + +
+
+ +
+
+ +
+ + +
+
+
+ +
+ + +
+
+
+ +
+ + +
+
+
+ + + + + +
Settings saved!
+ + + + diff --git a/chrome-extension/popup.js b/chrome-extension/popup.js new file mode 100644 index 0000000..f55edb0 --- /dev/null +++ b/chrome-extension/popup.js @@ -0,0 +1,75 @@ +// Pokemon Stock Monitor - Popup Script + +document.addEventListener("DOMContentLoaded", async () => { + // Load current config + const config = await chrome.runtime.sendMessage({ type: "getConfig" }); + const stats = await chrome.runtime.sendMessage({ type: "getStats" }); + + // Populate form + document.getElementById("webhook").value = config.discordWebhook || ""; + document.getElementById("urls").value = (config.urls || []).join("\n"); + document.getElementById("keywords").value = (config.keywords || []).join(", "); + document.getElementById("interval").value = config.checkIntervalMinutes || 1; + document.getElementById("enabled").checked = config.enabled !== false; + document.getElementById("notifyNew").checked = config.notifyNewProducts !== false; + document.getElementById("notifyRestock").checked = config.notifyRestocks !== false; + + // Update stats + document.getElementById("productCount").textContent = stats.totalProducts || 0; + updateStatus(stats.enabled !== false); + + // Save button + document.getElementById("saveBtn").addEventListener("click", async () => { + const urlsText = document.getElementById("urls").value; + const keywordsText = document.getElementById("keywords").value; + + const newConfig = { + discordWebhook: document.getElementById("webhook").value.trim(), + urls: urlsText.split("\n").map(u => u.trim()).filter(u => u), + keywords: keywordsText.split(",").map(k => k.trim()).filter(k => k), + checkIntervalMinutes: parseInt(document.getElementById("interval").value) || 1, + enabled: document.getElementById("enabled").checked, + notifyNewProducts: document.getElementById("notifyNew").checked, + notifyRestocks: document.getElementById("notifyRestock").checked + }; + + await chrome.runtime.sendMessage({ type: "saveConfig", config: newConfig }); + updateStatus(newConfig.enabled); + showSaved(); + }); + + // Check Now button + document.getElementById("checkNowBtn").addEventListener("click", async () => { + await chrome.runtime.sendMessage({ type: "runCheck" }); + showSaved("Check started!"); + }); + + // Clear button + document.getElementById("clearBtn").addEventListener("click", async () => { + if (confirm("Clear all tracked products? This will treat all products as new on next check.")) { + await chrome.runtime.sendMessage({ type: "clearProducts" }); + document.getElementById("productCount").textContent = "0"; + showSaved("History cleared!"); + } + }); +}); + +function updateStatus(enabled) { + const dot = document.getElementById("statusDot"); + const text = document.getElementById("statusText"); + + if (enabled) { + dot.className = "status-dot active"; + text.textContent = "Active"; + } else { + dot.className = "status-dot inactive"; + text.textContent = "Disabled"; + } +} + +function showSaved(msg = "Settings saved!") { + const el = document.getElementById("savedMsg"); + el.textContent = msg; + el.classList.add("show"); + setTimeout(() => el.classList.remove("show"), 2000); +} diff --git a/config.py b/config.py new file mode 100644 index 0000000..a201149 --- /dev/null +++ b/config.py @@ -0,0 +1,55 @@ +""" +Configuration for Pokemon Stock Monitor +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 = "YOUR_WEBHOOK_URL_HERE" + +# How often to check each site (in seconds) +CHECK_INTERVAL_SECONDS = 60 + +# Which sites to monitor +# Note: PokemonCenter has strong bot protection (Imperva) - may need manual workarounds +SITES_ENABLED = ["target"] # Options: "pokemoncenter", "target", "walmart", "bestbuy" + +# PokemonCenter URLs to monitor +POKEMON_CENTER_URLS = [ + "https://www.pokemoncenter.com/category/tcg-cards?sort=relevance", +] + +# Target URLs to monitor +TARGET_URLS = [ + "https://www.target.com/s?searchTerm=pokemon+tcg", +] + +# Keyword filtering (disabled by default) +KEYWORD_FILTER_ENABLED = False +KEYWORDS = [ + "chaos rising", + "booster", + "etb", + "elite trainer", + "booster bundle", +] + +# Browser settings +HEADLESS = False # Set to False to see the browser (useful for debugging) +SLOW_MO = 0 # Milliseconds to slow down browser actions (for debugging) + +# Chrome profile for PokemonCenter (optional - helps bypass bot detection) +# Set to your Chrome user data directory to use your real browser profile +# Windows: C:\Users\USERNAME\AppData\Local\Google\Chrome\User Data +# Linux: ~/.config/google-chrome +# Mac: ~/Library/Application Support/Google/Chrome +# Leave as None to use a fresh browser profile +# Set to None to use a fresh profile, or a path to use an existing profile +# Note: Chrome must be closed when using an existing profile +CHROME_USER_DATA_DIR = None + +# Use real Chrome instead of Playwright's Chromium (helps with bot detection) +USE_REAL_CHROME = True + +# Logging +LOG_LEVEL = "INFO" diff --git a/discord_notifier.py b/discord_notifier.py new file mode 100644 index 0000000..8f30c19 --- /dev/null +++ b/discord_notifier.py @@ -0,0 +1,183 @@ +""" +Discord webhook notifications for stock alerts +Sends rich embeds with product info and direct links +""" + +import logging +import requests +from datetime import datetime +from typing import Optional +from config import DISCORD_WEBHOOK_URL + +logger = logging.getLogger(__name__) + +# Colors for different notification types +COLOR_RESTOCK = 0x00FF00 # Green - item back in stock +COLOR_NEW_DROP = 0x0099FF # Blue - new product listing +COLOR_PREORDER = 0xFFAA00 # Orange - pre-order available +COLOR_ERROR = 0xFF0000 # Red - error notification + +# Site icons/emojis +SITE_EMOJIS = { + "pokemoncenter": "\U0001F7E1", # Yellow circle + "target": "\U0001F534", # Red circle + "walmart": "\U0001F535", # Blue circle + "bestbuy": "\U0001F7E1", # Yellow circle +} + + +def send_stock_alert( + product_name: str, + product_url: str, + price: str, + site: str, + alert_type: str = "restock", + image_url: Optional[str] = None, +): + """ + Send a Discord notification for a stock alert + + Args: + product_name: Name of the product + product_url: Direct link to the product + price: Price string (e.g., "$49.99") + site: Site name (pokemoncenter, target, etc.) + alert_type: "restock", "new_drop", or "preorder" + image_url: Optional product image URL + """ + if DISCORD_WEBHOOK_URL == "YOUR_WEBHOOK_URL_HERE": + logger.error("Discord webhook URL not configured! Update config.py") + return False + + # Choose color based on alert type + if alert_type == "restock": + color = COLOR_RESTOCK + title = f"\U0001F6A8 RESTOCK ALERT" + elif alert_type == "new_drop": + color = COLOR_NEW_DROP + title = f"\U0001F195 NEW DROP" + elif alert_type == "preorder": + color = COLOR_PREORDER + title = f"\u23F0 PRE-ORDER AVAILABLE" + else: + color = COLOR_RESTOCK + title = f"\U0001F514 STOCK ALERT" + + site_emoji = SITE_EMOJIS.get(site.lower(), "\U0001F6D2") + site_display = site.replace("pokemoncenter", "Pokemon Center").title() + + # Build the embed + embed = { + "title": title, + "description": f"**{product_name}**", + "url": product_url, + "color": color, + "fields": [ + {"name": "Price", "value": price or "See link", "inline": True}, + {"name": "Store", "value": f"{site_emoji} {site_display}", "inline": True}, + {"name": "Link", "value": f"[\U0001F6D2 BUY NOW]({product_url})", "inline": False}, + ], + "footer": {"text": "Pokemon Stock Monitor"}, + "timestamp": datetime.utcnow().isoformat(), + } + + if image_url: + embed["thumbnail"] = {"url": image_url} + + payload = { + "content": "@everyone", # Ping everyone + "embeds": [embed], + } + + try: + response = requests.post( + DISCORD_WEBHOOK_URL, + json=payload, + timeout=10, + ) + response.raise_for_status() + logger.info(f"Discord notification sent for: {product_name}") + return True + except requests.exceptions.RequestException as e: + logger.error(f"Failed to send Discord notification: {e}") + return False + + +def send_error_notification(error_message: str, site: str = "Unknown"): + """Send an error notification to Discord""" + if DISCORD_WEBHOOK_URL == "YOUR_WEBHOOK_URL_HERE": + return False + + embed = { + "title": "\u26A0\uFE0F Monitor Error", + "description": error_message, + "color": COLOR_ERROR, + "fields": [{"name": "Site", "value": site, "inline": True}], + "footer": {"text": "Pokemon Stock Monitor"}, + "timestamp": datetime.utcnow().isoformat(), + } + + payload = {"embeds": [embed]} + + try: + response = requests.post(DISCORD_WEBHOOK_URL, json=payload, timeout=10) + response.raise_for_status() + return True + except requests.exceptions.RequestException as e: + logger.error(f"Failed to send error notification: {e}") + return False + + +def send_startup_notification(): + """Send a notification that the monitor has started""" + if DISCORD_WEBHOOK_URL == "YOUR_WEBHOOK_URL_HERE": + return False + + embed = { + "title": "\u2705 Monitor Started", + "description": "Pokemon Stock Monitor is now running and watching for restocks!", + "color": 0x00FF00, + "footer": {"text": "Pokemon Stock Monitor"}, + "timestamp": datetime.utcnow().isoformat(), + } + + payload = {"embeds": [embed]} + + try: + response = requests.post(DISCORD_WEBHOOK_URL, json=payload, timeout=10) + response.raise_for_status() + return True + except requests.exceptions.RequestException as e: + logger.error(f"Failed to send startup notification: {e}") + return False + + +def test_webhook(): + """Test the Discord webhook connection""" + print("Testing Discord webhook...") + + if DISCORD_WEBHOOK_URL == "YOUR_WEBHOOK_URL_HERE": + print("ERROR: Webhook URL not configured!") + print("Edit config.py and set DISCORD_WEBHOOK_URL") + return False + + # Send a test notification + success = send_stock_alert( + product_name="Test Product - Pokemon TCG Booster", + product_url="https://www.pokemoncenter.com/test", + price="$4.99", + site="pokemoncenter", + alert_type="restock", + ) + + if success: + print("SUCCESS! Check your Discord channel for the test message.") + else: + print("FAILED! Check the webhook URL and try again.") + + return success + + +if __name__ == "__main__": + # Run webhook test + test_webhook() diff --git a/main.py b/main.py new file mode 100644 index 0000000..f61e738 --- /dev/null +++ b/main.py @@ -0,0 +1,232 @@ +""" +Pokemon Stock Monitor - Main entry point +Monitors retail sites for Pokemon card restocks and new drops +""" + +import logging +import time +import signal +import sys +from datetime import datetime + +import schedule + +from config import ( + CHECK_INTERVAL_SECONDS, + SITES_ENABLED, + POKEMON_CENTER_URLS, + TARGET_URLS, + KEYWORD_FILTER_ENABLED, + KEYWORDS, + LOG_LEVEL, +) +from browser import get_browser, shutdown_browser +from product_tracker import ProductTracker +from discord_notifier import send_stock_alert, send_startup_notification, send_error_notification +from scrapers import PokemonCenterScraper, TargetScraper + +# Setup logging +logging.basicConfig( + level=getattr(logging, LOG_LEVEL), + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + handlers=[ + logging.StreamHandler(), + logging.FileHandler("monitor.log", encoding="utf-8"), + ], +) +logger = logging.getLogger(__name__) + +# Global tracker +tracker = ProductTracker() + +# Scrapers +scrapers = { + "pokemoncenter": PokemonCenterScraper(), + "target": TargetScraper(), +} + + +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) + + # 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 + 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"] + + try: + for url in TARGET_URLS: + logger.info(f"Scraping: {url}") + products = scraper.scrape_category_page(url) + + # 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 + 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="target", + 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="target", + 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 Target: {e}") + send_error_notification(f"Error checking Target: {str(e)}", "Target") + + +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() + + if "target" in SITES_ENABLED: + check_target() + + logger.info("Check cycle complete") + + +def graceful_shutdown(signum, frame): + """Handle shutdown gracefully""" + logger.info("Shutting down...") + shutdown_browser() + sys.exit(0) + + +def main(): + """Main entry point""" + print("=" * 50) + print("Pokemon Stock Monitor") + print("=" * 50) + print(f"Monitoring: {', '.join(SITES_ENABLED)}") + print(f"Check interval: {CHECK_INTERVAL_SECONDS} seconds") + print(f"Keyword filter: {'ON' if KEYWORD_FILTER_ENABLED else 'OFF'}") + print("=" * 50) + print() + + # Register signal handlers + signal.signal(signal.SIGINT, graceful_shutdown) + signal.signal(signal.SIGTERM, graceful_shutdown) + + # Initialize browser + logger.info("Initializing browser...") + try: + get_browser() + except Exception as e: + logger.error(f"Failed to initialize browser: {e}") + print(f"\nERROR: Failed to start browser. Make sure Playwright is installed:") + print(" pip install playwright") + print(" playwright install chromium") + return + + # Send startup notification + send_startup_notification() + + # Run initial check + print("Running initial check...") + run_checks() + + # Schedule periodic checks + schedule.every(CHECK_INTERVAL_SECONDS).seconds.do(run_checks) + + print(f"\nMonitor running! Checking every {CHECK_INTERVAL_SECONDS} seconds.") + print("Press Ctrl+C to stop.\n") + + # Main loop + try: + while True: + schedule.run_pending() + time.sleep(1) + except KeyboardInterrupt: + pass + finally: + graceful_shutdown(None, None) + + +if __name__ == "__main__": + main() diff --git a/product_tracker.py b/product_tracker.py new file mode 100644 index 0000000..18c7f79 --- /dev/null +++ b/product_tracker.py @@ -0,0 +1,136 @@ +""" +Product tracker - keeps track of known products to detect new drops and restocks +""" + +import json +import logging +from pathlib import Path +from typing import Dict, List, Set, Optional +from dataclasses import asdict +from datetime import datetime + +from scrapers.base import Product + +logger = logging.getLogger(__name__) + +# File to store known products +PRODUCTS_FILE = Path(__file__).parent / "products.json" + + +class ProductTracker: + """Tracks known products to detect new listings and stock changes""" + + def __init__(self, products_file: Path = PRODUCTS_FILE): + self.products_file = products_file + self.products: Dict[str, dict] = {} # URL -> product data + self.load() + + def load(self): + """Load products from file""" + if self.products_file.exists(): + try: + with open(self.products_file, "r", encoding="utf-8") as f: + self.products = json.load(f) + logger.info(f"Loaded {len(self.products)} tracked products") + except (json.JSONDecodeError, IOError) as e: + logger.error(f"Error loading products file: {e}") + self.products = {} + else: + self.products = {} + logger.info("No existing products file, starting fresh") + + def save(self): + """Save products to file""" + try: + with open(self.products_file, "w", encoding="utf-8") as f: + json.dump(self.products, f, indent=2, ensure_ascii=False) + logger.debug(f"Saved {len(self.products)} products") + except IOError as e: + logger.error(f"Error saving products file: {e}") + + def process_products(self, products: List[Product]) -> tuple[List[Product], List[Product]]: + """ + Process a list of scraped products and detect changes + + Args: + products: List of products from scraper + + Returns: + Tuple of (new_products, restocked_products) + """ + new_products = [] + restocked_products = [] + + for product in products: + url = product.url + + if url not in self.products: + # New product! + new_products.append(product) + self.products[url] = { + "name": product.name, + "url": url, + "price": product.price, + "in_stock": product.in_stock, + "image_url": product.image_url, + "site": product.site, + "product_id": product.product_id, + "first_seen": datetime.now().isoformat(), + "last_seen": datetime.now().isoformat(), + "last_in_stock": datetime.now().isoformat() if product.in_stock else None, + } + logger.info(f"NEW PRODUCT: {product.name}") + + else: + # Existing product - check for restock + existing = self.products[url] + was_in_stock = existing.get("in_stock", False) + + # Update last seen + existing["last_seen"] = datetime.now().isoformat() + existing["price"] = product.price or existing.get("price") + existing["image_url"] = product.image_url or existing.get("image_url") + + if product.in_stock and not was_in_stock: + # RESTOCK! + restocked_products.append(product) + existing["last_in_stock"] = datetime.now().isoformat() + logger.info(f"RESTOCK: {product.name}") + + existing["in_stock"] = product.in_stock + self.products[url] = existing + + self.save() + return new_products, restocked_products + + def get_known_urls(self) -> Set[str]: + """Get all known product URLs""" + return set(self.products.keys()) + + def get_product(self, url: str) -> Optional[dict]: + """Get a specific product by URL""" + return self.products.get(url) + + def mark_out_of_stock(self, url: str): + """Mark a product as out of stock""" + if url in self.products: + self.products[url]["in_stock"] = False + self.save() + + def clear(self): + """Clear all tracked products""" + self.products = {} + self.save() + logger.info("Cleared all tracked products") + + def get_stats(self) -> dict: + """Get tracking statistics""" + total = len(self.products) + in_stock = sum(1 for p in self.products.values() if p.get("in_stock", False)) + out_of_stock = total - in_stock + + return { + "total_products": total, + "in_stock": in_stock, + "out_of_stock": out_of_stock, + } diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c88db0d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +playwright>=1.45.0 +playwright-stealth>=1.0.6 +requests>=2.31.0 +schedule>=1.2.1 +beautifulsoup4>=4.12.0 diff --git a/scrapers/__init__.py b/scrapers/__init__.py new file mode 100644 index 0000000..dd14901 --- /dev/null +++ b/scrapers/__init__.py @@ -0,0 +1,5 @@ +# Scrapers package +from .pokemoncenter import PokemonCenterScraper +from .target import TargetScraper + +__all__ = ["PokemonCenterScraper", "TargetScraper"] diff --git a/scrapers/base.py b/scrapers/base.py new file mode 100644 index 0000000..d70c017 --- /dev/null +++ b/scrapers/base.py @@ -0,0 +1,75 @@ +""" +Base scraper class with common functionality +""" + +import logging +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import List, Optional + +logger = logging.getLogger(__name__) + + +@dataclass +class Product: + """Represents a product listing""" + name: str + url: str + price: Optional[str] + in_stock: bool + image_url: Optional[str] = None + site: str = "" + product_id: Optional[str] = None # Unique identifier for tracking + + def __hash__(self): + return hash(self.url) + + def __eq__(self, other): + if isinstance(other, Product): + return self.url == other.url + return False + + +class BaseScraper(ABC): + """Base class for site-specific scrapers""" + + site_name: str = "unknown" + + @abstractmethod + def scrape_category_page(self, url: str) -> List[Product]: + """ + Scrape a category/search page and return all products found + + Args: + url: URL of the category page + + Returns: + List of Product objects + """ + pass + + @abstractmethod + def check_product_stock(self, product_url: str) -> tuple[bool, Optional[str]]: + """ + Check if a specific product is in stock + + Args: + product_url: URL of the product page + + Returns: + Tuple of (is_in_stock, price) + """ + pass + + def filter_by_keywords(self, products: List[Product], keywords: List[str]) -> List[Product]: + """Filter products by keywords in name""" + if not keywords: + return products + + filtered = [] + for product in products: + name_lower = product.name.lower() + if any(kw.lower() in name_lower for kw in keywords): + filtered.append(product) + + return filtered diff --git a/scrapers/pokemoncenter.py b/scrapers/pokemoncenter.py new file mode 100644 index 0000000..87bccd4 --- /dev/null +++ b/scrapers/pokemoncenter.py @@ -0,0 +1,276 @@ +""" +PokemonCenter.com scraper +Handles bot protection with Playwright stealth +""" + +import re +import logging +from typing import List, Optional +from bs4 import BeautifulSoup + +from .base import BaseScraper, Product +from browser import get_browser + +logger = logging.getLogger(__name__) + + +class PokemonCenterScraper(BaseScraper): + """Scraper for PokemonCenter.com""" + + site_name = "pokemoncenter" + base_url = "https://www.pokemoncenter.com" + + def scrape_category_page(self, url: str) -> List[Product]: + """ + Scrape a PokemonCenter category page for all products + + Args: + url: Category page URL + + Returns: + List of Product objects + """ + browser = get_browser() + products = [] + + try: + # Navigate and wait for page to load (don't wait for specific selector) + 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) + + 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 = get_browser() + + try: + 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() + + 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/scrapers/target.py b/scrapers/target.py new file mode 100644 index 0000000..026ebdd --- /dev/null +++ b/scrapers/target.py @@ -0,0 +1,344 @@ +""" +Target.com scraper +""" + +import re +import json +import logging +from typing import List, Optional +from bs4 import BeautifulSoup + +from .base import BaseScraper, Product +from browser import get_browser + +logger = logging.getLogger(__name__) + + +class TargetScraper(BaseScraper): + """Scraper for Target.com""" + + site_name = "target" + base_url = "https://www.target.com" + + def scrape_category_page(self, url: str) -> List[Product]: + """ + Scrape a Target search/category page for all products + + Args: + url: Search/category page URL + + Returns: + List of Product objects + """ + browser = get_browser() + products = [] + + try: + page, html = browser.get_page_content( + url, + wait_for_selector=None, + timeout=60000, + ) + + # Save debug screenshot + try: + page.screenshot(path="debug_target.png") + logger.info("Saved debug screenshot to debug_target.png") + except: + pass + + soup = BeautifulSoup(html, "html.parser") + + # Try to find product data in page scripts (Target uses React/hydration) + scripts = soup.find_all("script", type="application/json") + for script in scripts: + try: + data = json.loads(script.string) + # Look for product data in the JSON + products.extend(self._extract_products_from_json(data)) + except (json.JSONDecodeError, TypeError): + continue + + # Parse product links from the page + product_links = soup.select("a[href*='/p/']") + logger.info(f"Found {len(product_links)} product links") + + # Group links by href and pick the best one (with aria-label or text) + href_to_links = {} + for link in product_links: + href = link.get("href", "") + if not href: + continue + if href not in href_to_links: + href_to_links[href] = [] + href_to_links[href].append(link) + + logger.info(f"Found {len(href_to_links)} unique hrefs") + + for href, links in href_to_links.items(): + # Find the best link (one with aria-label or text content) + best_link = None + for link in links: + aria = link.get("aria-label", "") + text = link.get_text(strip=True) + if aria or (text and len(text) > 10): + best_link = link + break + if not best_link: + best_link = links[0] # Fallback to first link + + product = self._parse_product_link(best_link, soup) + if product: + products.append(product) + + page.close() + + except Exception as e: + logger.error(f"Error scraping Target 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 Target") + return unique_products + + def _extract_products_from_json(self, data, depth=0) -> List[Product]: + """Recursively search JSON for product data""" + products = [] + if depth > 10: # Prevent infinite recursion + return products + + if isinstance(data, dict): + # Check if this looks like a product + if "tcin" in data or ("title" in data and "price" in data): + product = self._parse_product_json(data) + if product: + products.append(product) + + # Recurse into nested objects + for value in data.values(): + products.extend(self._extract_products_from_json(value, depth + 1)) + + elif isinstance(data, list): + for item in data: + products.extend(self._extract_products_from_json(item, depth + 1)) + + return products + + def _parse_product_json(self, data: dict) -> Optional[Product]: + """Parse a product from Target's JSON data""" + try: + name = data.get("title") or data.get("product_description", {}).get("title", "") + if not name: + return None + + # Build URL + tcin = data.get("tcin", "") + slug = data.get("url_slug", name.lower().replace(" ", "-")) + url = f"{self.base_url}/p/{slug}/-/A-{tcin}" if tcin else "" + if not url: + return None + + # Get price + price_data = data.get("price", {}) + price = None + if isinstance(price_data, dict): + price = price_data.get("formatted_current_price") or price_data.get("current_retail") + elif isinstance(price_data, (int, float)): + price = f"${price_data:.2f}" + + # Check availability + availability = data.get("availability_status", "") + fulfillment = data.get("fulfillment", {}) + in_stock = availability not in ["OUT_OF_STOCK", "UNAVAILABLE"] + if fulfillment: + in_stock = fulfillment.get("is_out_of_stock_in_all_store_locations", True) is False + + # Get image + images = data.get("images", []) + image_url = images[0].get("base_url") if images else None + + return Product( + name=name, + url=url, + price=str(price) if price else None, + in_stock=in_stock, + image_url=image_url, + site=self.site_name, + product_id=tcin, + ) + + except Exception as e: + logger.debug(f"Error parsing Target product JSON: {e}") + return None + + def _parse_product_link(self, link, soup) -> Optional[Product]: + """Parse a product from a product link element""" + try: + href = link.get("href", "") + if not href: + return None + + url = href if href.startswith("http") else f"{self.base_url}{href}" + + # Get name - prefer aria-label as it's usually clean + name = link.get("aria-label", "") + + # If no aria-label, try link text + if not name or len(name) < 5: + name = link.get_text(strip=True) + + # Clean up the name + if name: + # Remove rating text patterns + name = re.sub(r'\d+\.?\d*\s*out of \d+ stars.*$', '', name, flags=re.IGNORECASE) + name = re.sub(r'\s*\d+ ratings?.*$', '', name, flags=re.IGNORECASE) + name = re.sub(r'\s*\d+ reviews?.*$', '', name, flags=re.IGNORECASE) + name = name.strip() + + # Skip empty/short names + if not name or len(name) < 10: + return None + + # Skip if name looks like navigation/rating text only + lower_name = name.lower() + if any(lower_name.startswith(skip) for skip in ['rating', 'stars', 'review', 'filter', 'sort']): + return None + + # Extract product ID from URL (A-12345678) + product_id = "" + match = re.search(r"/A-(\d+)", url) + if match: + product_id = match.group(1) + + # Try to find price near this link + parent = link.find_parent() + price = None + for _ in range(5): # Go up 5 levels max + if parent: + price_text = parent.get_text() + price_match = re.search(r"\$[\d,]+\.?\d*", price_text) + if price_match: + price = price_match.group() + break + parent = parent.find_parent() + + # Try to find image + image_url = None + img = link.select_one("img") + if img: + image_url = img.get("src") or img.get("data-src") + + # Check stock (assume in stock unless we see otherwise) + in_stock = True + parent = link.find_parent() + for _ in range(5): + if parent: + text = parent.get_text().lower() + if "out of stock" in text or "unavailable" in text: + in_stock = False + break + parent = parent.find_parent() + + return Product( + name=name, + url=url, + price=price, + in_stock=in_stock, + image_url=image_url, + site=self.site_name, + product_id=product_id, + ) + + except Exception as e: + logger.debug(f"Error parsing Target product link: {e}") + return None + + def _parse_product_card(self, card) -> Optional[Product]: + """Parse a product card element""" + try: + # Find link + link = card.select_one("a[href*='/p/']") + if not link: + return None + + href = link.get("href", "") + url = href if href.startswith("http") else f"{self.base_url}{href}" + + # Get name + name_elem = card.select_one("[data-test='product-title']") or card.select_one("a") + name = name_elem.get_text(strip=True) if name_elem else "Unknown" + + # Get price + price_elem = card.select_one("[data-test='current-price']") or card.select_one("[class*='price']") + price = None + if price_elem: + price_match = re.search(r"\$[\d,]+\.?\d*", price_elem.get_text()) + if price_match: + price = price_match.group() + + # Check stock + card_text = card.get_text().lower() + out_of_stock = "out of stock" in card_text or "unavailable" in card_text + in_stock = not out_of_stock + + # Get image + img = card.select_one("img") + image_url = img.get("src") if img else None + + # Extract product ID from URL + product_id = "" + match = re.search(r"/A-(\d+)", url) + if match: + product_id = match.group(1) + + return Product( + name=name, + url=url, + price=price, + in_stock=in_stock, + image_url=image_url, + site=self.site_name, + product_id=product_id, + ) + + except Exception as e: + logger.debug(f"Error parsing Target product card: {e}") + return None + + def check_product_stock(self, product_url: str) -> tuple[bool, Optional[str]]: + """Check if a specific product is in stock""" + browser = get_browser() + + try: + page, html = browser.get_page_content(product_url, timeout=30000) + soup = BeautifulSoup(html, "html.parser") + page_text = soup.get_text().lower() + + out_of_stock = any( + phrase in page_text + for phrase in ["out of stock", "unavailable", "not available"] + ) + + in_stock = not out_of_stock + + # Get price + price = None + price_elem = soup.select_one("[data-test='product-price']") + if price_elem: + price_match = re.search(r"\$[\d,]+\.?\d*", price_elem.get_text()) + if price_match: + price = price_match.group() + + page.close() + return in_stock, price + + except Exception as e: + logger.error(f"Error checking Target product stock: {e}") + return False, None