|
|
@@ -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<token>/<file_path>,
|
|
|
- where <file_path> 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.
|
|
|
|
