Compare commits

..

No commits in common. "037f5bebf88aef19453a9a06355703eeec595003" and "f9bc3dfe3217fdb1e48f8024a8e149a031728aaa" have entirely different histories.

7 changed files with 73 additions and 264 deletions

View File

@ -1,19 +1,17 @@
# Nvidia Stock Bot - WIP # Nvidia Stock Bot - WIP -
Par KevOut & Djeex Par KevOut & Djeex
[![](https://img.shields.io/badge/JV%20hardware-rejoindre-green?style=flat-square&logo=discord&logoColor=%23fff&label=JV%20hardware&link=https%3A%2F%2Fdiscord.gg%2Fgxffg3GA96)](https://discord.gg/gxffg3GA96) [![](https://img.shields.io/badge/JV%20hardware-rejoindre-green?style=flat-square&logo=discord&logoColor=%23fff&label=JV%20hardware&link=https%3A%2F%2Fdiscord.gg%2Fgxffg3GA96)](https://discord.gg/gxffg3GA96)
Ce robot : Ce robot :
- Appelle l'API de Nvidia listant le produit (par défaut toutes les 60s) - Appelle régulièrement l'api des stocks français de nvidia FE (par défaut toutes les 60s)
- Récupère le SKU du produit concerné - Vérifie si RTX 5090, RTX 5080, RTX 5070ti et RTX 5070 sont en stock
- Appelle le stock lié à ce SKU
- Si du stock est trouvé, envoie une notification discord via le webhook paramétré - Si du stock est trouvé, envoie une notification discord via le webhook paramétré
- Si le produit était déjà en stock, il n'envoie plus de notification
- Si le produit était en stock mais ne l'est plus, envoie une notification discord signifiant la fin du stock
<img src="https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/nvbot.png" align="center"> <img src="https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/nvbot.png" align="center">
Trois modes d'installation : Trois modes d'installation :
- [Avec le dépot Git et Docker](https://git.djeex.fr/Djeex/nvidia-stock-bot/#installation-avec-le-d%C3%A9pot) - [Avec le dépot Git et Docker](https://git.djeex.fr/Djeex/nvidia-stock-bot/#installation-avec-le-d%C3%A9pot)
- [Sans le dépot Git et avec notre image docker fournie](https://git.djeex.fr/Djeex/nvidia-stock-bot/#installation-sans-le-d%C3%A9pot-avec-docker-compose) - [Sans le dépot Git et avec notre image docker fournie](https://git.djeex.fr/Djeex/nvidia-stock-bot/#installation-sans-le-d%C3%A9pot-avec-docker-compose)
@ -75,11 +73,8 @@ services:
environment: environment:
- DISCORD_WEBHOOK_URL= # URL de votre webhook Discord - DISCORD_WEBHOOK_URL= # URL de votre webhook Discord
- REFRESH_TIME= # Durée de rafraichissement du script en secondes - REFRESH_TIME= # Durée de rafraichissement du script en secondes
- API_URL_SKU= # API listant le produit par exemple https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia&gpu=RTX%205090 - GPU_TARGETS= #SKU
- API_URL_STOCK= # API appelant le stock sans préciser la valeur du sku, par exemple https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus= - API_URL= #URL de l'API
- PRODUCT_URL= # URL d'achat du GPU
- PRODUCT_NAME= #Le nom du GPU qui s'affiche dans les notifications
- TEST_MODE= #true pour tester les notifications discord. false par défaut.
- PYTHONUNBUFFERED=1 # Permet d'afficher les logs en temps réel - PYTHONUNBUFFERED=1 # Permet d'afficher les logs en temps réel
command: python nvidia-stock-bot.py # Lance le script Python au démarrage du conteneur command: python nvidia-stock-bot.py # Lance le script Python au démarrage du conteneur
``` ```
@ -107,9 +102,6 @@ Vous trouverez ci-dessous comment exécuter directement le script Python. Avec c
```sh ```sh
export DISCORD_WEBHOOK_URL="https://votre_url_discord" export DISCORD_WEBHOOK_URL="https://votre_url_discord"
export REFRESH_TIME="60" export REFRESH_TIME="60"
export API_URL_SKU="https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia&gpu=RTX%205090"
export API_URL_STOCK="https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus="
export TEST_MODE=false
``` ```
- Lancez le script - Lancez le script

Binary file not shown.

Before

Width:  |  Height:  |  Size: 19 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 18 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 4.3 MiB

After

Width:  |  Height:  |  Size: 322 KiB

View File

@ -1,6 +1,4 @@
DS_HOOK= #votre url du webhook Discord DS_HOOK="votre url du webhook discord"
FREQ= #frequence de rafraichissement en secondes FREQ=60 #frequence de rafraichissement en secondes
API_URL_SKU= # API listant le produit par exemple https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale=fr-fr&Manufacturer=Nvidia&gpu=RTX%205090 GPU=
API_URL_STOCK= # API appelant le stock sans préciser la valeur du sku, par exemple https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus= URL=""
PRODUCT_URL= # URL d'achat du GPU
PRODUCT_NAME= #Le nom du GPU qui s'affiche dans les notifications

View File

@ -9,7 +9,7 @@ services:
environment: environment:
- DISCORD_WEBHOOK_URL=${DS_HOOK} - DISCORD_WEBHOOK_URL=${DS_HOOK}
- REFRESH_TIME=${FREQ} - REFRESH_TIME=${FREQ}
- API_URL_SKU=${API_URL_SKU} - GPU_TARGETS=${GPU} #SKU
- API_URL_STOCK=${API_URL_STOCK} - API_URL=${URL} #URL de l'API
- PYTHONUNBUFFERED=1 # Permet d'afficher les logs en temps réel - PYTHONUNBUFFERED=1 # Permet d'afficher les logs en temps réel
command: python nvidia-stock-bot.py # Lance le script Python command: python nvidia-stock-bot.py # Lance le script Python

View File

@ -2,59 +2,42 @@ import requests
import logging import logging
import time import time
import os import os
import re
from requests.adapters import HTTPAdapter, Retry
# Configuration du logger # Configuration du logger
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("Démarrage du script")
# Récupération des variables d'environnement # Récupération des variables d'environnement
try: try:
DISCORD_WEBHOOK_URL = os.environ['DISCORD_WEBHOOK_URL'] DISCORD_WEBHOOK_URL = os.environ['DISCORD_WEBHOOK_URL']
API_URL_SKU = os.environ['API_URL_SKU'] API_URL = os.environ['API_URL']
API_URL_STOCK = os.environ['API_URL_STOCK'] GPU_TARGETS = os.environ['GPU_TARGETS'].split(",") # Séparer en liste
GPU_TARGETS = [gpu.strip() for gpu in GPU_TARGETS] # Nettoyer les espaces
REFRESH_TIME = int(os.environ['REFRESH_TIME']) # Convertir en entier REFRESH_TIME = int(os.environ['REFRESH_TIME']) # Convertir en entier
TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true'
PRODUCT_URL = os.environ['PRODUCT_URL']
PRODUCT_NAME = os.environ['PRODUCT_NAME']
# Regex pour extraire l'ID et le token
match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL)
if match:
webhook_id = match.group(1)
webhook_token = match.group(2)
# Masquer derniers caractères de l'ID
masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10
# Masquer derniers caractères du token
masked_webhook_token = webhook_token[:len(webhook_token) - 120] + '*' * 10
# Reconstruction de l'url masquée
wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}"
except KeyError as e: except KeyError as e:
logging.error(f"Variable d'environnement manquante : {e}") logging.error(f"Variable d'environnement manquante : {e}")
exit(1) exit(1) # Quitter le script proprement en cas d'erreur
except ValueError: except ValueError:
logging.error("REFRESH_TIME doit être un entier valide.") logging.error("REFRESH_TIME doit être un entier valide.")
exit(1) exit(1)
# Affichage des URLs et configurations # Afficher les valeurs des variables d'environnement
logging.info(f"GPU: {PRODUCT_NAME}") print(f"url du webhook Discord: {DISCORD_WEBHOOK_URL}")
logging.info(f"URL Webhook Discord: {wh_masked_url}") print(f"url de l'API: {API_URL}")
logging.info(f"URL API SKU: {API_URL_SKU}") print(f"GPU recherché: {GPU_TARGETS}")
logging.info(f"URL API Stock: {API_URL_STOCK}") print(f"Temps d'actualisation (en secondes) : {REFRESH_TIME}")
logging.info(f"URL produit: {PRODUCT_URL}")
logging.info(f"Temps d'actualisation: {REFRESH_TIME} secondes")
logging.info(f"Mode Test: {TEST_MODE}")
# LURL de lAPI (exemple)
#API_URL = "https://api.store.nvidia.com/partner/v1/feinventory?locale=fr-fr&skus=5090LDLCFE"
# Entêtes HTTP # GPUs à surveiller
#GPU_TARGETS = ["5090LDLCFE_FR"]
# Entêtes HTTP pour la requête
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) "
@ -73,248 +56,84 @@ HEADERS = {
"Sec-GPC": "1", "Sec-GPC": "1",
} }
# Session avec retries # Dictionnaire stockant l'état de stock
stock_status = {gpu.upper(): False for gpu in GPU_TARGETS}
session = requests.Session() session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
# Stockage de l'état des stocks
global_stock_status = {}
# Stocke le dernier SKU connu
last_sku = None
first_run = True # Before calling check_rtx_50_founders
# Notifications Discord
def send_discord_notification(gpu_name: str, product_link: str, products_price: str):
# Récupérer le timestamp UNIX actuel
timestamp_unix = int(time.time())
if TEST_MODE:
logging.info(f"[TEST MODE] Notification Discord: {gpu_name} disponible !")
return
def send_discord_notification(gpu_name: str, product_link: str):
"""Envoie une notification Discord avec un embed via un webhook."""
embed = { embed = {
"title": f"🚀 {PRODUCT_NAME} EN STOCK !", "title": f"🚀 {gpu_name} en stock !",
"color": 3066993, "description": f":point_right: **[Achetez ici](https://marketplace.nvidia.com/fr-fr/consumer/graphics-cards/?locale=fr-fr&page=1&limit=12&gpu=RTX%205090,RTX%205080&manufacturer=NVIDIA)**",
"thumbnail": { "color": 3066993, # Couleur verte
"url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()),
}, #"thumbnail": {
"author": { # "url": "https://www.nvidia.com/content/dam/en-zz/Solutions/geforce/graphic-cards/50-series/rtx-5090/geforce-rtx-5090-learn-more-og-1200x630.jpg"
"name": "Nvidia Founder Editions" #}
},
"fields": [
{
"name": "Prix",
"value": f"`{products_price} €`",
"inline": True
},
{
"name": "Heure",
"value": f"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>",
"inline": True
},
{
"name": "Lien",
"value": f"{PRODUCT_URL}"
}
],
"description": f"**:point_right: [Acheter maintenant]({product_link})**",
"url": f"{product_link}"
"footer": {
"text": "Par KevOut & Djeex"
}
}
payload = {"content": "@everyone", "username": "NviBot", "avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000_pp.jpg", "embeds": [embed]}
try:
response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
if response.status_code == 204:
logging.info("✅ Notification envoyée sur Discord.")
else:
logging.error(f"❌ Erreur Webhook : {response.status_code} - {response.text}")
except Exception as e:
logging.error(f"🚨 Erreur lors de l'envoi du webhook : {e}")
def send_out_of_stock_notification(gpu_name: str, product_link: str, products_price: str):
# Récupérer le timestamp UNIX actuel
timestamp_unix = int(time.time())
if TEST_MODE:
logging.info(f"[TEST MODE] Notification Discord: {gpu_name} hors stock !")
return
embed = {
"title": f"{PRODUCT_NAME} n'est plus en stock",
"color": 15158332, # Rouge pour hors 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": "Par KevOut & Djeex"
},
"fields": [
{
"name": "Heure",
"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/RTX5000_pp.jpg", "embeds": [embed]}
try:
response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
if response.status_code == 204:
logging.info("✅ Notification 'hors stock' envoyée sur Discord.")
else:
logging.error(f"❌ Erreur Webhook : {response.status_code} - {response.text}")
except Exception as e:
logging.error(f"🚨 Erreur lors de l'envoi du webhook : {e}")
def send_sku_change_notification(old_sku: str, new_sku: str):
# Récupérer le timestamp UNIX actuel
timestamp_unix = int(time.time())
if TEST_MODE:
logging.info(f"[TEST MODE] Changement de SKU détecté : {old_sku}{new_sku}")
return
embed = {
"title": f"🔄 {PRODUCT_NAME} Changement de SKU détecté",
"description": f"**Ancien SKU** : `{old_sku}`\n**Nouveau SKU** : `{new_sku}`",
"color": 16776960, # Jaune
"footer": {
"text": "Par KevOut & Djeex"
},
"fields": [
{
"name": "Heure",
"value": f"<t:{timestamp_unix}:d> <t:{timestamp_unix}:T>",
"inline": True
}
]
} }
payload = { payload = {
"content": "@everyone ⚠️ Potentiel drop imminent !", "content": "@everyone",
"username": "NviBot", "username": "Nvidia Bot",
"avatar_url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000_pp.jpg",
"embeds": [embed] "embeds": [embed]
} }
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("✅ Embed envoyé sur Discord.")
else: else:
logging.error(f"❌ Erreur Webhook : {response.status_code} - {response.text}") logging.error(f"❌ Erreur d'envoi du webhook : {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"🚨 Erreur lors de l'envoi du webhook : {e}")
# Recherche du stock
def check_rtx_50_founders(): def check_rtx_50_founders():
global global_stock_status, last_sku, first_run """Vérifie l'état de stock des GPU Founders Edition et notifie Discord si un GPU repasse en stock."""
# Appel vers l'API produit pour récupérer le sku et l'upc
try:
response = session.get(API_URL_SKU, headers=HEADERS, timeout=10)
logging.info(f"Réponse de l'API SKU : {response.status_code}")
response.raise_for_status()
data = response.json()
except requests.exceptions.RequestException as e:
logging.error(f"Erreur API SKU : {e}")
return
product_details = data['searchedProducts']['productDetails'][0]
product_sku = product_details['productSKU']
# Vérifier si c'est la première exécution
if last_sku is not None and product_sku != last_sku:
if not first_run: # Évite d'envoyer une notification au premier appel
logging.warning(f"⚠️ SKU modifié : {last_sku}{product_sku}")
send_sku_change_notification(last_sku, product_sku)
# Mettre à jour le SKU stocké
last_sku = product_sku
first_run = False # Désactive la protection après la première exécution
product_details = data['searchedProducts']['productDetails'][0]
product_sku = product_details['productSKU']
product_upc = product_details.get('productUPC', "")
if not isinstance(product_upc, list):
product_upc = [product_upc]
# Construction de l'url de l'API de stock et appel pour vérifier le statut
API_URL = API_URL_STOCK + product_sku
logging.info(f"URL de l'API de stock appelée : {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"Réponse de l'API : {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"Erreur lors de l'appel API : {e}")
return return
products = data.get("listMap", []) products = data.get("listMap", [])
products_price = 'Prix non disponible' # Valeur par défaut
# Vérification de la liste des produits et récupération du prix
if isinstance(products, list) and len(products) > 0:
for product in products:
price = product.get("price", 'Prix non disponible')
if price != 'Prix non disponible':
products_price = price # Utiliser le prix trouvé
break # Sortir dès qu'on trouve un prix
else:
logging.error("La liste des produits est vide ou mal formée.")
found_in_stock = set() found_in_stock = set()
# Recherche du statut et notifications selon le statut
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"
if is_active and any(target.upper() in gpu_name for target in product_upc):
if is_active:
if any(target.upper() in gpu_name for target in GPU_TARGETS):
found_in_stock.add(gpu_name) found_in_stock.add(gpu_name)
for gpu in product_upc: for gpu in GPU_TARGETS:
gpu_upper = gpu.upper() gpu_upper = gpu.upper()
currently_in_stock = gpu_upper in found_in_stock currently_in_stock = (gpu_upper in found_in_stock)
previously_in_stock = global_stock_status.get(gpu_upper, False) previously_in_stock = stock_status[gpu_upper]
if currently_in_stock and not previously_in_stock: if currently_in_stock and not previously_in_stock:
product_link = PRODUCT_URL for product in products:
send_discord_notification(gpu_upper, product_link, products_price) product_name = product.get("fe_sku", "").upper()
global_stock_status[gpu_upper] = True if product_name == gpu_upper:
logging.info(f"{gpu} est maintenant en stock!") real_gpu_name = product.get("fe_sku", "Inconnu")
elif not currently_in_stock and previously_in_stock: product_link = "https://marketplace.nvidia.com/fr-fr/consumer/graphics-cards/?locale=fr-fr&page=1&limit=12&gpu=RTX%205090,RTX%205080"
product_link = PRODUCT_URL send_discord_notification(real_gpu_name, product_link)
send_out_of_stock_notification(gpu_upper, product_link, products_price)
global_stock_status[gpu_upper] = False stock_status[gpu_upper] = True
print(f"{gpu} est maintenant en stock!")
elif (not currently_in_stock) and previously_in_stock:
logging.info(f"{gpu} n'est plus en stock.") logging.info(f"{gpu} n'est plus en stock.")
stock_status[gpu_upper] = False
print(f"{gpu} est hors stock !")
elif currently_in_stock and previously_in_stock: elif not currently_in_stock:
logging.info(f"{gpu} est actuellement en stock.") print(f"{gpu} est actuellement hors stock.")
else:
logging.info(f"{gpu} est actuellement hors stock.")
# Boucle
if __name__ == "__main__": if __name__ == "__main__":
while True: while True:
check_rtx_50_founders() check_rtx_50_founders()