diff --git a/VERSION b/VERSION
index de197cc..6aba2b2 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-4.1.3
+4.2.0
diff --git a/app/env_config.py b/app/env_config.py
index 23f75c7..17fea40 100644
--- a/app/env_config.py
+++ b/app/env_config.py
@@ -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}")
diff --git a/app/gpu_checker.py b/app/gpu_checker.py
index b8fd047..9267b88 100644
--- a/app/gpu_checker.py
+++ b/app/gpu_checker.py
@@ -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:
diff --git a/app/notifier.py b/app/notifier.py
index e040a14..823a04c 100644
--- a/app/notifier.py
+++ b/app/notifier.py
@@ -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'
{m.group(1)}', text)
+ text = _MD_BOLD.sub(lambda m: f"
{m.group(1)}", text)
+ text = _MD_CODE.sub(lambda m: f"
{m.group(1)}", 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"
{title}\n"
+ f"{description}\n\n"
+ f"{html.escape(price_label)}:
{price}\n"
+ f"{html.escape(time_label)}: {_telegram_time()}\n\n"
+ f"
{_telegram_footer()}"
+ )
+ _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'
{title}\n'
+ f"{html.escape(time_label)}: {_telegram_time()}\n\n"
+ f"
{_telegram_footer()}"
+ )
+ _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'
{title}\n'
+ f"{description}\n"
+ f"{html.escape(time_label)}: {_telegram_time()}\n\n"
+ f"
{_telegram_footer()}"
+ )
+ _send_telegram(text, "✅ SKU change notification sent to Telegram.")
diff --git a/docker/.env b/docker/.env
index b0336a3..4df3be9 100644
--- a/docker/.env
+++ b/docker/.env
@@ -1,3 +1,5 @@
PRODUCT_NAMES= # Exact GPU name (e.g. "RTX 5080, RTX 5090")
-DISCORD_WEBHOOK_URL= # Your Discord webhook URL
+DISCORD_WEBHOOK_URL= # Your Discord webhook URL (optional)
+TELEGRAM_BOT_TOKEN= # Your Telegram bot token (optional)
+TELEGRAM_CHAT_ID= # Your Telegram chat ID (optional)
COUNTRY= # Set your country (e.g. 'US')
\ No newline at end of file
diff --git a/docker/compose.yaml b/docker/compose.yaml
index 9c377f9..1736ed9 100644
--- a/docker/compose.yaml
+++ b/docker/compose.yaml
@@ -9,5 +9,7 @@ services:
# Minimal environment variables
- PRODUCT_NAMES=${PRODUCT_NAMES}
- DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL}
+ - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
+ - TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID}
- COUNTRY=${COUNTRY}
- PYTHONUNBUFFERED=1
\ No newline at end of file
diff --git a/tests/test_env_config.py b/tests/test_env_config.py
index 45a9c47..4de4e04 100644
--- a/tests/test_env_config.py
+++ b/tests/test_env_config.py
@@ -11,6 +11,8 @@ ENV_KEYS = [
"DISCORD_WEBHOOK_URL",
"DISCORD_SERVER_NAME",
"DISCORD_ROLES",
+ "TELEGRAM_BOT_TOKEN",
+ "TELEGRAM_CHAT_ID",
"COUNTRY",
"REFRESH_TIME",
"TEST_MODE",
@@ -27,9 +29,28 @@ def _reload_env_config(monkeypatch, env):
return importlib.import_module(MODULE_NAME)
-def test_missing_webhook_exits(monkeypatch):
- with pytest.raises(SystemExit):
- _reload_env_config(monkeypatch, {"PRODUCT_NAMES": "RTX 5090"})
+def test_no_notification_channel_does_not_exit(monkeypatch, caplog):
+ cfg = _reload_env_config(monkeypatch, {"PRODUCT_NAMES": "RTX 5090"})
+ assert not cfg.DISCORD_ENABLED
+ assert not cfg.TELEGRAM_ENABLED
+ assert "DISCORD IS NOT CONFIGURED" in caplog.text
+ assert "TELEGRAM IS NOT CONFIGURED" in caplog.text
+
+
+def test_telegram_only_is_enabled(monkeypatch):
+ cfg = _reload_env_config(
+ monkeypatch,
+ {"PRODUCT_NAMES": "RTX 5090", "TELEGRAM_BOT_TOKEN": "123:abc", "TELEGRAM_CHAT_ID": "-100"},
+ )
+ assert cfg.TELEGRAM_ENABLED
+ assert not cfg.DISCORD_ENABLED
+
+
+def test_telegram_partial_config_is_disabled(monkeypatch):
+ cfg = _reload_env_config(
+ monkeypatch, {"PRODUCT_NAMES": "RTX 5090", "TELEGRAM_BOT_TOKEN": "123:abc"}
+ )
+ assert not cfg.TELEGRAM_ENABLED
def test_missing_product_names_exits(monkeypatch):
diff --git a/tests/test_notifier.py b/tests/test_notifier.py
index 49f12ad..8443bb4 100644
--- a/tests/test_notifier.py
+++ b/tests/test_notifier.py
@@ -4,10 +4,13 @@ import sys
VALID_WEBHOOK = "https://discord.com/api/webhooks/123456789012345678/abcdef"
-def _import_notifier(monkeypatch, test_mode="False", discord_roles=None):
- monkeypatch.setenv("DISCORD_WEBHOOK_URL", VALID_WEBHOOK)
+def _import_notifier(monkeypatch, test_mode="False", discord_roles=None, keep_discord=True):
+ if keep_discord:
+ monkeypatch.setenv("DISCORD_WEBHOOK_URL", VALID_WEBHOOK)
monkeypatch.setenv("PRODUCT_NAMES", "RTX 5090 Founders Edition")
monkeypatch.setenv("TEST_MODE", test_mode)
+ if not keep_discord:
+ monkeypatch.delenv("DISCORD_WEBHOOK_URL", raising=False)
if discord_roles is not None:
monkeypatch.setenv("DISCORD_ROLES", discord_roles)
else:
@@ -185,3 +188,74 @@ def test_sku_change_notification_mentions_role_and_skus(monkeypatch):
description = captured["json"]["embeds"][0]["description"]
assert "old-sku-123" in description
assert "new-sku-456" in description
+
+
+def _import_telegram_notifier(monkeypatch, with_discord=False):
+ if not with_discord:
+ monkeypatch.delenv("DISCORD_WEBHOOK_URL", raising=False)
+ monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "123:abc")
+ monkeypatch.setenv("TELEGRAM_CHAT_ID", "-100123")
+ return _import_notifier(monkeypatch, test_mode="False", keep_discord=with_discord)
+
+
+def test_telegram_in_stock_posts_to_bot_api(monkeypatch):
+ notifier = _import_telegram_notifier(monkeypatch)
+ captured = {}
+
+ def fake_post(url, json=None, **kwargs):
+ captured["url"] = url
+ captured["json"] = json
+ return FakeResponse(status_code=200)
+
+ monkeypatch.setattr(notifier.requests, "post", fake_post)
+
+ notifier.send_telegram_notification("RTX 5090 Founders Edition", "https://example.com/buy", "1999")
+
+ assert captured["url"] == "https://api.telegram.org/bot123:abc/sendMessage"
+ assert captured["json"]["chat_id"] == "-100123"
+ assert "RTX 5090 Founders Edition" in captured["json"]["text"]
+ assert "https://example.com/buy" in captured["json"]["text"]
+
+
+def test_telegram_disabled_sends_nothing(monkeypatch):
+ notifier = _import_notifier(monkeypatch, test_mode="False")
+ calls = []
+ monkeypatch.setattr(notifier.requests, "post", lambda *a, **k: calls.append(a))
+
+ notifier.send_telegram_notification("RTX 5090 Founders Edition", "https://example.com", "1999")
+ notifier.send_telegram_out_of_stock_notification("RTX 5090", "https://example.com", "1999")
+ notifier.send_telegram_sku_change_notification("RTX 5090", "a", "b", "https://example.com")
+
+ assert calls == []
+
+
+def test_discord_disabled_sends_nothing(monkeypatch):
+ notifier = _import_telegram_notifier(monkeypatch)
+ calls = []
+ monkeypatch.setattr(notifier.requests, "post", lambda *a, **k: calls.append(a))
+
+ notifier.send_discord_notification("RTX 5090 Founders Edition", "https://example.com", "1999")
+
+ assert calls == []
+
+
+def test_telegram_error_does_not_leak_token_or_raise(monkeypatch, caplog):
+ notifier = _import_telegram_notifier(monkeypatch)
+
+ def raise_error(*a, **k):
+ raise notifier.requests.exceptions.ConnectionError("https://api.telegram.org/bot123:abc/")
+
+ monkeypatch.setattr(notifier.requests, "post", raise_error)
+
+ notifier.send_telegram_out_of_stock_notification("RTX 5090", "https://example.com", "1999")
+
+ assert "123:abc" not in caplog.text
+
+
+def test_telegram_sku_change_and_http_error(monkeypatch):
+ notifier = _import_telegram_notifier(monkeypatch)
+ monkeypatch.setattr(
+ notifier.requests, "post", lambda *a, **k: FakeResponse(status_code=400, text="bad")
+ )
+
+ notifier.send_telegram_sku_change_notification("RTX 5090", "old", "new", "https://example.com")