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 <noreply@anthropic.com>
This commit is contained in:
2026-03-24 12:39:09 -04:00
commit 9d49a99916
18 changed files with 2653 additions and 0 deletions
+20
View File
@@ -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
+286
View File
@@ -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
+183
View File
@@ -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
+68
View File
@@ -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
+317
View File
@@ -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);
+109
View File
@@ -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;
}
})();
+30
View File
@@ -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"
}
]
}
+254
View File
@@ -0,0 +1,254 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<style>
body {
width: 320px;
padding: 15px;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: #1a1a2e;
color: #eee;
margin: 0;
}
h1 {
font-size: 16px;
margin: 0 0 15px 0;
color: #ffcb05;
display: flex;
align-items: center;
gap: 8px;
}
h1 span {
font-size: 20px;
}
.section {
margin-bottom: 15px;
}
label {
display: block;
font-size: 12px;
color: #aaa;
margin-bottom: 4px;
}
input[type="text"], input[type="number"], textarea {
width: 100%;
padding: 8px;
border: 1px solid #333;
border-radius: 4px;
background: #16213e;
color: #eee;
font-size: 13px;
box-sizing: border-box;
}
input:focus, textarea:focus {
outline: none;
border-color: #ffcb05;
}
textarea {
height: 60px;
resize: vertical;
}
.row {
display: flex;
gap: 10px;
}
.row > div {
flex: 1;
}
.toggle-row {
display: flex;
justify-content: space-between;
align-items: center;
padding: 8px 0;
}
.toggle-row label {
margin: 0;
color: #eee;
font-size: 13px;
}
.toggle {
position: relative;
width: 44px;
height: 24px;
}
.toggle input {
opacity: 0;
width: 0;
height: 0;
}
.slider {
position: absolute;
cursor: pointer;
top: 0;
left: 0;
right: 0;
bottom: 0;
background-color: #333;
transition: .3s;
border-radius: 24px;
}
.slider:before {
position: absolute;
content: "";
height: 18px;
width: 18px;
left: 3px;
bottom: 3px;
background-color: #eee;
transition: .3s;
border-radius: 50%;
}
input:checked + .slider {
background-color: #00c853;
}
input:checked + .slider:before {
transform: translateX(20px);
}
button {
width: 100%;
padding: 10px;
border: none;
border-radius: 4px;
font-size: 13px;
font-weight: 600;
cursor: pointer;
transition: background 0.2s;
}
.btn-primary {
background: #ffcb05;
color: #1a1a2e;
}
.btn-primary:hover {
background: #ffd633;
}
.btn-secondary {
background: #333;
color: #eee;
margin-top: 8px;
}
.btn-secondary:hover {
background: #444;
}
.btn-danger {
background: #c62828;
color: #fff;
margin-top: 8px;
}
.btn-danger:hover {
background: #d32f2f;
}
.stats {
background: #16213e;
padding: 10px;
border-radius: 4px;
font-size: 12px;
margin-bottom: 15px;
}
.stats-row {
display: flex;
justify-content: space-between;
margin-bottom: 4px;
}
.stats-row:last-child {
margin-bottom: 0;
}
.status-dot {
display: inline-block;
width: 8px;
height: 8px;
border-radius: 50%;
margin-right: 6px;
}
.status-dot.active {
background: #00c853;
}
.status-dot.inactive {
background: #c62828;
}
.saved-msg {
text-align: center;
color: #00c853;
font-size: 12px;
margin-top: 8px;
opacity: 0;
transition: opacity 0.3s;
}
.saved-msg.show {
opacity: 1;
}
small {
color: #666;
font-size: 11px;
}
</style>
</head>
<body>
<h1><span>&#9889;</span> Pokemon Stock Monitor</h1>
<div class="stats" id="stats">
<div class="stats-row">
<span>Status:</span>
<span><span class="status-dot active" id="statusDot"></span><span id="statusText">Active</span></span>
</div>
<div class="stats-row">
<span>Products tracked:</span>
<span id="productCount">0</span>
</div>
</div>
<div class="section">
<label>Discord Webhook URL</label>
<input type="text" id="webhook" placeholder="https://discord.com/api/webhooks/...">
</div>
<div class="section">
<label>URLs to Monitor (one per line)</label>
<textarea id="urls" placeholder="https://www.pokemoncenter.com/category/tcg-cards"></textarea>
</div>
<div class="section">
<label>Keywords (comma separated, leave empty for all)</label>
<input type="text" id="keywords" placeholder="chaos rising, booster, etb">
</div>
<div class="row">
<div>
<label>Check Interval (min)</label>
<input type="number" id="interval" min="1" max="60" value="1">
</div>
</div>
<div class="section">
<div class="toggle-row">
<label>Monitor Enabled</label>
<div class="toggle">
<input type="checkbox" id="enabled" checked>
<span class="slider"></span>
</div>
</div>
<div class="toggle-row">
<label>Notify New Products</label>
<div class="toggle">
<input type="checkbox" id="notifyNew" checked>
<span class="slider"></span>
</div>
</div>
<div class="toggle-row">
<label>Notify Restocks</label>
<div class="toggle">
<input type="checkbox" id="notifyRestock" checked>
<span class="slider"></span>
</div>
</div>
</div>
<button class="btn-primary" id="saveBtn">Save Settings</button>
<button class="btn-secondary" id="checkNowBtn">Check Now</button>
<button class="btn-danger" id="clearBtn">Clear Product History</button>
<div class="saved-msg" id="savedMsg">Settings saved!</div>
<script src="popup.js"></script>
</body>
</html>
+75
View File
@@ -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);
}
+55
View File
@@ -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"
+183
View File
@@ -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()
+232
View File
@@ -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()
+136
View File
@@ -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,
}
+5
View File
@@ -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
+5
View File
@@ -0,0 +1,5 @@
# Scrapers package
from .pokemoncenter import PokemonCenterScraper
from .target import TargetScraper
__all__ = ["PokemonCenterScraper", "TargetScraper"]
+75
View File
@@ -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
+276
View File
@@ -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]
+344
View File
@@ -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