Djeex/nvidia-stock-bot
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Activity

Fully translated

Browse Source
This commit is contained in:
Djeex
2025-06-03 12:58:34 +00:00
parent 8124ad6a2d
commit e2d86cb787
5 changed files with 91 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -31,8 +31,6 @@
- Selectable check frequency - Selectable check frequency
- Selectable GPU model - Selectable GPU model
<img src="https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/nvbot_schematics.png" align="center">
## 🐳 Docker Installation without the repo (quick) ## 🐳 Docker Installation without the repo (quick)
Below are the instructions to set up the container using our pre-built image. With this setup, your bot will run independently as long as the container is active. Below are the instructions to set up the container using our pre-built image. With this setup, your bot will run independently as long as the container is active.

BIN
assets/img/nvbot_schematics.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 499 KiB

12
docker/.env
View File

@ -1,6 +1,6 @@
DS_HOOK= # Votre url du webhook Discord DS_HOOK= # Your Discord webhook URL
FREQ= # Frequence de rafraichissement en secondes, par défaut 30 FREQ= # Refresh frequency in seconds, default is 30
API_URL_SKU= # API listant le produit, par défaut https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia API_URL_SKU= # API listing the product, default is https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia
API_URL_STOCK= # API appelant le stock sans préciser la valeur du sku, par défaut https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus= API_URL_STOCK= # API used to check stock without specifying the SKU, default is https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus=
PRODUCT_URL= # URL d'achat du GPU PRODUCT_URL= # Purchase URL of the GPU
PRODUCT_NAME= # Le nom exact du GPU que vous recherchez comme : "RTX 5080" PRODUCT_NAME= # The exact name of the GPU you're looking for, e.g., "RTX 5080"

4
docker/compose.yaml
View File

@ -12,5 +12,5 @@ services:
- API_URL_STOCK=${API_URL_STOCK} - API_URL_STOCK=${API_URL_STOCK}
- PRODUCT_URL=${PRODUCT_URL} - PRODUCT_URL=${PRODUCT_URL}
- PRODUCT_NAME=${PRODUCT_NAME} - PRODUCT_NAME=${PRODUCT_NAME}
- PYTHONUNBUFFERED=1 # Permet d'afficher les logs en temps réel - PYTHONUNBUFFERED=1 # Allow to display log in real-time
command: python nvidia-stock-bot.py # Lance le script Python command: python nvidia-stock-bot.py # Run the script

171
nvidia-stock-bot.py
View File

@ -5,14 +5,14 @@ import os
import re import re
from requests.adapters import HTTPAdapter, Retry from requests.adapters import HTTPAdapter, Retry
# Configuration du logger # Logger configuration
logging.basicConfig( logging.basicConfig(
level=logging.INFO, level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s", format="%(asctime)s [%(levelname)s] %(message)s",
) )
logging.info("Démarrage du script") logging.info("Script started")
# Récupération des variables d'environnement # Retrieve environment variables
try: try:
DISCORD_WEBHOOK_URL = os.environ.get('DISCORD_WEBHOOK_URL') DISCORD_WEBHOOK_URL = os.environ.get('DISCORD_WEBHOOK_URL')
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_SKU = os.environ.get('API_URL_SKU', 'https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia')
@ -22,49 +22,48 @@ try:
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') 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')
PRODUCT_NAME = os.environ.get('PRODUCT_NAME') PRODUCT_NAME = os.environ.get('PRODUCT_NAME')
# Logging des erreurs # Error logging
if not DISCORD_WEBHOOK_URL: if not DISCORD_WEBHOOK_URL:
logging.error("❌ DISCORD_WEBHOOK_URL est requis mais non défini.") logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.")
exit(1) exit(1)
if not PRODUCT_NAME: if not PRODUCT_NAME:
logging.error("❌ PRODUCT_NAME est requis mais non défini.") logging.error("❌ PRODUCT_NAME is required but not defined.")
exit(1) exit(1)
# Regex pour extraire l'ID et le token # Regex to extract ID and token
match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL) match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL)
if match: if match:
webhook_id = match.group(1) webhook_id = match.group(1)
webhook_token = match.group(2) webhook_token = match.group(2)
# Masquer derniers caractères de l'ID # Mask last characters of the ID
masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10 masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10
# Masquer derniers caractères du token # Mask last characters of the token
masked_webhook_token = webhook_token[:len(webhook_token) - 120] + '*' * 10 masked_webhook_token = webhook_token[:len(webhook_token) - 120] + '*' * 10
# Reconstruction de l'url masquée # Rebuild masked URL
wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}" wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}"
# Logging des erreurs # Error logging
except KeyError as e: except KeyError as e:
logging.error(f"Variable d'environnement manquante : {e}") logging.error(f"Missing environment variable: {e}")
exit(1) exit(1)
except ValueError: except ValueError:
logging.error("REFRESH_TIME doit être un entier valide.") logging.error("REFRESH_TIME must be a valid integer.")
exit(1) exit(1)
# Affichage des URLs et configurations # Display URLs and configurations
logging.info(f"GPU: {PRODUCT_NAME}") logging.info(f"GPU: {PRODUCT_NAME}")
logging.info(f"URL Webhook Discord: {wh_masked_url}") logging.info(f"Discord Webhook URL: {wh_masked_url}")
logging.info(f"URL API SKU: {API_URL_SKU}") logging.info(f"API URL SKU: {API_URL_SKU}")
logging.info(f"URL API Stock: {API_URL_STOCK}") logging.info(f"API URL Stock: {API_URL_STOCK}")
logging.info(f"URL produit: {PRODUCT_URL}") logging.info(f"Product URL: {PRODUCT_URL}")
logging.info(f"Temps d'actualisation: {REFRESH_TIME} secondes") logging.info(f"Refresh time: {REFRESH_TIME} seconds")
logging.info(f"Mode Test: {TEST_MODE}") logging.info(f"Test Mode: {TEST_MODE}")
# HTTP headers
# Entêtes HTTP
HEADERS = { HEADERS = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) " "AppleWebKit/537.36 (KHTML, like Gecko) "
@ -83,30 +82,30 @@ HEADERS = {
"Sec-GPC": "1", "Sec-GPC": "1",
} }
# Session avec retries # Session with retries
session = requests.Session() session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries)) session.mount('https://', HTTPAdapter(max_retries=retries))
# Stockage de l'état des stocks # Store stock status
global_stock_status = {} global_stock_status = {}
# Stocke le dernier SKU connu # Store the last known SKU
last_sku = None last_sku = None
first_run = True # Before calling check_rtx_50_founders first_run = True # Before calling check_rtx_50_founders
# Notifications Discord # Discord notifications
def send_discord_notification(gpu_name: str, product_link: str, products_price: str): def send_discord_notification(gpu_name: str, product_link: str, products_price: str):
# Récupérer le timestamp UNIX actuel # Get current UNIX timestamp
timestamp_unix = int(time.time()) timestamp_unix = int(time.time())
if TEST_MODE: if TEST_MODE:
logging.info(f"[TEST MODE] Notification Discord: {gpu_name} disponible !") logging.info(f"[TEST MODE] Discord Notification: {gpu_name} available!")
return return
embed = { embed = {
"title": f"🚀 {PRODUCT_NAME} EN STOCK !", "title": f"🚀 {PRODUCT_NAME} IN STOCK!",
"color": 3066993, "color": 3066993,
"thumbnail": { "thumbnail": {
"url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" "url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg"
@ -117,18 +116,18 @@ def send_discord_notification(gpu_name: str, product_link: str, products_price:
"fields": [ "fields": [
{ {
"name": "Prix", "name": "Price",
"value": f"`{products_price}€`", "value": f"`{products_price}€`",
"inline": True "inline": True
}, },
{ {
"name": "Heure", "name": "Time",
"value": f"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>", "value": f"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>",
"inline": True "inline": True
}, },
], ],
"description": f"**:point_right: [Acheter maintenant]({product_link})**", "description": f"**:point_right: [Buy now]({product_link})**",
"footer": { "footer": {
"text": "NviBot • JV Hardware 2.0", "text": "NviBot • JV Hardware 2.0",
"icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" "icon_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg"
@ -138,24 +137,24 @@ def send_discord_notification(gpu_name: str, product_link: str, products_price:
try: try:
response = requests.post(DISCORD_WEBHOOK_URL, json=payload) response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
if response.status_code == 204: if response.status_code == 204:
logging.info("✅ Notification envoyée sur Discord.") logging.info("✅ Notification sent to Discord.")
else: else:
logging.error(f"❌ Erreur Webhook : {response.status_code} - {response.text}") logging.error(f"❌ Webhook error: {response.status_code} - {response.text}")
except Exception as e: except Exception as e:
logging.error(f"🚨 Erreur lors de l'envoi du webhook : {e}") logging.error(f"🚨 Error sending webhook: {e}")
def send_out_of_stock_notification(gpu_name: str, product_link: str, products_price: str): def send_out_of_stock_notification(gpu_name: str, product_link: str, products_price: str):
# Récupérer le timestamp UNIX actuel # Get current UNIX timestamp
timestamp_unix = int(time.time()) timestamp_unix = int(time.time())
if TEST_MODE: if TEST_MODE:
logging.info(f"[TEST MODE] Notification Discord: {gpu_name} hors stock !") logging.info(f"[TEST MODE] Discord Notification: {gpu_name} out of stock!")
return return
embed = { embed = {
"title": f"❌ {PRODUCT_NAME} n'est plus en stock", "title": f"❌ {PRODUCT_NAME} is out of stock",
"color": 15158332, # Rouge pour hors stock "color": 15158332, # Red for out of stock
"thumbnail": { "thumbnail": {
"url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" "url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg"
}, },
@ -171,7 +170,7 @@ def send_out_of_stock_notification(gpu_name: str, product_link: str, products_pr
"fields": [ "fields": [
{ {
"name": "Heure", "name": "Time",
"value": f"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>", "value": f"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>",
"inline": True "inline": True
} }
@ -181,26 +180,26 @@ def send_out_of_stock_notification(gpu_name: str, product_link: str, products_pr
try: try:
response = requests.post(DISCORD_WEBHOOK_URL, json=payload) response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
if response.status_code == 204: if response.status_code == 204:
logging.info("✅ Notification 'hors stock' envoyée sur Discord.") logging.info("✅ 'Out of stock' notification sent to Discord.")
else: else:
logging.error(f"❌ Erreur Webhook : {response.status_code} - {response.text}") logging.error(f"❌ Webhook error: {response.status_code} - {response.text}")
except Exception as e: except Exception as e:
logging.error(f"🚨 Erreur lors de l'envoi du webhook : {e}") logging.error(f"🚨 Error sending webhook: {e}")
def send_sku_change_notification(old_sku: str, new_sku: str, product_link: str): def send_sku_change_notification(old_sku: str, new_sku: str, product_link: str):
# Récupérer le timestamp UNIX actuel # Get current UNIX timestamp
timestamp_unix = int(time.time()) timestamp_unix = int(time.time())
if TEST_MODE: if TEST_MODE:
logging.info(f"[TEST MODE] Changement de SKU détecté : {old_sku} → {new_sku}") logging.info(f"[TEST MODE] SKU change detected: {old_sku} → {new_sku}")
return return
embed = { embed = {
"title": f"🔄 {PRODUCT_NAME} Changement de SKU détecté", "title": f"🔄 {PRODUCT_NAME} SKU change detected",
"url": f"{product_link}", "url": f"{product_link}",
"description": f"**Ancien SKU** : `{old_sku}`\n**Nouveau SKU** : `{new_sku}`", "description": f"**Old SKU** : `{old_sku}`\n**New SKU** : `{new_sku}`",
"color": 16776960, # Jaune "color": 16776960, # Yellow
"footer": { "footer": {
"text": "NviBot • JV Hardware 2.0", "text": "NviBot • JV Hardware 2.0",
@ -209,7 +208,7 @@ def send_sku_change_notification(old_sku: str, new_sku: str, product_link: str):
"fields": [ "fields": [
{ {
"name": "Heure", "name": "Time",
"value": f"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>", "value": f"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>",
"inline": True "inline": True
} }
@ -217,7 +216,7 @@ def send_sku_change_notification(old_sku: str, new_sku: str, product_link: str):
} }
payload = { payload = {
"content": "@everyone ⚠️ Potentiel drop imminent !", "content": "@everyone ⚠️ Possible imminent drop!",
"username": "NviBot", "username": "NviBot",
"avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg", "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg",
"embeds": [embed] "embeds": [embed]
@ -226,90 +225,88 @@ def send_sku_change_notification(old_sku: str, new_sku: str, product_link: str):
try: try:
response = requests.post(DISCORD_WEBHOOK_URL, json=payload) response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
if response.status_code == 204: if response.status_code == 204:
logging.info("✅ Notification de changement de SKU envoyée sur Discord.") logging.info("✅ SKU change notification sent to Discord.")
else: else:
logging.error(f"❌ Erreur Webhook : {response.status_code} - {response.text}") logging.error(f"❌ Webhook error: {response.status_code} - {response.text}")
except Exception as e: except Exception as e:
logging.error(f"🚨 Erreur lors de l'envoi du webhook : {e}") logging.error(f"🚨 Error sending webhook: {e}")
# Recherche du stock # Stock search
def check_rtx_50_founders(): def check_rtx_50_founders():
global global_stock_status, last_sku, first_run global global_stock_status, last_sku, first_run
# Appel vers l'API produit pour récupérer le sku et l'upc # Call product API to retrieve SKU and UPC
try: try:
response = session.get(API_URL_SKU, headers=HEADERS, timeout=10) response = session.get(API_URL_SKU, headers=HEADERS, timeout=10)
logging.info(f"Réponse de l'API SKU : {response.status_code}") logging.info(f"SKU API response: {response.status_code}")
response.raise_for_status() response.raise_for_status()
data = response.json() data = response.json()
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
logging.error(f"Erreur API SKU : {e}") logging.error(f"SKU API error: {e}")
return return
# Recherche du produit dont le GPU correspond à PRODUCT_NAME # Look for product whose GPU matches PRODUCT_NAME
product_details = None product_details = None
for p in data['searchedProducts']['productDetails']: for p in data['searchedProducts']['productDetails']:
gpu_name = p.get("gpu", "").strip() gpu_name = p.get("gpu", "").strip()
# Si le GPU correspond exactement à PRODUCT_NAME # If the GPU matches exactly PRODUCT_NAME
if gpu_name == PRODUCT_NAME.strip(): if gpu_name == PRODUCT_NAME.strip():
product_details = p product_details = p
break # Sortir dès qu'on trouve le bon produit break # Exit as soon as the correct product is found
if not product_details: if not product_details:
logging.warning(f"⚠️ Aucun produit avec le GPU '{PRODUCT_NAME}' trouvé.") logging.warning(f"⚠️ No product with GPU '{PRODUCT_NAME}' found.")
return return
# Récupérer le SKU pour le GPU trouvé # Retrieve the SKU for the found GPU
product_sku = product_details['productSKU'] product_sku = product_details['productSKU']
product_upc = product_details.get('productUPC', "") product_upc = product_details.get('productUPC', "")
# Check if this is the first execution
# Vérifier si c'est la première exécution
if last_sku is not None and product_sku != last_sku: if last_sku is not None and product_sku != last_sku:
if not first_run: # Évite d'envoyer une notification au premier appel if not first_run: # Prevent sending notification on first run
product_link = PRODUCT_URL product_link = PRODUCT_URL
logging.warning(f"⚠️ SKU modifié : {last_sku} → {product_sku}") logging.warning(f"⚠️ SKU changed: {last_sku} → {product_sku}")
send_sku_change_notification(last_sku, product_sku, product_link) send_sku_change_notification(last_sku, product_sku, product_link)
# Mettre à jour le SKU stocké # Update stored SKU
last_sku = product_sku last_sku = product_sku
first_run = False # Désactive la protection après la première exécution first_run = False # Disable first-run protection
if not isinstance(product_upc, list): if not isinstance(product_upc, list):
product_upc = [product_upc] product_upc = [product_upc]
# Construction de l'url de l'API de stock et appel pour vérifier le statut # Build stock API URL and call to check status
API_URL = API_URL_STOCK + product_sku API_URL = API_URL_STOCK + product_sku
logging.info(f"URL de l'API de stock appelée : {API_URL}") logging.info(f"Stock API URL called: {API_URL}")
try: try:
response = session.get(API_URL, headers=HEADERS, timeout=10) response = session.get(API_URL, headers=HEADERS, timeout=10)
logging.info(f"Réponse de l'API : {response.status_code}") logging.info(f"Stock API response: {response.status_code}")
response.raise_for_status() response.raise_for_status()
data = response.json() data = response.json()
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
logging.error(f"Erreur API Stock : {e}") logging.error(f"Stock API error: {e}")
return return
products = data.get("listMap", []) products = data.get("listMap", [])
products_price = 'Prix non disponible' # Valeur par défaut products_price = 'Price not available' # Default value
# Vérification de la liste des produits et récupération du prix # Check product list and retrieve price
if isinstance(products, list) and len(products) > 0: if isinstance(products, list) and len(products) > 0:
for product in products: for product in products:
price = product.get("price", 'Prix non disponible') price = product.get("price", 'Price not available')
if price != 'Prix non disponible': if price != 'Price not available':
products_price = price # Utiliser le prix trouvé products_price = price # Use found price
break # Sortir dès qu'on trouve un prix break # Stop at first found price
else: else:
logging.error("La liste des produits est vide ou mal formée.") logging.error("Product list is empty or malformed.")
found_in_stock = set() found_in_stock = set()
# Recherche du statut et notifications selon le statut # Check status and send notifications accordingly
for p in products: for p in products:
gpu_name = p.get("fe_sku", "").upper() gpu_name = p.get("fe_sku", "").upper()
is_active = p.get("is_active") == "true" is_active = p.get("is_active") == "true"
@ -325,20 +322,18 @@ def check_rtx_50_founders():
product_link = PRODUCT_URL product_link = PRODUCT_URL
send_discord_notification(gpu_upper, product_link, products_price) send_discord_notification(gpu_upper, product_link, products_price)
global_stock_status[gpu_upper] = True global_stock_status[gpu_upper] = True
logging.info(f"{gpu} est maintenant en stock!") logging.info(f"{gpu} is now in stock!")
elif not currently_in_stock and previously_in_stock: elif not currently_in_stock and previously_in_stock:
product_link = PRODUCT_URL product_link = PRODUCT_URL
send_out_of_stock_notification(gpu_upper, product_link, products_price) send_out_of_stock_notification(gpu_upper, product_link, products_price)
global_stock_status[gpu_upper] = False global_stock_status[gpu_upper] = False
logging.info(f"{gpu} n'est plus en stock.") logging.info(f"{gpu} is no longer in stock.")
elif currently_in_stock and previously_in_stock: elif currently_in_stock and previously_in_stock:
logging.info(f"{gpu} est actuellement en stock.") logging.info(f"{gpu} is currently in stock.")
else: else:
logging.info(f"{gpu} est actuellement hors stock.") logging.info(f"{gpu} is currently out of stock.")
# Boucle # Loop
if __name__ == "__main__": if __name__ == "__main__":
while True: while True:
check_rtx_50_founders() check_rtx_50_founders()