python-telegram-bot/telegram/bot.py

#!/usr/bin/env python
# encoding: utf-8

"""A library that provides a Python interface to the Telegram Bot API"""

import json
try:
    from urllib.parse import urlencode
    from urllib.request import urlopen, Request
    from urllib.error import HTTPError, URLError
except ImportError:
    from urllib import urlencode
    from urllib2 import urlopen, Request
    from urllib2 import HTTPError, URLError
import functools
import logging

from telegram import (User, Message, Update, UserProfilePhotos, TelegramError,
                      ReplyMarkup, InputFile, TelegramObject, NullHandler)

h = NullHandler()
logging.getLogger(__name__).addHandler(h)


class Bot(TelegramObject):

    def __init__(self,
                 token,
                 base_url=None):
        self.token = token

        if base_url is None:
            self.base_url = 'https://api.telegram.org/bot%s' % self.token
        else:
            self.base_url = base_url + self.token

        self.log = logging.getLogger(__name__)

        try:
            bot = self.getMe()

            self.id = bot.id
            self.first_name = bot.first_name
            self.last_name = bot.last_name
            self.username = bot.username

            self.__auth = True

            self.log.info('Starting bot %s' % self.name)
        except TelegramError:
            raise TelegramError({'message': 'Bad token'})

    @property
    def name(self):
        return '@%s' % self.username

    def log(func):
        logger = logging.getLogger(func.__module__)

        @functools.wraps(func)
        def decorator(self, *args, **kwargs):
            logger.debug('Entering: %s' % func.__name__)
            result = func(self, *args, **kwargs)
            logger.debug(result)
            logger.debug('Exiting: %s' % func.__name__)
            return result
        return decorator

    def message(func):
        """
        Returns:
          A telegram.Message instance representing the message posted.
        """
        @functools.wraps(func)
        def decorator(self, *args, **kwargs):
            url, data = func(self, *args, **kwargs)

            if kwargs.get('reply_to_message_id'):
                reply_to_message_id = kwargs.get('reply_to_message_id')
                data['reply_to_message_id'] = reply_to_message_id

            if kwargs.get('reply_markup'):
                reply_markup = kwargs.get('reply_markup')
                if isinstance(reply_markup, ReplyMarkup):
                    data['reply_markup'] = reply_markup.to_json()
                else:
                    data['reply_markup'] = reply_markup

            json_data = self._requestUrl(url, 'POST', data=data)
            data = self._parseAndCheckTelegram(json_data)

            if data is True:
                return data

            return Message.de_json(data)
        return decorator

    def require_authentication(func):
        @functools.wraps(func)
        def decorator(self, *args, **kwargs):
            if not self.__auth:
                raise TelegramError({'message': "API must be authenticated."})

            return func(self, *args, **kwargs)
        return decorator

    @log
    @require_authentication
    def clearCredentials(self):
        """Clear any credentials for this instance.
        """
        self.__auth = False

    @log
    def getMe(self):
        """A simple method for testing your bot's auth token.

        Returns:
          A telegram.User instance representing that bot if the
          credentials are valid, None otherwise.
        """
        url = '%s/getMe' % self.base_url

        json_data = self._requestUrl(url, 'GET')
        data = self._parseAndCheckTelegram(json_data)

        return User.de_json(data)

    @log
    @message
    @require_authentication
    def sendMessage(self,
                    chat_id,
                    text,
                    disable_web_page_preview=None,
                    reply_to_message_id=None,
                    reply_markup=None):
        """Use this method to send text messages.

        Args:
          chat_id:
            Unique identifier for the message recipient — telegram.User or
            telegram.GroupChat id.
          text:
            Text of the message to be sent.
          disable_web_page_preview:
            Disables link previews for links in this message. [Optional]
          reply_to_message_id:
            If the message is a reply, ID of the original message. [Optional]
          reply_markup:
            Additional interface options. A JSON-serialized object for a custom
            reply keyboard, instructions to hide keyboard or to force a reply
            from the user. [Optional]

        Returns:
          A telegram.Message instance representing the message posted.
        """

        url = '%s/sendMessage' % self.base_url

        data = {'chat_id': chat_id,
                'text': text}

        if disable_web_page_preview:
            data['disable_web_page_preview'] = disable_web_page_preview

        return url, data

    @log
    @message
    @require_authentication
    def forwardMessage(self,
                       chat_id,
                       from_chat_id,
                       message_id):
        """Use this method to forward messages of any kind.

        Args:
          chat_id:
            Unique identifier for the message recipient — User or GroupChat id.
          from_chat_id:
            Unique identifier for the chat where the original message was sent
            — User or GroupChat id.
          message_id:
            Unique message identifier.

        Returns:
          A telegram.Message instance representing the message forwarded.
        """

        url = '%s/forwardMessage' % self.base_url

        data = {}
        if chat_id:
            data['chat_id'] = chat_id
        if from_chat_id:
            data['from_chat_id'] = from_chat_id
        if message_id:
            data['message_id'] = message_id

        return url, data

    @log
    @message
    @require_authentication
    def sendPhoto(self,
                  chat_id,
                  photo,
                  caption=None,
                  reply_to_message_id=None,
                  reply_markup=None):
        """Use this method to send photos.

        Args:
          chat_id:
            Unique identifier for the message recipient — User or GroupChat id.
          photo:
            Photo to send. You can either pass a file_id as String to resend a
            photo that is already on the Telegram servers, or upload a new
            photo using multipart/form-data.
          caption:
            Photo caption (may also be used when resending photos by file_id).
            [Optional]
          reply_to_message_id:
            If the message is a reply, ID of the original message. [Optional]
          reply_markup:
            Additional interface options. A JSON-serialized object for a custom
            reply keyboard, instructions to hide keyboard or to force a reply
            from the user. [Optional]

        Returns:
          A telegram.Message instance representing the message posted.
        """

        url = '%s/sendPhoto' % self.base_url

        data = {'chat_id': chat_id,
                'photo': photo}

        if caption:
            data['caption'] = caption

        return url, data

    @log
    @message
    @require_authentication
    def sendAudio(self,
                  chat_id,
                  audio,
                  reply_to_message_id=None,
                  reply_markup=None):
        """Use this method to send audio files, if you want Telegram clients to
        display the file as a playable voice message. For this to work, your
        audio must be in an .ogg file encoded with OPUS (other formats may be
        sent as telegram.Document).

        Args:
          chat_id:
            Unique identifier for the message recipient — User or GroupChat id.
          audio:
            Audio file to send. You can either pass a file_id as String to
            resend an audio that is already on the Telegram servers, or upload
            a new audio file using multipart/form-data.
          reply_to_message_id:
            If the message is a reply, ID of the original message. [Optional]
          reply_markup:
            Additional interface options. A JSON-serialized object for a
            custom reply keyboard, instructions to hide keyboard or to force a
            reply from the user. [Optional]

        Returns:
          A telegram.Message instance representing the message posted.
        """

        url = '%s/sendAudio' % self.base_url

        data = {'chat_id': chat_id,
                'audio': audio}

        return url, data

    @log
    @message
    @require_authentication
    def sendDocument(self,
                     chat_id,
                     document,
                     reply_to_message_id=None,
                     reply_markup=None):
        """Use this method to send general files.

        Args:
          chat_id:
            Unique identifier for the message recipient — User or GroupChat id.
          document:
            File to send. You can either pass a file_id as String to resend a
            file that is already on the Telegram servers, or upload a new file
            using multipart/form-data.
          reply_to_message_id:
            If the message is a reply, ID of the original message. [Optional]
          reply_markup:
            Additional interface options. A JSON-serialized object for a
            custom reply keyboard, instructions to hide keyboard or to force a
            reply from the user. [Optional]

        Returns:
          A telegram.Message instance representing the message posted.
        """

        url = '%s/sendDocument' % self.base_url

        data = {'chat_id': chat_id,
                'document': document}

        return url, data

    @log
    @message
    @require_authentication
    def sendSticker(self,
                    chat_id,
                    sticker,
                    reply_to_message_id=None,
                    reply_markup=None):
        """Use this method to send .webp stickers.

        Args:
          chat_id:
            Unique identifier for the message recipient — User or GroupChat id.
          sticker:
            Sticker to send. You can either pass a file_id as String to resend
            a sticker that is already on the Telegram servers, or upload a new
            sticker using multipart/form-data.
          reply_to_message_id:
            If the message is a reply, ID of the original message. [Optional]
          reply_markup:
            Additional interface options. A JSON-serialized object for a
            custom reply keyboard, instructions to hide keyboard or to force a
            reply from the user. [Optional]

        Returns:
          A telegram.Message instance representing the message posted.
        """

        url = '%s/sendSticker' % self.base_url

        data = {'chat_id': chat_id,
                'sticker': sticker}

        return url, data

    @log
    @message
    @require_authentication
    def sendVideo(self,
                  chat_id,
                  video,
                  reply_to_message_id=None,
                  reply_markup=None):
        """Use this method to send video files, Telegram clients support mp4
        videos (other formats may be sent as telegram.Document).

        Args:
          chat_id:
            Unique identifier for the message recipient — User or GroupChat id.
          video:
            Video to send. You can either pass a file_id as String to resend a
            video that is already on the Telegram servers, or upload a new
            video file using multipart/form-data.
          reply_to_message_id:
            If the message is a reply, ID of the original message. [Optional]
          reply_markup:
            Additional interface options. A JSON-serialized object for a
            custom reply keyboard, instructions to hide keyboard or to force a
            reply from the user. [Optional]

        Returns:
          A telegram.Message instance representing the message posted.
        """

        url = '%s/sendVideo' % self.base_url

        data = {'chat_id': chat_id,
                'video': video}

        return url, data

    @log
    @message
    @require_authentication
    def sendLocation(self,
                     chat_id,
                     latitude,
                     longitude,
                     reply_to_message_id=None,
                     reply_markup=None):
        """Use this method to send point on the map.

        Args:
          chat_id:
            Unique identifier for the message recipient — User or GroupChat id.
          latitude:
            Latitude of location.
          longitude:
            Longitude of location.
          reply_to_message_id:
            If the message is a reply, ID of the original message. [Optional]
          reply_markup:
            Additional interface options. A JSON-serialized object for a
            custom reply keyboard, instructions to hide keyboard or to force a
            reply from the user. [Optional]

        Returns:
          A telegram.Message instance representing the message posted.
        """

        url = '%s/sendLocation' % self.base_url

        data = {'chat_id': chat_id,
                'latitude': latitude,
                'longitude': longitude}

        return url, data

    @log
    @message
    @require_authentication
    def sendChatAction(self,
                       chat_id,
                       action):
        """Use this method when you need to tell the user that something is
        happening on the bot's side. The status is set for 5 seconds or less
        (when a message arrives from your bot, Telegram clients clear its
        typing status).

        Args:
          chat_id:
            Unique identifier for the message recipient — User or GroupChat id.
          action:
            Type of action to broadcast. Choose one, depending on what the user
            is about to receive:
            - ChatAction.TYPING for text messages,
            - ChatAction.UPLOAD_PHOTO for photos,
            - ChatAction.UPLOAD_VIDEO or upload_video for videos,
            - ChatAction.UPLOAD_AUDIO or upload_audio for audio files,
            - ChatAction.UPLOAD_DOCUMENT for general files,
            - ChatAction.FIND_LOCATION for location data.
        """

        url = '%s/sendChatAction' % self.base_url

        data = {'chat_id': chat_id,
                'action': action}

        return url, data

    @log
    @require_authentication
    def getUserProfilePhotos(self,
                             user_id,
                             offset=None,
                             limit=100):
        """Use this method to get a list of profile pictures for a user.

        Args:
          user_id:
            Unique identifier of the target user.
          offset:
            Sequential number of the first photo to be returned. By default,
            all photos are returned. [Optional]
          limit:
            Limits the number of photos to be retrieved. Values between 1—100
            are accepted. Defaults to 100. [Optional]

        Returns:
          Returns a telegram.UserProfilePhotos object.
        """

        url = '%s/getUserProfilePhotos' % self.base_url

        data = {'user_id': user_id}

        if offset:
            data['offset'] = offset
        if limit:
            data['limit'] = limit

        json_data = self._requestUrl(url, 'POST', data=data)
        data = self._parseAndCheckTelegram(json_data)

        return UserProfilePhotos.de_json(data)

    @log
    @require_authentication
    def getUpdates(self,
                   offset=None,
                   limit=100,
                   timeout=0):
        """Use this method to receive incoming updates using long polling.

        Args:
          offset:
            Identifier of the first update to be returned. Must be greater by
            one than the highest among the identifiers of previously received
            updates. By default, updates starting with the earliest unconfirmed
            update are returned. An update is considered confirmed as soon as
            getUpdates is called with an offset higher than its update_id.
          limit:
            Limits the number of updates to be retrieved. Values between 1—100
            are accepted. Defaults to 100.
          timeout:
            Timeout in seconds for long polling. Defaults to 0, i.e. usual
            short polling.

        Returns:
          A list of telegram.Update objects are returned.
        """

        url = '%s/getUpdates' % self.base_url

        data = {}
        if offset:
            data['offset'] = offset
        if limit:
            data['limit'] = limit
        if timeout:
            data['timeout'] = timeout

        json_data = self._requestUrl(url, 'POST', data=data)
        data = self._parseAndCheckTelegram(json_data)

        if data:
            self.log.info(
                'Getting updates: %s' % [u['update_id'] for u in data])
        else:
            self.log.info('No new updates found.')

        return [Update.de_json(x) for x in data]

    @log
    @require_authentication
    def setWebhook(self,
                   webhook_url):
        """Use this method to specify a url and receive incoming updates via an
        outgoing webhook. Whenever there is an update for the bot, we will send
        an HTTPS POST request to the specified url, containing a
        JSON-serialized Update. In case of an unsuccessful request, we will
        give up after a reasonable amount of attempts.

        Args:
          url:
            HTTPS url to send updates to.
            Use an empty string to remove webhook integration

        Returns:
          True if successful else TelegramError was raised
        """
        url = '%s/setWebhook' % self.base_url

        data = {'url': webhook_url}

        json_data = self._requestUrl(url, 'POST', data=data)
        data = self._parseAndCheckTelegram(json_data)

        return True

    def _requestUrl(self,
                    url,
                    method,
                    data=None):
        """Request an URL.

        Args:
          url:
            The web location we want to retrieve.
          method:
            Either POST or GET.
          data:
            A dict of (str, unicode) key/value pairs.

        Returns:
          A JSON object.
        """
        if method not in ('POST', 'GET'):
            raise ValueError(
                "Method '%s' is neither 'POST' nor 'GET'" % method)

        if method == 'POST':
            try:
                if InputFile.is_inputfile(data):
                    data = InputFile(data)

                    request = Request(
                        url,
                        data=data.to_form(),
                        headers=data.headers
                    )

                    return urlopen(request).read()
                else:
                    return urlopen(
                        url,
                        urlencode(data).encode()
                    ).read()
            except IOError as e:
                raise TelegramError(str(e))
            except HTTPError as e:
                raise TelegramError(str(e))
            except URLError as e:
                raise TelegramError(str(e))

        if method == 'GET':
            try:
                return urlopen(url).read()
            except URLError as e:
                raise TelegramError(str(e))

    def _parseAndCheckTelegram(self,
                               json_data):
        """Try and parse the JSON returned from Telegram and return an empty
        dictionary if there is any error.

        Args:
          json_data:
            JSON results from Telegram Bot API.

        Returns:
          A JSON parsed as Python dict with results.
        """

        try:
            data = json.loads(json_data.decode())
            if not data['ok']:
                raise TelegramError(data['description'])
        except ValueError:
            if '<title>403 Forbidden</title>' in json_data:
                raise TelegramError({'message': 'API must be authenticated'})
            raise TelegramError({'message': 'JSON decoding'})

        return data['result']

    def to_dict(self):
        data = {'id': self.id,
                'username': self.username,
                'first_name': self.username}
        if self.last_name:
            data['last_name'] = self.last_name
        return data