mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
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)}
|