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

Multi GPU merge #10

Merged
Djeex merged 10 commits from wip into main 2025-07-08 19:15:29 +02:00
Conversation 0 Commits 10 Files Changed 3 +114 -111
2 changed files with 114 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 6c4be66f3d - Show all commits

194
nvidia-stock-bot.py
View File

@ -21,22 +21,23 @@ try:
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')
PRODUCT_NAME = os.environ.get('PRODUCT_NAME')
PRODUCT_NAMES = os.environ.get('PRODUCT_NAMES')
# Errors logging
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(',')]
# Validate role format
if DISCORD_ROLE != '@everyone' and not re.match(r'^<@&\d{17,20}>$', DISCORD_ROLE):
logging.error("❌ DISCORD_ROLE format is invalid. Use '@everyone' or '<@&ROLE_ID>'.")
exit(1)
# Error logging
if not DISCORD_WEBHOOK_URL:
logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.")
exit(1)
if not PRODUCT_NAME:
logging.error("❌ PRODUCT_NAME is required but not defined.")
exit(1)
# Regex to extract ID and token
match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL)
if match:
@ -61,7 +62,7 @@ except ValueError:
exit(1)
# Display URLs and configurations
logging.info(f"GPU: {PRODUCT_NAME}")
logging.info(f"GPU: {PRODUCT_NAMES}")
logging.info(f"Discord Webhook URL: {wh_masked_url}")
logging.info(f"Discord Role Mention: {DISCORD_ROLE}")
logging.info(f"API URL SKU: {API_URL_SKU}")
@ -94,12 +95,9 @@ session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
# Store stock status
global_stock_status = {}
# Store the last known SKU
last_sku = None
first_run = True # Before calling check_rtx_50_founders
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):
@ -112,7 +110,7 @@ def send_discord_notification(gpu_name: str, product_link: str, products_price:
return
embed = {
"title": f"🚀 {PRODUCT_NAME} IN STOCK!",
"title": f"🚀 {gpu_name} IN STOCK!",
"color": 3066993,
"thumbnail": {
"url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg"
@ -160,7 +158,7 @@ def send_out_of_stock_notification(gpu_name: str, product_link: str, products_pr
return
embed = {
"title": f"❌ {PRODUCT_NAME} is out of stock",
"title": f"❌ {gpu_name} is out of stock",
"color": 15158332, # Red for out of stock
"thumbnail": {
"url": "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg"
@ -203,7 +201,7 @@ def send_sku_change_notification(old_sku: str, new_sku: str, product_link: str):
return
embed = {
"title": f"🔄 {PRODUCT_NAME} SKU change detected",
"title": f"🔄 {gpu_name} SKU change detected",
"url": f"{product_link}",
"description": f"**Old SKU** : `{old_sku}`\n**New SKU** : `{new_sku}`",
"color": 16776960, # Yellow
@ -240,9 +238,8 @@ def send_sku_change_notification(old_sku: str, new_sku: str, product_link: str):
# Stock search
def check_rtx_50_founders():
global global_stock_status, last_sku, first_run
global last_sku_dict, global_stock_status_dict, first_run_dict
# Call product API to retrieve SKU and UPC
try:
response = session.get(API_URL_SKU, headers=HEADERS, timeout=10)
logging.info(f"SKU API response: {response.status_code}")
@ -252,101 +249,90 @@ def check_rtx_50_founders():
logging.error(f"SKU API error: {e}")
return
# Look for product whose GPU matches PRODUCT_NAME
product_details = None
# All available products
all_products = data['searchedProducts']['productDetails']
for p in data['searchedProducts']['productDetails']:
gpu_name = p.get("gpu", "").strip()
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 the GPU matches exactly PRODUCT_NAME
if gpu_name == PRODUCT_NAME.strip():
product_details = p
break # Exit as soon as the correct product is found
if not product_details:
logging.warning(f"⚠️ No product with GPU '{product_name}' found.")
continue
if not product_details:
logging.warning(f"⚠️ No product with GPU '{PRODUCT_NAME}' found.")
return
product_sku = product_details['productSKU']
product_upc = product_details.get('productUPC', "")
if not isinstance(product_upc, list):
product_upc = [product_upc]
# Retrieve the SKU for the found GPU
product_sku = product_details['productSKU']
product_upc = product_details.get('productUPC', "")
# 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(old_sku, product_sku, PRODUCT_URL)
# Check if this is the first execution
if last_sku is not None and product_sku != last_sku:
if not first_run: # Prevent sending notification on first run
product_link = PRODUCT_URL
logging.warning(f"⚠️ SKU changed: {last_sku} → {product_sku}")
send_sku_change_notification(last_sku, product_sku, product_link)
last_sku_dict[product_name] = product_sku
first_run_dict[product_name] = False
# Update stored SKU
last_sku = product_sku
first_run = False # Disable first-run protection
# Stock check
api_stock_url = API_URL_STOCK + product_sku
logging.info(f"[{product_name}] Checking stock: {api_stock_url}")
if not isinstance(product_upc, list):
product_upc = [product_upc]
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
# Build stock API URL and call to check status
API_URL = API_URL_STOCK + product_sku
logging.info(f"Stock API URL called: {API_URL}")
try:
response = session.get(API_URL, headers=HEADERS, timeout=10)
logging.info(f"Stock API response: {response.status_code}")
response.raise_for_status()
data = response.json()
except requests.exceptions.RequestException as e:
logging.error(f"Stock API error: {e}")
return
products = data.get("listMap", [])
products_price = 'Price not available' # Default value
# Check product list and retrieve price
if isinstance(products, list) and len(products) > 0:
for product in products:
price = product.get("price", 'Price not available')
if price != 'Price not available':
products_price = price # Use found price
break # Stop at first found price
else:
logging.error("Product list is empty or malformed.")
found_in_stock = set()
# Check status and send notifications accordingly
for p in products:
gpu_name = p.get("fe_sku", "").upper()
is_active = p.get("is_active") == "true"
if is_active and any(target.upper() in gpu_name for target in product_upc):
found_in_stock.add(gpu_name)
for gpu in product_upc:
gpu_upper = gpu.upper()
currently_in_stock = gpu_upper in found_in_stock
previously_in_stock = global_stock_status.get(gpu_upper, False)
if currently_in_stock and not previously_in_stock:
product_link = PRODUCT_URL
send_discord_notification(gpu_upper, product_link, products_price)
global_stock_status[gpu_upper] = True
logging.info(f"{gpu} is now in stock!")
elif not currently_in_stock and previously_in_stock:
product_link = PRODUCT_URL
send_out_of_stock_notification(gpu_upper, product_link, products_price)
global_stock_status[gpu_upper] = False
logging.info(f"{gpu} is no longer in stock.")
elif currently_in_stock and previously_in_stock:
logging.info(f"{gpu} is currently in stock.")
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.info(f"{gpu} is currently out of stock.")
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__":
while True:
check_rtx_50_founders()
time.sleep(REFRESH_TIME)
try:
while True:
check_rtx_50_founders()
time.sleep(REFRESH_TIME)
# Gracefully shut down
except KeyboardInterrupt:
logging.info("🛑 Script interrupted by user. Exiting gracefully.")
exit(0)
# Gracefully shut down
except KeyboardInterrupt:
logging.info("🛑 Script interrupted by user. Exiting gracefully.")
exit(0)