From 70a7df1f97d66531ec330903970eb4d882182177 Mon Sep 17 00:00:00 2001 From: Djeex Date: Fri, 11 Jul 2025 10:24:38 +0000 Subject: [PATCH 1/2] Modules --- Dockerfile | 8 +-- app/env_config.py | 93 +++++++++++++++++++++++++++++ app/gpu_checker.py | 38 ++++++++++++ app/localization.py | 50 ++++++++++++++++ app/main.py | 53 +++++++++++++++++ app/notifier.py | 140 ++++++++++++++++++++++++++++++++++++++++++++ docker/compose.yaml | 3 +- 7 files changed, 377 insertions(+), 8 deletions(-) create mode 100644 app/env_config.py create mode 100644 app/gpu_checker.py create mode 100644 app/localization.py create mode 100644 app/main.py create mode 100644 app/notifier.py diff --git a/Dockerfile b/Dockerfile index 22ab619..3eb1582 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,15 +4,11 @@ RUN apk add --no-cache ca-certificates WORKDIR /app -COPY app/nvidia-stock-bot.py /app/ - -COPY app/requirements.txt /app/ - -COPY app/localization.json /app/ +COPY /app/ /app/ RUN pip install --no-cache-dir -r requirements.txt ENV DISCORD_WEBHOOK_URL="https://example.com/webhook" \ REFRESH_TIME="30" -CMD ["python", "nvidia-stock-bot.py"] \ No newline at end of file +CMD ["python", "main.py"] \ No newline at end of file diff --git a/app/env_config.py b/app/env_config.py new file mode 100644 index 0000000..f9cf33a --- /dev/null +++ b/app/env_config.py @@ -0,0 +1,93 @@ +import os +import re +import logging +import requests +from requests.adapters import HTTPAdapter, Retry + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", +) +logging.info("Script started") + +# Load environment variables +DISCORD_WEBHOOK_URL = os.environ.get('DISCORD_WEBHOOK_URL') +DISCORD_SERVER_NAME = os.environ.get('DISCORD_SERVER_NAME', 'Shared for free') +API_URL_SKU = os.environ.get('API_URL_SKU', 'https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia') +API_URL_STOCK = os.environ.get('API_URL_STOCK', 'https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus=') +PRODUCT_URL = os.environ.get('PRODUCT_URL', 'https://marketplace.nvidia.com/fr-fr/consumer/graphics-cards/?locale=fr-fr&page=1&limit=12&manufacturer=NVIDIA') + +REFRESH_TIME = int(os.environ.get('REFRESH_TIME', '60')) # default 60 seconds if missing +TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true' + +PRODUCT_NAMES = os.environ.get('PRODUCT_NAMES') +if not PRODUCT_NAMES: + logging.error("❌ PRODUCT_NAMES is required but not defined.") + exit(1) +PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(',')] + +DISCORD_ROLES = os.environ.get('DISCORD_ROLES') + +# Validate roles and map them to product names +DISCORD_ROLE_MAP = {} +if not DISCORD_ROLES or not DISCORD_ROLES.strip(): + logging.warning("⚠️ DISCORD_ROLES not defined or empty. Defaulting all roles to @everyone.") + for name in PRODUCT_NAMES: + DISCORD_ROLE_MAP[name] = '@everyone' +else: + roles = [r.strip() if r.strip() else '@everyone' for r in DISCORD_ROLES.split(',')] + if len(roles) != len(PRODUCT_NAMES): + logging.error("❌ The number of DISCORD_ROLES must match PRODUCT_NAMES.") + exit(1) + for name, role in zip(PRODUCT_NAMES, roles): + if role != '@everyone' and not re.match(r'^<@&\d{17,20}>$', role): + logging.error(f"❌ Invalid DISCORD_ROLE format for {name}: {role}") + exit(1) + DISCORD_ROLE_MAP[name] = role + +if not DISCORD_WEBHOOK_URL: + logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.") + exit(1) + +# Mask webhook URL for logging +match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL) +if match: + webhook_id = match.group(1) + webhook_token = match.group(2) + masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10 + masked_webhook_token = webhook_token[:len(webhook_token) - 10] + '*' * 10 + wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}" +else: + wh_masked_url = "Invalid Webhook URL" + +logging.info(f"GPU: {PRODUCT_NAMES}") +logging.info(f"Discord Webhook URL: {wh_masked_url}") +logging.info(f"Discord Role Mention: {DISCORD_ROLES}") +logging.info(f"API URL SKU: {API_URL_SKU}") +logging.info(f"API URL Stock: {API_URL_STOCK}") +logging.info(f"Product URL: {PRODUCT_URL}") +logging.info(f"Refresh time: {REFRESH_TIME} seconds") +logging.info(f"Test Mode: {TEST_MODE}") + +HEADERS = { + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/131.0.0.0 Safari/537.36", + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9," + "image/avif,image/webp,image/apng,*/*;q=0.8", + "Accept-Encoding": "gzip, deflate, br, zstd", + "Accept-Language": "en-US,en;q=0.5", + "Cache-Control": "max-age=0", + "Connection": "keep-alive", + "Sec-Fetch-Dest": "document", + "Sec-Fetch-Mode": "navigate", + "Sec-Fetch-Site": "none", + "Sec-Fetch-User": "?1", + "Upgrade-Insecure-Requests": "1", + "Sec-GPC": "1", +} + +# Setup requests session with retries +session = requests.Session() +retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) +session.mount('https://', HTTPAdapter(max_retries=retries)) \ No newline at end of file diff --git a/app/gpu_checker.py b/app/gpu_checker.py new file mode 100644 index 0000000..553ae99 --- /dev/null +++ b/app/gpu_checker.py @@ -0,0 +1,38 @@ +import logging +from env_config import session, API_URL_SKU, API_URL_STOCK, PRODUCT_NAMES + +def get_current_skus(): + try: + response = session.get(API_URL_SKU) + response.raise_for_status() + data = response.json() + sku_map = {} + for product in data.get('products', []): + name = product.get('name') + sku = product.get('productSKU') + if name and sku and any(pn.lower() in name.lower() for pn in PRODUCT_NAMES): + sku_map[name] = sku + logging.info(f"✅ Fetched current SKUs: {sku_map}") + return sku_map + except Exception as e: + logging.error(f"🚨 Error fetching SKUs: {e}") + return {} + +def check_stock_for_skus(skus): + try: + sku_list = ','.join(skus) + url = f"{API_URL_STOCK}{sku_list}" + response = session.get(url) + response.raise_for_status() + data = response.json() + stock_status = {} + for sku_data in data.get('inventory', []): + sku = sku_data.get('sku') + available = sku_data.get('availability') == 'IN_STOCK' + price = sku_data.get('price', 'N/A') + stock_status[sku] = {'available': available, 'price': price} + logging.info(f"✅ Stock check results: {stock_status}") + return stock_status + except Exception as e: + logging.error(f"🚨 Error checking stock: {e}") + return {} \ No newline at end of file diff --git a/app/localization.py b/app/localization.py new file mode 100644 index 0000000..7896b26 --- /dev/null +++ b/app/localization.py @@ -0,0 +1,50 @@ +import json +import logging +import os +import sys + +required_keys = [ + "in_stock_title", "out_of_stock_title", "sku_change_title", + "buy_now", "price", "time", "footer", + "sku_description", "imminent_drop" +] + +try: + with open("localization.json", "r", encoding="utf-8") as f: + localization = json.load(f) +except FileNotFoundError: + logging.error("❌ localization.json file not found.") + sys.exit(1) + +language = os.environ.get("DISCORD_NOTIFICATION_LANGUAGE", "en").lower() +loc = localization.get(language) +fallback = localization.get("en") + +if not loc: + logging.warning(f"⚠️ Language '{language}' not found. Falling back to English.") + loc = fallback + +if not loc: + logging.error("❌ No localization found for language 'en'. Cannot continue.") + sys.exit(1) + +missing_keys = [key for key in required_keys if key not in loc] +if missing_keys: + logging.warning(f"⚠️ Missing keys in localization for '{language}': {', '.join(missing_keys)}. Falling back to English for those.") + for key in missing_keys: + if key in fallback: + loc[key] = fallback[key] + else: + logging.error(f"❌ Missing required key '{key}' in both '{language}' and fallback 'en'.") + sys.exit(1) + +# Export localization strings for import convenience +in_stock_title = loc["in_stock_title"] +out_of_stock_title = loc["out_of_stock_title"] +sku_change_title = loc["sku_change_title"] +buy_now = loc["buy_now"] +price_label = loc["price"] +time_label = loc["time"] +footer = loc["footer"] +sku_description = loc["sku_description"] +imminent_drop = loc["imminent_drop"] \ No newline at end of file diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..7ba4689 --- /dev/null +++ b/app/main.py @@ -0,0 +1,53 @@ +import time +import logging +from env_config import REFRESH_TIME, PRODUCT_NAMES, PRODUCT_URL +from gpu_checker import get_current_skus, check_stock_for_skus +from notifier import send_discord_notification, send_out_of_stock_notification, send_sku_change_notification + +def main(): + previous_sku_map = {} + while True: + try: + current_sku_map = get_current_skus() + if not current_sku_map: + logging.warning("⚠️ No SKUs found, skipping this cycle.") + time.sleep(REFRESH_TIME) + continue + + sku_list = list(current_sku_map.values()) + stock_data = check_stock_for_skus(sku_list) + if not stock_data: + logging.warning("⚠️ No stock data found, skipping this cycle.") + time.sleep(REFRESH_TIME) + continue + + for gpu_name in PRODUCT_NAMES: + old_sku = previous_sku_map.get(gpu_name) + new_sku = current_sku_map.get(gpu_name) + + # Detect SKU changes + if old_sku and new_sku and old_sku != new_sku: + send_sku_change_notification(gpu_name, old_sku, new_sku, PRODUCT_URL) + + sku_to_check = new_sku or old_sku + if sku_to_check and sku_to_check in stock_data: + availability = stock_data[sku_to_check]['available'] + price = stock_data[sku_to_check]['price'] + if availability: + send_discord_notification(gpu_name, PRODUCT_URL, price) + else: + send_out_of_stock_notification(gpu_name, PRODUCT_URL, price) + + previous_sku_map = current_sku_map + logging.info(f"Waiting {REFRESH_TIME}s before next check...") + time.sleep(REFRESH_TIME) + + except KeyboardInterrupt: + logging.info("Stopping script due to keyboard interrupt.") + break + except Exception as e: + logging.error(f"🚨 Unexpected error: {e}") + time.sleep(REFRESH_TIME) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/app/notifier.py b/app/notifier.py new file mode 100644 index 0000000..5c11917 --- /dev/null +++ b/app/notifier.py @@ -0,0 +1,140 @@ +import time +import logging +import requests +from env_config import DISCORD_WEBHOOK_URL, DISCORD_ROLE_MAP, TEST_MODE, DISCORD_SERVER_NAME +from localization import ( + in_stock_title, out_of_stock_title, sku_change_title, + buy_now, price_label, time_label, footer, + sku_description, imminent_drop +) + +def send_discord_notification(gpu_name: str, product_link: str, products_price: str): + timestamp_unix = int(time.time()) + if TEST_MODE: + logging.info(f"[TEST MODE] Discord Notification: {gpu_name} available!") + return + + embed = { + "title": in_stock_title.format(gpu_name=gpu_name), + "color": 3066993, + "thumbnail": { + "url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" + }, + "author": { + "name": "Nvidia Founder Editions" + }, + "fields": [ + { + "name": price_label, + "value": f"`{products_price}€`", + "inline": True + }, + { + "name": time_label, + "value": f" ", + "inline": True + }, + ], + "description": buy_now.format(product_link=product_link), + "footer": { + "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), + "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" + } + } + + payload = { + "content": DISCORD_ROLE_MAP.get(gpu_name, '@everyone'), + "username": "NviBot", + "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", + "embeds": [embed] + } + try: + response = requests.post(DISCORD_WEBHOOK_URL, json=payload) + if response.status_code == 204: + logging.info("✅ Notification sent to Discord.") + else: + logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") + except Exception as e: + logging.error(f"🚨 Error sending webhook: {e}") + +def send_out_of_stock_notification(gpu_name: str, product_link: str, products_price: str): + timestamp_unix = int(time.time()) + if TEST_MODE: + logging.info(f"[TEST MODE] Discord Notification: {gpu_name} out of stock!") + return + + embed = { + "title": out_of_stock_title.format(gpu_name=gpu_name), + "color": 15158332, + "thumbnail": { + "url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" + }, + "url": product_link, + "author": { + "name": "Nvidia Founder Editions" + }, + "footer": { + "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), + "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" + }, + "fields": [ + { + "name": time_label, + "value": f" ", + "inline": True + } + ] + } + payload = { + "username": "NviBot", + "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", + "embeds": [embed] + } + try: + response = requests.post(DISCORD_WEBHOOK_URL, json=payload) + if response.status_code == 204: + logging.info("✅ 'Out of stock' notification sent to Discord.") + else: + logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") + except Exception as e: + logging.error(f"🚨 Error sending webhook: {e}") + +def send_sku_change_notification(gpu_name: str, old_sku: str, new_sku: str, product_link: str): + timestamp_unix = int(time.time()) + if TEST_MODE: + logging.info(f"[TEST MODE] Discord Notification: SKU change for {gpu_name}: {old_sku} -> {new_sku}") + return + + embed = { + "title": sku_change_title.format(gpu_name=gpu_name), + "color": 3447003, + "author": { + "name": "Nvidia Founder Editions" + }, + "description": sku_description.format(old_sku=old_sku, new_sku=new_sku), + "url": product_link, + "footer": { + "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), + "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" + }, + "fields": [ + { + "name": time_label, + "value": f" ", + "inline": True + } + ] + } + payload = { + "username": "NviBot", + "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", + "embeds": [embed] + } + try: + response = requests.post(DISCORD_WEBHOOK_URL, json=payload) + if response.status_code == 204: + logging.info("✅ SKU change notification sent to Discord.") + else: + logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") + except Exception as e: + logging.error(f"🚨 Error sending webhook: {e}") \ No newline at end of file diff --git a/docker/compose.yaml b/docker/compose.yaml index 2cbd3ff..095be85 100644 --- a/docker/compose.yaml +++ b/docker/compose.yaml @@ -12,5 +12,4 @@ services: - API_URL_SKU=${API_URL_SKU} # API listing the product for your country - API_URL_STOCK=${API_URL_STOCK} # API providing stock data for your country - PRODUCT_URL=${PRODUCT_URL} # GPU purchase URL for your country - - PYTHONUNBUFFERED=1 # Enables real-time log output - command: python nvidia-stock-bot.py # Run the script \ No newline at end of file + - PYTHONUNBUFFERED=1 # Enables real-time log output \ No newline at end of file From 584c2220764d7e62d9d64a19bb295208fd7a4229 Mon Sep 17 00:00:00 2001 From: Djeex Date: Fri, 11 Jul 2025 11:42:52 +0000 Subject: [PATCH 2/2] Revamp --- README.md | 1 - app/env_config.py | 115 +++++++---- app/gpu_checker.py | 130 +++++++++---- app/localization.py | 50 ----- app/main.py | 68 ++----- app/notifier.py | 115 ++++------- app/nvidia-stock-bot.py | 411 ---------------------------------------- 7 files changed, 241 insertions(+), 649 deletions(-) delete mode 100644 app/localization.py delete mode 100644 app/nvidia-stock-bot.py diff --git a/README.md b/README.md index 23c0be8..5edd1a4 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,6 @@ services: - API_URL_STOCK= # API providing stock data for your country - PRODUCT_URL= # GPU purchase URL for your country - PYTHONUNBUFFERED=1 # Enables real-time log output - command: python nvidia-stock-bot.py ``` **Environment Variables:** diff --git a/app/env_config.py b/app/env_config.py index f9cf33a..bc66402 100644 --- a/app/env_config.py +++ b/app/env_config.py @@ -1,34 +1,41 @@ import os import re import logging -import requests -from requests.adapters import HTTPAdapter, Retry +import json +import sys +# Logger setup logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logging.info("Script started") -# Load environment variables -DISCORD_WEBHOOK_URL = os.environ.get('DISCORD_WEBHOOK_URL') -DISCORD_SERVER_NAME = os.environ.get('DISCORD_SERVER_NAME', 'Shared for free') -API_URL_SKU = os.environ.get('API_URL_SKU', 'https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia') -API_URL_STOCK = os.environ.get('API_URL_STOCK', 'https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus=') -PRODUCT_URL = os.environ.get('PRODUCT_URL', 'https://marketplace.nvidia.com/fr-fr/consumer/graphics-cards/?locale=fr-fr&page=1&limit=12&manufacturer=NVIDIA') +try: + # Env variables + DISCORD_WEBHOOK_URL = os.environ['DISCORD_WEBHOOK_URL'] + DISCORD_SERVER_NAME = os.environ.get('DISCORD_SERVER_NAME', 'Shared for free') + API_URL_SKU = os.environ.get('API_URL_SKU', 'https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia') + API_URL_STOCK = os.environ.get('API_URL_STOCK', 'https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus=') + REFRESH_TIME = int(os.environ['REFRESH_TIME']) + TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true' + PRODUCT_URL = os.environ.get('PRODUCT_URL', 'https://marketplace.nvidia.com/fr-fr/consumer/graphics-cards/?locale=fr-fr&page=1&limit=12&manufacturer=NVIDIA') + DISCORD_ROLES = os.environ.get('DISCORD_ROLES') + PRODUCT_NAMES = os.environ['PRODUCT_NAMES'] +except KeyError as e: + logging.error(f"Missing environment variable: {e}") + sys.exit(1) +except ValueError: + logging.error("REFRESH_TIME must be a valid integer.") + sys.exit(1) -REFRESH_TIME = int(os.environ.get('REFRESH_TIME', '60')) # default 60 seconds if missing -TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true' - -PRODUCT_NAMES = os.environ.get('PRODUCT_NAMES') if not PRODUCT_NAMES: logging.error("❌ PRODUCT_NAMES is required but not defined.") - exit(1) + sys.exit(1) + PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(',')] -DISCORD_ROLES = os.environ.get('DISCORD_ROLES') - -# Validate roles and map them to product names +# Role mapping DISCORD_ROLE_MAP = {} if not DISCORD_ROLES or not DISCORD_ROLES.strip(): logging.warning("⚠️ DISCORD_ROLES not defined or empty. Defaulting all roles to @everyone.") @@ -38,37 +45,29 @@ else: roles = [r.strip() if r.strip() else '@everyone' for r in DISCORD_ROLES.split(',')] if len(roles) != len(PRODUCT_NAMES): logging.error("❌ The number of DISCORD_ROLES must match PRODUCT_NAMES.") - exit(1) + sys.exit(1) for name, role in zip(PRODUCT_NAMES, roles): if role != '@everyone' and not re.match(r'^<@&\d{17,20}>$', role): logging.error(f"❌ Invalid DISCORD_ROLE format for {name}: {role}") - exit(1) + sys.exit(1) DISCORD_ROLE_MAP[name] = role if not DISCORD_WEBHOOK_URL: logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.") - exit(1) + sys.exit(1) -# Mask webhook URL for logging +# Masked webhook for display match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL) if match: webhook_id = match.group(1) webhook_token = match.group(2) masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10 - masked_webhook_token = webhook_token[:len(webhook_token) - 10] + '*' * 10 + masked_webhook_token = webhook_token[:len(webhook_token) - 120] + '*' * 10 wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}" else: - wh_masked_url = "Invalid Webhook URL" - -logging.info(f"GPU: {PRODUCT_NAMES}") -logging.info(f"Discord Webhook URL: {wh_masked_url}") -logging.info(f"Discord Role Mention: {DISCORD_ROLES}") -logging.info(f"API URL SKU: {API_URL_SKU}") -logging.info(f"API URL Stock: {API_URL_STOCK}") -logging.info(f"Product URL: {PRODUCT_URL}") -logging.info(f"Refresh time: {REFRESH_TIME} seconds") -logging.info(f"Test Mode: {TEST_MODE}") + wh_masked_url = "[Invalid webhook URL]" +# HTTP headers HEADERS = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " @@ -87,7 +86,53 @@ HEADERS = { "Sec-GPC": "1", } -# Setup requests session with retries -session = requests.Session() -retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) -session.mount('https://', HTTPAdapter(max_retries=retries)) \ No newline at end of file +# Localization loading +language = os.environ.get("DISCORD_NOTIFICATION_LANGUAGE", "en").lower() +try: + with open("localization.json", "r", encoding="utf-8") as f: + localization = json.load(f) +except FileNotFoundError: + logging.error("❌ localization.json file not found.") + sys.exit(1) + +required_keys = [ + "in_stock_title", "out_of_stock_title", "sku_change_title", + "buy_now", "price", "time", "footer", + "sku_description", "imminent_drop" +] +loc = localization.get(language, localization.get("en", {})) +logging.info(f"Notification language: {language}") + +if not loc: + logging.warning(f"⚠️ Language '{language}' not found. Falling back to English.") + loc = localization.get("en", {}) + language = "en" + +missing_keys = [key for key in required_keys if key not in loc] +fallback = localization.get("en", {}) +for key in missing_keys: + if key in fallback: + loc[key] = fallback[key] + else: + logging.error(f"❌ Missing required key '{key}' in both '{language}' and fallback 'en'.") + sys.exit(1) + +# Public constants +in_stock_title = loc["in_stock_title"] +out_of_stock_title = loc["out_of_stock_title"] +sku_change_title = loc["sku_change_title"] +buy_now = loc["buy_now"] +price_label = loc["price"] +time_label = loc["time"] +footer = loc["footer"] +sku_description = loc["sku_description"] +imminent_drop = loc["imminent_drop"] + +logging.info(f"GPU: {PRODUCT_NAMES}") +logging.info(f"Discord Webhook URL: {wh_masked_url}") +logging.info(f"Discord Role Mention: {DISCORD_ROLES}") +logging.info(f"API URL SKU: {API_URL_SKU}") +logging.info(f"API URL Stock: {API_URL_STOCK}") +logging.info(f"Product URL: {PRODUCT_URL}") +logging.info(f"Refresh time: {REFRESH_TIME} seconds") +logging.info(f"Test Mode: {TEST_MODE}") diff --git a/app/gpu_checker.py b/app/gpu_checker.py index 553ae99..817d4af 100644 --- a/app/gpu_checker.py +++ b/app/gpu_checker.py @@ -1,38 +1,104 @@ +import requests import logging -from env_config import session, API_URL_SKU, API_URL_STOCK, PRODUCT_NAMES +from env_config import API_URL_SKU, API_URL_STOCK, HEADERS, PRODUCT_NAMES, PRODUCT_URL +from notifier import send_discord_notification, send_out_of_stock_notification, send_sku_change_notification + +from requests.adapters import HTTPAdapter, Retry + +session = requests.Session() +retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) +session.mount('https://', HTTPAdapter(max_retries=retries)) + +last_sku_dict = {} +global_stock_status_dict = {} +first_run_dict = {name: True for name in PRODUCT_NAMES} + +def check_rtx_50_founders(): + global last_sku_dict, global_stock_status_dict, first_run_dict -def get_current_skus(): try: - response = session.get(API_URL_SKU) + response = session.get(API_URL_SKU, headers=HEADERS, timeout=10) + logging.info(f"SKU API response: {response.status_code}") response.raise_for_status() data = response.json() - sku_map = {} - for product in data.get('products', []): - name = product.get('name') - sku = product.get('productSKU') - if name and sku and any(pn.lower() in name.lower() for pn in PRODUCT_NAMES): - sku_map[name] = sku - logging.info(f"✅ Fetched current SKUs: {sku_map}") - return sku_map - except Exception as e: - logging.error(f"🚨 Error fetching SKUs: {e}") - return {} + except requests.exceptions.RequestException as e: + logging.error(f"SKU API error: {e}") + return -def check_stock_for_skus(skus): - try: - sku_list = ','.join(skus) - url = f"{API_URL_STOCK}{sku_list}" - response = session.get(url) - response.raise_for_status() - data = response.json() - stock_status = {} - for sku_data in data.get('inventory', []): - sku = sku_data.get('sku') - available = sku_data.get('availability') == 'IN_STOCK' - price = sku_data.get('price', 'N/A') - stock_status[sku] = {'available': available, 'price': price} - logging.info(f"✅ Stock check results: {stock_status}") - return stock_status - except Exception as e: - logging.error(f"🚨 Error checking stock: {e}") - return {} \ No newline at end of file + # All available products + all_products = data['searchedProducts']['productDetails'] + + for product_name in PRODUCT_NAMES: + product_details = None + for p in all_products: + if p.get("gpu", "").strip() == product_name: + product_details = p + break + + if not product_details: + logging.warning(f"⚠️ No product with GPU '{product_name}' found.") + continue + + product_sku = product_details['productSKU'] + product_upc = product_details.get('productUPC', "") + if not isinstance(product_upc, list): + product_upc = [product_upc] + + # Check SKU change + old_sku = last_sku_dict.get(product_name) + if old_sku and old_sku != product_sku and not first_run_dict[product_name]: + logging.warning(f"⚠️ SKU changed for {product_name}: {old_sku} → {product_sku}") + send_sku_change_notification(product_name, old_sku, product_sku, PRODUCT_URL) + + last_sku_dict[product_name] = product_sku + first_run_dict[product_name] = False + + # Stock check + api_stock_url = API_URL_STOCK + product_sku + logging.info(f"[{product_name}] Checking stock: {api_stock_url}") + + try: + response = session.get(api_stock_url, headers=HEADERS, timeout=10) + logging.info(f"[{product_name}] Stock API response: {response.status_code}") + response.raise_for_status() + stock_data = response.json() + except requests.exceptions.RequestException as e: + logging.error(f"Stock API error: {e}") + continue + + products = stock_data.get("listMap", []) + products_price = "Price not available" + if isinstance(products, list) and len(products) > 0: + for p in products: + price = p.get("price", 'Price not available') + if price != 'Price not available': + products_price = price + break + else: + logging.error(f"[{product_name}] Product list is empty or malformed.") + + found_in_stock = set() + for p in products: + gpu_name = p.get("fe_sku", "").upper() + is_active = p.get("is_active") == "true" + if is_active and any(upc.upper() in gpu_name for upc in product_upc): + found_in_stock.add(gpu_name) + + for upc in product_upc: + upc_upper = upc.upper() + currently_in_stock = upc_upper in found_in_stock + previously_in_stock = global_stock_status_dict.get((product_name, upc_upper), False) + + if currently_in_stock and not previously_in_stock: + send_discord_notification(product_name, PRODUCT_URL, products_price) + global_stock_status_dict[(product_name, upc_upper)] = True + logging.info(f"[{product_name}] {upc} is now in stock!") + elif not currently_in_stock and previously_in_stock: + send_out_of_stock_notification(product_name, PRODUCT_URL, products_price) + global_stock_status_dict[(product_name, upc_upper)] = False + logging.info(f"[{product_name}] {upc} is now out of stock.") + elif currently_in_stock: + logging.info(f"[{product_name}] {upc} still in stock.") + else: + logging.info(f"[{product_name}] {upc} still out of stock.") + \ No newline at end of file diff --git a/app/localization.py b/app/localization.py deleted file mode 100644 index 7896b26..0000000 --- a/app/localization.py +++ /dev/null @@ -1,50 +0,0 @@ -import json -import logging -import os -import sys - -required_keys = [ - "in_stock_title", "out_of_stock_title", "sku_change_title", - "buy_now", "price", "time", "footer", - "sku_description", "imminent_drop" -] - -try: - with open("localization.json", "r", encoding="utf-8") as f: - localization = json.load(f) -except FileNotFoundError: - logging.error("❌ localization.json file not found.") - sys.exit(1) - -language = os.environ.get("DISCORD_NOTIFICATION_LANGUAGE", "en").lower() -loc = localization.get(language) -fallback = localization.get("en") - -if not loc: - logging.warning(f"⚠️ Language '{language}' not found. Falling back to English.") - loc = fallback - -if not loc: - logging.error("❌ No localization found for language 'en'. Cannot continue.") - sys.exit(1) - -missing_keys = [key for key in required_keys if key not in loc] -if missing_keys: - logging.warning(f"⚠️ Missing keys in localization for '{language}': {', '.join(missing_keys)}. Falling back to English for those.") - for key in missing_keys: - if key in fallback: - loc[key] = fallback[key] - else: - logging.error(f"❌ Missing required key '{key}' in both '{language}' and fallback 'en'.") - sys.exit(1) - -# Export localization strings for import convenience -in_stock_title = loc["in_stock_title"] -out_of_stock_title = loc["out_of_stock_title"] -sku_change_title = loc["sku_change_title"] -buy_now = loc["buy_now"] -price_label = loc["price"] -time_label = loc["time"] -footer = loc["footer"] -sku_description = loc["sku_description"] -imminent_drop = loc["imminent_drop"] \ No newline at end of file diff --git a/app/main.py b/app/main.py index 7ba4689..eeae05d 100644 --- a/app/main.py +++ b/app/main.py @@ -1,53 +1,27 @@ import time import logging -from env_config import REFRESH_TIME, PRODUCT_NAMES, PRODUCT_URL -from gpu_checker import get_current_skus, check_stock_for_skus -from notifier import send_discord_notification, send_out_of_stock_notification, send_sku_change_notification +import signal +import sys +from gpu_checker import check_rtx_50_founders +from env_config import REFRESH_TIME -def main(): - previous_sku_map = {} - while True: - try: - current_sku_map = get_current_skus() - if not current_sku_map: - logging.warning("⚠️ No SKUs found, skipping this cycle.") - time.sleep(REFRESH_TIME) - continue +# Signal handler function +def handle_exit(signum, frame): + logging.info(f"🛑 Received signal {signum}. Exiting gracefully...") + sys.exit(0) - sku_list = list(current_sku_map.values()) - stock_data = check_stock_for_skus(sku_list) - if not stock_data: - logging.warning("⚠️ No stock data found, skipping this cycle.") - time.sleep(REFRESH_TIME) - continue - - for gpu_name in PRODUCT_NAMES: - old_sku = previous_sku_map.get(gpu_name) - new_sku = current_sku_map.get(gpu_name) - - # Detect SKU changes - if old_sku and new_sku and old_sku != new_sku: - send_sku_change_notification(gpu_name, old_sku, new_sku, PRODUCT_URL) - - sku_to_check = new_sku or old_sku - if sku_to_check and sku_to_check in stock_data: - availability = stock_data[sku_to_check]['available'] - price = stock_data[sku_to_check]['price'] - if availability: - send_discord_notification(gpu_name, PRODUCT_URL, price) - else: - send_out_of_stock_notification(gpu_name, PRODUCT_URL, price) - - previous_sku_map = current_sku_map - logging.info(f"Waiting {REFRESH_TIME}s before next check...") - time.sleep(REFRESH_TIME) - - except KeyboardInterrupt: - logging.info("Stopping script due to keyboard interrupt.") - break - except Exception as e: - logging.error(f"🚨 Unexpected error: {e}") - time.sleep(REFRESH_TIME) +# Register signal handlers +signal.signal(signal.SIGINT, handle_exit) # Ctrl+C +signal.signal(signal.SIGTERM, handle_exit) # docker stop / kill -15 if __name__ == "__main__": - main() \ No newline at end of file + try: + while True: + start = time.time() + check_rtx_50_founders() + elapsed = time.time() - start + time.sleep(max(0, REFRESH_TIME - elapsed)) + except KeyboardInterrupt: + logging.info("🛑 Script interrupted by user (KeyboardInterrupt). Exiting gracefully.") + sys.exit(0) + \ No newline at end of file diff --git a/app/notifier.py b/app/notifier.py index 5c11917..aadcb86 100644 --- a/app/notifier.py +++ b/app/notifier.py @@ -1,53 +1,41 @@ import time import logging import requests -from env_config import DISCORD_WEBHOOK_URL, DISCORD_ROLE_MAP, TEST_MODE, DISCORD_SERVER_NAME -from localization import ( +from env_config import ( + DISCORD_WEBHOOK_URL, DISCORD_SERVER_NAME, DISCORD_ROLE_MAP, TEST_MODE, in_stock_title, out_of_stock_title, sku_change_title, - buy_now, price_label, time_label, footer, - sku_description, imminent_drop + buy_now, price_label, time_label, footer, sku_description, imminent_drop ) -def send_discord_notification(gpu_name: str, product_link: str, products_price: str): - timestamp_unix = int(time.time()) +AVATAR = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" +THUMBNAIL = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" + +def send_discord_notification(gpu_name, product_link, products_price): + timestamp = int(time.time()) if TEST_MODE: - logging.info(f"[TEST MODE] Discord Notification: {gpu_name} available!") + logging.info(f"[TEST MODE] Notification: {gpu_name} available!") return embed = { "title": in_stock_title.format(gpu_name=gpu_name), "color": 3066993, - "thumbnail": { - "url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" - }, - "author": { - "name": "Nvidia Founder Editions" - }, + "thumbnail": {"url": THUMBNAIL}, + "author": {"name": "Nvidia Founder Editions"}, "fields": [ - { - "name": price_label, - "value": f"`{products_price}€`", - "inline": True - }, - { - "name": time_label, - "value": f" ", - "inline": True - }, + {"name": price_label, "value": f"`{products_price}€`", "inline": True}, + {"name": time_label, "value": f" ", "inline": True} ], "description": buy_now.format(product_link=product_link), - "footer": { - "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), - "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" - } + "footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR} } payload = { - "content": DISCORD_ROLE_MAP.get(gpu_name, '@everyone'), + "content": DISCORD_ROLE_MAP.get(gpu_name, "@everyone"), "username": "NviBot", - "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", + "avatar_url": AVATAR, "embeds": [embed] } + try: response = requests.post(DISCORD_WEBHOOK_URL, json=payload) if response.status_code == 204: @@ -57,84 +45,65 @@ def send_discord_notification(gpu_name: str, product_link: str, products_price: except Exception as e: logging.error(f"🚨 Error sending webhook: {e}") -def send_out_of_stock_notification(gpu_name: str, product_link: str, products_price: str): - timestamp_unix = int(time.time()) +def send_out_of_stock_notification(gpu_name, product_link, products_price): + timestamp = int(time.time()) if TEST_MODE: - logging.info(f"[TEST MODE] Discord Notification: {gpu_name} out of stock!") + logging.info(f"[TEST MODE] Out of stock: {gpu_name}") return embed = { "title": out_of_stock_title.format(gpu_name=gpu_name), "color": 15158332, - "thumbnail": { - "url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" - }, + "thumbnail": {"url": THUMBNAIL}, "url": product_link, - "author": { - "name": "Nvidia Founder Editions" - }, - "footer": { - "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), - "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" - }, - "fields": [ - { - "name": time_label, - "value": f" ", - "inline": True - } - ] + "author": {"name": "Nvidia Founder Editions"}, + "footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR}, + "fields": [{"name": time_label, "value": f" ", "inline": True}] } + payload = { "username": "NviBot", - "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", + "avatar_url": AVATAR, "embeds": [embed] } + try: response = requests.post(DISCORD_WEBHOOK_URL, json=payload) if response.status_code == 204: - logging.info("✅ 'Out of stock' notification sent to Discord.") + logging.info("✅ Out-of-stock notification sent.") else: logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") except Exception as e: logging.error(f"🚨 Error sending webhook: {e}") -def send_sku_change_notification(gpu_name: str, old_sku: str, new_sku: str, product_link: str): - timestamp_unix = int(time.time()) +def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link): + timestamp = int(time.time()) if TEST_MODE: - logging.info(f"[TEST MODE] Discord Notification: SKU change for {gpu_name}: {old_sku} -> {new_sku}") + logging.info(f"[TEST MODE] SKU change: {old_sku} → {new_sku}") return embed = { "title": sku_change_title.format(gpu_name=gpu_name), - "color": 3447003, - "author": { - "name": "Nvidia Founder Editions" - }, - "description": sku_description.format(old_sku=old_sku, new_sku=new_sku), "url": product_link, - "footer": { - "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), - "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" - }, - "fields": [ - { - "name": time_label, - "value": f" ", - "inline": True - } - ] + "description": sku_description.format(old_sku=old_sku, new_sku=new_sku), + "color": 16776960, + "footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR}, + "fields": [{"name": time_label, "value": f" ", "inline": True}] } + payload = { + "content": imminent_drop.format(DISCORD_ROLE=DISCORD_ROLE_MAP.get(gpu_name, '@everyone')), "username": "NviBot", - "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", + "avatar_url": AVATAR, "embeds": [embed] } + try: response = requests.post(DISCORD_WEBHOOK_URL, json=payload) if response.status_code == 204: - logging.info("✅ SKU change notification sent to Discord.") + logging.info("✅ SKU change notification sent.") else: logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") except Exception as e: - logging.error(f"🚨 Error sending webhook: {e}") \ No newline at end of file + logging.error(f"🚨 Error sending webhook: {e}") + \ No newline at end of file diff --git a/app/nvidia-stock-bot.py b/app/nvidia-stock-bot.py deleted file mode 100644 index 023a529..0000000 --- a/app/nvidia-stock-bot.py +++ /dev/null @@ -1,411 +0,0 @@ -import requests -import logging -import time -import os -import re -import json -from requests.adapters import HTTPAdapter, Retry - -# Logger configuration -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(message)s", -) -logging.info("Script started") - -# Retrieve environment variables -try: - DISCORD_WEBHOOK_URL = os.environ.get('DISCORD_WEBHOOK_URL') - DISCORD_SERVER_NAME = os.environ.get('DISCORD_SERVER_NAME', 'Shared for free') - API_URL_SKU = os.environ.get('API_URL_SKU', 'https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia') - API_URL_STOCK = os.environ.get('API_URL_STOCK', 'https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus=') - REFRESH_TIME = int(os.environ.get('REFRESH_TIME')) - TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true' - PRODUCT_URL = os.environ.get('PRODUCT_URL', 'https://marketplace.nvidia.com/fr-fr/consumer/graphics-cards/?locale=fr-fr&page=1&limit=12&manufacturer=NVIDIA') - DISCORD_ROLES = os.environ.get('DISCORD_ROLES') - PRODUCT_NAMES = os.environ.get('PRODUCT_NAMES') - - if not PRODUCT_NAMES: - logging.error("❌ PRODUCT_NAMES is required but not defined.") - exit(1) - - PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(',')] - DISCORD_ROLE_MAP = {} - - if not DISCORD_ROLES or not DISCORD_ROLES.strip(): - logging.warning("⚠️ DISCORD_ROLES not defined or empty. Defaulting all roles to @everyone.") - for name in PRODUCT_NAMES: - DISCORD_ROLE_MAP[name] = '@everyone' - else: - roles = [r.strip() if r.strip() else '@everyone' for r in DISCORD_ROLES.split(',')] - - if len(roles) != len(PRODUCT_NAMES): - logging.error("❌ The number of DISCORD_ROLES must match PRODUCT_NAMES.") - exit(1) - - for name, role in zip(PRODUCT_NAMES, roles): - if role != '@everyone' and not re.match(r'^<@&\d{17,20}>$', role): - logging.error(f"❌ Invalid DISCORD_ROLE format for {name}: {role}") - exit(1) - DISCORD_ROLE_MAP[name] = role - - - if not DISCORD_WEBHOOK_URL: - logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.") - exit(1) - - # Regex to extract ID and token - match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL) - if match: - webhook_id = match.group(1) - webhook_token = match.group(2) - - # Mask last characters of the ID - masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10 - - # Mask last characters of the token - masked_webhook_token = webhook_token[:len(webhook_token) - 120] + '*' * 10 - - # Rebuild masked URL - wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}" - -# Error logging -except KeyError as e: - logging.error(f"Missing environment variable: {e}") - exit(1) -except ValueError: - logging.error("REFRESH_TIME must be a valid integer.") - exit(1) - -# Localization -try: - with open("localization.json", "r", encoding="utf-8") as f: - localization = json.load(f) -except FileNotFoundError: - logging.error("❌ localization.json file not found.") - exit(1) - -language = os.environ.get("DISCORD_NOTIFICATION_LANGUAGE", "en").lower() -required_keys = [ - "in_stock_title", "out_of_stock_title", "sku_change_title", - "buy_now", "price", "time", "footer", - "sku_description", "imminent_drop" -] -loc = localization.get(language, localization["en"]) -logging.info(f"Notification language: {language}") - -if not loc: - logging.warning(f"⚠️ Language '{language}' not found. Falling back to English.") - loc = localization.get("en") - language = "en" - -if not loc: - logging.error("❌ No localization found for language 'en'. Cannot continue.") - sys.exit(1) -fallback = localization.get("en", {}) -missing_keys = [key for key in required_keys if key not in loc] - -if missing_keys: - logging.warning(f"⚠️ Missing keys in localization for '{language}': {', '.join(missing_keys)}. Falling back to English for those.") - - for key in missing_keys: - if key in fallback: - loc[key] = fallback[key] - else: - logging.error(f"❌ Missing required key '{key}' in both '{language}' and fallback 'en'.") - sys.exit(1) - -in_stock_title = loc["in_stock_title"] -out_of_stock_title = loc["out_of_stock_title"] -sku_change_title = loc["sku_change_title"] -buy_now = loc["buy_now"] -price_label = loc["price"] -time_label = loc["time"] -footer = loc["footer"] -sku_description = loc["sku_description"] -imminent_drop = loc["imminent_drop"] - -# Display URLs and configurations -logging.info(f"GPU: {PRODUCT_NAMES}") -logging.info(f"Discord Webhook URL: {wh_masked_url}") -logging.info(f"Discord Role Mention: {DISCORD_ROLES}") -logging.info(f"API URL SKU: {API_URL_SKU}") -logging.info(f"API URL Stock: {API_URL_STOCK}") -logging.info(f"Product URL: {PRODUCT_URL}") -logging.info(f"Refresh time: {REFRESH_TIME} seconds") -logging.info(f"Test Mode: {TEST_MODE}") - -# HTTP headers -HEADERS = { - "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/131.0.0.0 Safari/537.36", - "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9," - "image/avif,image/webp,image/apng,*/*;q=0.8", - "Accept-Encoding": "gzip, deflate, br, zstd", - "Accept-Language": "en-US,en;q=0.5", - "Cache-Control": "max-age=0", - "Connection": "keep-alive", - "Sec-Fetch-Dest": "document", - "Sec-Fetch-Mode": "navigate", - "Sec-Fetch-Site": "none", - "Sec-Fetch-User": "?1", - "Upgrade-Insecure-Requests": "1", - "Sec-GPC": "1", -} - -# Session with retries -session = requests.Session() -retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) -session.mount('https://', HTTPAdapter(max_retries=retries)) - -last_sku_dict = {} -global_stock_status_dict = {} -first_run_dict = {name: True for name in PRODUCT_NAMES} - -# Discord notifications -def send_discord_notification(gpu_name: str, product_link: str, products_price: str): - - # Get current UNIX timestamp - timestamp_unix = int(time.time()) - - if TEST_MODE: - logging.info(f"[TEST MODE] Discord Notification: {gpu_name} available!") - return - - embed = { - "title": in_stock_title.format(gpu_name=gpu_name), - "color": 3066993, - "thumbnail": { - "url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" - }, - "author": { - "name": "Nvidia Founder Editions" - }, - - "fields": [ - { - "name": price_label, - "value": f"`{products_price}€`", - "inline": True - }, - - { - "name": time_label, - "value": f" ", - "inline": True - }, - ], - "description": buy_now.format(product_link=product_link), - "footer": { - "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), - "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" - } - } - - payload = { - "content": f"{DISCORD_ROLE_MAP.get(gpu_name, '@everyone')}", - "username": "NviBot", - "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", - "embeds": [embed] - } - - try: - response = requests.post(DISCORD_WEBHOOK_URL, json=payload) - if response.status_code == 204: - logging.info("✅ Notification sent to Discord.") - else: - logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") - except Exception as e: - logging.error(f"🚨 Error sending webhook: {e}") - -def send_out_of_stock_notification(gpu_name: str, product_link: str, products_price: str): - - # Get current UNIX timestamp - timestamp_unix = int(time.time()) - - if TEST_MODE: - logging.info(f"[TEST MODE] Discord Notification: {gpu_name} out of stock!") - return - - embed = { - "title": out_of_stock_title.format(gpu_name=gpu_name), - "color": 15158332, # Red for out of stock - "thumbnail": { - "url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" - }, - "url": f"{product_link}", - "author": { - "name": "Nvidia Founder Editions" - }, - - "footer": { - "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), - "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" - }, - - "fields": [ - { - "name": time_label, - "value": f" ", - "inline": True - } - ] - } - payload = {"username": "NviBot", "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", "embeds": [embed]} - try: - response = requests.post(DISCORD_WEBHOOK_URL, json=payload) - if response.status_code == 204: - logging.info("✅ 'Out of stock' notification sent to Discord.") - else: - logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") - except Exception as e: - logging.error(f"🚨 Error sending webhook: {e}") - -def send_sku_change_notification(gpu_name: str, old_sku: str, new_sku: str, product_link: str): - - # Get current UNIX timestamp - timestamp_unix = int(time.time()) - - if TEST_MODE: - logging.info(f"[TEST MODE] SKU change detected: {old_sku} → {new_sku}") - return - - embed = { - "title": sku_change_title.format(gpu_name=gpu_name), - "url": f"{product_link}", - "description": sku_description.format(old_sku=old_sku, new_sku=new_sku), - "color": 16776960, # Yellow - - "footer": { - "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), - "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" - }, - - "fields": [ - { - "name": time_label, - "value": f" ", - "inline": True - } - ] - } - - payload = { - "content": imminent_drop.format(DISCORD_ROLE=DISCORD_ROLE_MAP.get(gpu_name, '@everyone')), - "username": "NviBot", - "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", - "embeds": [embed] - } - - try: - response = requests.post(DISCORD_WEBHOOK_URL, json=payload) - if response.status_code == 204: - logging.info("✅ SKU change notification sent to Discord.") - else: - logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") - except Exception as e: - logging.error(f"🚨 Error sending webhook: {e}") - -# Stock search -def check_rtx_50_founders(): - global last_sku_dict, global_stock_status_dict, first_run_dict - - try: - response = session.get(API_URL_SKU, headers=HEADERS, timeout=10) - logging.info(f"SKU API response: {response.status_code}") - response.raise_for_status() - data = response.json() - except requests.exceptions.RequestException as e: - logging.error(f"SKU API error: {e}") - return - - # All available products - all_products = data['searchedProducts']['productDetails'] - - for product_name in PRODUCT_NAMES: - product_details = None - for p in all_products: - if p.get("gpu", "").strip() == product_name: - product_details = p - break - - if not product_details: - logging.warning(f"⚠️ No product with GPU '{product_name}' found.") - continue - - product_sku = product_details['productSKU'] - product_upc = product_details.get('productUPC', "") - if not isinstance(product_upc, list): - product_upc = [product_upc] - - # Check SKU change - old_sku = last_sku_dict.get(product_name) - if old_sku and old_sku != product_sku and not first_run_dict[product_name]: - logging.warning(f"⚠️ SKU changed for {product_name}: {old_sku} → {product_sku}") - send_sku_change_notification(product_name, old_sku, product_sku, PRODUCT_URL) - - last_sku_dict[product_name] = product_sku - first_run_dict[product_name] = False - - # Stock check - api_stock_url = API_URL_STOCK + product_sku - logging.info(f"[{product_name}] Checking stock: {api_stock_url}") - - try: - response = session.get(api_stock_url, headers=HEADERS, timeout=10) - logging.info(f"[{product_name}] Stock API response: {response.status_code}") - response.raise_for_status() - stock_data = response.json() - except requests.exceptions.RequestException as e: - logging.error(f"Stock API error: {e}") - continue - - products = stock_data.get("listMap", []) - products_price = "Price not available" - if isinstance(products, list) and len(products) > 0: - for p in products: - price = p.get("price", 'Price not available') - if price != 'Price not available': - products_price = price - break - else: - logging.error(f"[{product_name}] Product list is empty or malformed.") - - found_in_stock = set() - for p in products: - gpu_name = p.get("fe_sku", "").upper() - is_active = p.get("is_active") == "true" - if is_active and any(upc.upper() in gpu_name for upc in product_upc): - found_in_stock.add(gpu_name) - - for upc in product_upc: - upc_upper = upc.upper() - currently_in_stock = upc_upper in found_in_stock - previously_in_stock = global_stock_status_dict.get((product_name, upc_upper), False) - - if currently_in_stock and not previously_in_stock: - send_discord_notification(product_name, PRODUCT_URL, products_price) - global_stock_status_dict[(product_name, upc_upper)] = True - logging.info(f"[{product_name}] {upc} is now in stock!") - elif not currently_in_stock and previously_in_stock: - send_out_of_stock_notification(product_name, PRODUCT_URL, products_price) - global_stock_status_dict[(product_name, upc_upper)] = False - logging.info(f"[{product_name}] {upc} is now out of stock.") - elif currently_in_stock: - logging.info(f"[{product_name}] {upc} still in stock.") - else: - logging.info(f"[{product_name}] {upc} still out of stock.") -# Loop -if __name__ == "__main__": - try: - while True: - start = time.time() - check_rtx_50_founders() - elapsed = time.time() - start - time.sleep(max(0, REFRESH_TIME - elapsed)) - - # Gracefully shut down - except KeyboardInterrupt: - logging.info("🛑 Script interrupted by user. Exiting gracefully.") - exit(0) \ No newline at end of file