mmcghen
9d49a99916
Chrome extension for PokemonCenter monitoring with Discord notifications. Includes Python scripts for Target monitoring. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
137 lines
4.7 KiB
Python
137 lines
4.7 KiB
Python
"""
|
|
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,
|
|
}
|