Used to notify user of events via telegram.
| 11 | |
| 12 | |
| 13 | class TelegramNotifier(NotifierUtils): |
| 14 | """Used to notify user of events via telegram. |
| 15 | """ |
| 16 | |
| 17 | def __init__(self, token, chat_id, parse_mode): |
| 18 | """Initialize TelegramNotifier class |
| 19 | |
| 20 | Args: |
| 21 | token (str): The telegram API token. |
| 22 | chat_id (str): The chat ID you want the bot to send messages to. |
| 23 | """ |
| 24 | |
| 25 | self.logger = structlog.get_logger() |
| 26 | self.bot = telegram.Bot(token=token) |
| 27 | self.chat_id = chat_id |
| 28 | self.parse_mode = parse_mode |
| 29 | |
| 30 | |
| 31 | @retry( |
| 32 | retry=retry_if_exception_type(telegram.error.TimedOut), |
| 33 | stop=stop_after_attempt(3), |
| 34 | wait=wait_fixed(5) |
| 35 | ) |
| 36 | def notify(self, message): |
| 37 | """Send the notification. |
| 38 | |
| 39 | Args: |
| 40 | message (str): The message to send. |
| 41 | """ |
| 42 | |
| 43 | max_message_size = 4096 |
| 44 | message_chunks = self.chunk_message(message=message, max_message_size=max_message_size) |
| 45 | #print(message_chunks) |
| 46 | #exit() |
| 47 | for message_chunk in message_chunks: |
| 48 | self.bot.send_message(chat_id=self.chat_id, text=message_chunk, parse_mode=self.parse_mode) |