diff --git a/davtelepot/api.py b/davtelepot/api.py new file mode 100644 index 0000000..de899e4 --- /dev/null +++ b/davtelepot/api.py @@ -0,0 +1,1167 @@ +"""This module provides a glow-like middleware for Telegram bot API. + +All methods and parameters are the same as the original json API. +A simple aiohttp asyncronous web client is used to make requests. +""" + +# Standard library modules +import asyncio +import logging + +# Third party modules +import aiohttp +from aiohttp import web + + +class TelegramError(Exception): + """Telegram API exceptions class.""" + + def __init__(self, error_code=0, description=None, ok=False): + """Get an error response and return corresponding Exception.""" + self._code = error_code + if description is None: + self._description = 'Generic error' + else: + self._description = description + super().__init__(self.description) + + @property + def code(self): + """Telegram error code.""" + return self._code + + @property + def description(self): + """Human-readable description of error.""" + return f"Error {self.code}: {self._description}" + + +class TelegramBot(object): + """Provide python method having the same signature as Telegram API methods. + + All mirrored methods are camelCase. + """ + + loop = asyncio.get_event_loop() + app = web.Application(loop=loop) + sessions_timeouts = { + 'getUpdates': dict( + timeout=35, + close=False + ), + 'sendMessage': dict( + timeout=20, + close=False + ) + } + + def __init__(self, token): + """Set bot token and store HTTP sessions.""" + self._token = token + self.sessions = dict() + + @property + def token(self): + """Telegram API bot token.""" + return self._token + + @staticmethod + def check_telegram_api_json(response): + """Take a json Telegram response, check it and return its content. + + Example of well-formed json Telegram responses: + { + "ok": False, + "error_code": 401, + "description": "Unauthorized" + } + { + "ok": True, + "result": ... + } + """ + assert 'ok' in response, ( + "All Telegram API responses have an `ok` field." + ) + if not response['ok']: + raise TelegramError(**response) + return response['result'] + + @staticmethod + def adapt_parameters(parameters, exclude=[]): + """Build a aiohttp.FormData object from given `paramters`. + + Exclude `self`, empty values and parameters in `exclude` list. + Cast integers to string to avoid TypeError during json serialization. + """ + exclude.append('self') + data = aiohttp.FormData() + for key, value in parameters.items(): + if not (key in exclude or value is None): + if type(value) is int: + value = str(value) + data.add_field(key, value) + return data + + def get_session(self, api_method): + """According to API method, return proper session and information. + + Return a tuple (session, session_must_be_closed) + session : aiohttp.ClientSession + Client session with proper timeout + session_must_be_closed : bool + True if session must be closed after being used once + """ + cls = self.__class__ + if api_method in cls.sessions_timeouts: + if api_method not in self.sessions: + self.sessions[api_method] = aiohttp.ClientSession( + loop=cls.loop, + timeout=aiohttp.ClientTimeout( + total=cls.sessions_timeouts[api_method]['timeout'] + ) + ) + session = self.sessions[api_method] + session_must_be_closed = cls.sessions_timeouts[api_method]['close'] + else: + session = aiohttp.ClientSession( + loop=cls.loop, + timeout=aiohttp.ClientTimeout(total=None) + ) + session_must_be_closed = True + return session, session_must_be_closed + + async def api_request(self, method, parameters={}, exclude=[]): + """Return the result of a Telegram bot API request, or an Exception. + + Opened sessions will be used more than one time (if appropriate) and + will be closed on `Bot.app.cleanup`. + Result may be a Telegram API json response, None, or Exception. + """ + response_object = None + session, session_must_be_closed = self.get_session(method) + parameters = self.adapt_parameters(parameters, exclude=exclude) + try: + async with session.post( + "https://api.telegram.org/bot" + f"{self.token}/{method}", + data=parameters + ) as response: + try: + response_object = self.check_telegram_api_json( + await response.json() # Telegram returns json objects + ) + except TelegramError as e: + logging.error(f"{e}") + return e + except Exception as e: + logging.error(f"{e}", exc_info=True) + return e + except asyncio.TimeoutError as e: + logging.info(f"{e}: {method} API call timed out") + finally: + if session_must_be_closed: + await session.close() + return response_object + + async def getMe(self): + """Get basic information about the bot in form of a User object. + + Useful to test `self.token`. + See https://core.telegram.org/bots/api#getme for details. + """ + return await self.api_request( + 'getMe', + ) + + async def getUpdates(self, offset, timeout, limit, allowed_updates): + """Get a list of updates starting from `offset`. + + If there are no updates, keep the request hanging until `timeout`. + If there are more than `limit` updates, retrieve them in packs of + `limit`. + Allowed update types (empty list to allow all). + See https://core.telegram.org/bots/api#getupdates for details. + """ + return await self.api_request( + method='getUpdates', + parameters=locals() + ) + + async def setWebhook(self, url=None, certificate=None, + max_connections=None, allowed_updates=None): + """Set or remove a webhook. Telegram will post to `url` new updates. + + See https://core.telegram.org/bots/api#setwebhook for details. + """ + if url is None: + url = self.webhook_url + if allowed_updates is None: + allowed_updates = self.allowed_updates + if max_connections is None: + max_connections = self.max_connections + if certificate is None: + certificate = self.certificate + if type(certificate) is str: + try: + certificate = open(certificate, 'r') + except FileNotFoundError as e: + logging.error(f"{e}") + certificate = None + certificate = dict( + file=certificate + ) + return await self.api_request( + 'setWebhook', + parameters=locals() + ) + + async def deleteWebhook(self): + """Remove webhook integration and switch back to getUpdate. + + See https://core.telegram.org/bots/api#deletewebhook for details. + """ + return await self.api_request( + 'deleteWebhook', + ) + + async def getWebhookInfo(self): + """Get current webhook status. + + See https://core.telegram.org/bots/api#getwebhookinfo for details. + """ + return await self.api_request( + 'getWebhookInfo', + ) + + async def sendMessage(self, chat_id, text, + parse_mode=None, + disable_web_page_preview=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a text message. On success, return it. + + See https://core.telegram.org/bots/api#sendmessage for details. + """ + return await self.api_request( + 'sendMessage', + parameters=locals() + ) + + async def forwardMessage(self, chat_id, from_chat_id, message_id, + disable_notification=None): + """Forward a message. + + See https://core.telegram.org/bots/api#forwardmessage for details. + """ + return await self.api_request( + 'forwardMessage', + parameters=locals() + ) + + async def sendPhoto(self, chat_id, photo, + caption=None, + parse_mode=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a photo from file_id, HTTP url or file. + + See https://core.telegram.org/bots/api#sendphoto for details. + """ + return await self.api_request( + 'sendPhoto', + parameters=locals() + ) + + async def sendAudio(self, chat_id, audio, + caption=None, + parse_mode=None, + duration=None, + performer=None, + title=None, + thumb=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send an audio file from file_id, HTTP url or file. + + See https://core.telegram.org/bots/api#sendaudio for details. + """ + return await self.api_request( + 'sendAudio', + parameters=locals() + ) + + async def sendDocument(self, chat_id, document, + thumb=None, + caption=None, + parse_mode=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a document from file_id, HTTP url or file. + + See https://core.telegram.org/bots/api#senddocument for details. + """ + return await self.api_request( + 'sendDocument', + parameters=locals() + ) + + async def sendVideo(self, chat_id, video, + duration=None, + width=None, + height=None, + thumb=None, + caption=None, + parse_mode=None, + supports_streaming=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a video from file_id, HTTP url or file. + + See https://core.telegram.org/bots/api#sendvideo for details. + """ + return await self.api_request( + 'sendVideo', + parameters=locals() + ) + + async def sendAnimation(self, chat_id, animation, + duration=None, + width=None, + height=None, + thumb=None, + caption=None, + parse_mode=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send animation files (GIF or H.264/MPEG-4 AVC video without sound). + + See https://core.telegram.org/bots/api#sendanimation for details. + """ + return await self.api_request( + 'sendAnimation', + parameters=locals() + ) + + async def sendVoice(self, chat_id, voice, + caption=None, + parse_mode=None, + duration=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send an audio file to be displayed as playable voice message. + + `voice` must be in an .ogg file encoded with OPUS. + See https://core.telegram.org/bots/api#sendvoice for details. + """ + return await self.api_request( + 'sendVoice', + parameters=locals() + ) + + async def sendVideoNote(self, chat_id, video_note, + duration=None, + length=None, + thumb=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a rounded square mp4 video message of up to 1 minute long. + + See https://core.telegram.org/bots/api#sendvideonote for details. + """ + return await self.api_request( + 'sendVideoNote', + parameters=locals() + ) + + async def sendMediaGroup(self, chat_id, media, + disable_notification=None, + reply_to_message_id=None): + """Send a group of photos or videos as an album. + + `media` must be a list of `InputMediaPhoto` and/or `InputMediaVideo` + objects. + See https://core.telegram.org/bots/api#sendmediagroup for details. + """ + return await self.api_request( + 'sendMediaGroup', + parameters=locals() + ) + + async def sendLocation(self, chat_id, latitude, longitude, + live_period=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a point on the map. May be kept updated for a `live_period`. + + See https://core.telegram.org/bots/api#sendlocation for details. + """ + return await self.api_request( + 'sendLocation', + parameters=locals() + ) + + async def editMessageLiveLocation(self, latitude, longitude, + chat_id=None, message_id=None, + inline_message_id=None, + reply_markup=None): + """Edit live location messages. + + A location can be edited until its live_period expires or editing is + explicitly disabled by a call to stopMessageLiveLocation. + The message to be edited may be identified through `inline_message_id` + OR the couple (`chat_id`, `message_id`). + See https://core.telegram.org/bots/api#editmessagelivelocation + for details. + """ + return await self.api_request( + 'editMessageLiveLocation', + parameters=locals() + ) + + async def stopMessageLiveLocation(self, + chat_id=None, message_id=None, + inline_message_id=None, + reply_markup=None): + """Stop updating a live location message before live_period expires. + + The position to be stopped may be identified through + `inline_message_id` OR the couple (`chat_id`, `message_id`). + `reply_markup` type may be only `InlineKeyboardMarkup`. + See https://core.telegram.org/bots/api#stopmessagelivelocation + for details. + """ + return await self.api_request( + 'stopMessageLiveLocation', + parameters=locals() + ) + + async def sendVenue(self, chat_id, latitude, longitude, title, address, + foursquare_id=None, + foursquare_type=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send information about a venue. + + Integrated with FourSquare. + See https://core.telegram.org/bots/api#sendvenue for details. + """ + return await self.api_request( + 'sendVenue', + parameters=locals() + ) + + async def sendContact(self, chat_id, phone_number, first_name, + last_name=None, + vcard=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a phone contact. + + See https://core.telegram.org/bots/api#sendcontact for details. + """ + return await self.api_request( + 'sendContact', + parameters=locals() + ) + + async def sendPoll(self, chat_id, question, options, + dummy=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a native poll in a group, a supergroup or channel. + + See https://core.telegram.org/bots/api#sendpoll for details. + """ + return await self.api_request( + 'sendPoll', + parameters=locals() + ) + + async def sendChatAction(self, chat_id, action): + """Fake a typing status or similar. + + See https://core.telegram.org/bots/api#sendchataction for details. + """ + return await self.api_request( + 'sendChatAction', + parameters=locals() + ) + + async def getUserProfilePhotos(self, user_id, + offset=None, + limit=None,): + """Get a list of profile pictures for a user. + + See https://core.telegram.org/bots/api#getuserprofilephotos + for details. + """ + return await self.api_request( + 'getUserProfilePhotos', + parameters=locals() + ) + + async def getFile(self, file_id): + """Get basic info about a file and prepare it for downloading. + + For the moment, bots can download files of up to + 20MB in size. + On success, a File object is returned. The file can then be downloaded + via the link https://api.telegram.org/file/bot/, + where is taken from the response. + + See https://core.telegram.org/bots/api#getfile for details. + """ + return await self.api_request( + 'getFile', + parameters=locals() + ) + + async def kickChatMember(self, chat_id, user_id, + until_date=None): + """Kick a user from a group, a supergroup or a channel. + + In the case of supergroups and channels, the user will not be able to + return to the group on their own using invite links, etc., unless + unbanned first. + Note: In regular groups (non-supergroups), this method will only work + if the ‘All Members Are Admins’ setting is off in the target group. + Otherwise members may only be removed by the group's creator or by + the member that added them. + See https://core.telegram.org/bots/api#kickchatmember for details. + """ + return await self.api_request( + 'kickChatMember', + parameters=locals() + ) + + async def unbanChatMember(self, chat_id, user_id): + """Unban a previously kicked user in a supergroup or channel. + + The user will not return to the group or channel automatically, but + will be able to join via link, etc. + The bot must be an administrator for this to work. + Return True on success. + See https://core.telegram.org/bots/api#unbanchatmember for details. + """ + return await self.api_request( + 'unbanChatMember', + parameters=locals() + ) + + async def restrictChatMember(self, chat_id, user_id, + until_date=None, + can_send_messages=None, + can_send_media_messages=None, + can_send_other_messages=None, + can_add_web_page_previews=None): + """Restrict a user in a supergroup. + + The bot must be an administrator in the supergroup for this to work + and must have the appropriate admin rights. + Pass True for all boolean parameters to lift restrictions from a + user. + Return True on success. + See https://core.telegram.org/bots/api#restrictchatmember for details. + """ + return await self.api_request( + 'restrictChatMember', + parameters=locals() + ) + + async def promoteChatMember(self, chat_id, user_id, + can_change_info=None, + can_post_messages=None, + can_edit_messages=None, + can_delete_messages=None, + can_invite_users=None, + can_restrict_members=None, + can_pin_messages=None, + can_promote_members=None): + """Promote or demote a user in a supergroup or a channel. + + The bot must be an administrator in the chat for this to work and must + have the appropriate admin rights. + Pass False for all boolean parameters to demote a user. + Return True on success. + See https://core.telegram.org/bots/api#promotechatmember for details. + """ + return await self.api_request( + 'promoteChatMember', + parameters=locals() + ) + + async def exportChatInviteLink(self, chat_id): + """Generate a new invite link for a chat and revoke any active link. + + The bot must be an administrator in the chat for this to work and must + have the appropriate admin rights. + Return the new invite link as String on success. + NOTE: to get the current invite link, use `getChat` method. + See https://core.telegram.org/bots/api#exportchatinvitelink + for details. + """ + return await self.api_request( + 'exportChatInviteLink', + parameters=locals() + ) + + async def setChatPhoto(self, chat_id, photo): + """Set a new profile photo for the chat. + + Photos can't be changed for private chats. + `photo` must be an input file (file_id and urls are not allowed). + The bot must be an administrator in the chat for this to work and must + have the appropriate admin rights. + Return True on success. + See https://core.telegram.org/bots/api#setchatphoto for details. + """ + return await self.api_request( + 'setChatPhoto', + parameters=locals() + ) + + async def deleteChatPhoto(self, chat_id): + """Delete a chat photo. + + Photos can't be changed for private chats. + The bot must be an administrator in the chat for this to work and must + have the appropriate admin rights. + Return True on success. + See https://core.telegram.org/bots/api#deletechatphoto for details. + """ + return await self.api_request( + 'deleteChatPhoto', + parameters=locals() + ) + + async def setChatTitle(self, chat_id, title): + """Change the title of a chat. + + Titles can't be changed for private chats. + The bot must be an administrator in the chat for this to work and must + have the appropriate admin rights. + Return True on success. + See https://core.telegram.org/bots/api#setchattitle for details. + """ + return await self.api_request( + 'setChatTitle', + parameters=locals() + ) + + async def setChatDescription(self, chat_id, description): + """Change the description of a supergroup or a channel. + + The bot must be an administrator in the chat for this to work and must + have the appropriate admin rights. + Return True on success. + See https://core.telegram.org/bots/api#setchatdescription for details. + """ + return await self.api_request( + 'setChatDescription', + parameters=locals() + ) + + async def pinChatMessage(self, chat_id, message_id, + disable_notification=None): + """Pin a message in a group, a supergroup, or a channel. + + The bot must be an administrator in the chat for this to work and must + have the ‘can_pin_messages’ admin right in the supergroup or + ‘can_edit_messages’ admin right in the channel. + Return True on success. + See https://core.telegram.org/bots/api#pinchatmessage for details. + """ + return await self.api_request( + 'pinChatMessage', + parameters=locals() + ) + + async def unpinChatMessage(self, chat_id): + """Unpin a message in a group, a supergroup, or a channel. + + The bot must be an administrator in the chat for this to work and must + have the ‘can_pin_messages’ admin right in the supergroup or + ‘can_edit_messages’ admin right in the channel. + Return True on success. + See https://core.telegram.org/bots/api#unpinchatmessage for details. + """ + return await self.api_request( + 'unpinChatMessage', + parameters=locals() + ) + + async def leaveChat(self, chat_id): + """Make the bot leave a group, supergroup or channel. + + Return True on success. + See https://core.telegram.org/bots/api#leavechat for details. + """ + return await self.api_request( + 'leaveChat', + parameters=locals() + ) + + async def getChat(self, chat_id): + """Get up to date information about the chat. + + Return a Chat object on success. + See https://core.telegram.org/bots/api#getchat for details. + """ + return await self.api_request( + 'getChat', + parameters=locals() + ) + + async def getChatAdministrators(self, chat_id): + """Get a list of administrators in a chat. + + On success, return an Array of ChatMember objects that contains + information about all chat administrators except other bots. + If the chat is a group or a supergroup and no administrators were + appointed, only the creator will be returned. + + See https://core.telegram.org/bots/api#getchatadministrators + for details. + """ + return await self.api_request( + 'getChatAdministrators', + parameters=locals() + ) + + async def getChatMembersCount(self, chat_id): + """Get the number of members in a chat. + + Returns Int on success. + See https://core.telegram.org/bots/api#getchatmemberscount for details. + """ + return await self.api_request( + 'getChatMembersCount', + parameters=locals() + ) + + async def getChatMember(self, chat_id, user_id): + """Get information about a member of a chat. + + Returns a ChatMember object on success. + See https://core.telegram.org/bots/api#getchatmember for details. + """ + return await self.api_request( + 'getChatMember', + parameters=locals() + ) + + async def setChatStickerSet(self, chat_id, sticker_set_name): + """Set a new group sticker set for a supergroup. + + The bot must be an administrator in the chat for this to work and must + have the appropriate admin rights. + Use the field `can_set_sticker_set` optionally returned in getChat + requests to check if the bot can use this method. + Returns True on success. + See https://core.telegram.org/bots/api#setchatstickerset for details. + """ + return await self.api_request( + 'setChatStickerSet', + parameters=locals() + ) + + async def deleteChatStickerSet(self, chat_id): + """Delete a group sticker set from a supergroup. + + The bot must be an administrator in the chat for this to work and must + have the appropriate admin rights. + Use the field `can_set_sticker_set` optionally returned in getChat + requests to check if the bot can use this method. + Returns True on success. + See https://core.telegram.org/bots/api#deletechatstickerset for + details. + """ + return await self.api_request( + 'deleteChatStickerSet', + parameters=locals() + ) + + async def answerCallbackQuery(self, callback_query_id, + text=None, + show_alert=None, + url=None, + cache_time=None): + """Send answers to callback queries sent from inline keyboards. + + The answer will be displayed to the user as a notification at the top + of the chat screen or as an alert. + On success, True is returned. + See https://core.telegram.org/bots/api#answercallbackquery for details. + """ + return await self.api_request( + 'answerCallbackQuery', + parameters=locals() + ) + + async def editMessageText(self, text, + chat_id=None, message_id=None, + inline_message_id=None, + parse_mode=None, + disable_web_page_preview=None, + reply_markup=None): + """Edit text and game messages. + + On success, if edited message is sent by the bot, the edited Message + is returned, otherwise True is returned. + See https://core.telegram.org/bots/api#editmessagetext for details. + """ + return await self.api_request( + 'editMessageText', + parameters=locals() + ) + + async def editMessageCaption(self, + chat_id=None, message_id=None, + inline_message_id=None, + caption=None, + parse_mode=None, + reply_markup=None): + """Edit captions of messages. + + On success, if edited message is sent by the bot, the edited Message is + returned, otherwise True is returned. + See https://core.telegram.org/bots/api#editmessagecaption for details. + """ + return await self.api_request( + 'editMessageCaption', + parameters=locals() + ) + + async def editMessageMedia(self, + chat_id=None, message_id=None, + inline_message_id=None, + media=None, + reply_markup=None): + """Edit animation, audio, document, photo, or video messages. + + If a message is a part of a message album, then it can be edited only + to a photo or a video. Otherwise, message type can be changed + arbitrarily. + When inline message is edited, new file can't be uploaded. + Use previously uploaded file via its file_id or specify a URL. + On success, if the edited message was sent by the bot, the edited + Message is returned, otherwise True is returned. + See https://core.telegram.org/bots/api#editmessagemedia for details. + """ + return await self.api_request( + 'editMessageMedia', + parameters=locals() + ) + + async def editMessageReplyMarkup(self, + chat_id=None, message_id=None, + inline_message_id=None, + reply_markup=None): + """Edit only the reply markup of messages. + + On success, if edited message is sent by the bot, the edited Message is + returned, otherwise True is returned. + See https://core.telegram.org/bots/api#editmessagereplymarkup for + details. + """ + return await self.api_request( + 'editMessageReplyMarkup', + parameters=locals() + ) + + async def stopPoll(self, chat_id, message_id, + reply_markup=None): + """Stop a poll which was sent by the bot. + + On success, the stopped Poll with the final results is returned. + `reply_markup` type may be only `InlineKeyboardMarkup`. + See https://core.telegram.org/bots/api#stoppoll for details. + """ + return await self.api_request( + 'stopPoll', + parameters=locals() + ) + + async def deleteMessage(self, chat_id, message_id): + """Delete a message, including service messages. + + - A message can only be deleted if it was sent less than 48 hours + ago. + - Bots can delete outgoing messages in private chats, groups, and + supergroups. + - Bots can delete incoming messages in private chats. + - Bots granted can_post_messages permissions can delete outgoing + messages in channels. + - If the bot is an administrator of a group, it can delete any + message there. + - If the bot has can_delete_messages permission in a supergroup or + a channel, it can delete any message there. + Returns True on success. + + See https://core.telegram.org/bots/api#deletemessage for details. + """ + return await self.api_request( + 'deleteMessage', + parameters=locals() + ) + + async def sendSticker(self, chat_id, sticker, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send .webp stickers. + + On success, the sent Message is returned. + See https://core.telegram.org/bots/api#sendsticker for details. + """ + return await self.api_request( + 'sendSticker', + parameters=locals() + ) + + async def getStickerSet(self, name): + """Get a sticker set. + + On success, a StickerSet object is returned. + See https://core.telegram.org/bots/api#getstickerset for details. + """ + return await self.api_request( + 'getStickerSet', + parameters=locals() + ) + + async def uploadStickerFile(self, user_id, png_sticker): + """Upload a .png file as a sticker. + + Use it later via `createNewStickerSet` and `addStickerToSet` methods + (can be used multiple times). + Return the uploaded File on success. + `png_sticker` must be a *.png image up to 512 kilobytes in size, + dimensions must not exceed 512px, and either width or height must + be exactly 512px. + See https://core.telegram.org/bots/api#uploadstickerfile for details. + """ + return await self.api_request( + 'uploadStickerFile', + parameters=locals() + ) + + async def createNewStickerSet(self, user_id, + name, title, png_sticker, emojis, + contains_masks=None, + mask_position=None): + """Create new sticker set owned by a user. + + The bot will be able to edit the created sticker set. + Returns True on success. + See https://core.telegram.org/bots/api#createnewstickerset for details. + """ + return await self.api_request( + 'createNewStickerSet', + parameters=locals() + ) + + async def addStickerToSet(self, user_id, name, png_sticker, emojis, + mask_position=None): + """Add a new sticker to a set created by the bot. + + Returns True on success. + See https://core.telegram.org/bots/api#addstickertoset for details. + """ + return await self.api_request( + 'addStickerToSet', + parameters=locals() + ) + + async def setStickerPositionInSet(self, sticker, position): + """Move a sticker in a set created by the bot to a specific position . + + Position is 0-based. + Returns True on success. + See https://core.telegram.org/bots/api#setstickerpositioninset for + details. + """ + return await self.api_request( + 'setStickerPositionInSet', + parameters=locals() + ) + + async def deleteStickerFromSet(self, sticker): + """Delete a sticker from a set created by the bot. + + Returns True on success. + See https://core.telegram.org/bots/api#deletestickerfromset for + details. + """ + return await self.api_request( + 'deleteStickerFromSet', + parameters=locals() + ) + + async def answerInlineQuery(self, inline_query_id, results, + cache_time=None, + is_personal=None, + next_offset=None, + switch_pm_text=None, + switch_pm_parameter=None): + """Send answers to an inline query. + + On success, True is returned. + No more than 50 results per query are allowed. + See https://core.telegram.org/bots/api#answerinlinequery for details. + """ + return await self.api_request( + 'answerInlineQuery', + parameters=locals() + ) + + async def sendInvoice(self, chat_id, title, description, payload, + provider_token, start_parameter, currency, prices, + provider_data=None, + photo_url=None, + photo_size=None, + photo_width=None, + photo_height=None, + need_name=None, + need_phone_number=None, + need_email=None, + need_shipping_address=None, + send_phone_number_to_provider=None, + send_email_to_provider=None, + is_flexible=None, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send an invoice. + + On success, the sent Message is returned. + See https://core.telegram.org/bots/api#sendinvoice for details. + """ + return await self.api_request( + 'sendInvoice', + parameters=locals() + ) + + async def answerShippingQuery(self, shipping_query_id, ok, + shipping_options=None, + error_message=None): + """Reply to shipping queries. + + On success, True is returned. + If you sent an invoice requesting a shipping address and the parameter + is_flexible was specified, the Bot API will send an Update with a + shipping_query field to the bot. + See https://core.telegram.org/bots/api#answershippingquery for details. + """ + return await self.api_request( + 'answerShippingQuery', + parameters=locals() + ) + + async def answerPreCheckoutQuery(self, pre_checkout_query_id, ok, + error_message=None): + """Respond to pre-checkout queries. + + Once the user has confirmed their payment and shipping details, the Bot + API sends the final confirmation in the form of an Update with the + field pre_checkout_query. + On success, True is returned. + Note: The Bot API must receive an answer within 10 seconds after the + pre-checkout query was sent. + See https://core.telegram.org/bots/api#answerprecheckoutquery for + details. + """ + return await self.api_request( + 'answerPreCheckoutQuery', + parameters=locals() + ) + + async def setPassportDataErrors(self, user_id, errors): + """Refuse a Telegram Passport element with `errors`. + + Inform a user that some of the Telegram Passport elements they provided + contains errors. + The user will not be able to re-submit their Passport to you until the + errors are fixed (the contents of the field for which you returned + the error must change). + Returns True on success. + Use this if the data submitted by the user doesn't satisfy the + standards your service requires for any reason. + For example, if a birthday date seems invalid, a submitted document + is blurry, a scan shows evidence of tampering, etc. + Supply some details in the error message to make sure the user knows + how to correct the issues. + See https://core.telegram.org/bots/api#setpassportdataerrors for + details. + """ + return await self.api_request( + 'setPassportDataErrors', + parameters=locals() + ) + + async def sendGame(self, chat_id, game_short_name, + disable_notification=None, + reply_to_message_id=None, + reply_markup=None): + """Send a game. + + On success, the sent Message is returned. + See https://core.telegram.org/bots/api#sendgame for + details. + """ + return await self.api_request( + 'sendGame', + parameters=locals() + ) + + async def setGameScore(self, user_id, score, + force=None, + disable_edit_message=None, + chat_id=None, message_id=None, + inline_message_id=None): + """Set the score of the specified user in a game. + + On success, if the message was sent by the bot, returns the edited + Message, otherwise returns True. + Returns an error, if the new score is not greater than the user's + current score in the chat and force is False. + See https://core.telegram.org/bots/api#setgamescore for + details. + """ + return await self.api_request( + 'setGameScore', + parameters=locals() + ) + + async def getGameHighScores(self, user_id, + chat_id=None, message_id=None, + inline_message_id=None): + """Get data for high score tables. + + Will return the score of the specified user and several of his + neighbors in a game. + On success, returns an Array of GameHighScore objects. + This method will currently return scores for the target user, plus two + of his closest neighbors on each side. Will also return the top + three users if the user and his neighbors are not among them. + Please note that this behavior is subject to change. + See https://core.telegram.org/bots/api#getgamehighscores for + details. + """ + return await self.api_request( + 'getGameHighScores', + parameters=locals() + ) diff --git a/davtelepot/bot.py b/davtelepot/bot.py index 381c008..0486425 100644 --- a/davtelepot/bot.py +++ b/davtelepot/bot.py @@ -9,1170 +9,16 @@ import asyncio import logging # Third party modules -import aiohttp from aiohttp import web # Project modules +from api import TelegramBot, TelegramError from utilities import get_secure_key # Do not log aiohttp `INFO` and `DEBUG` levels logging.getLogger('aiohttp').setLevel(logging.WARNING) -class TelegramError(Exception): - """Telegram API exceptions class.""" - - def __init__(self, error_code=0, description=None, ok=False): - """Get an error response and return corresponding Exception.""" - self._code = error_code - if description is None: - self._description = 'Generic error' - else: - self._description = description - super().__init__(self.description) - - @property - def code(self): - """Telegram error code.""" - return self._code - - @property - def description(self): - """Human-readable description of error.""" - return f"Error {self.code}: {self._description}" - - -class TelegramBot(object): - """Provide python method having the same signature as Telegram API methods. - - All mirrored methods are camelCase. - """ - - loop = asyncio.get_event_loop() - app = web.Application(loop=loop) - sessions_timeouts = { - 'getUpdates': dict( - timeout=35, - close=False - ), - 'sendMessage': dict( - timeout=20, - close=False - ) - } - - def __init__(self, token): - """Set bot token and store HTTP sessions.""" - self._token = token - self.sessions = dict() - - @property - def token(self): - """Telegram API bot token.""" - return self._token - - @staticmethod - def check_telegram_api_json(response): - """Take a json Telegram response, check it and return its content. - - Example of well-formed json Telegram responses: - { - "ok": False, - "error_code": 401, - "description": "Unauthorized" - } - { - "ok": True, - "result": ... - } - """ - assert 'ok' in response, ( - "All Telegram API responses have an `ok` field." - ) - if not response['ok']: - raise TelegramError(**response) - return response['result'] - - @staticmethod - def adapt_parameters(parameters, exclude=[]): - """Build a aiohttp.FormData object from given `paramters`. - - Exclude `self`, empty values and parameters in `exclude` list. - Cast integers to string to avoid TypeError during json serialization. - """ - exclude.append('self') - data = aiohttp.FormData() - for key, value in parameters.items(): - if not (key in exclude or value is None): - if type(value) is int: - value = str(value) - data.add_field(key, value) - return data - - def get_session(self, api_method): - """According to API method, return proper session and information. - - Return a tuple (session, session_must_be_closed) - session : aiohttp.ClientSession - Client session with proper timeout - session_must_be_closed : bool - True if session must be closed after being used once - """ - cls = self.__class__ - if api_method in cls.sessions_timeouts: - if api_method not in self.sessions: - self.sessions[api_method] = aiohttp.ClientSession( - loop=cls.loop, - timeout=aiohttp.ClientTimeout( - total=cls.sessions_timeouts[api_method]['timeout'] - ) - ) - session = self.sessions[api_method] - session_must_be_closed = cls.sessions_timeouts[api_method]['close'] - else: - session = aiohttp.ClientSession( - loop=cls.loop, - timeout=aiohttp.ClientTimeout(total=None) - ) - session_must_be_closed = True - return session, session_must_be_closed - - async def api_request(self, method, parameters={}, exclude=[]): - """Return the result of a Telegram bot API request, or an Exception. - - Opened sessions will be used more than one time (if appropriate) and - will be closed on `Bot.app.cleanup`. - Result may be a Telegram API json response, None, or Exception. - """ - response_object = None - session, session_must_be_closed = self.get_session(method) - parameters = self.adapt_parameters(parameters, exclude=exclude) - try: - async with session.post( - "https://api.telegram.org/bot" - f"{self.token}/{method}", - data=parameters - ) as response: - try: - response_object = self.check_telegram_api_json( - await response.json() # Telegram returns json objects - ) - except TelegramError as e: - logging.error(f"{e}") - return e - except Exception as e: - logging.error(f"{e}", exc_info=True) - return e - except asyncio.TimeoutError as e: - logging.info(f"{e}: {method} API call timed out") - finally: - if session_must_be_closed: - await session.close() - return response_object - - async def getMe(self): - """Get basic information about the bot in form of a User object. - - Useful to test `self.token`. - See https://core.telegram.org/bots/api#getme for details. - """ - return await self.api_request( - 'getMe', - ) - - async def getUpdates(self, offset, timeout, limit, allowed_updates): - """Get a list of updates starting from `offset`. - - If there are no updates, keep the request hanging until `timeout`. - If there are more than `limit` updates, retrieve them in packs of - `limit`. - Allowed update types (empty list to allow all). - See https://core.telegram.org/bots/api#getupdates for details. - """ - return await self.api_request( - method='getUpdates', - parameters=locals() - ) - - async def setWebhook(self, url=None, certificate=None, - max_connections=None, allowed_updates=None): - """Set or remove a webhook. Telegram will post to `url` new updates. - - See https://core.telegram.org/bots/api#setwebhook for details. - """ - if url is None: - url = self.webhook_url - if allowed_updates is None: - allowed_updates = self.allowed_updates - if max_connections is None: - max_connections = self.max_connections - if certificate is None: - certificate = self.certificate - if type(certificate) is str: - try: - certificate = open(certificate, 'r') - except FileNotFoundError as e: - logging.error(f"{e}") - certificate = None - certificate = dict( - file=certificate - ) - return await self.api_request( - 'setWebhook', - parameters=locals() - ) - - async def deleteWebhook(self): - """Remove webhook integration and switch back to getUpdate. - - See https://core.telegram.org/bots/api#deletewebhook for details. - """ - return await self.api_request( - 'deleteWebhook', - ) - - async def getWebhookInfo(self): - """Get current webhook status. - - See https://core.telegram.org/bots/api#getwebhookinfo for details. - """ - return await self.api_request( - 'getWebhookInfo', - ) - - async def sendMessage(self, chat_id, text, - parse_mode=None, - disable_web_page_preview=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a text message. On success, return it. - - See https://core.telegram.org/bots/api#sendmessage for details. - """ - return await self.api_request( - 'sendMessage', - parameters=locals() - ) - - async def forwardMessage(self, chat_id, from_chat_id, message_id, - disable_notification=None): - """Forward a message. - - See https://core.telegram.org/bots/api#forwardmessage for details. - """ - return await self.api_request( - 'forwardMessage', - parameters=locals() - ) - - async def sendPhoto(self, chat_id, photo, - caption=None, - parse_mode=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a photo from file_id, HTTP url or file. - - See https://core.telegram.org/bots/api#sendphoto for details. - """ - return await self.api_request( - 'sendPhoto', - parameters=locals() - ) - - async def sendAudio(self, chat_id, audio, - caption=None, - parse_mode=None, - duration=None, - performer=None, - title=None, - thumb=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send an audio file from file_id, HTTP url or file. - - See https://core.telegram.org/bots/api#sendaudio for details. - """ - return await self.api_request( - 'sendAudio', - parameters=locals() - ) - - async def sendDocument(self, chat_id, document, - thumb=None, - caption=None, - parse_mode=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a document from file_id, HTTP url or file. - - See https://core.telegram.org/bots/api#senddocument for details. - """ - return await self.api_request( - 'sendDocument', - parameters=locals() - ) - - async def sendVideo(self, chat_id, video, - duration=None, - width=None, - height=None, - thumb=None, - caption=None, - parse_mode=None, - supports_streaming=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a video from file_id, HTTP url or file. - - See https://core.telegram.org/bots/api#sendvideo for details. - """ - return await self.api_request( - 'sendVideo', - parameters=locals() - ) - - async def sendAnimation(self, chat_id, animation, - duration=None, - width=None, - height=None, - thumb=None, - caption=None, - parse_mode=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send animation files (GIF or H.264/MPEG-4 AVC video without sound). - - See https://core.telegram.org/bots/api#sendanimation for details. - """ - return await self.api_request( - 'sendAnimation', - parameters=locals() - ) - - async def sendVoice(self, chat_id, voice, - caption=None, - parse_mode=None, - duration=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send an audio file to be displayed as playable voice message. - - `voice` must be in an .ogg file encoded with OPUS. - See https://core.telegram.org/bots/api#sendvoice for details. - """ - return await self.api_request( - 'sendVoice', - parameters=locals() - ) - - async def sendVideoNote(self, chat_id, video_note, - duration=None, - length=None, - thumb=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a rounded square mp4 video message of up to 1 minute long. - - See https://core.telegram.org/bots/api#sendvideonote for details. - """ - return await self.api_request( - 'sendVideoNote', - parameters=locals() - ) - - async def sendMediaGroup(self, chat_id, media, - disable_notification=None, - reply_to_message_id=None): - """Send a group of photos or videos as an album. - - `media` must be a list of `InputMediaPhoto` and/or `InputMediaVideo` - objects. - See https://core.telegram.org/bots/api#sendmediagroup for details. - """ - return await self.api_request( - 'sendMediaGroup', - parameters=locals() - ) - - async def sendLocation(self, chat_id, latitude, longitude, - live_period=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a point on the map. May be kept updated for a `live_period`. - - See https://core.telegram.org/bots/api#sendlocation for details. - """ - return await self.api_request( - 'sendLocation', - parameters=locals() - ) - - async def editMessageLiveLocation(self, latitude, longitude, - chat_id=None, message_id=None, - inline_message_id=None, - reply_markup=None): - """Edit live location messages. - - A location can be edited until its live_period expires or editing is - explicitly disabled by a call to stopMessageLiveLocation. - The message to be edited may be identified through `inline_message_id` - OR the couple (`chat_id`, `message_id`). - See https://core.telegram.org/bots/api#editmessagelivelocation - for details. - """ - return await self.api_request( - 'editMessageLiveLocation', - parameters=locals() - ) - - async def stopMessageLiveLocation(self, - chat_id=None, message_id=None, - inline_message_id=None, - reply_markup=None): - """Stop updating a live location message before live_period expires. - - The position to be stopped may be identified through - `inline_message_id` OR the couple (`chat_id`, `message_id`). - `reply_markup` type may be only `InlineKeyboardMarkup`. - See https://core.telegram.org/bots/api#stopmessagelivelocation - for details. - """ - return await self.api_request( - 'stopMessageLiveLocation', - parameters=locals() - ) - - async def sendVenue(self, chat_id, latitude, longitude, title, address, - foursquare_id=None, - foursquare_type=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send information about a venue. - - Integrated with FourSquare. - See https://core.telegram.org/bots/api#sendvenue for details. - """ - return await self.api_request( - 'sendVenue', - parameters=locals() - ) - - async def sendContact(self, chat_id, phone_number, first_name, - last_name=None, - vcard=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a phone contact. - - See https://core.telegram.org/bots/api#sendcontact for details. - """ - return await self.api_request( - 'sendContact', - parameters=locals() - ) - - async def sendPoll(self, chat_id, question, options, - dummy=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a native poll in a group, a supergroup or channel. - - See https://core.telegram.org/bots/api#sendpoll for details. - """ - return await self.api_request( - 'sendPoll', - parameters=locals() - ) - - async def sendChatAction(self, chat_id, action): - """Fake a typing status or similar. - - See https://core.telegram.org/bots/api#sendchataction for details. - """ - return await self.api_request( - 'sendChatAction', - parameters=locals() - ) - - async def getUserProfilePhotos(self, user_id, - offset=None, - limit=None,): - """Get a list of profile pictures for a user. - - See https://core.telegram.org/bots/api#getuserprofilephotos - for details. - """ - return await self.api_request( - 'getUserProfilePhotos', - parameters=locals() - ) - - async def getFile(self, file_id): - """Get basic info about a file and prepare it for downloading. - - For the moment, bots can download files of up to - 20MB in size. - On success, a File object is returned. The file can then be downloaded - via the link https://api.telegram.org/file/bot/, - where is taken from the response. - - See https://core.telegram.org/bots/api#getfile for details. - """ - return await self.api_request( - 'getFile', - parameters=locals() - ) - - async def kickChatMember(self, chat_id, user_id, - until_date=None): - """Kick a user from a group, a supergroup or a channel. - - In the case of supergroups and channels, the user will not be able to - return to the group on their own using invite links, etc., unless - unbanned first. - Note: In regular groups (non-supergroups), this method will only work - if the ‘All Members Are Admins’ setting is off in the target group. - Otherwise members may only be removed by the group's creator or by - the member that added them. - See https://core.telegram.org/bots/api#kickchatmember for details. - """ - return await self.api_request( - 'kickChatMember', - parameters=locals() - ) - - async def unbanChatMember(self, chat_id, user_id): - """Unban a previously kicked user in a supergroup or channel. - - The user will not return to the group or channel automatically, but - will be able to join via link, etc. - The bot must be an administrator for this to work. - Return True on success. - See https://core.telegram.org/bots/api#unbanchatmember for details. - """ - return await self.api_request( - 'unbanChatMember', - parameters=locals() - ) - - async def restrictChatMember(self, chat_id, user_id, - until_date=None, - can_send_messages=None, - can_send_media_messages=None, - can_send_other_messages=None, - can_add_web_page_previews=None): - """Restrict a user in a supergroup. - - The bot must be an administrator in the supergroup for this to work - and must have the appropriate admin rights. - Pass True for all boolean parameters to lift restrictions from a - user. - Return True on success. - See https://core.telegram.org/bots/api#restrictchatmember for details. - """ - return await self.api_request( - 'restrictChatMember', - parameters=locals() - ) - - async def promoteChatMember(self, chat_id, user_id, - can_change_info=None, - can_post_messages=None, - can_edit_messages=None, - can_delete_messages=None, - can_invite_users=None, - can_restrict_members=None, - can_pin_messages=None, - can_promote_members=None): - """Promote or demote a user in a supergroup or a channel. - - The bot must be an administrator in the chat for this to work and must - have the appropriate admin rights. - Pass False for all boolean parameters to demote a user. - Return True on success. - See https://core.telegram.org/bots/api#promotechatmember for details. - """ - return await self.api_request( - 'promoteChatMember', - parameters=locals() - ) - - async def exportChatInviteLink(self, chat_id): - """Generate a new invite link for a chat and revoke any active link. - - The bot must be an administrator in the chat for this to work and must - have the appropriate admin rights. - Return the new invite link as String on success. - NOTE: to get the current invite link, use `getChat` method. - See https://core.telegram.org/bots/api#exportchatinvitelink - for details. - """ - return await self.api_request( - 'exportChatInviteLink', - parameters=locals() - ) - - async def setChatPhoto(self, chat_id, photo): - """Set a new profile photo for the chat. - - Photos can't be changed for private chats. - `photo` must be an input file (file_id and urls are not allowed). - The bot must be an administrator in the chat for this to work and must - have the appropriate admin rights. - Return True on success. - See https://core.telegram.org/bots/api#setchatphoto for details. - """ - return await self.api_request( - 'setChatPhoto', - parameters=locals() - ) - - async def deleteChatPhoto(self, chat_id): - """Delete a chat photo. - - Photos can't be changed for private chats. - The bot must be an administrator in the chat for this to work and must - have the appropriate admin rights. - Return True on success. - See https://core.telegram.org/bots/api#deletechatphoto for details. - """ - return await self.api_request( - 'deleteChatPhoto', - parameters=locals() - ) - - async def setChatTitle(self, chat_id, title): - """Change the title of a chat. - - Titles can't be changed for private chats. - The bot must be an administrator in the chat for this to work and must - have the appropriate admin rights. - Return True on success. - See https://core.telegram.org/bots/api#setchattitle for details. - """ - return await self.api_request( - 'setChatTitle', - parameters=locals() - ) - - async def setChatDescription(self, chat_id, description): - """Change the description of a supergroup or a channel. - - The bot must be an administrator in the chat for this to work and must - have the appropriate admin rights. - Return True on success. - See https://core.telegram.org/bots/api#setchatdescription for details. - """ - return await self.api_request( - 'setChatDescription', - parameters=locals() - ) - - async def pinChatMessage(self, chat_id, message_id, - disable_notification=None): - """Pin a message in a group, a supergroup, or a channel. - - The bot must be an administrator in the chat for this to work and must - have the ‘can_pin_messages’ admin right in the supergroup or - ‘can_edit_messages’ admin right in the channel. - Return True on success. - See https://core.telegram.org/bots/api#pinchatmessage for details. - """ - return await self.api_request( - 'pinChatMessage', - parameters=locals() - ) - - async def unpinChatMessage(self, chat_id): - """Unpin a message in a group, a supergroup, or a channel. - - The bot must be an administrator in the chat for this to work and must - have the ‘can_pin_messages’ admin right in the supergroup or - ‘can_edit_messages’ admin right in the channel. - Return True on success. - See https://core.telegram.org/bots/api#unpinchatmessage for details. - """ - return await self.api_request( - 'unpinChatMessage', - parameters=locals() - ) - - async def leaveChat(self, chat_id): - """Make the bot leave a group, supergroup or channel. - - Return True on success. - See https://core.telegram.org/bots/api#leavechat for details. - """ - return await self.api_request( - 'leaveChat', - parameters=locals() - ) - - async def getChat(self, chat_id): - """Get up to date information about the chat. - - Return a Chat object on success. - See https://core.telegram.org/bots/api#getchat for details. - """ - return await self.api_request( - 'getChat', - parameters=locals() - ) - - async def getChatAdministrators(self, chat_id): - """Get a list of administrators in a chat. - - On success, return an Array of ChatMember objects that contains - information about all chat administrators except other bots. - If the chat is a group or a supergroup and no administrators were - appointed, only the creator will be returned. - - See https://core.telegram.org/bots/api#getchatadministrators - for details. - """ - return await self.api_request( - 'getChatAdministrators', - parameters=locals() - ) - - async def getChatMembersCount(self, chat_id): - """Get the number of members in a chat. - - Returns Int on success. - See https://core.telegram.org/bots/api#getchatmemberscount for details. - """ - return await self.api_request( - 'getChatMembersCount', - parameters=locals() - ) - - async def getChatMember(self, chat_id, user_id): - """Get information about a member of a chat. - - Returns a ChatMember object on success. - See https://core.telegram.org/bots/api#getchatmember for details. - """ - return await self.api_request( - 'getChatMember', - parameters=locals() - ) - - async def setChatStickerSet(self, chat_id, sticker_set_name): - """Set a new group sticker set for a supergroup. - - The bot must be an administrator in the chat for this to work and must - have the appropriate admin rights. - Use the field `can_set_sticker_set` optionally returned in getChat - requests to check if the bot can use this method. - Returns True on success. - See https://core.telegram.org/bots/api#setchatstickerset for details. - """ - return await self.api_request( - 'setChatStickerSet', - parameters=locals() - ) - - async def deleteChatStickerSet(self, chat_id): - """Delete a group sticker set from a supergroup. - - The bot must be an administrator in the chat for this to work and must - have the appropriate admin rights. - Use the field `can_set_sticker_set` optionally returned in getChat - requests to check if the bot can use this method. - Returns True on success. - See https://core.telegram.org/bots/api#deletechatstickerset for - details. - """ - return await self.api_request( - 'deleteChatStickerSet', - parameters=locals() - ) - - async def answerCallbackQuery(self, callback_query_id, - text=None, - show_alert=None, - url=None, - cache_time=None): - """Send answers to callback queries sent from inline keyboards. - - The answer will be displayed to the user as a notification at the top - of the chat screen or as an alert. - On success, True is returned. - See https://core.telegram.org/bots/api#answercallbackquery for details. - """ - return await self.api_request( - 'answerCallbackQuery', - parameters=locals() - ) - - async def editMessageText(self, text, - chat_id=None, message_id=None, - inline_message_id=None, - parse_mode=None, - disable_web_page_preview=None, - reply_markup=None): - """Edit text and game messages. - - On success, if edited message is sent by the bot, the edited Message - is returned, otherwise True is returned. - See https://core.telegram.org/bots/api#editmessagetext for details. - """ - return await self.api_request( - 'editMessageText', - parameters=locals() - ) - - async def editMessageCaption(self, - chat_id=None, message_id=None, - inline_message_id=None, - caption=None, - parse_mode=None, - reply_markup=None): - """Edit captions of messages. - - On success, if edited message is sent by the bot, the edited Message is - returned, otherwise True is returned. - See https://core.telegram.org/bots/api#editmessagecaption for details. - """ - return await self.api_request( - 'editMessageCaption', - parameters=locals() - ) - - async def editMessageMedia(self, - chat_id=None, message_id=None, - inline_message_id=None, - media=None, - reply_markup=None): - """Edit animation, audio, document, photo, or video messages. - - If a message is a part of a message album, then it can be edited only - to a photo or a video. Otherwise, message type can be changed - arbitrarily. - When inline message is edited, new file can't be uploaded. - Use previously uploaded file via its file_id or specify a URL. - On success, if the edited message was sent by the bot, the edited - Message is returned, otherwise True is returned. - See https://core.telegram.org/bots/api#editmessagemedia for details. - """ - return await self.api_request( - 'editMessageMedia', - parameters=locals() - ) - - async def editMessageReplyMarkup(self, - chat_id=None, message_id=None, - inline_message_id=None, - reply_markup=None): - """Edit only the reply markup of messages. - - On success, if edited message is sent by the bot, the edited Message is - returned, otherwise True is returned. - See https://core.telegram.org/bots/api#editmessagereplymarkup for - details. - """ - return await self.api_request( - 'editMessageReplyMarkup', - parameters=locals() - ) - - async def stopPoll(self, chat_id, message_id, - reply_markup=None): - """Stop a poll which was sent by the bot. - - On success, the stopped Poll with the final results is returned. - `reply_markup` type may be only `InlineKeyboardMarkup`. - See https://core.telegram.org/bots/api#stoppoll for details. - """ - return await self.api_request( - 'stopPoll', - parameters=locals() - ) - - async def deleteMessage(self, chat_id, message_id): - """Delete a message, including service messages. - - - A message can only be deleted if it was sent less than 48 hours - ago. - - Bots can delete outgoing messages in private chats, groups, and - supergroups. - - Bots can delete incoming messages in private chats. - - Bots granted can_post_messages permissions can delete outgoing - messages in channels. - - If the bot is an administrator of a group, it can delete any - message there. - - If the bot has can_delete_messages permission in a supergroup or - a channel, it can delete any message there. - Returns True on success. - - See https://core.telegram.org/bots/api#deletemessage for details. - """ - return await self.api_request( - 'deleteMessage', - parameters=locals() - ) - - async def sendSticker(self, chat_id, sticker, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send .webp stickers. - - On success, the sent Message is returned. - See https://core.telegram.org/bots/api#sendsticker for details. - """ - return await self.api_request( - 'sendSticker', - parameters=locals() - ) - - async def getStickerSet(self, name): - """Get a sticker set. - - On success, a StickerSet object is returned. - See https://core.telegram.org/bots/api#getstickerset for details. - """ - return await self.api_request( - 'getStickerSet', - parameters=locals() - ) - - async def uploadStickerFile(self, user_id, png_sticker): - """Upload a .png file as a sticker. - - Use it later via `createNewStickerSet` and `addStickerToSet` methods - (can be used multiple times). - Return the uploaded File on success. - `png_sticker` must be a *.png image up to 512 kilobytes in size, - dimensions must not exceed 512px, and either width or height must - be exactly 512px. - See https://core.telegram.org/bots/api#uploadstickerfile for details. - """ - return await self.api_request( - 'uploadStickerFile', - parameters=locals() - ) - - async def createNewStickerSet(self, user_id, - name, title, png_sticker, emojis, - contains_masks=None, - mask_position=None): - """Create new sticker set owned by a user. - - The bot will be able to edit the created sticker set. - Returns True on success. - See https://core.telegram.org/bots/api#createnewstickerset for details. - """ - return await self.api_request( - 'createNewStickerSet', - parameters=locals() - ) - - async def addStickerToSet(self, user_id, name, png_sticker, emojis, - mask_position=None): - """Add a new sticker to a set created by the bot. - - Returns True on success. - See https://core.telegram.org/bots/api#addstickertoset for details. - """ - return await self.api_request( - 'addStickerToSet', - parameters=locals() - ) - - async def setStickerPositionInSet(self, sticker, position): - """Move a sticker in a set created by the bot to a specific position . - - Position is 0-based. - Returns True on success. - See https://core.telegram.org/bots/api#setstickerpositioninset for - details. - """ - return await self.api_request( - 'setStickerPositionInSet', - parameters=locals() - ) - - async def deleteStickerFromSet(self, sticker): - """Delete a sticker from a set created by the bot. - - Returns True on success. - See https://core.telegram.org/bots/api#deletestickerfromset for - details. - """ - return await self.api_request( - 'deleteStickerFromSet', - parameters=locals() - ) - - async def answerInlineQuery(self, inline_query_id, results, - cache_time=None, - is_personal=None, - next_offset=None, - switch_pm_text=None, - switch_pm_parameter=None): - """Send answers to an inline query. - - On success, True is returned. - No more than 50 results per query are allowed. - See https://core.telegram.org/bots/api#answerinlinequery for details. - """ - return await self.api_request( - 'answerInlineQuery', - parameters=locals() - ) - - async def sendInvoice(self, chat_id, title, description, payload, - provider_token, start_parameter, currency, prices, - provider_data=None, - photo_url=None, - photo_size=None, - photo_width=None, - photo_height=None, - need_name=None, - need_phone_number=None, - need_email=None, - need_shipping_address=None, - send_phone_number_to_provider=None, - send_email_to_provider=None, - is_flexible=None, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send an invoice. - - On success, the sent Message is returned. - See https://core.telegram.org/bots/api#sendinvoice for details. - """ - return await self.api_request( - 'sendInvoice', - parameters=locals() - ) - - async def answerShippingQuery(self, shipping_query_id, ok, - shipping_options=None, - error_message=None): - """Reply to shipping queries. - - On success, True is returned. - If you sent an invoice requesting a shipping address and the parameter - is_flexible was specified, the Bot API will send an Update with a - shipping_query field to the bot. - See https://core.telegram.org/bots/api#answershippingquery for details. - """ - return await self.api_request( - 'answerShippingQuery', - parameters=locals() - ) - - async def answerPreCheckoutQuery(self, pre_checkout_query_id, ok, - error_message=None): - """Respond to pre-checkout queries. - - Once the user has confirmed their payment and shipping details, the Bot - API sends the final confirmation in the form of an Update with the - field pre_checkout_query. - On success, True is returned. - Note: The Bot API must receive an answer within 10 seconds after the - pre-checkout query was sent. - See https://core.telegram.org/bots/api#answerprecheckoutquery for - details. - """ - return await self.api_request( - 'answerPreCheckoutQuery', - parameters=locals() - ) - - async def setPassportDataErrors(self, user_id, errors): - """Refuse a Telegram Passport element with `errors`. - - Inform a user that some of the Telegram Passport elements they provided - contains errors. - The user will not be able to re-submit their Passport to you until the - errors are fixed (the contents of the field for which you returned - the error must change). - Returns True on success. - Use this if the data submitted by the user doesn't satisfy the - standards your service requires for any reason. - For example, if a birthday date seems invalid, a submitted document - is blurry, a scan shows evidence of tampering, etc. - Supply some details in the error message to make sure the user knows - how to correct the issues. - See https://core.telegram.org/bots/api#setpassportdataerrors for - details. - """ - return await self.api_request( - 'setPassportDataErrors', - parameters=locals() - ) - - async def sendGame(self, chat_id, game_short_name, - disable_notification=None, - reply_to_message_id=None, - reply_markup=None): - """Send a game. - - On success, the sent Message is returned. - See https://core.telegram.org/bots/api#sendgame for - details. - """ - return await self.api_request( - 'sendGame', - parameters=locals() - ) - - async def setGameScore(self, user_id, score, - force=None, - disable_edit_message=None, - chat_id=None, message_id=None, - inline_message_id=None): - """Set the score of the specified user in a game. - - On success, if the message was sent by the bot, returns the edited - Message, otherwise returns True. - Returns an error, if the new score is not greater than the user's - current score in the chat and force is False. - See https://core.telegram.org/bots/api#setgamescore for - details. - """ - return await self.api_request( - 'setGameScore', - parameters=locals() - ) - - async def getGameHighScores(self, user_id, - chat_id=None, message_id=None, - inline_message_id=None): - """Get data for high score tables. - - Will return the score of the specified user and several of his - neighbors in a game. - On success, returns an Array of GameHighScore objects. - This method will currently return scores for the target user, plus two - of his closest neighbors on each side. Will also return the top - three users if the user and his neighbors are not among them. - Please note that this behavior is subject to change. - See https://core.telegram.org/bots/api#getgamehighscores for - details. - """ - return await self.api_request( - 'getGameHighScores', - parameters=locals() - ) - - class Bot(TelegramBot): """Simple Bot object, providing methods corresponding to Telegram bot API.