add Telegram() class

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.
     """