mmcghen/pokemon-stock-checker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add multi-channel Discord webhook routing

Browse Source 
Route notifications to both site-specific and type-specific channels simultaneously. Includes site name normalization for matching config keys.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Michael McGhen
2026-04-01 22:16:50 -04:00
parent d33639fdc7
commit 74c2a45d50
2 changed files with 284 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.py
+49 -2
View File
@@ -5,13 +5,60 @@ 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 = "https://discord.com/api/webhooks/1471927173353963570/8Xh4Y_D8zRDi5MAP6ZlX0rf2Fc5wEtRKCup74kBZltk2qaatxArFp-yrk7ZWh5EOZbHp"
# This is the default/fallback webhook for anything not specifically routed below
DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/1488982276455796756/4HEzQDBF-Hs4a9Iyd7yvj7-SfP0BwrbQGdGBmQ9TL9urWzkTRaHelkvrqX4v9NTf2ZJ_"
# =============================================================================
# MULTI-CHANNEL WEBHOOK ROUTING (Optional)
# =============================================================================
# Route notifications to different Discord channels based on site and/or alert type.
# Leave webhooks as None to use the default DISCORD_WEBHOOK_URL above.
#
# Create webhooks in Discord: Server Settings -> Integrations -> Webhooks -> New Webhook
#
# Lookup priority:
# 1. overrides (specific site + type combo)
# 2. sites (all alerts for a specific store)
# 3. types (all alerts of a specific type)
# 4. default (DISCORD_WEBHOOK_URL)
DISCORD_WEBHOOKS = {
# Per-site channels - all alerts from a specific store go here first
"sites": {
"target": "https://discord.com/api/webhooks/1488982720901025852/O0KkVFHrQ6-k6FmCqQE4QpuWF6YFJ8YyFgojBPKcQu9g0UHzTpnAU9G1gbhOftARS11R",
"bestbuy": "https://discord.com/api/webhooks/1488982786781216798/AtG6x7FWzJjOsmosq9oPMuwlPrSwNHZJJTUuB3AApjyT7LIOxmEcYeiMFP7HiA_Ov1pn",
"gamestop": "https://discord.com/api/webhooks/1488982854036619355/NWZtydrKEhj-kxKbFi78f1Gpf3h16W_1kAJEcSuIgtjJUoAO8Sk7inOI2T8fb9GXo9xY",
"walmart": "https://discord.com/api/webhooks/1488982929450340375/0g16tJXw8f9UtUrRrLd2jDPWWD__LTDYO_X2VT2WL4F5GBYB8ykVxrjV-bCjzKKjSA5C",
"pokemoncenter": "https://discord.com/api/webhooks/1488983050090975364/DwY9QQvBZa6V-zz8mIg302sukbISSbkXOhDwe1Q6yDG0zAwvaGQcn8iTzNB4D676e3cA",
},
# Per-type channels - used when no site-specific webhook is set
"types": {
"restock": "https://discord.com/api/webhooks/1488982403048542269/s1J_p8R6IMGQGypPh_ZYTHHBD0cMnd_BjLGggVWaL2OcyDQAR42KS-CyuHcVr5s21vl3",
"new_drop": "https://discord.com/api/webhooks/1488982511613771898/RLDoHxNjsAx0PfHfT0kZe6-IW3g9PR_ckX_sI-CALcv4Lt0USslvI0rqePqW9gVP7D7v",
"price_change": "https://discord.com/api/webhooks/1488982573723025633/J864UL4vk_q9Rmj4i4wtkGfzB3nKAOc0y7jzBFDgUbhwg4XKm1Z7sWduqRc6x_ThWetc",
"error": "https://discord.com/api/webhooks/1488983128641900657/RP5FRSWwbVaRAf7RdXPs66ETbY8ziI1ekIx7lrHqxp8n200KpzUiX_EylCxXBH5nJD3B",
"captcha": "https://discord.com/api/webhooks/1488983128641900657/RP5FRSWwbVaRAf7RdXPs66ETbY8ziI1ekIx7lrHqxp8n200KpzUiX_EylCxXBH5nJD3B",
},
# Override specific site + type combinations (highest priority)
# Use tuples: ("site", "type"): "webhook_url"
"overrides": {
# ("bestbuy", "restock"): "https://discord.com/api/webhooks/xxx/bestbuy-restocks",
# ("pokemoncenter", "new_drop"): "https://discord.com/api/webhooks/xxx/pokemon-drops",
},
}
# Enable/disable Discord notifications
NOTIFICATIONS_ENABLED = True # Set to False to disable all Discord alerts
NEW_DROP_NOTIFICATIONS = True # Set to False to silence new drop notifications
PRICE_CHANGE_NOTIFICATIONS = True # Set to False to silence price change notifications
SKIP_DUPLICATE_SKUS = True # Skip notifications for products we've already notified about
# Restock notification cooldown (prevents duplicate alerts for same product)
# Set to number of hours before a product can trigger another restock notification
RESTOCK_COOLDOWN_HOURS = 6 # Minimum hours between restock alerts for same product
# Discord Bot Token (for interactive bot commands)
# Create a bot at https://discord.com/developers/applications
# Required for !setlocation, !stores, !stock commands
@@ -36,7 +83,7 @@ SORT_BY_NEWEST = True
# Which sites to monitor
# Note: PokemonCenter has strong bot protection (Imperva) - may need manual workarounds
# Note: Walmart has aggressive bot protection (PerimeterX) - may need stealth mode
SITES_ENABLED = ["target", "gamestop", "bestbuy"] # Options: "pokemoncenter", "target", "gamestop", "bestbuy" (disabled - needs rework), "walmart"
SITES_ENABLED = ["target", "bestbuy", "gamestop"] # Options: "pokemoncenter", "target", "gamestop", "bestbuy" (disabled - needs rework), "walmart"
# PokemonCenter URLs to monitor
POKEMON_CENTER_URLS = [
src/discord_notifier.py
+225 -13
View File
@@ -10,10 +10,69 @@ import requests
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Set
from config import DISCORD_WEBHOOK_URL, NOTIFICATIONS_ENABLED, SKIP_DUPLICATE_SKUS
from config import DISCORD_WEBHOOK_URL, DISCORD_WEBHOOKS, NOTIFICATIONS_ENABLED, SKIP_DUPLICATE_SKUS
logger = logging.getLogger(__name__)
def _normalize_site_name(site: str) -> str:
"""Normalize site name to match config keys (lowercase, no spaces)."""
if not site:
return ""
# Remove spaces and convert to lowercase
# "Best Buy" -> "bestbuy", "Pokemon Center" -> "pokemoncenter"
return site.lower().replace(" ", "").replace("-", "")
def get_webhooks(site: str = None, alert_type: str = None) -> list:
"""
Get all applicable webhook URLs based on site and alert type.
Sends to BOTH site-specific AND type-specific channels.
Args:
site: Site name (target, bestbuy, gamestop, walmart, pokemoncenter)
alert_type: Alert type (restock, new_drop, price_change, error, captcha)
Returns:
List of webhook URLs to send to (deduplicated)
"""
webhooks = []
normalized_site = _normalize_site_name(site)
normalized_type = alert_type.lower() if alert_type else None
# Check for specific override first (highest priority - replaces both site and type)
if normalized_site and normalized_type:
overrides = DISCORD_WEBHOOKS.get("overrides", {})
override_url = overrides.get((normalized_site, normalized_type))
if override_url:
return [override_url] # Override replaces all other routing
# Add site-specific webhook
if normalized_site:
sites = DISCORD_WEBHOOKS.get("sites", {})
site_url = sites.get(normalized_site)
if site_url:
webhooks.append(site_url)
# Add type-specific webhook
if normalized_type:
types = DISCORD_WEBHOOKS.get("types", {})
type_url = types.get(normalized_type)
if type_url and type_url not in webhooks:
webhooks.append(type_url)
# Fall back to default if no specific webhooks found
if not webhooks:
webhooks.append(DISCORD_WEBHOOK_URL)
return webhooks
def get_webhook(site: str = None, alert_type: str = None) -> str:
"""Legacy single-webhook function for backwards compatibility."""
webhooks = get_webhooks(site, alert_type)
return webhooks[0] if webhooks else DISCORD_WEBHOOK_URL
# Track notified products to avoid duplicates
NOTIFIED_FILE = Path(__file__).parent.parent / "data" / "notified_products.json"
_notified_urls: Set[str] = set()
@@ -56,6 +115,7 @@ def mark_as_notified(product_url: str):
COLOR_RESTOCK = 0x00FF00 # Green - item back in stock
COLOR_NEW_DROP = 0x0099FF # Blue - new product listing
COLOR_PREORDER = 0xFFAA00 # Orange - pre-order available
COLOR_PRICE_CHANGE = 0x9B59B6 # Purple - price changed
COLOR_ERROR = 0xFF0000 # Red - error notification
COLOR_FAVORITE_HIGH = 0xFF0000 # Red - high priority favorite
COLOR_FAVORITE_MEDIUM = 0xFFAA00 # Orange - medium priority favorite
@@ -109,12 +169,13 @@ def send_stock_alert(
logger.debug(f"Already notified about this product - skipping: {product_name}")
return True
# Determine which webhook to use
webhook_url = DISCORD_WEBHOOK_URL
# Determine which webhooks to use (priority: custom > routing > default)
if priority_settings and priority_settings.get('custom_webhook'):
webhook_url = priority_settings['custom_webhook']
webhook_urls = [priority_settings['custom_webhook']]
else:
webhook_urls = get_webhooks(site=site, alert_type=alert_type)
if webhook_url == "YOUR_WEBHOOK_URL_HERE":
if not webhook_urls or webhook_urls[0] == "YOUR_WEBHOOK_URL_HERE":
logger.error("Discord webhook URL not configured! Update config.py")
return False
@@ -189,6 +250,9 @@ def send_stock_alert(
"embeds": [embed],
}
# Send to all applicable webhooks
success = False
for webhook_url in webhook_urls:
try:
response = requests.post(
webhook_url,
@@ -196,14 +260,16 @@ def send_stock_alert(
timeout=10,
)
response.raise_for_status()
logger.info(f"Discord notification sent for: {product_name} (priority: {priority})")
success = True
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send Discord notification to {webhook_url}: {e}")
if success:
logger.info(f"Discord notification sent for: {product_name} to {len(webhook_urls)} channel(s)")
# Track that we notified about this product
if alert_type == "new_drop":
mark_as_notified(product_url)
return True
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send Discord notification: {e}")
return False
return success
def send_stock_alert_with_priority(product, alert_type: str = "restock",
@@ -231,7 +297,9 @@ def send_error_notification(error_message: str, site: str = "Unknown"):
"""Send an error notification to Discord"""
if not NOTIFICATIONS_ENABLED:
return True
if DISCORD_WEBHOOK_URL == "YOUR_WEBHOOK_URL_HERE":
webhook_urls = get_webhooks(site=site, alert_type="error")
if not webhook_urls or webhook_urls[0] == "YOUR_WEBHOOK_URL_HERE":
return False
embed = {
@@ -245,14 +313,158 @@ def send_error_notification(error_message: str, site: str = "Unknown"):
payload = {"embeds": [embed]}
success = False
for webhook_url in webhook_urls:
try:
response = requests.post(DISCORD_WEBHOOK_URL, json=payload, timeout=10)
response = requests.post(webhook_url, json=payload, timeout=10)
response.raise_for_status()
return True
success = True
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send error notification: {e}")
return success
def send_captcha_alert(site: str, url: str, message: str = None):
"""
Send a Discord notification when CAPTCHA is detected and human intervention is needed.
Args:
site: Site name where CAPTCHA was detected (e.g., "PokemonCenter")
url: URL that triggered the CAPTCHA
message: Optional custom message
"""
if not NOTIFICATIONS_ENABLED:
return True
webhook_urls = get_webhooks(site=site, alert_type="captcha")
if not webhook_urls or webhook_urls[0] == "YOUR_WEBHOOK_URL_HERE":
return False
default_message = "Human intervention required to solve CAPTCHA. The scraper is paused until resolved."
description = message or default_message
embed = {
"title": "\U0001F6A7 CAPTCHA DETECTED - Action Required",
"description": description,
"color": 0xFF6600, # Orange color for warning
"fields": [
{"name": "Site", "value": site, "inline": True},
{"name": "URL", "value": f"[Open Page]({url})", "inline": True},
{"name": "Action Needed", "value": "Please solve the CAPTCHA manually to resume monitoring.", "inline": False},
],
"footer": {"text": "Pokemon Stock Monitor"},
"timestamp": datetime.utcnow().isoformat(),
}
payload = {
"content": "@everyone \U0001F6A7 CAPTCHA requires manual intervention!",
"embeds": [embed]
}
success = False
for webhook_url in webhook_urls:
try:
response = requests.post(webhook_url, json=payload, timeout=10)
response.raise_for_status()
success = True
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send CAPTCHA alert: {e}")
if success:
logger.info(f"CAPTCHA alert sent for {site}")
return success
def send_price_change_alert(
product_name: str,
product_url: str,
old_price: str,
new_price: str,
site: str,
in_stock: bool = True,
image_url: Optional[str] = None,
):
"""
Send a Discord notification when a product's price changes.
Args:
product_name: Name of the product
product_url: Direct link to the product
old_price: Previous price string
new_price: New price string
site: Site name (target, bestbuy, etc.)
in_stock: Whether the product is currently in stock
image_url: Optional product image URL
"""
if not NOTIFICATIONS_ENABLED:
logger.debug(f"Notifications disabled - skipping price change alert for: {product_name}")
return True
webhook_urls = get_webhooks(site=site, alert_type="price_change")
if not webhook_urls or webhook_urls[0] == "YOUR_WEBHOOK_URL_HERE":
logger.error("Discord webhook URL not configured! Update config.py")
return False
# Determine if price went up or down
try:
old_val = float(old_price.replace("$", "").replace(",", "").strip()) if old_price else 0
new_val = float(new_price.replace("$", "").replace(",", "").strip()) if new_price else 0
price_dropped = new_val < old_val
except (ValueError, AttributeError):
price_dropped = False
# Use green for price drop, purple for price increase
color = 0x00FF00 if price_dropped else COLOR_PRICE_CHANGE
if price_dropped:
title = "\U0001F4B0 PRICE DROP!"
price_change_text = f"~~{old_price}~~ → **{new_price}**"
else:
title = "\U0001F4C8 Price Changed"
price_change_text = f"{old_price} → {new_price}"
site_emoji = SITE_EMOJIS.get(site.lower(), "\U0001F6D2")
site_display = site.replace("pokemoncenter", "Pokemon Center").title()
stock_status = "\u2705 In Stock" if in_stock else "\u274C Out of Stock"
embed = {
"title": title,
"description": f"**{product_name}**",
"url": product_url,
"color": color,
"fields": [
{"name": "Price Change", "value": price_change_text, "inline": True},
{"name": "Store", "value": f"{site_emoji} {site_display}", "inline": True},
{"name": "Stock Status", "value": stock_status, "inline": True},
{"name": "Link", "value": f"[\U0001F6D2 View Product]({product_url})", "inline": False},
],
"footer": {"text": "Pokemon Stock Monitor"},
"timestamp": datetime.utcnow().isoformat(),
}
if image_url:
embed["thumbnail"] = {"url": image_url}
# Only ping everyone for price drops on in-stock items
content = "@everyone" if price_dropped and in_stock else None
payload = {"embeds": [embed]}
if content:
payload["content"] = content
success = False
for webhook_url in webhook_urls:
try:
response = requests.post(webhook_url, json=payload, timeout=10)
response.raise_for_status()
success = True
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send price change notification: {e}")
if success:
logger.info(f"Price change notification sent for: {product_name} ({old_price} -> {new_price}) to {len(webhook_urls)} channel(s)")
return success
def send_startup_notification():
"""Send a notification that the monitor has started"""