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

Add Telegram notifications, make Discord optional

Browse Source
This commit is contained in:
Djeex
2026-10-28 17:47:36 +01:00
parent b61feb4fef
commit 775dd4178d
9 changed files with 313 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/env_config.py
+33 -6
View File
@@ -25,9 +25,11 @@ logging.info("=" * 60)
# Env variables
try:
DISCORD_WEBHOOK_URL = os.environ["DISCORD_WEBHOOK_URL"]
DISCORD_WEBHOOK_URL = (os.environ.get("DISCORD_WEBHOOK_URL") or "").strip()
DISCORD_SERVER_NAME = os.environ.get("DISCORD_SERVER_NAME", "Shared for free")
DISCORD_ROLES = os.environ.get("DISCORD_ROLES")
TELEGRAM_BOT_TOKEN = (os.environ.get("TELEGRAM_BOT_TOKEN") or "").strip()
TELEGRAM_CHAT_ID = (os.environ.get("TELEGRAM_CHAT_ID") or "").strip()
COUNTRY = os.environ.get("COUNTRY") or "US"
REFRESH_TIME = int(os.environ.get("REFRESH_TIME") or 30)
TEST_MODE = os.environ.get("TEST_MODE", "False").lower() == "true"
@@ -47,15 +49,34 @@ if TEST_MODE:
if not PRODUCT_NAMES:
logging.error("❌ PRODUCT_NAMES is required but not defined.")
sys.exit(1)
if not DISCORD_WEBHOOK_URL:
logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.")
sys.exit(1)
# Notification channels: each one is optional, but at least one must be configured
if bool(TELEGRAM_BOT_TOKEN) != bool(TELEGRAM_CHAT_ID):
logging.warning(
"⚠️ TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID must both be defined. Telegram is disabled."
)
TELEGRAM_BOT_TOKEN = TELEGRAM_CHAT_ID = ""
DISCORD_ENABLED = bool(DISCORD_WEBHOOK_URL)
TELEGRAM_ENABLED = bool(TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID)
if not DISCORD_ENABLED:
logging.warning("⚠️ DISCORD IS NOT CONFIGURED: no Discord notification will be sent.")
if not TELEGRAM_ENABLED:
logging.warning("⚠️ TELEGRAM IS NOT CONFIGURED: no Telegram notification will be sent.")
if not DISCORD_ENABLED and not TELEGRAM_ENABLED:
logging.warning(
"🚨 NO NOTIFICATION CHANNEL CONFIGURED: the bot will run but will not alert you! "
"Define DISCORD_WEBHOOK_URL and/or TELEGRAM_BOT_TOKEN + TELEGRAM_CHAT_ID."
)
PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(",")]
# Role mapping
DISCORD_ROLE_MAP = {}
if not DISCORD_ROLES or not DISCORD_ROLES.strip():
if not DISCORD_ENABLED:
pass
elif 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"
@@ -72,7 +93,9 @@ else:
# Masked webhook in terminal
match = re.search(r"/(\d+)/(.*)", DISCORD_WEBHOOK_URL)
if match:
if not DISCORD_ENABLED:
wh_masked_url = "[Disabled]"
elif match:
webhook_id = match.group(1)
webhook_token = match.group(2)
masked_webhook_id = webhook_id[: len(webhook_id) - 10] + "*" * 10
@@ -198,6 +221,10 @@ imminent_drop = loc["imminent_drop"]
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(
"Telegram: %s",
f"enabled (chat {TELEGRAM_CHAT_ID})" if TELEGRAM_ENABLED else "[Disabled]",
)
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}")
app/gpu_checker.py
+10
View File
@@ -9,6 +9,9 @@ from notifier import (
send_discord_notification,
send_out_of_stock_notification,
send_sku_change_notification,
send_telegram_notification,
send_telegram_out_of_stock_notification,
send_telegram_sku_change_notification,
)
# HTTP session
@@ -64,6 +67,9 @@ def check_rtx_50_founders():
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(product_name, old_sku, product_sku, PRODUCT_URL)
send_telegram_sku_change_notification(
product_name, old_sku, product_sku, PRODUCT_URL
)
last_sku_dict[product_name] = product_sku
first_run_dict[product_name] = False
@@ -109,10 +115,14 @@ def check_rtx_50_founders():
if currently_in_stock and not previously_in_stock:
send_discord_notification(product_name, PRODUCT_URL, products_price)
send_telegram_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)
send_telegram_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:
app/notifier.py
+108
View File
@@ -1,12 +1,18 @@
import html
import logging
import re
import time
import requests
from env_config import (
DISCORD_ENABLED,
DISCORD_ROLE_MAP,
DISCORD_SERVER_NAME,
DISCORD_WEBHOOK_URL,
TELEGRAM_BOT_TOKEN,
TELEGRAM_CHAT_ID,
TELEGRAM_ENABLED,
TEST_MODE,
buy_now,
currency,
@@ -24,8 +30,27 @@ AVATAR = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img
THUMBNAIL = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg"
_MD_LINK = re.compile(r"\[([^\]]*)\]\(([^)\s]*)\)")
_MD_BOLD = re.compile(r"\*\*(.+?)\*\*")
_MD_CODE = re.compile(r"`(.+?)`")
_EMOJI_SHORTCODES = {":point_right:": "👉"}
def _md_to_telegram_html(text):
"""Convert the Discord markdown of the translations to Telegram HTML."""
text = html.escape(text)
text = _MD_LINK.sub(lambda m: f'<a href="{m.group(2)}">{m.group(1)}</a>', text)
text = _MD_BOLD.sub(lambda m: f"<b>{m.group(1)}</b>", text)
text = _MD_CODE.sub(lambda m: f"<code>{m.group(1)}</code>", text)
for shortcode, emoji in _EMOJI_SHORTCODES.items():
text = text.replace(shortcode, emoji)
return text
# In stock
def send_discord_notification(gpu_name, product_link, products_price):
if not DISCORD_ENABLED:
return
timestamp = int(time.time())
if TEST_MODE:
logging.info(f"[TEST MODE] Notification: {gpu_name} available!")
@@ -66,6 +91,8 @@ def send_discord_notification(gpu_name, product_link, products_price):
# Out of stock
def send_out_of_stock_notification(gpu_name, product_link, products_price):
if not DISCORD_ENABLED:
return
timestamp = int(time.time())
if TEST_MODE:
logging.info(f"[TEST MODE] Out of stock: {gpu_name}")
@@ -100,6 +127,8 @@ def send_out_of_stock_notification(gpu_name, product_link, products_price):
# SKU change
def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
if not DISCORD_ENABLED:
return
timestamp = int(time.time())
if TEST_MODE:
logging.info(f"[TEST MODE] SKU change: {old_sku} → {new_sku}")
@@ -134,3 +163,82 @@ def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
logging.error(f"❌ Webhook error: {response.status_code} - {response.text}")
except Exception as e:
logging.error(f"🚨 Error sending webhook: {e}")
# Telegram
def _send_telegram(text, success_log):
if TEST_MODE:
logging.info(f"[TEST MODE] Telegram message: {text}")
return
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": text,
"parse_mode": "HTML",
"disable_web_page_preview": True,
}
try:
response = requests.post(url, json=payload, timeout=10)
if response.status_code == 200:
logging.info(success_log)
else:
logging.error(f"❌ Telegram error: {response.status_code} - {response.text}")
except Exception as e:
# The exception message may contain the bot token (it is part of the URL)
logging.error(f"🚨 Error sending Telegram message: {type(e).__name__}")
def _telegram_footer():
return html.escape(footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME))
def _telegram_time():
return html.escape(time.strftime("%Y-%m-%d %H:%M:%S %Z"))
def send_telegram_notification(gpu_name, product_link, products_price):
if not TELEGRAM_ENABLED:
return
title = html.escape(in_stock_title.format(gpu_name=gpu_name))
description = _md_to_telegram_html(buy_now.format(product_link=product_link))
price = html.escape(f"{currency}{products_price}")
text = (
f"<b>{title}</b>\n"
f"{description}\n\n"
f"{html.escape(price_label)}: <code>{price}</code>\n"
f"{html.escape(time_label)}: {_telegram_time()}\n\n"
f"<i>{_telegram_footer()}</i>"
)
_send_telegram(text, "✅ Notification sent to Telegram.")
def send_telegram_out_of_stock_notification(gpu_name, product_link, products_price):
if not TELEGRAM_ENABLED:
return
title = html.escape(out_of_stock_title.format(gpu_name=gpu_name))
text = (
f'<b><a href="{html.escape(product_link)}">{title}</a></b>\n'
f"{html.escape(time_label)}: {_telegram_time()}\n\n"
f"<i>{_telegram_footer()}</i>"
)
_send_telegram(text, "✅ Out-of-stock notification sent to Telegram.")
def send_telegram_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
if not TELEGRAM_ENABLED:
return
alert = html.escape(imminent_drop.format(DISCORD_ROLE="").strip())
title = html.escape(sku_change_title.format(gpu_name=gpu_name))
description = _md_to_telegram_html(
sku_description.format(old_sku=old_sku, new_sku=new_sku)
)
text = (
f"{alert}\n"
f'<b><a href="{html.escape(product_link)}">{title}</a></b>\n'
f"{description}\n"
f"{html.escape(time_label)}: {_telegram_time()}\n\n"
f"<i>{_telegram_footer()}</i>"
)
_send_telegram(text, "✅ SKU change notification sent to Telegram.")