Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Djeex
|
775dd4178d | ||
 nvidia-stock-bot-ci
|
b61feb4fef | ||
|
Djeex
|
2825c2dbfd |
+1
-1
@@ -1,4 +1,4 @@
|
||||
FROM python:3.15.0-alpine AS base
|
||||
FROM python:3.15.1-alpine AS base
|
||||
|
||||
RUN apk add --no-cache ca-certificates
|
||||
|
||||
|
||||
@@ -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">
|
||||
</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]
|
||||
>_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<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
|
||||
|
||||
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
|
||||
|
||||
<div align="center" >
|
||||
|
||||
+33
-6
@@ -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}")
|
||||
|
||||
@@ -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:
|
||||
|
||||
+108
@@ -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.")
|
||||
|
||||
+3
-1
@@ -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')
|
||||
@@ -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
|
||||
@@ -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):
|
||||
|
||||
+76
-2
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user