150 lines
5.6 KiB
Python
150 lines
5.6 KiB
Python
import logging
|
|
import os
|
|
|
|
import telepot
|
|
from alerta.plugins import PluginBase
|
|
from jinja2 import Template, UndefinedError
|
|
|
|
try:
|
|
from alerta.plugins import app # alerta >= 5.0
|
|
except ImportError:
|
|
from alerta.app import app # alerta < 5.0
|
|
|
|
|
|
DEFAULT_TMPL = r"""
|
|
{% if customer %}Customer: `{{customer}}` {% endif %}
|
|
|
|
*[{{ status.capitalize() }}] {{ environment }} {{ severity.capitalize() }}*
|
|
{{ event | replace("_","\_") }} {{ resource.capitalize() }}
|
|
|
|
```
|
|
{{ text }}
|
|
```
|
|
"""
|
|
|
|
LOG = logging.getLogger('alerta.plugins.telegram')
|
|
|
|
TELEGRAM_TOKEN = app.config.get('TELEGRAM_TOKEN') \
|
|
or os.environ.get('TELEGRAM_TOKEN')
|
|
TELEGRAM_CHAT_ID = app.config.get('TELEGRAM_CHAT_ID') \
|
|
or os.environ.get('TELEGRAM_CHAT_ID')
|
|
TELEGRAM_WEBHOOK_URL = app.config.get('TELEGRAM_WEBHOOK_URL', None) \
|
|
or os.environ.get('TELEGRAM_WEBHOOK_URL')
|
|
TELEGRAM_TEMPLATE = app.config.get('TELEGRAM_TEMPLATE') \
|
|
or os.environ.get('TELEGRAM_TEMPLATE')
|
|
TELEGRAM_PROXY = app.config.get('TELEGRAM_PROXY') \
|
|
or os.environ.get('TELEGRAM_PROXY')
|
|
TELEGRAM_PROXY_USERNAME = app.config.get('TELEGRAM_PROXY_USERNAME') \
|
|
or os.environ.get('TELEGRAM_PROXY_USERNAME')
|
|
TELEGRAM_PROXY_PASSWORD = app.config.get('TELEGRAM_PROXY_PASSWORD') \
|
|
or os.environ.get('TELEGRAM_PROXY_PASSWORD')
|
|
TELEGRAM_SOUND_NOTIFICATION_SEVERITY = app.config.get('TELEGRAM_SOUND_NOTIFICATION_SEVERITY') \
|
|
or os.environ.get('TELEGRAM_SOUND_NOTIFICATION_SEVERITY')
|
|
TELEGRAM_DISABLE_NOTIFICATION_SEVERITY = app.config.get('TELEGRAM_DISABLE_NOTIFICATION_SEVERITY') \
|
|
or os.environ.get('TELEGRAM_DISABLE_NOTIFICATION_SEVERITY')
|
|
|
|
DASHBOARD_URL = app.config.get('DASHBOARD_URL', '') \
|
|
or os.environ.get('DASHBOARD_URL')
|
|
|
|
# use all the same, but telepot.aio.api.set_proxy for async telepot
|
|
if all([TELEGRAM_PROXY, TELEGRAM_PROXY_USERNAME, TELEGRAM_PROXY_PASSWORD]):
|
|
telepot.api.set_proxy(
|
|
TELEGRAM_PROXY, (TELEGRAM_PROXY_USERNAME, TELEGRAM_PROXY_PASSWORD))
|
|
LOG.debug('Telegram: using proxy %s', TELEGRAM_PROXY)
|
|
elif TELEGRAM_PROXY is not None:
|
|
telepot.api.set_proxy(TELEGRAM_PROXY)
|
|
LOG.debug('Telegram: using proxy %s', TELEGRAM_PROXY)
|
|
|
|
|
|
class TelegramBot(PluginBase):
|
|
def __init__(self, name=None):
|
|
|
|
self.bot = telepot.Bot(TELEGRAM_TOKEN)
|
|
LOG.debug('Telegram: %s', self.bot.getMe())
|
|
|
|
if TELEGRAM_WEBHOOK_URL and \
|
|
TELEGRAM_WEBHOOK_URL != self.bot.getWebhookInfo()['url']:
|
|
self.bot.setWebhook(TELEGRAM_WEBHOOK_URL)
|
|
LOG.debug('Telegram: %s', self.bot.getWebhookInfo())
|
|
|
|
super().__init__(name)
|
|
if TELEGRAM_TEMPLATE:
|
|
if os.path.exists(TELEGRAM_TEMPLATE):
|
|
with open(TELEGRAM_TEMPLATE) as f:
|
|
self.template = Template(f.read())
|
|
else:
|
|
self.template = Template(TELEGRAM_TEMPLATE)
|
|
else:
|
|
self.template = Template(DEFAULT_TMPL)
|
|
|
|
def pre_receive(self, alert):
|
|
return alert
|
|
|
|
def post_receive(self, alert):
|
|
|
|
if alert.repeat:
|
|
return
|
|
|
|
try:
|
|
text = self.template.render(alert.__dict__)
|
|
except UndefinedError:
|
|
text = 'Something bad has happened but also we ' \
|
|
"can't handle your telegram template message."
|
|
|
|
LOG.debug('Telegram: message=%s', text)
|
|
|
|
if TELEGRAM_WEBHOOK_URL:
|
|
keyboard = {
|
|
'inline_keyboard': [
|
|
[
|
|
{'text': 'ack', 'callback_data': '/ack ' + alert.id},
|
|
{'text': 'close', 'callback_data': '/close ' + alert.id},
|
|
{'text': 'blackout',
|
|
'callback_data': '/blackout ' + alert.id}
|
|
]
|
|
]
|
|
}
|
|
else:
|
|
keyboard = None
|
|
|
|
if TELEGRAM_SOUND_NOTIFICATION_SEVERITY:
|
|
disable_notification = True
|
|
if alert.severity in TELEGRAM_SOUND_NOTIFICATION_SEVERITY:
|
|
disable_notification = False
|
|
else:
|
|
disable_notification = False
|
|
|
|
if TELEGRAM_DISABLE_NOTIFICATION_SEVERITY:
|
|
if alert.severity in TELEGRAM_DISABLE_NOTIFICATION_SEVERITY:
|
|
return
|
|
|
|
if alert.severity in ['ok', 'normal', 'cleared', app.config.get('DEFAULT_NORMAL_SEVERITY')] and alert.previous_severity in TELEGRAM_DISABLE_NOTIFICATION_SEVERITY:
|
|
LOG.debug('Alert severity is {} but previous_severity was {} (included in DEFAULT_NORMAL_SEVERITY list), thus it will not be forwarded to Telegram.'.format(
|
|
alert.severity, alert.previous_severity))
|
|
return
|
|
|
|
LOG.debug('Telegram: post_receive sendMessage disable_notification=%s', str(
|
|
disable_notification))
|
|
|
|
chat_ids = TELEGRAM_CHAT_ID.split(',')
|
|
for chat_id in chat_ids:
|
|
try:
|
|
response = self.bot.sendMessage(chat_id,
|
|
text,
|
|
parse_mode='Markdown',
|
|
disable_notification=disable_notification,
|
|
reply_markup=keyboard)
|
|
except telepot.exception.TelegramError as e:
|
|
raise RuntimeError('Telegram (ChatId: %s): ERROR - %s, description= %s, json=%s',
|
|
chat_id,
|
|
e.error_code,
|
|
e.description,
|
|
e.json)
|
|
except Exception as e:
|
|
raise RuntimeError('Telegram: ERROR - %s', e)
|
|
|
|
LOG.debug('Telegram (ChatId: %s): %s', chat_id, response)
|
|
|
|
def status_change(self, alert, status, summary):
|
|
return
|