v4.0 - Full localization #21
@ -4,15 +4,23 @@ import logging
|
||||
import json
|
||||
import sys
|
||||
|
||||
VERSION = "4.0.0"
|
||||
|
||||
# Logger setup
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||
)
|
||||
logging.info("Script started")
|
||||
|
||||
# Logging starter
|
||||
logging.info("=" * 60)
|
||||
logging.info("🟩 Nvidia Stock Bot - Version %s", VERSION)
|
||||
logging.info("Source: https://git.djeex.fr/Djeex/nvidia-stock-bot")
|
||||
logging.info("Mirror: https://github.com/Djeex/nvidia-stock-bot")
|
||||
logging.info("=" * 60)
|
||||
|
||||
# Env variables
|
||||
try:
|
||||
# Env variables
|
||||
DISCORD_WEBHOOK_URL = os.environ['DISCORD_WEBHOOK_URL']
|
||||
DISCORD_SERVER_NAME = os.environ.get('DISCORD_SERVER_NAME', 'Shared for free')
|
||||
DISCORD_ROLES = os.environ.get('DISCORD_ROLES')
|
||||
@ -20,6 +28,8 @@ try:
|
||||
REFRESH_TIME = int(os.environ.get('REFRESH_TIME') or 30)
|
||||
TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true'
|
||||
PRODUCT_NAMES = os.environ['PRODUCT_NAMES']
|
||||
|
||||
# Errors and warning
|
||||
except KeyError as e:
|
||||
logging.error(f"Missing environment variable: {e}")
|
||||
sys.exit(1)
|
||||
@ -27,6 +37,9 @@ except ValueError:
|
||||
logging.error("REFRESH_TIME must be a valid integer.")
|
||||
sys.exit(1)
|
||||
|
||||
if TEST_MODE:
|
||||
logging.warning("🚧 Test mode is active. No real alerts will be sent.")
|
||||
|
||||
if not PRODUCT_NAMES:
|
||||
logging.error("❌ PRODUCT_NAMES is required but not defined.")
|
||||
sys.exit(1)
|
||||
@ -53,7 +66,7 @@ else:
|
||||
sys.exit(1)
|
||||
DISCORD_ROLE_MAP[name] = role
|
||||
|
||||
# Masked webhook for display
|
||||
# Masked webhook in terminal
|
||||
match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL)
|
||||
if match:
|
||||
webhook_id = match.group(1)
|
||||
@ -64,10 +77,6 @@ if match:
|
||||
else:
|
||||
wh_masked_url = "[Invalid webhook URL]"
|
||||
|
||||
# Test mode
|
||||
if TEST_MODE:
|
||||
logging.warning("🚧 Test mode is active. No real alerts will be sent.")
|
||||
|
||||
# HTTP headers
|
||||
HEADERS = {
|
||||
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||||
|
@ -4,17 +4,21 @@ from env_config import HEADERS, PRODUCT_NAMES, API_URL_SKU, API_URL_STOCK, PRODU
|
||||
from notifier import send_discord_notification, send_out_of_stock_notification, send_sku_change_notification
|
||||
from requests.adapters import HTTPAdapter, Retry
|
||||
|
||||
# HTTP session
|
||||
session = requests.Session()
|
||||
retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
|
||||
session.mount('https://', HTTPAdapter(max_retries=retries))
|
||||
|
||||
# Keeping memory of last run
|
||||
last_sku_dict = {}
|
||||
global_stock_status_dict = {}
|
||||
first_run_dict = {name: True for name in PRODUCT_NAMES}
|
||||
|
||||
# Stock check function
|
||||
def check_rtx_50_founders():
|
||||
global last_sku_dict, global_stock_status_dict, first_run_dict
|
||||
|
||||
# Fetching nvidia API data
|
||||
try:
|
||||
response = session.get(API_URL_SKU, headers=HEADERS, timeout=10)
|
||||
logging.info(f"SKU API response: {response.status_code}")
|
||||
@ -23,7 +27,8 @@ def check_rtx_50_founders():
|
||||
except requests.exceptions.RequestException as e:
|
||||
logging.error(f"SKU API error: {e}")
|
||||
return
|
||||
|
||||
|
||||
# Checking productSKU and productUPC for all GPU set in PRODUCT_NAME
|
||||
all_products = data['searchedProducts']['productDetails']
|
||||
|
||||
for product_name in PRODUCT_NAMES:
|
||||
@ -42,6 +47,7 @@ def check_rtx_50_founders():
|
||||
if not isinstance(product_upc, list):
|
||||
product_upc = [product_upc]
|
||||
|
||||
# Detect SKU changes
|
||||
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}")
|
||||
@ -49,7 +55,8 @@ def check_rtx_50_founders():
|
||||
|
||||
last_sku_dict[product_name] = product_sku
|
||||
first_run_dict[product_name] = False
|
||||
|
||||
|
||||
# Check product availability in API_URL_STOCK for each SKU
|
||||
api_stock_url = API_URL_STOCK + product_sku
|
||||
logging.info(f"[{product_name}] Checking stock: {api_stock_url}")
|
||||
|
||||
@ -62,6 +69,7 @@ def check_rtx_50_founders():
|
||||
logging.error(f"Stock API error: {e}")
|
||||
continue
|
||||
|
||||
# Retrieve availibilty and price for each SKU
|
||||
products = stock_data.get("listMap", [])
|
||||
products_price = "Price not available"
|
||||
if isinstance(products, list) and len(products) > 0:
|
||||
@ -79,7 +87,8 @@ def check_rtx_50_founders():
|
||||
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)
|
||||
|
||||
|
||||
# Comparing previous state and notify
|
||||
for upc in product_upc:
|
||||
upc_upper = upc.upper()
|
||||
currently_in_stock = upc_upper in found_in_stock
|
||||
|
Reference in New Issue
Block a user