import logging import requests logger = logging.getLogger(__name__) _ICONS = {'rate_limit': '🚨', 'error_rate': '⚠️'} def send_alert_telegram(profile, alert) -> bool: if not profile.telegram_bot_token or not profile.telegram_chat_id: return False icon = _ICONS.get(alert.alert_type, '🔔') geo = f' ({alert.city}, {alert.country})' if alert.country else '' text = ( f'{icon} *NginxMon Alert* — {profile.name}\n' f'Type: {alert.get_alert_type_display()}\n' f'IP: `{alert.remote_addr}`{geo}\n' f'{alert.detail}\n' f'_Detected at {alert.detected_at.strftime("%Y-%m-%d %H:%M:%S UTC") if alert.detected_at else "now"}_' ) url = f'https://api.telegram.org/bot{profile.telegram_bot_token}/sendMessage' try: resp = requests.post(url, json={ 'chat_id': profile.telegram_chat_id, 'text': text, 'parse_mode': 'Markdown', }, timeout=10) resp.raise_for_status() alert.notified = True alert.save(update_fields=['notified']) logger.info('nginxmon: Telegram alert sent for alert #%s', alert.pk) return True except Exception as exc: logger.error('nginxmon: Telegram send failed: %s', exc) return False def send_test_telegram(profile) -> dict: if not profile.telegram_bot_token or not profile.telegram_chat_id: return {'ok': False, 'error': 'Bot token or chat ID not set.'} try: url = f'https://api.telegram.org/bot{profile.telegram_bot_token}/sendMessage' resp = requests.post(url, json={ 'chat_id': profile.telegram_chat_id, 'text': f'✅ *NginxMon test* from profile _{profile.name}_. Alerts are working.', 'parse_mode': 'Markdown', }, timeout=10) resp.raise_for_status() return {'ok': True} except Exception as e: return {'ok': False, 'error': str(e)}