mmcghen
8d382e723f
- Added Walmart scraper to scrape product data from Walmart.com, including category pages and product details. - Introduced a stealth browser module to handle bot protection and improve scraping reliability. - Created a SQLite database for tracking product history, price changes, stock events, and user favorites. - Developed a Discord bot for user interaction, allowing location setting and stock checking at local stores. - Implemented a favorites system to manage priority products and categories with custom notification settings. - Added news aggregation module to fetch and analyze Pokemon TCG news from various sources. - Created tools for API discovery and monitoring, including a backend monitor for detecting new products. - Added unit tests for database operations, product filtering, and API endpoints to ensure functionality. - Enhanced existing modules with improved error handling and logging for better maintainability.
416 lines
13 KiB
Python
416 lines
13 KiB
Python
"""
|
|
Smart Pokemon Center Monitor
|
|
|
|
Uses stealth browser with adaptive modes and human-like behavior.
|
|
Detects new products before they're announced.
|
|
|
|
Modes:
|
|
- STEALTH: Normal monitoring (~1 min intervals, human-like)
|
|
- ALERT: Fast checking when new product detected (~20 sec)
|
|
- COOLDOWN: Gradual return to stealth after alert
|
|
|
|
Usage:
|
|
python smart_monitor.py
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
import random
|
|
import logging
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Set, List, Dict
|
|
from enum import Enum
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s [%(levelname)s] %(message)s',
|
|
datefmt='%H:%M:%S'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# File paths
|
|
DATA_DIR = Path(__file__).parent.parent / "data"
|
|
KNOWN_SKUS_FILE = DATA_DIR / "known_skus.json"
|
|
DETECTIONS_FILE = DATA_DIR / "detections.json"
|
|
|
|
# URLs to monitor
|
|
MONITOR_URLS = [
|
|
"https://www.pokemoncenter.com/category/tcg-cards?sort=newest",
|
|
"https://www.pokemoncenter.com/category/new-releases",
|
|
]
|
|
|
|
|
|
class MonitorMode(Enum):
|
|
STEALTH = "stealth"
|
|
ALERT = "alert"
|
|
COOLDOWN = "cooldown"
|
|
|
|
|
|
class SmartMonitor:
|
|
"""Adaptive monitor with human-like behavior"""
|
|
|
|
def __init__(self):
|
|
self.browser = None
|
|
self.known_skus: Set[str] = set()
|
|
self.mode = MonitorMode.STEALTH
|
|
self.alert_triggered_at: Optional[datetime] = None
|
|
self.check_count = 0
|
|
self.last_detection: Optional[Dict] = None
|
|
|
|
# Mode timing settings
|
|
self.timing = {
|
|
MonitorMode.STEALTH: (45, 90), # 45-90 seconds
|
|
MonitorMode.ALERT: (15, 30), # 15-30 seconds
|
|
MonitorMode.COOLDOWN: (60, 120), # 60-120 seconds
|
|
}
|
|
|
|
# Alert mode duration
|
|
self.alert_duration = timedelta(minutes=30)
|
|
self.cooldown_duration = timedelta(minutes=15)
|
|
|
|
def start_browser(self):
|
|
"""Start the stealth browser"""
|
|
if self.browser:
|
|
return
|
|
|
|
logger.info("Starting stealth browser...")
|
|
|
|
try:
|
|
from stealth_browser import StealthBrowser
|
|
self.browser = StealthBrowser(
|
|
headless=False,
|
|
session_name="smart_monitor"
|
|
)
|
|
self.browser.start()
|
|
logger.info("Browser started successfully")
|
|
except Exception as e:
|
|
logger.error(f"Failed to start browser: {e}")
|
|
raise
|
|
|
|
def stop_browser(self):
|
|
"""Stop the browser"""
|
|
if self.browser:
|
|
logger.info("Stopping browser...")
|
|
self.browser.stop()
|
|
self.browser = None
|
|
|
|
def load_known_skus(self):
|
|
"""Load previously seen SKUs"""
|
|
if KNOWN_SKUS_FILE.exists():
|
|
with open(KNOWN_SKUS_FILE, 'r') as f:
|
|
data = json.load(f)
|
|
self.known_skus = set(data.get('skus', []))
|
|
logger.info(f"Loaded {len(self.known_skus)} known SKUs")
|
|
else:
|
|
self.known_skus = set()
|
|
logger.info("No known SKUs file - starting fresh")
|
|
|
|
def save_known_skus(self):
|
|
"""Save known SKUs"""
|
|
with open(KNOWN_SKUS_FILE, 'w') as f:
|
|
json.dump({
|
|
'skus': list(self.known_skus),
|
|
'updated': datetime.now().isoformat(),
|
|
'count': len(self.known_skus)
|
|
}, f, indent=2)
|
|
|
|
def log_detection(self, sku: str, name: str, url: str):
|
|
"""Log a new product detection"""
|
|
detection = {
|
|
'sku': sku,
|
|
'name': name,
|
|
'url': url,
|
|
'detected_at': datetime.now().isoformat(),
|
|
'mode': self.mode.value
|
|
}
|
|
|
|
self.last_detection = detection
|
|
|
|
# Load existing detections
|
|
detections = []
|
|
if DETECTIONS_FILE.exists():
|
|
with open(DETECTIONS_FILE, 'r') as f:
|
|
detections = json.load(f)
|
|
|
|
detections.append(detection)
|
|
|
|
with open(DETECTIONS_FILE, 'w') as f:
|
|
json.dump(detections, f, indent=2)
|
|
|
|
logger.info(f"Detection logged: {sku}")
|
|
|
|
def get_interval(self) -> float:
|
|
"""Get randomized check interval based on current mode"""
|
|
min_sec, max_sec = self.timing[self.mode]
|
|
|
|
# Add gaussian jitter for more natural timing
|
|
base = (min_sec + max_sec) / 2
|
|
jitter = random.gauss(0, (max_sec - min_sec) / 4)
|
|
interval = base + jitter
|
|
|
|
# Clamp to bounds
|
|
return max(min_sec * 0.8, min(max_sec * 1.2, interval))
|
|
|
|
def maybe_do_human_action(self):
|
|
"""Occasionally do something human-like"""
|
|
if not self.browser or not self.browser.driver:
|
|
return
|
|
|
|
action = random.random()
|
|
|
|
if action < 0.3:
|
|
# Scroll randomly
|
|
self.browser.human_scroll()
|
|
elif action < 0.4:
|
|
# Small mouse movement
|
|
self.browser.human_mouse_move()
|
|
elif action < 0.45:
|
|
# Longer pause (human distraction)
|
|
pause = random.uniform(3, 8)
|
|
logger.debug(f"Human pause: {pause:.1f}s")
|
|
time.sleep(pause)
|
|
|
|
def maybe_take_break(self) -> bool:
|
|
"""Occasionally take a longer break"""
|
|
# 3% chance of a break
|
|
if random.random() < 0.03:
|
|
break_time = random.randint(120, 300) # 2-5 minutes
|
|
logger.info(f"Taking a break for {break_time}s (human-like pause)")
|
|
time.sleep(break_time)
|
|
return True
|
|
return False
|
|
|
|
def extract_skus_from_page(self) -> Set[str]:
|
|
"""Extract product SKUs from the current page"""
|
|
if not self.browser or not self.browser.driver:
|
|
return set()
|
|
|
|
skus = set()
|
|
|
|
try:
|
|
# Get page source and parse
|
|
from bs4 import BeautifulSoup
|
|
html = self.browser.driver.page_source
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
|
|
# Look for product links - Pokemon Center format: /product/SKU/name
|
|
import re
|
|
product_links = soup.select('a[href*="/product/"]')
|
|
|
|
for link in product_links:
|
|
href = link.get('href', '')
|
|
# Extract SKU from URL like /product/699-17113/product-name
|
|
match = re.search(r'/product/([0-9]+-?[0-9]+)', href)
|
|
if match:
|
|
skus.add(match.group(1))
|
|
|
|
# Also look for data attributes
|
|
for elem in soup.select('[data-sku], [data-product-id]'):
|
|
sku = elem.get('data-sku') or elem.get('data-product-id')
|
|
if sku and re.match(r'^[0-9]+-?[0-9]+', sku):
|
|
skus.add(sku)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error extracting SKUs: {e}")
|
|
|
|
return skus
|
|
|
|
def check_page(self, url: str) -> Set[str]:
|
|
"""Load a page and extract SKUs"""
|
|
if not self.browser:
|
|
self.start_browser()
|
|
|
|
try:
|
|
# Human-like delay before navigation
|
|
self.browser.human_delay(0.5, 2.0)
|
|
|
|
# Navigate
|
|
logger.debug(f"Loading: {url}")
|
|
self.browser.driver.get(url)
|
|
|
|
# Wait for page load
|
|
time.sleep(random.uniform(3, 6))
|
|
|
|
# Check for CAPTCHA
|
|
if self.browser.check_for_captcha():
|
|
logger.warning("CAPTCHA detected!")
|
|
print("\n" + "!" * 50)
|
|
print("CAPTCHA DETECTED - Please solve it in the browser")
|
|
print("!" * 50 + "\n")
|
|
self.browser.wait_for_captcha_solve(timeout=120)
|
|
|
|
# Human actions
|
|
self.maybe_do_human_action()
|
|
|
|
# Extract SKUs
|
|
skus = self.extract_skus_from_page()
|
|
|
|
return skus
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking page: {e}")
|
|
return set()
|
|
|
|
def update_mode(self):
|
|
"""Update monitoring mode based on state"""
|
|
now = datetime.now()
|
|
|
|
if self.mode == MonitorMode.ALERT:
|
|
# Check if alert period is over
|
|
if self.alert_triggered_at:
|
|
elapsed = now - self.alert_triggered_at
|
|
if elapsed > self.alert_duration:
|
|
logger.info("Alert period over, entering cooldown")
|
|
self.mode = MonitorMode.COOLDOWN
|
|
|
|
elif self.mode == MonitorMode.COOLDOWN:
|
|
# Check if cooldown is over
|
|
if self.alert_triggered_at:
|
|
elapsed = now - self.alert_triggered_at
|
|
if elapsed > (self.alert_duration + self.cooldown_duration):
|
|
logger.info("Cooldown over, returning to stealth mode")
|
|
self.mode = MonitorMode.STEALTH
|
|
self.alert_triggered_at = None
|
|
|
|
def trigger_alert_mode(self):
|
|
"""Switch to alert mode"""
|
|
logger.info("!!! ENTERING ALERT MODE - Faster checks !!!")
|
|
self.mode = MonitorMode.ALERT
|
|
self.alert_triggered_at = datetime.now()
|
|
|
|
def send_discord_alert(self, sku: str, name: str, url: str):
|
|
"""Send Discord notification"""
|
|
try:
|
|
from src.discord_notifier import DiscordNotifier
|
|
from config import DISCORD_WEBHOOK_URL
|
|
|
|
if DISCORD_WEBHOOK_URL and DISCORD_WEBHOOK_URL != "YOUR_WEBHOOK_URL_HERE":
|
|
notifier = DiscordNotifier(DISCORD_WEBHOOK_URL)
|
|
# Create a simple product dict
|
|
product = {
|
|
'name': name,
|
|
'url': f"https://www.pokemoncenter.com/product/{sku}",
|
|
'price': 'Check site',
|
|
'site': 'pokemoncenter',
|
|
}
|
|
notifier.send_stock_alert(product, "NEW BACKEND DETECTION")
|
|
logger.info("Discord alert sent!")
|
|
except Exception as e:
|
|
logger.warning(f"Could not send Discord alert: {e}")
|
|
|
|
def run_check(self) -> List[str]:
|
|
"""Run a single check across all monitored URLs"""
|
|
self.check_count += 1
|
|
all_skus = set()
|
|
new_skus = []
|
|
|
|
logger.info(f"Check #{self.check_count} | Mode: {self.mode.value.upper()}")
|
|
|
|
for url in MONITOR_URLS:
|
|
skus = self.check_page(url)
|
|
all_skus.update(skus)
|
|
|
|
# Brief pause between pages
|
|
if url != MONITOR_URLS[-1]:
|
|
time.sleep(random.uniform(2, 5))
|
|
|
|
logger.info(f"Found {len(all_skus)} total SKUs")
|
|
|
|
# Find new SKUs
|
|
if self.known_skus: # Only check if we have a baseline
|
|
new = all_skus - self.known_skus
|
|
if new:
|
|
for sku in new:
|
|
logger.info(f"!!! NEW SKU DETECTED: {sku}")
|
|
new_skus.append(sku)
|
|
|
|
# Log and alert
|
|
url = f"https://www.pokemoncenter.com/product/{sku}"
|
|
self.log_detection(sku, f"New Product {sku}", url)
|
|
self.send_discord_alert(sku, f"New Product {sku}", url)
|
|
|
|
# Trigger alert mode
|
|
self.trigger_alert_mode()
|
|
|
|
# Update known SKUs
|
|
self.known_skus.update(all_skus)
|
|
self.save_known_skus()
|
|
|
|
return new_skus
|
|
|
|
def run(self):
|
|
"""Main monitoring loop"""
|
|
print()
|
|
print("=" * 60)
|
|
print(" SMART POKEMON CENTER MONITOR")
|
|
print("=" * 60)
|
|
print()
|
|
print("Modes:")
|
|
print(f" STEALTH: {self.timing[MonitorMode.STEALTH]} sec (normal)")
|
|
print(f" ALERT: {self.timing[MonitorMode.ALERT]} sec (after detection)")
|
|
print(f" COOLDOWN: {self.timing[MonitorMode.COOLDOWN]} sec (transition)")
|
|
print()
|
|
print("Press Ctrl+C to stop")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
try:
|
|
# Initialize
|
|
self.load_known_skus()
|
|
self.start_browser()
|
|
|
|
# First check - build baseline if needed
|
|
if not self.known_skus:
|
|
logger.info("First run - building SKU baseline...")
|
|
self.run_check()
|
|
logger.info(f"Baseline: {len(self.known_skus)} products")
|
|
print()
|
|
|
|
# Main loop
|
|
while True:
|
|
# Maybe take a break
|
|
if self.maybe_take_break():
|
|
continue
|
|
|
|
# Run check
|
|
new_skus = self.run_check()
|
|
|
|
if new_skus:
|
|
print()
|
|
print("!" * 60)
|
|
print("!!! NEW PRODUCT(S) DETECTED !!!")
|
|
for sku in new_skus:
|
|
print(f" -> {sku}")
|
|
print("!" * 60)
|
|
print()
|
|
|
|
# Update mode
|
|
self.update_mode()
|
|
|
|
# Get interval and wait
|
|
interval = self.get_interval()
|
|
logger.info(f"Next check in {interval:.0f}s")
|
|
time.sleep(interval)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\nStopping monitor...")
|
|
except Exception as e:
|
|
logger.error(f"Monitor error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
self.save_known_skus()
|
|
self.stop_browser()
|
|
print("Monitor stopped.")
|
|
|
|
|
|
def main():
|
|
monitor = SmartMonitor()
|
|
monitor.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|