From 094a3802ec57d31f0e28904e719f0d449a338bdb Mon Sep 17 00:00:00 2001 From: Pavel Muhortov Date: Sun, 18 Jun 2023 10:55:19 +0300 Subject: [PATCH] add Telegram() class --- cctv-scheduler.py | 339 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 339 insertions(+) diff --git a/cctv-scheduler.py b/cctv-scheduler.py index 39f0e13..d23d652 100644 --- a/cctv-scheduler.py +++ b/cctv-scheduler.py @@ -15,6 +15,7 @@ from os import path, sep, makedirs, remove, replace, environ from subprocess import Popen, PIPE, STDOUT from sys import platform from time import sleep +import requests from paramiko import SSHClient, AutoAddPolicy @@ -1310,6 +1311,344 @@ class Wordpress(Connect): return {"success": False, "result": "ERROR"} +class Telegram(): + """Set of methods (functions) for Telegram Bot API. + Reference: https://core.telegram.org/bots/api#available-methods + """ + def __init__(self, token: str): + """Object constructor. + + Args: + token (str): Telegram Bot API access token. + """ + if Do.args_valid(locals(), self.__init__.__annotations__): + self._token = token + self.api_root = 'https://api.telegram.org' + self.api_path = self.api_root + '/bot' + self._token + + def send_message(self, chat: str, text: str, parse_mode: str = 'HTML') -> dict: + """Send text message. + + Args: + chat (str): unique identifier for the target chat or username of the target channel. + text (str): text of the message to be sent, 1-4096 characters after entities parsing. + parse_mode (str, optional): 'HTML', 'Markdown', 'MarkdownV2'. Defaults to 'HTML'. + + Returns: + dict: {"success":bool,"result":"API response" or "ERROR"} + """ + if Do.args_valid(locals(), self.send_message.__annotations__): + url=self.api_path + '/sendMessage' + data = { + "chat_id": chat, + "text": text, + "parse_mode": parse_mode, + "disable_notification": True + } + response = requests.post(url=url, json=data, timeout=15) + if response.status_code == 200: + response = response.json() + logging.info(msg="" + + "message '" + + str(response['result']['message_id']) + + "' sent to telegram chat " + + str(chat) + ) + return {'success': True, 'result': response} + else: + logging.warning(msg="message didn't send to telegram chat " + str(chat)) + return {'success': False, 'result': response} + + def delete_message(self, chat: str, message_id: int) -> dict: + """Delete message. + + Args: + chat (str): unique identifier for the target chat or username of the target channel. + message_id (int): identifier of the message to delete. + + Returns: + dict: {"success":bool,"result":"API response" or "ERROR"} + """ + if Do.args_valid(locals(), self.delete_message.__annotations__): + url=self.api_path + '/deleteMessage' + data = {"chat_id": chat, "message_id": message_id} + response = requests.post(url=url, json=data, timeout=15) + if response.status_code == 200: + response = response.json() + logging.info(msg="" + + "message '" + str(message_id) + "' deleted from telegram chat " + + str(chat) + ) + return {'success': True, 'result': response} + else: + logging.warning(msg="" + + "message '" + str(message_id) + "' didn't deleted from telegram chat " + + str(chat) + ) + return {'success': False, 'result': response} + + def __send_media( + self, + chat: str, + media_path: str, + media_type: str, + caption: (str, type(None)), + parse_mode: str, + disable_notification: bool, + additional_url_param: (str, type(None)) + ) -> dict: + """Send media by api.telegram.org. + + Args: + chat (str): unique identifier for the target chat or username of the target channel. + media_path (str): /local/path/to/file, https://url/to/file, file_id=EXISTFILEID. + media_type (str): 'document', 'photo', 'video', 'audio'. + caption (str, None): media caption less 1024 characters. + parse_mode (str): caption 'HTML', 'Markdown', 'MarkdownV2' parse mode. + disable_notification (bool): send silently. + additional_url_param (str, None): example: '&duration=30&width=960&height=540'. + + Raises: + ValueError: "'media_type' value is wrong" + + Returns: + dict: {'success':bool,'result':response}. + """ + if Do.args_valid(locals(), self.__send_media.__annotations__): + if ( + media_type == 'document' or + media_type == 'photo' or + media_type == 'video' or + media_type == 'audio' + ): + url = self.api_path + '/send' + media_type + '?chat_id=' + chat + else: + raise ValueError("'media_type' value is wrong: " + media_type) + + if caption: + url = url + '&caption=' + caption + '&parse_mode=' + parse_mode + if disable_notification: + url = url + "&disable_notification=True" + if additional_url_param: + url = url + additional_url_param + if re.match("^(?:http://|https://|file_id=)", media_path): + media_path = media_path.replace('file_id=', '') + response = requests.post( + url=url + "&" + media_type + "=" + media_path, + timeout=60 + ) + if response.status_code == 200: + response = response.json() + if media_type == 'photo': + file_id = response['result'][media_type][-1]['file_id'] + else: + file_id = response['result'][media_type]['file_id'] + logging.info(msg="" + + media_type + + " '" + + str(file_id) + + "' sent to telegram chat " + + chat + ) + return {'success': True, 'result': response} + else: + response = requests.post( + url=url, + files={media_type: open(media_path, "rb")}, + timeout=60 + ) + if response.status_code == 200: + response = response.json() + if media_type == 'photo': + file_id = response['result'][media_type][-1]['file_id'] + else: + file_id = response['result'][media_type]['file_id'] + logging.info(msg="" + + media_type + + " '" + + str(file_id) + + "' sent to telegram chat " + + chat + ) + return {'success': True, 'result': response} + logging.warning( + msg=media_type + " " + media_path + " didn't send to telegram chat " + str(chat) + ) + return {'success': False, 'result': response} + + def send_document( + self, + chat: str, + document: str, + caption: (str, type(None)) = None, + parse_mode: str = 'HTML', + disable_notification: bool = True + ) -> dict: + """Send document. See self.__send_media(). + """ + if Do.args_valid(locals(), self.send_document.__annotations__): + return self.__send_media( + chat=chat, + media_path=document, + media_type='document', + caption=caption, + parse_mode=parse_mode, + disable_notification=disable_notification, + additional_url_param=None + ) + + def send_photo( + self, + chat: str, + photo: str, + caption: (str, type(None)) = None, + parse_mode: str = 'HTML', + disable_notification: bool = True + ) -> dict: + """Send photo. See self.__send_media(). + """ + if Do.args_valid(locals(), self.send_photo.__annotations__): + return self.__send_media( + chat=chat, + media_path=photo, + media_type='photo', + caption=caption, + parse_mode=parse_mode, + disable_notification=disable_notification, + additional_url_param=None + ) + + def send_video( + self, + chat: str, + video: str, + width: (int, type(None)) = None, + height: (int, type(None)) = None, + duration: (int, type(None)) = None, + caption: (str, type(None)) = None, + parse_mode: str = 'HTML', + disable_notification: bool = True + ) -> dict: + """Send video. See self.__send_media(). + """ + if Do.args_valid(locals(), self.send_video.__annotations__): + + if width or height or duration: + additional_url_param = '' + if width: + additional_url_param += '&width=' + str(width) + if height: + additional_url_param += '&height=' + str(height) + if duration: + additional_url_param += '&duration=' + str(duration) + else: + additional_url_param = None + + return self.__send_media( + chat=chat, + media_path=video, + media_type='video', + caption=caption, + parse_mode=parse_mode, + disable_notification=disable_notification, + additional_url_param=additional_url_param + ) + + def send_audio( + self, + chat: str, + audio: str, + caption: (str, type(None)) = None, + parse_mode: str = 'HTML', + disable_notification: bool = True + ) -> dict: + """Send audio. See self.__send_media(). + """ + if Do.args_valid(locals(), self.send_audio.__annotations__): + return self.__send_media( + chat=chat, + media_path=audio, + media_type='audio', + caption=caption, + parse_mode=parse_mode, + disable_notification=disable_notification, + additional_url_param=None + ) + + def send_mediagroup( + self, + chat: str, + media: dict, + caption: (str, type(None)) = None, + parse_mode: str = 'HTML', + disable_notification: bool = True + ) -> dict: + """Send media group of photo, video, audio, documents. + + Args: + chat (str): unique identifier for the target chat or username of the target channel. + media (dict): { + name:{'type':'photo',path:'https://url/to/file',caption:text}, + name:{'type':'video',path:'/local/path/to/file',caption:text}, + name:{'type':'audio',path:'file_id=EXISTFILEID',caption:text}, + }. + caption (str, type, optional): media caption less 1024 characters. Defaults to None. + parse_mode (str): caption 'HTML', 'Markdown', 'MarkdownV2' parse mode. + disable_notification (bool, optional): send silently. Defaults to True. + + Returns: + dict: {'success':bool,'result':response}. + """ + if Do.args_valid(locals(), self.send_mediagroup.__annotations__): + url=self.api_path + '/sendMediaGroup' + files = {} + group = [] + + for media_name in media.keys(): + if re.match("^(?:http://|https://|file_id=)", media[media_name]['path']): + files[media_name] = None + media_source = media[media_name]['path'].replace('file_id=', '') + else: + with open(media[media_name]['path'], mode='rb') as file: + files[media_name] = file.read() + media_source = "attach://" + media_name + + if not caption and media[media_name]['caption']: + media_caption = media[media_name]['caption'] + else: + media_caption = '' + + group.append({ + "type": media[media_name]['type'], + "media": media_source, + "caption": media_caption + } + ) + + if caption: + group[0]['caption'] = caption + group[0]['parse_mode'] = parse_mode + + data = { + 'chat_id': chat, + 'media': json.dumps(group), + "disable_notification": disable_notification + } + + response = requests.post(url=url, data=data, files=files, timeout=300) + if response.status_code == 200: + response = response.json() + logging.info(msg="" + + "mediagroup '" + + str(response['result'][0]['media_group_id']) + + "' sent to telegram chat " + + str(chat) + ) + return {'success': True, 'result': response} + logging.warning(msg="mediagroup didn't send to telegram chat " + str(chat)) + return {'success': False, 'result': response} + + class Sequence: """Sequence handling. """