From 775dd4178d50bf5b17a382b69c9b3cb4b2743119 Mon Sep 17 00:00:00 2001 From: Djeex Date: Wed, 28 Oct 2026 17:47:36 +0100 Subject: [PATCH] Add Telegram notifications, make Discord optional --- README.md | 62 +++++++++++++++++++--- VERSION | 2 +- app/env_config.py | 39 +++++++++++--- app/gpu_checker.py | 10 ++++ app/notifier.py | 108 +++++++++++++++++++++++++++++++++++++++ docker/.env | 4 +- docker/compose.yaml | 2 + tests/test_env_config.py | 27 ++++++++-- tests/test_notifier.py | 78 +++++++++++++++++++++++++++- 9 files changed, 313 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index da94a97..4c23465 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Nvidia Stock Bot -**🤖 Nvidia Stock Bot** - A bot that alerts you in real-time about **Nvidia RTX FE** GPU stock availability through Discord notifications. +**🤖 Nvidia Stock Bot** - A bot that alerts you in real-time about **Nvidia RTX FE** GPU stock availability through Discord and/or Telegram notifications. > [!NOTE] >_Github repo is a mirror of https://git.djeex.fr/Djeex/nvidia-stock-bot. You'll find full package, history and release note there. LLM used for bugs check and languages files generation._ @@ -25,6 +25,7 @@ - [🐳 Docker Installation without cloning the repo (quick)](#-docker-installation-without-the-repo-quick) - [🐙 Docker Installation with the repo (developer)](#-docker-installation-with-the-repo) - [🐍 Python Installation (developer)](#-python-installation) +- [📨 Notification channels](#-notification-channels) - [🐞 Common issues](#-common-issues) - [❓ How it works](#-how-it-works) - [🧑‍💻 Contributors](#-contributors) @@ -36,6 +37,8 @@ - Discord `@everyone` or specified role on SKU change (possible imminent drop) - Discord `@everyone` or specified role notification when stock is detected, including model, price, and link - Silent Discord notification when no stock is detected +- Telegram notifications (stock, out of stock, SKU change) in the same language as Discord +- Discord and Telegram are both optional and can be used together or separately - Selectable notification server name in notification footer - Selectable check frequency - Test mode to check without sending notifications @@ -61,7 +64,9 @@ services: environment: # Minimal environment variables - 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') - PYTHONUNBUFFERED=1 # Enables real-time log output ``` @@ -71,7 +76,9 @@ services: | Variable | Description | Possible Values | Default Value | |---------------------|-------------------------------------------------|------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------| | `PRODUCT_NAMES` | The exact GPU names you're searching for | `RTX 5080, RTX 5090` | | -| `DISCORD_WEBHOOK_URL` | Your Discord webhook URL | A valid URL | | +| `DISCORD_WEBHOOK_URL` | Your Discord webhook URL. Optional: if empty, Discord notifications are disabled (a warning is logged) | A valid URL | | +| `TELEGRAM_BOT_TOKEN` | Token of your Telegram bot (created with @BotFather). Optional, must be set together with `TELEGRAM_CHAT_ID`; if empty, Telegram notifications are disabled (a warning is logged) | `123456:ABC-DEF...` | | +| `TELEGRAM_CHAT_ID` | ID of the Telegram chat, group or channel receiving the notifications | `123456789`, `-1001234567890` | | | `COUNTRY` | Your country (only one). API localization, Discord notification currency and language will be set accordingly | `GB`, `US`, `CA`, `BE`, `NL`, `DK`, `DE`, `ES`, `FR`, `IT`, `NO`, `AT`, `PL`, `FI`, `SE`, `KR`, `JP` | `US` | | `DISCORD_SERVER_NAME` | The name of your server, displayed in notification's footer | A text | Shared for free | | `DISCORD_ROLES` | List of Discord roles ID in the same order than `PRODUCT_NAMES` values, found in your discord server settings (with user profile developer mode enabled) | `<@&12345><@&6789>` | @everyone | @@ -116,7 +123,7 @@ docker build -t nvidia-stock-bot . ``` - Then go to `nvidia-stock-bot/docker` and edit the `.env` file with: - - Your Discord webhook URL + - Your Discord webhook URL and/or your Telegram bot token and chat ID - The API and product URLs - Stock checking frequency (default: 60s; lowering too much may get your IP blocked by Nvidia) @@ -150,10 +157,12 @@ Instructions to directly run the Python script. Note: the bot stops when you clo git clone https://git.djeex.fr/Djeex/nvidia-stock-bot.git ``` - Navigate to `nvidia-stock-bot` and create a virtual environment (e.g., `python3 -m venv env_name`) -- Export the environment variables with your webhook and refresh time, for exemple: +- Export the environment variables with your webhook and/or Telegram settings and refresh time, for example: ```sh -export DISCORD_WEBHOOK_URL="https://your_discord_url" +export DISCORD_WEBHOOK_URL="https://your_discord_url" # optional +export TELEGRAM_BOT_TOKEN="123456:ABC-DEF" # optional +export TELEGRAM_CHAT_ID="123456789" # optional export COUNTRY=US export PRODUCT_NAMES=RTX 5080, RTX 5090 export DISCORD_ROLES=<@&12345>, <@&6789> @@ -168,6 +177,42 @@ export PYTHONUNBUFFERED=1 python nvidia-stock-bot.py ``` +## 📨 Notification channels + +Discord and Telegram are both optional and independent: configure one, both or none. The bot never stops because a channel is missing. For each channel that is not configured, a warning is written in the logs at startup: + +``` +[WARNING] ⚠️ DISCORD IS NOT CONFIGURED: no Discord notification will be sent. +[WARNING] ⚠️ TELEGRAM IS NOT CONFIGURED: no Telegram notification will be sent. +[WARNING] 🚨 NO NOTIFICATION CHANNEL CONFIGURED: the bot will run but will not alert you! +``` + +**Discord** + +Set `DISCORD_WEBHOOK_URL` (and optionally `DISCORD_ROLES` and `DISCORD_SERVER_NAME`). + +**Telegram** + +1. Create a bot with [@BotFather](https://t.me/BotFather) and copy its token (`TELEGRAM_BOT_TOKEN`). +2. Send a message to the bot, or add it to your group / channel (as administrator for a channel). Otherwise Telegram rejects the messages. +3. Get your chat ID (`TELEGRAM_CHAT_ID`), for example by opening `https://api.telegram.org/bot/getUpdates` after writing to the bot. Group and channel IDs are negative (e.g. `-1001234567890`). +4. Set both `TELEGRAM_BOT_TOKEN` and `TELEGRAM_CHAT_ID`. If only one of them is set, Telegram is disabled and a warning is logged. + +`DISCORD_SERVER_NAME` is also used in the footer of Telegram messages. Discord roles (`DISCORD_ROLES`) have no effect on Telegram. + +**Testing the notifications** + +`TEST_MODE=true` disables all sending, so it can't be used to check a channel. To send real test messages without touching the Nvidia API, call the notifier from the running container (keep `TEST_MODE` to `False`): + +```sh +docker exec nvidia-stock-bot python -c " +import notifier as n +u = 'https://marketplace.nvidia.com' +n.send_telegram_notification('RTX 5090', u, '1999') +n.send_discord_notification('RTX 5090', u, '1999') +" +``` + ## 🐞 Common issues Error when trying to reach product API url : @@ -175,6 +220,11 @@ Error when trying to reach product API url : - Your IP may be blacklisted by nvidia. Try to use a VPN. - nvidia API may be down +No Telegram message received: +- Check the logs: the Telegram API error (e.g. `chat not found`, `Forbidden`) is displayed there +- The bot must have received a message from you, or be a member of the group / channel +- `TELEGRAM_BOT_TOKEN` and `TELEGRAM_CHAT_ID` must both be set + ## ❓ How it works
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")