links/pricemon/notifications.py

import logging

import requests

logger = logging.getLogger(__name__)


def _get_telegram_credentials():
    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() -> dict:
    token, chat_id = _get_telegram_credentials()
    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