150 lines
5.7 KiB
Python
150 lines
5.7 KiB
Python
from typing import Optional, Tuple, Union
|
|
from urllib.parse import urljoin
|
|
|
|
from telegram import Bot, InlineKeyboardMarkup, Message, ParseMode
|
|
from telegram.error import InvalidToken, Unauthorized
|
|
from telegram.utils.request import Request
|
|
|
|
from apps.alerts.models import AlertGroup
|
|
from apps.base.utils import live_settings
|
|
from apps.telegram.models import TelegramMessage
|
|
from apps.telegram.renderers.keyboard import TelegramKeyboardRenderer
|
|
from apps.telegram.renderers.message import TelegramMessageRenderer
|
|
|
|
|
|
class TelegramClient:
|
|
ALLOWED_UPDATES = ("message", "callback_query")
|
|
PARSE_MODE = ParseMode.HTML
|
|
|
|
def __init__(self, token: Optional[str] = None):
|
|
self.token = token or live_settings.TELEGRAM_TOKEN
|
|
|
|
if self.token is None:
|
|
raise InvalidToken()
|
|
|
|
@property
|
|
def api_client(self) -> Bot:
|
|
return Bot(self.token, request=Request(read_timeout=15))
|
|
|
|
def is_chat_member(self, chat_id: Union[int, str]) -> bool:
|
|
try:
|
|
self.api_client.get_chat(chat_id=chat_id)
|
|
return True
|
|
except Unauthorized:
|
|
return False
|
|
|
|
def register_webhook(self, webhook_url: Optional[str] = None) -> None:
|
|
webhook_url = webhook_url or urljoin(live_settings.TELEGRAM_WEBHOOK_HOST, "/telegram/")
|
|
|
|
if webhook_url is None:
|
|
webhook_url = live_settings.TELEGRAM_WEBHOOK_URL
|
|
|
|
webhook_info = self.api_client.get_webhook_info()
|
|
if webhook_info.url == webhook_url:
|
|
return
|
|
|
|
self.api_client.set_webhook(webhook_url, allowed_updates=self.ALLOWED_UPDATES)
|
|
|
|
def send_message(
|
|
self,
|
|
chat_id: Union[int, str],
|
|
message_type: int,
|
|
alert_group: AlertGroup,
|
|
reply_to_message_id: Optional[int] = None,
|
|
) -> TelegramMessage:
|
|
text, keyboard = self._get_message_and_keyboard(message_type=message_type, alert_group=alert_group)
|
|
|
|
raw_message = self.send_raw_message(
|
|
chat_id=chat_id, text=text, keyboard=keyboard, reply_to_message_id=reply_to_message_id
|
|
)
|
|
message = TelegramMessage.create_from_message(
|
|
message=raw_message, alert_group=alert_group, message_type=message_type
|
|
)
|
|
|
|
return message
|
|
|
|
def send_raw_message(
|
|
self,
|
|
chat_id: Union[int, str],
|
|
text: str,
|
|
keyboard: Optional[InlineKeyboardMarkup] = None,
|
|
reply_to_message_id: Optional[int] = None,
|
|
) -> Message:
|
|
message = self.api_client.send_message(
|
|
chat_id=chat_id,
|
|
text=text,
|
|
reply_markup=keyboard,
|
|
reply_to_message_id=reply_to_message_id,
|
|
parse_mode=self.PARSE_MODE,
|
|
disable_web_page_preview=False,
|
|
)
|
|
return message
|
|
|
|
def edit_message(self, message: TelegramMessage) -> TelegramMessage:
|
|
text, keyboard = self._get_message_and_keyboard(
|
|
message_type=message.message_type, alert_group=message.alert_group
|
|
)
|
|
|
|
self.edit_raw_message(chat_id=message.chat_id, message_id=message.message_id, text=text, keyboard=keyboard)
|
|
return message
|
|
|
|
def edit_raw_message(
|
|
self,
|
|
chat_id: Union[int, str],
|
|
message_id: Union[int, str],
|
|
text: str,
|
|
keyboard: Optional[InlineKeyboardMarkup] = None,
|
|
) -> Message:
|
|
message = self.api_client.edit_message_text(
|
|
chat_id=chat_id,
|
|
message_id=message_id,
|
|
text=text,
|
|
reply_markup=keyboard,
|
|
parse_mode=self.PARSE_MODE,
|
|
disable_web_page_preview=False,
|
|
)
|
|
return message
|
|
|
|
@staticmethod
|
|
def _get_message_and_keyboard(
|
|
message_type: int, alert_group: AlertGroup
|
|
) -> Tuple[str, Optional[InlineKeyboardMarkup]]:
|
|
message_renderer = TelegramMessageRenderer(alert_group=alert_group)
|
|
keyboard_renderer = TelegramKeyboardRenderer(alert_group=alert_group)
|
|
|
|
if message_type == TelegramMessage.ALERT_GROUP_MESSAGE:
|
|
text = message_renderer.render_alert_group_message()
|
|
keyboard = None
|
|
elif message_type == TelegramMessage.LOG_MESSAGE:
|
|
text = message_renderer.render_log_message()
|
|
keyboard = None
|
|
elif message_type == TelegramMessage.ACTIONS_MESSAGE:
|
|
text = message_renderer.render_actions_message()
|
|
keyboard = keyboard_renderer.render_actions_keyboard()
|
|
elif message_type == TelegramMessage.PERSONAL_MESSAGE:
|
|
text = message_renderer.render_personal_message()
|
|
keyboard = keyboard_renderer.render_actions_keyboard()
|
|
elif message_type == TelegramMessage.FORMATTING_ERROR:
|
|
text = message_renderer.render_formatting_error_message()
|
|
keyboard = None
|
|
elif message_type in (
|
|
TelegramMessage.LINK_TO_CHANNEL_MESSAGE,
|
|
TelegramMessage.LINK_TO_CHANNEL_MESSAGE_WITHOUT_TITLE,
|
|
):
|
|
alert_group_message = alert_group.telegram_messages.filter(
|
|
chat_id__startswith="-",
|
|
message_type__in=[TelegramMessage.ALERT_GROUP_MESSAGE, TelegramMessage.FORMATTING_ERROR],
|
|
).first()
|
|
|
|
if alert_group_message is None:
|
|
raise Exception("No alert group message found, probably it is not saved to database yet")
|
|
|
|
include_title = message_type == TelegramMessage.LINK_TO_CHANNEL_MESSAGE
|
|
link = alert_group_message.link
|
|
|
|
text = message_renderer.render_link_to_channel_message(include_title=include_title)
|
|
keyboard = keyboard_renderer.render_link_to_channel_keyboard(link=link)
|
|
else:
|
|
raise Exception(f"_get_message_and_keyboard with type {message_type} is not implemented")
|
|
|
|
return text, keyboard
|