Modules

Dockerfile

@@ -4,15 +4,11 @@ RUN apk add --no-cache ca-certificates
 
 WORKDIR /app
 
-COPY app/nvidia-stock-bot.py /app/
+COPY /app/ /app/
-
-COPY app/requirements.txt /app/
-
-COPY app/localization.json /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"]
+CMD ["python", "main.py"]

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))

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 {}

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"]

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()

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"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>",
+                "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"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>",
+                "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"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>",
+                "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}")

docker/compose.yaml

@@ -13,4 +13,3 @@ services:
       - 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