579 lines
22 KiB
Python
579 lines
22 KiB
Python
"""
|
|
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,
|
|
TARGET_URLS,
|
|
GAMESTOP_URLS,
|
|
BESTBUY_URLS,
|
|
WALMART_URLS,
|
|
KEYWORD_FILTER_ENABLED,
|
|
KEYWORDS,
|
|
LOG_LEVEL,
|
|
NEW_DROP_NOTIFICATIONS,
|
|
PRICE_CHANGE_NOTIFICATIONS,
|
|
MAX_PAGES,
|
|
AUTO_BUY_ENABLED,
|
|
AUTO_BUY_SITES,
|
|
AUTO_BUY_DRY_RUN,
|
|
)
|
|
from src.browser import get_browser, shutdown_browser
|
|
from src.product_tracker import ProductTracker
|
|
from src.discord_notifier import send_stock_alert, send_startup_notification, send_error_notification, send_price_change_alert
|
|
from src.scraper_state import scraper_state
|
|
from src.buyers import get_buyer
|
|
# Note: Pokemon Center uses Chrome Extension (chrome-extension/), not a Python scraper
|
|
from scrapers import (
|
|
TargetScraper,
|
|
GameStopScraper,
|
|
BestBuyScraper,
|
|
WalmartScraper,
|
|
)
|
|
|
|
# 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 (Pokemon Center uses Chrome Extension instead)
|
|
scrapers = {
|
|
"target": TargetScraper(),
|
|
"gamestop": GameStopScraper(),
|
|
"bestbuy": BestBuyScraper(),
|
|
"walmart": WalmartScraper(),
|
|
}
|
|
|
|
|
|
def attempt_auto_buy(site: str, products: list):
|
|
"""
|
|
Attempt to purchase in-stock products on the given site.
|
|
Only runs when AUTO_BUY_ENABLED=True and site is in AUTO_BUY_SITES.
|
|
Logs and notifies on every attempt — never silently buys anything.
|
|
"""
|
|
if not AUTO_BUY_ENABLED:
|
|
return
|
|
if site not in AUTO_BUY_SITES:
|
|
return
|
|
if not products:
|
|
return
|
|
|
|
buyer = get_buyer(site)
|
|
if not buyer:
|
|
logger.warning(f"[auto-buy] No buyer implemented for {site}")
|
|
return
|
|
|
|
mode = "DRY RUN" if AUTO_BUY_DRY_RUN else "LIVE"
|
|
for product in products:
|
|
if not product.in_stock:
|
|
continue
|
|
logger.info(f"[auto-buy] {mode} — {site} — {product.name}")
|
|
try:
|
|
result = buyer.buy_product(product)
|
|
if result.success:
|
|
logger.info(f"[auto-buy] SUCCESS ({mode}): {product.name} — {result.message}")
|
|
send_stock_alert(
|
|
product_name=product.name,
|
|
product_url=product.url,
|
|
price=product.price or "See link",
|
|
site=site,
|
|
alert_type="auto_buy_success" if not result.dry_run else "auto_buy_dry_run",
|
|
image_url=product.image_url,
|
|
)
|
|
else:
|
|
logger.warning(f"[auto-buy] FAILED: {product.name} — {result.message}")
|
|
send_error_notification(
|
|
f"Auto-buy failed for {product.name} on {site}: {result.message}",
|
|
site.title(),
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"[auto-buy] Error buying {product.name} on {site}: {e}", exc_info=True)
|
|
|
|
|
|
def check_target():
|
|
"""Check Target for restocks and new drops"""
|
|
logger.info("Checking Target...")
|
|
scraper = scrapers["target"]
|
|
start_time = time.time()
|
|
total_products = 0
|
|
total_new = 0
|
|
total_restocks = 0
|
|
|
|
try:
|
|
for url in TARGET_URLS:
|
|
logger.info(f"Scraping: {url}")
|
|
products = scraper.scrape_category_page(url, max_pages=MAX_PAGES.get("target", 1))
|
|
|
|
# Always filter to Pokemon products only
|
|
products = scraper.filter_pokemon_products(products)
|
|
|
|
# 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")
|
|
|
|
total_products += len(products)
|
|
|
|
# Process and detect changes
|
|
new_products, restocked_products, price_changed_products = tracker.process_products(products)
|
|
total_new += len(new_products)
|
|
total_restocks += len(restocked_products)
|
|
|
|
# Send notifications for new products (if enabled)
|
|
if NEW_DROP_NOTIFICATIONS:
|
|
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}")
|
|
|
|
# Send notifications for price changes (if enabled)
|
|
if PRICE_CHANGE_NOTIFICATIONS:
|
|
for product, old_price, new_price in price_changed_products:
|
|
send_price_change_alert(
|
|
product_name=product.name,
|
|
product_url=product.url,
|
|
old_price=old_price,
|
|
new_price=new_price,
|
|
site="target",
|
|
in_stock=product.in_stock,
|
|
image_url=product.image_url,
|
|
)
|
|
logger.info(f"Sent notification for price change: {product.name} ({old_price} -> {new_price})")
|
|
|
|
# Auto-buy in-stock new drops and restocks
|
|
attempt_auto_buy("target", [p for p in new_products if p.in_stock] + restocked_products)
|
|
|
|
# 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"
|
|
)
|
|
|
|
# Log successful check to database
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
tracker.log_check("target", total_products, total_new, total_restocks, duration_ms, success=True)
|
|
scraper_state.record_check("target", success=True)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking Target: {e}")
|
|
send_error_notification(f"Error checking Target: {str(e)}", "Target")
|
|
# Log failed check to database
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
tracker.log_check("target", total_products, total_new, total_restocks, duration_ms, success=False, error_message=str(e))
|
|
scraper_state.record_check("target", success=False, error=str(e))
|
|
|
|
|
|
def check_gamestop():
|
|
"""Check GameStop for restocks and new drops"""
|
|
logger.info("Checking GameStop...")
|
|
scraper = scrapers["gamestop"]
|
|
start_time = time.time()
|
|
total_products = 0
|
|
total_new = 0
|
|
total_restocks = 0
|
|
|
|
try:
|
|
for url in GAMESTOP_URLS:
|
|
logger.info(f"Scraping: {url}")
|
|
products = scraper.scrape_category_page(url, max_pages=MAX_PAGES.get("gamestop", 1))
|
|
|
|
# Always filter to Pokemon products only
|
|
products = scraper.filter_pokemon_products(products)
|
|
|
|
# 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")
|
|
|
|
total_products += len(products)
|
|
|
|
# Process and detect changes
|
|
new_products, restocked_products, price_changed_products = tracker.process_products(products)
|
|
total_new += len(new_products)
|
|
total_restocks += len(restocked_products)
|
|
|
|
# Send notifications for new products (if enabled)
|
|
if NEW_DROP_NOTIFICATIONS:
|
|
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="gamestop",
|
|
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="gamestop",
|
|
alert_type="restock",
|
|
image_url=product.image_url,
|
|
)
|
|
logger.info(f"Sent notification for restock: {product.name}")
|
|
|
|
# Send notifications for price changes (if enabled)
|
|
if PRICE_CHANGE_NOTIFICATIONS:
|
|
for product, old_price, new_price in price_changed_products:
|
|
send_price_change_alert(
|
|
product_name=product.name,
|
|
product_url=product.url,
|
|
old_price=old_price,
|
|
new_price=new_price,
|
|
site="gamestop",
|
|
in_stock=product.in_stock,
|
|
image_url=product.image_url,
|
|
)
|
|
logger.info(f"Sent notification for price change: {product.name} ({old_price} -> {new_price})")
|
|
|
|
# Auto-buy in-stock new drops and restocks
|
|
attempt_auto_buy("gamestop", [p for p in new_products if p.in_stock] + restocked_products)
|
|
|
|
# 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"
|
|
)
|
|
|
|
# Log successful check to database
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
tracker.log_check("gamestop", total_products, total_new, total_restocks, duration_ms, success=True)
|
|
scraper_state.record_check("gamestop", success=True)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking GameStop: {e}")
|
|
send_error_notification(f"Error checking GameStop: {str(e)}", "GameStop")
|
|
# Log failed check to database
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
tracker.log_check("gamestop", total_products, total_new, total_restocks, duration_ms, success=False, error_message=str(e))
|
|
scraper_state.record_check("gamestop", success=False, error=str(e))
|
|
|
|
|
|
def check_bestbuy():
|
|
"""Check Best Buy for restocks and new drops"""
|
|
logger.info("Checking Best Buy...")
|
|
scraper = scrapers["bestbuy"]
|
|
start_time = time.time()
|
|
total_products = 0
|
|
total_new = 0
|
|
total_restocks = 0
|
|
|
|
try:
|
|
for url in BESTBUY_URLS:
|
|
logger.info(f"Scraping: {url}")
|
|
products = scraper.scrape_category_page(url, max_pages=MAX_PAGES.get("bestbuy", 1))
|
|
|
|
# Always filter to Pokemon products only
|
|
products = scraper.filter_pokemon_products(products)
|
|
|
|
# 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")
|
|
|
|
total_products += len(products)
|
|
|
|
# Process and detect changes
|
|
new_products, restocked_products, price_changed_products = tracker.process_products(products)
|
|
total_new += len(new_products)
|
|
total_restocks += len(restocked_products)
|
|
|
|
# Send notifications for new products (if enabled)
|
|
if NEW_DROP_NOTIFICATIONS:
|
|
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="bestbuy",
|
|
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="bestbuy",
|
|
alert_type="restock",
|
|
image_url=product.image_url,
|
|
)
|
|
logger.info(f"Sent notification for restock: {product.name}")
|
|
|
|
# Send notifications for price changes (if enabled)
|
|
if PRICE_CHANGE_NOTIFICATIONS:
|
|
for product, old_price, new_price in price_changed_products:
|
|
send_price_change_alert(
|
|
product_name=product.name,
|
|
product_url=product.url,
|
|
old_price=old_price,
|
|
new_price=new_price,
|
|
site="bestbuy",
|
|
in_stock=product.in_stock,
|
|
image_url=product.image_url,
|
|
)
|
|
logger.info(f"Sent notification for price change: {product.name} ({old_price} -> {new_price})")
|
|
|
|
# Auto-buy in-stock new drops and restocks
|
|
attempt_auto_buy("bestbuy", [p for p in new_products if p.in_stock] + restocked_products)
|
|
|
|
# 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"
|
|
)
|
|
|
|
# Log successful check to database
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
tracker.log_check("bestbuy", total_products, total_new, total_restocks, duration_ms, success=True)
|
|
scraper_state.record_check("bestbuy", success=True)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking Best Buy: {e}")
|
|
send_error_notification(f"Error checking Best Buy: {str(e)}", "Best Buy")
|
|
# Log failed check to database
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
tracker.log_check("bestbuy", total_products, total_new, total_restocks, duration_ms, success=False, error_message=str(e))
|
|
scraper_state.record_check("bestbuy", success=False, error=str(e))
|
|
|
|
|
|
def check_walmart():
|
|
"""Check Walmart for restocks and new drops"""
|
|
logger.info("Checking Walmart...")
|
|
scraper = scrapers["walmart"]
|
|
start_time = time.time()
|
|
total_products = 0
|
|
total_new = 0
|
|
total_restocks = 0
|
|
|
|
try:
|
|
for url in WALMART_URLS:
|
|
logger.info(f"Scraping: {url}")
|
|
products = scraper.scrape_category_page(url, max_pages=MAX_PAGES.get("walmart", 1))
|
|
|
|
# Always filter to Pokemon products only
|
|
products = scraper.filter_pokemon_products(products)
|
|
|
|
# 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")
|
|
|
|
total_products += len(products)
|
|
|
|
# Process and detect changes
|
|
new_products, restocked_products, price_changed_products = tracker.process_products(products)
|
|
total_new += len(new_products)
|
|
total_restocks += len(restocked_products)
|
|
|
|
# Send notifications for new products (if enabled)
|
|
if NEW_DROP_NOTIFICATIONS:
|
|
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="walmart",
|
|
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="walmart",
|
|
alert_type="restock",
|
|
image_url=product.image_url,
|
|
)
|
|
logger.info(f"Sent notification for restock: {product.name}")
|
|
|
|
# Send notifications for price changes (if enabled)
|
|
if PRICE_CHANGE_NOTIFICATIONS:
|
|
for product, old_price, new_price in price_changed_products:
|
|
send_price_change_alert(
|
|
product_name=product.name,
|
|
product_url=product.url,
|
|
old_price=old_price,
|
|
new_price=new_price,
|
|
site="walmart",
|
|
in_stock=product.in_stock,
|
|
image_url=product.image_url,
|
|
)
|
|
logger.info(f"Sent notification for price change: {product.name} ({old_price} -> {new_price})")
|
|
|
|
# Log stats
|
|
stats = tracker.get_stats()
|
|
logger.info(
|
|
f"Stats: {stats['total_products']} total, "
|
|
f"{stats['in_stock']} in stock, "
|
|
f"{stats['out_of_stock']} out of stock"
|
|
)
|
|
|
|
# Log successful check to database
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
tracker.log_check("walmart", total_products, total_new, total_restocks, duration_ms, success=True)
|
|
scraper_state.record_check("walmart", success=True)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking Walmart: {e}")
|
|
send_error_notification(f"Error checking Walmart: {str(e)}", "Walmart")
|
|
# Log failed check to database
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
tracker.log_check("walmart", total_products, total_new, total_restocks, duration_ms, success=False, error_message=str(e))
|
|
scraper_state.record_check("walmart", success=False, error=str(e))
|
|
|
|
|
|
def run_checks():
|
|
"""Run all enabled site checks"""
|
|
logger.info(f"Running checks at {datetime.now().strftime('%H:%M:%S')}")
|
|
|
|
# Note: Pokemon Center is monitored via Chrome Extension, not here
|
|
|
|
if "target" in SITES_ENABLED:
|
|
check_target()
|
|
|
|
if "gamestop" in SITES_ENABLED:
|
|
check_gamestop()
|
|
|
|
if "bestbuy" in SITES_ENABLED:
|
|
check_bestbuy()
|
|
|
|
if "walmart" in SITES_ENABLED:
|
|
check_walmart()
|
|
|
|
logger.info("Check cycle complete")
|
|
|
|
|
|
def graceful_shutdown(signum, frame):
|
|
"""Handle shutdown gracefully"""
|
|
logger.info("Shutting down...")
|
|
shutdown_browser()
|
|
scraper_state.state["monitor_running"] = False
|
|
scraper_state.state["monitor_pid"] = None
|
|
for site in scraper_state.state["scrapers"]:
|
|
scraper_state.state["scrapers"][site]["running"] = False
|
|
scraper_state._save_state()
|
|
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
|
|
|
|
# Mark monitor as running in state file
|
|
import os
|
|
scraper_state.state["monitor_running"] = True
|
|
scraper_state.state["monitor_pid"] = os.getpid()
|
|
for site in SITES_ENABLED:
|
|
if site in scraper_state.state["scrapers"]:
|
|
scraper_state.state["scrapers"][site]["running"] = True
|
|
scraper_state._save_state()
|
|
|
|
# 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()
|