import logging import requests logger = logging.getLogger(__name__) def _get_telegram_credentials(token: str | None = None, chat_id: str | None = None): if token is not None and chat_id is not None: return token, chat_id from links.models import SiteSettings cfg = SiteSettings.get() return cfg.telegram_bot_token, cfg.telegram_chat_id def notify_telegram(watcher, alert) -> None: token, chat_id = _get_telegram_credentials() if not token or not chat_id: logger.warning('Pricemon: no global Telegram credentials configured, skipping alert') return text = ( f'\U0001f4b0 *Price Drop Alert\\!*\n' f'*{_esc(watcher.name)}*\n' f'${alert.old_price} → *${alert.new_price}* \\({float(alert.drop_pct):.1f}% off\\)\n' f'[View Product]({watcher.url})' ) try: resp = requests.post( f'https://api.telegram.org/bot{token}/sendMessage', json={'chat_id': chat_id, 'text': text, 'parse_mode': 'MarkdownV2'}, timeout=10, ) resp.raise_for_status() logger.info(f'Telegram alert sent for watcher {watcher.pk}') except Exception as exc: logger.error(f'Failed to send Telegram alert for watcher {watcher.pk}: {exc}') def send_test_telegram(token: str | None = None, chat_id: str | None = None) -> dict: token, chat_id = _get_telegram_credentials(token, chat_id) if not token or not chat_id: return {'ok': False, 'error': 'No Telegram credentials in global Settings.'} try: resp = requests.post( f'https://api.telegram.org/bot{token}/sendMessage', json={'chat_id': chat_id, 'text': '\u2705 *PriceMon test* — notifications are working\\!', 'parse_mode': 'MarkdownV2'}, timeout=10, ) resp.raise_for_status() return {'ok': True} except Exception as exc: return {'ok': False, 'error': str(exc)} def _esc(text: str) -> str: for ch in r'\_*[]()~`>#+-=|{}.!': text = text.replace(ch, f'\\{ch}') return text