Compare commits

...
3 Commits
Author SHA1 Message Date
Djeex 775dd4178d Add Telegram notifications, make Discord optional 2026-10-28 17:47:36 +01:00
nvidia-stock-bot-ci b61feb4fef Bump build version to 4.1.3 [skip ci] 2026-10-24 12:54:39 +00:00
Djeex 2825c2dbfd chore(deps): update python docker tag to v3.15.1 (#34)
CI / build-and-scan (push) Successful in 36s
This PR contains the following updates:

| Package | Type | Update | Change |
|---|---|---|---|
| python | final | minor | `3.14.8-alpine` -> `3.15.1-alpine` |

---

### Configuration

📅 **Schedule**: Branch creation - At any time (no schedule defined), Automerge - At any time (no schedule defined).

🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied.

♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox.

🔕 **Ignore**: Close this PR and you won't be reminded about this update again.

---

 - [ ] <!-- rebase-check -->If you want to rebase/retry this PR, check this box

---

This PR has been generated by [Renovate Bot](https://github.com/renovatebot/renovate).
<!--renovate-debug:eyJjcmVhdGVkSW5WZXIiOiIzOS4yNjQuMCIsInVwZGF0ZWRJblZlciI6IjM5LjI2NC4wIiwidGFyZ2V0QnJhbmNoIjoibWFpbiIsImxhYmVscyI6WyJib3QiLCJtaW5vciJdfQ==-->

Reviewed-on: #34
2026-10-24 14:52:24 +02:00
10 changed files with 314 additions and 20 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
FROM python:3.14.8-alpine AS base FROM python:3.15.1-alpine AS base
RUN apk add --no-cache ca-certificates RUN apk add --no-cache ca-certificates
+56 -6
View File
@@ -8,7 +8,7 @@
<img src="https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/nvidia-stock-bot-logo.png" alt="Nvidia Stock Bot" width="300"> <img src="https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/nvidia-stock-bot-logo.png" alt="Nvidia Stock Bot" width="300">
</div> </div>
**🤖 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] > [!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._ >_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 without cloning the repo (quick)](#-docker-installation-without-the-repo-quick)
- [🐙 Docker Installation with the repo (developer)](#-docker-installation-with-the-repo) - [🐙 Docker Installation with the repo (developer)](#-docker-installation-with-the-repo)
- [🐍 Python Installation (developer)](#-python-installation) - [🐍 Python Installation (developer)](#-python-installation)
- [📨 Notification channels](#-notification-channels)
- [🐞 Common issues](#-common-issues) - [🐞 Common issues](#-common-issues)
- [❓ How it works](#-how-it-works) - [❓ How it works](#-how-it-works)
- [🧑‍💻 Contributors](#-contributors) - [🧑‍💻 Contributors](#-contributors)
@@ -36,6 +37,8 @@
- Discord `@everyone` or specified role on SKU change (possible imminent drop) - 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 - Discord `@everyone` or specified role notification when stock is detected, including model, price, and link
- Silent Discord notification when no stock is detected - 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 notification server name in notification footer
- Selectable check frequency - Selectable check frequency
- Test mode to check without sending notifications - Test mode to check without sending notifications
@@ -61,7 +64,9 @@ services:
environment: environment:
# Minimal environment variables # Minimal environment variables
- PRODUCT_NAMES= # Exact GPU name (e.g. 'RTX 5080, RTX 5090') - 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') - COUNTRY= # Set your country (e.g. 'US')
- PYTHONUNBUFFERED=1 # Enables real-time log output - PYTHONUNBUFFERED=1 # Enables real-time log output
``` ```
@@ -71,7 +76,9 @@ services:
| Variable | Description | Possible Values | Default Value | | Variable | Description | Possible Values | Default Value |
|---------------------|-------------------------------------------------|------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------| |---------------------|-------------------------------------------------|------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------|
| `PRODUCT_NAMES` | The exact GPU names you're searching for | `RTX 5080, RTX 5090` | | | `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` | | `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_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 | | `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: - 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 - The API and product URLs
- Stock checking frequency (default: 60s; lowering too much may get your IP blocked by Nvidia) - 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 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`) - 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 ```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 COUNTRY=US
export PRODUCT_NAMES=RTX 5080, RTX 5090 export PRODUCT_NAMES=RTX 5080, RTX 5090
export DISCORD_ROLES=<@&12345>, <@&6789> export DISCORD_ROLES=<@&12345>, <@&6789>
@@ -168,6 +177,42 @@ export PYTHONUNBUFFERED=1
python nvidia-stock-bot.py 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<TOKEN>/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 ## 🐞 Common issues
Error when trying to reach product API url : 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. - Your IP may be blacklisted by nvidia. Try to use a VPN.
- nvidia API may be down - 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 ## ❓ How it works
<div align="center" > <div align="center" >
+1 -1
View File
@@ -1 +1 @@
4.1.2 4.2.0
+33 -6
View File
@@ -25,9 +25,11 @@ logging.info("=" * 60)
# Env variables # Env variables
try: 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_SERVER_NAME = os.environ.get("DISCORD_SERVER_NAME", "Shared for free")
DISCORD_ROLES = os.environ.get("DISCORD_ROLES") 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" COUNTRY = os.environ.get("COUNTRY") or "US"
REFRESH_TIME = int(os.environ.get("REFRESH_TIME") or 30) REFRESH_TIME = int(os.environ.get("REFRESH_TIME") or 30)
TEST_MODE = os.environ.get("TEST_MODE", "False").lower() == "true" TEST_MODE = os.environ.get("TEST_MODE", "False").lower() == "true"
@@ -47,15 +49,34 @@ if TEST_MODE:
if not PRODUCT_NAMES: if not PRODUCT_NAMES:
logging.error("❌ PRODUCT_NAMES is required but not defined.") logging.error("❌ PRODUCT_NAMES is required but not defined.")
sys.exit(1) sys.exit(1)
if not DISCORD_WEBHOOK_URL:
logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.") # Notification channels: each one is optional, but at least one must be configured
sys.exit(1) 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(",")] PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(",")]
# Role mapping # Role mapping
DISCORD_ROLE_MAP = {} 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.") logging.warning("⚠️ DISCORD_ROLES not defined or empty. Defaulting all roles to @everyone.")
for name in PRODUCT_NAMES: for name in PRODUCT_NAMES:
DISCORD_ROLE_MAP[name] = "@everyone" DISCORD_ROLE_MAP[name] = "@everyone"
@@ -72,7 +93,9 @@ else:
# Masked webhook in terminal # Masked webhook in terminal
match = re.search(r"/(\d+)/(.*)", DISCORD_WEBHOOK_URL) 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_id = match.group(1)
webhook_token = match.group(2) webhook_token = match.group(2)
masked_webhook_id = webhook_id[: len(webhook_id) - 10] + "*" * 10 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"GPU: {PRODUCT_NAMES}")
logging.info(f"Discord Webhook URL: {wh_masked_url}") logging.info(f"Discord Webhook URL: {wh_masked_url}")
logging.info(f"Discord Role Mention: {DISCORD_ROLES}") 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 SKU: {API_URL_SKU}")
logging.info(f"API URL Stock: {API_URL_STOCK}") logging.info(f"API URL Stock: {API_URL_STOCK}")
logging.info(f"Product URL: {PRODUCT_URL}") logging.info(f"Product URL: {PRODUCT_URL}")
+10
View File
@@ -9,6 +9,9 @@ from notifier import (
send_discord_notification, send_discord_notification,
send_out_of_stock_notification, send_out_of_stock_notification,
send_sku_change_notification, send_sku_change_notification,
send_telegram_notification,
send_telegram_out_of_stock_notification,
send_telegram_sku_change_notification,
) )
# HTTP session # 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]: 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}") 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_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 last_sku_dict[product_name] = product_sku
first_run_dict[product_name] = False first_run_dict[product_name] = False
@@ -109,10 +115,14 @@ def check_rtx_50_founders():
if currently_in_stock and not previously_in_stock: if currently_in_stock and not previously_in_stock:
send_discord_notification(product_name, PRODUCT_URL, products_price) 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 global_stock_status_dict[(product_name, upc_upper)] = True
logging.info(f"[{product_name}] {upc} is now in stock!") logging.info(f"[{product_name}] {upc} is now in stock!")
elif not currently_in_stock and previously_in_stock: elif not currently_in_stock and previously_in_stock:
send_out_of_stock_notification(product_name, PRODUCT_URL, products_price) 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 global_stock_status_dict[(product_name, upc_upper)] = False
logging.info(f"[{product_name}] {upc} is now out of stock.") logging.info(f"[{product_name}] {upc} is now out of stock.")
elif currently_in_stock: elif currently_in_stock:
+108
View File
@@ -1,12 +1,18 @@
import html
import logging import logging
import re
import time import time
import requests import requests
from env_config import ( from env_config import (
DISCORD_ENABLED,
DISCORD_ROLE_MAP, DISCORD_ROLE_MAP,
DISCORD_SERVER_NAME, DISCORD_SERVER_NAME,
DISCORD_WEBHOOK_URL, DISCORD_WEBHOOK_URL,
TELEGRAM_BOT_TOKEN,
TELEGRAM_CHAT_ID,
TELEGRAM_ENABLED,
TEST_MODE, TEST_MODE,
buy_now, buy_now,
currency, 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" 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 # In stock
def send_discord_notification(gpu_name, product_link, products_price): def send_discord_notification(gpu_name, product_link, products_price):
if not DISCORD_ENABLED:
return
timestamp = int(time.time()) timestamp = int(time.time())
if TEST_MODE: if TEST_MODE:
logging.info(f"[TEST MODE] Notification: {gpu_name} available!") 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 # Out of stock
def send_out_of_stock_notification(gpu_name, product_link, products_price): def send_out_of_stock_notification(gpu_name, product_link, products_price):
if not DISCORD_ENABLED:
return
timestamp = int(time.time()) timestamp = int(time.time())
if TEST_MODE: if TEST_MODE:
logging.info(f"[TEST MODE] Out of stock: {gpu_name}") 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 # SKU change
def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link): def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
if not DISCORD_ENABLED:
return
timestamp = int(time.time()) timestamp = int(time.time())
if TEST_MODE: if TEST_MODE:
logging.info(f"[TEST MODE] SKU change: {old_sku} → {new_sku}") 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}") logging.error(f"❌ Webhook error: {response.status_code} - {response.text}")
except Exception as e: except Exception as e:
logging.error(f"🚨 Error sending webhook: {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.")
+3 -1
View File
@@ -1,3 +1,5 @@
PRODUCT_NAMES= # Exact GPU name (e.g. "RTX 5080, RTX 5090") 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') COUNTRY= # Set your country (e.g. 'US')
+2
View File
@@ -9,5 +9,7 @@ services:
# Minimal environment variables # Minimal environment variables
- PRODUCT_NAMES=${PRODUCT_NAMES} - PRODUCT_NAMES=${PRODUCT_NAMES}
- DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL} - DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL}
- TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
- TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID}
- COUNTRY=${COUNTRY} - COUNTRY=${COUNTRY}
- PYTHONUNBUFFERED=1 - PYTHONUNBUFFERED=1
+24 -3
View File
@@ -11,6 +11,8 @@ ENV_KEYS = [
"DISCORD_WEBHOOK_URL", "DISCORD_WEBHOOK_URL",
"DISCORD_SERVER_NAME", "DISCORD_SERVER_NAME",
"DISCORD_ROLES", "DISCORD_ROLES",
"TELEGRAM_BOT_TOKEN",
"TELEGRAM_CHAT_ID",
"COUNTRY", "COUNTRY",
"REFRESH_TIME", "REFRESH_TIME",
"TEST_MODE", "TEST_MODE",
@@ -27,9 +29,28 @@ def _reload_env_config(monkeypatch, env):
return importlib.import_module(MODULE_NAME) return importlib.import_module(MODULE_NAME)
def test_missing_webhook_exits(monkeypatch): def test_no_notification_channel_does_not_exit(monkeypatch, caplog):
with pytest.raises(SystemExit): cfg = _reload_env_config(monkeypatch, {"PRODUCT_NAMES": "RTX 5090"})
_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): def test_missing_product_names_exits(monkeypatch):
+76 -2
View File
@@ -4,10 +4,13 @@ import sys
VALID_WEBHOOK = "https://discord.com/api/webhooks/123456789012345678/abcdef" VALID_WEBHOOK = "https://discord.com/api/webhooks/123456789012345678/abcdef"
def _import_notifier(monkeypatch, test_mode="False", discord_roles=None): def _import_notifier(monkeypatch, test_mode="False", discord_roles=None, keep_discord=True):
monkeypatch.setenv("DISCORD_WEBHOOK_URL", VALID_WEBHOOK) if keep_discord:
monkeypatch.setenv("DISCORD_WEBHOOK_URL", VALID_WEBHOOK)
monkeypatch.setenv("PRODUCT_NAMES", "RTX 5090 Founders Edition") monkeypatch.setenv("PRODUCT_NAMES", "RTX 5090 Founders Edition")
monkeypatch.setenv("TEST_MODE", test_mode) monkeypatch.setenv("TEST_MODE", test_mode)
if not keep_discord:
monkeypatch.delenv("DISCORD_WEBHOOK_URL", raising=False)
if discord_roles is not None: if discord_roles is not None:
monkeypatch.setenv("DISCORD_ROLES", discord_roles) monkeypatch.setenv("DISCORD_ROLES", discord_roles)
else: else:
@@ -185,3 +188,74 @@ def test_sku_change_notification_mentions_role_and_skus(monkeypatch):
description = captured["json"]["embeds"][0]["description"] description = captured["json"]["embeds"][0]["description"]
assert "old-sku-123" in description assert "old-sku-123" in description
assert "new-sku-456" 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")