Davte/davtelepot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compliance with Telegram Bot API 5.0 and improved type hinting for API methods.

Browse Source
This commit is contained in:
Davte
2020-11-16 22:54:25 +01:00
parent a3cc676432
commit dc9c2a7b14
3 changed files with 406 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
davtelepot/__init__.py
View File

@@ -11,7 +11,7 @@ __author__ = "Davide Testa"
__email__ = "davide@davte.it"
__credits__ = ["Marco Origlia", "Nick Lee @Nickoala"]
__license__ = "GNU General Public License v3.0"
__version__ = "2.6.20"
__version__ = "2.7.0"
__maintainer__ = "Davide Testa"
__contact__ = "t.me/davte"

490
davtelepot/api.py
View File

@@ -7,10 +7,11 @@ A simple aiohttp asynchronous web client is used to make requests.
# Standard library modules
import asyncio
import datetime
import io
import json
import logging
from typing import Union, List
from typing import Dict, Union, List, IO
# Third party modules
import aiohttp
@@ -199,6 +200,19 @@ class TelegramBot:
data.add_field(key, value)
return data
@staticmethod
def prepare_file_object(file: Union[str, IO, dict, None]
) -> Union[Dict[str, IO], None]:
if type(file) is str:
try:
file = open(file, 'r')
except FileNotFoundError as e:
logging.error(f"{e}")
file = None
if isinstance(file, io.IOBase):
file = dict(file=file)
return file
def get_session(self, api_method):
"""According to API method, return proper session and information.
@@ -368,7 +382,10 @@ class TelegramBot:
'getMe',
)
async def getUpdates(self, offset, timeout, limit, allowed_updates):
async def getUpdates(self, offset: int = None,
limit: int = None,
timeout: int = None,
allowed_updates: List[str] = None):
"""Get a list of updates starting from `offset`.
If there are no updates, keep the request hanging until `timeout`.
@@ -382,20 +399,17 @@ class TelegramBot:
parameters=locals()
)
async def setWebhook(self, url=None, certificate=None,
max_connections=None, allowed_updates=None):
async def setWebhook(self, url: str,
certificate: Union[str, IO] = None,
ip_address: str = None,
max_connections: int = None,
allowed_updates: List[str] = None,
drop_pending_updates: bool = None):
"""Set or remove a webhook. Telegram will post to `url` new updates.
See https://core.telegram.org/bots/api#setwebhook for details.
"""
if type(certificate) is str:
try:
certificate = dict(
file=open(certificate, 'r')
)
except FileNotFoundError as e:
logging.error(f"{e}\nCertificate set to `None`")
certificate = None
certificate = self.prepare_file_object(certificate)
result = await self.api_request(
'setWebhook',
parameters=locals()
@@ -404,13 +418,14 @@ class TelegramBot:
certificate['file'].close()
return result
async def deleteWebhook(self):
async def deleteWebhook(self, drop_pending_updates: bool = None):
"""Remove webhook integration and switch back to getUpdate.
See https://core.telegram.org/bots/api#deletewebhook for details.
"""
return await self.api_request(
'deleteWebhook',
parameters=locals()
)
async def getWebhookInfo(self):
@@ -422,11 +437,13 @@ class TelegramBot:
'getWebhookInfo',
)
async def sendMessage(self, chat_id, text,
parse_mode=None,
disable_web_page_preview=None,
disable_notification=None,
reply_to_message_id=None,
async def sendMessage(self, chat_id: Union[int, str], text: str,
parse_mode: str = None,
entities: List[dict] = None,
disable_web_page_preview: bool = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send a text message. On success, return it.
@@ -437,8 +454,10 @@ class TelegramBot:
parameters=locals()
)
async def forwardMessage(self, chat_id, from_chat_id, message_id,
disable_notification=None):
async def forwardMessage(self, chat_id: Union[int, str],
from_chat_id: Union[int, str],
message_id: int,
disable_notification: bool = None):
"""Forward a message.
See https://core.telegram.org/bots/api#forwardmessage for details.
@@ -448,11 +467,13 @@ class TelegramBot:
parameters=locals()
)
async def sendPhoto(self, chat_id, photo,
caption=None,
parse_mode=None,
disable_notification=None,
reply_to_message_id=None,
async def sendPhoto(self, chat_id: Union[int, str], photo,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send a photo from file_id, HTTP url or file.
@@ -463,15 +484,17 @@ class TelegramBot:
parameters=locals()
)
async def sendAudio(self, chat_id, audio,
caption=None,
parse_mode=None,
duration=None,
performer=None,
title=None,
async def sendAudio(self, chat_id: Union[int, str], audio,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
duration: int = None,
performer: str = None,
title: str = None,
thumb=None,
disable_notification=None,
reply_to_message_id=None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send an audio file from file_id, HTTP url or file.
@@ -482,12 +505,15 @@ class TelegramBot:
parameters=locals()
)
async def sendDocument(self, chat_id, document,
async def sendDocument(self, chat_id: Union[int, str], document,
thumb=None,
caption=None,
parse_mode=None,
disable_notification=None,
reply_to_message_id=None,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
disable_content_type_detection: bool = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send a document from file_id, HTTP url or file.
@@ -498,16 +524,18 @@ class TelegramBot:
parameters=locals()
)
async def sendVideo(self, chat_id, video,
duration=None,
width=None,
height=None,
async def sendVideo(self, chat_id: Union[int, str], video,
duration: int = None,
width: int = None,
height: int = None,
thumb=None,
caption=None,
parse_mode=None,
supports_streaming=None,
disable_notification=None,
reply_to_message_id=None,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
supports_streaming: bool = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send a video from file_id, HTTP url or file.
@@ -518,15 +546,17 @@ class TelegramBot:
parameters=locals()
)
async def sendAnimation(self, chat_id, animation,
duration=None,
width=None,
height=None,
async def sendAnimation(self, chat_id: Union[int, str], animation,
duration: int = None,
width: int = None,
height: int = None,
thumb=None,
caption=None,
parse_mode=None,
disable_notification=None,
reply_to_message_id=None,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send animation files (GIF or H.264/MPEG-4 AVC video without sound).
@@ -537,12 +567,14 @@ class TelegramBot:
parameters=locals()
)
async def sendVoice(self, chat_id, voice,
caption=None,
parse_mode=None,
duration=None,
disable_notification=None,
reply_to_message_id=None,
async def sendVoice(self, chat_id: Union[int, str], voice,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
duration: int = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send an audio file to be displayed as playable voice message.
@@ -554,12 +586,13 @@ class TelegramBot:
parameters=locals()
)
async def sendVideoNote(self, chat_id, video_note,
duration=None,
length=None,
async def sendVideoNote(self, chat_id: Union[int, str], video_note,
duration: int = None,
length: int = None,
thumb=None,
disable_notification=None,
reply_to_message_id=None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send a rounded square mp4 video message of up to 1 minute long.
@@ -570,9 +603,10 @@ class TelegramBot:
parameters=locals()
)
async def sendMediaGroup(self, chat_id, media,
disable_notification=None,
reply_to_message_id=None):
async def sendMediaGroup(self, chat_id: Union[int, str], media: list,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None):
"""Send a group of photos or videos as an album.
`media` must be a list of `InputMediaPhoto` and/or `InputMediaVideo`
@@ -584,23 +618,40 @@ class TelegramBot:
parameters=locals()
)
async def sendLocation(self, chat_id, latitude, longitude,
async def sendLocation(self, chat_id: Union[int, str],
latitude: float, longitude: float,
horizontal_accuracy: float = None,
live_period=None,
disable_notification=None,
reply_to_message_id=None,
heading: int = None,
proximity_alert_radius: int = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = 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.
"""
if horizontal_accuracy: # Horizontal accuracy: 0-1500 m [float].
horizontal_accuracy = max(0.0, min(horizontal_accuracy, 1500.0))
if live_period:
live_period = max(60, min(live_period, 86400))
if heading: # Direction in which the user is moving, 1-360°
heading = max(1, min(heading, 360))
if proximity_alert_radius: # Distance 1-100000 m
proximity_alert_radius = max(1, min(proximity_alert_radius, 100000))
return await self.api_request(
'sendLocation',
parameters=locals()
)
async def editMessageLiveLocation(self, latitude, longitude,
chat_id=None, message_id=None,
inline_message_id=None,
async def editMessageLiveLocation(self, latitude: float, longitude: float,
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: str = None,
horizontal_accuracy: float = None,
heading: int = None,
proximity_alert_radius: int = None,
reply_markup=None):
"""Edit live location messages.
@@ -611,14 +662,23 @@ class TelegramBot:
See https://core.telegram.org/bots/api#editmessagelivelocation
for details.
"""
if inline_message_id is None and (chat_id is None or message_id is None):
logging.error("Invalid target chat!")
if horizontal_accuracy: # Horizontal accuracy: 0-1500 m [float].
horizontal_accuracy = max(0.0, min(horizontal_accuracy, 1500.0))
if heading: # Direction in which the user is moving, 1-360°
heading = max(1, min(heading, 360))
if proximity_alert_radius: # Distance 1-100000 m
proximity_alert_radius = max(1, min(proximity_alert_radius, 100000))
return await self.api_request(
'editMessageLiveLocation',
parameters=locals()
)
async def stopMessageLiveLocation(self,
chat_id=None, message_id=None,
inline_message_id=None,
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: int = None,
reply_markup=None):
"""Stop updating a live location message before live_period expires.
@@ -633,11 +693,16 @@ class TelegramBot:
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,
async def sendVenue(self, chat_id: Union[int, str],
latitude: float, longitude: float,
title: str, address: str,
foursquare_id: str = None,
foursquare_type: str = None,
google_place_id: str = None,
google_place_type: str = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send information about a venue.
@@ -649,11 +714,14 @@ class TelegramBot:
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,
async def sendContact(self, chat_id: Union[int, str],
phone_number: str,
first_name: str,
last_name: str = None,
vcard: str = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send a phone contact.
@@ -674,16 +742,33 @@ class TelegramBot:
correct_option_id: int = None,
explanation: str = None,
explanation_parse_mode: str = None,
explanation_entities: List[dict] = None,
open_period: int = None,
close_date: int = None,
close_date: Union[int, datetime.datetime] = None,
is_closed: bool = None,
disable_notification: bool = None,
allow_sending_without_reply: bool = None,
reply_to_message_id: int = 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.
close_date: Unix timestamp; 5-600 seconds from now.
open_period (overwrites close_date): seconds (integer), 5-600.
"""
if open_period is not None:
close_date = None
open_period = min(max(5, open_period), 600)
elif isinstance(close_date, datetime.datetime):
now = datetime.datetime.now()
close_date = min(
max(
now + datetime.timedelta(seconds=5),
close_date
), now + datetime.timedelta(seconds=600)
)
close_date = int(close_date.timestamp())
# To avoid shadowing `type`, this workaround is required
parameters = locals().copy()
parameters['type'] = parameters['type_']
@@ -693,7 +778,7 @@ class TelegramBot:
parameters=parameters
)
async def sendChatAction(self, chat_id, action):
async def sendChatAction(self, chat_id: Union[int, str], action):
"""Fake a typing status or similar.
See https://core.telegram.org/bots/api#sendchataction for details.
@@ -732,7 +817,7 @@ class TelegramBot:
parameters=locals()
)
async def kickChatMember(self, chat_id, user_id,
async def kickChatMember(self, chat_id: Union[int, str], user_id,
until_date=None):
"""Kick a user from a group, a supergroup or a channel.
@@ -750,7 +835,8 @@ class TelegramBot:
parameters=locals()
)
async def unbanChatMember(self, chat_id, user_id):
async def unbanChatMember(self, chat_id: Union[int, str], user_id: int,
only_if_banned: bool = True):
"""Unban a previously kicked user in a supergroup or channel.
The user will not return to the group or channel automatically, but
@@ -758,18 +844,18 @@ class TelegramBot:
The bot must be an administrator for this to work.
Return True on success.
See https://core.telegram.org/bots/api#unbanchatmember for details.
If `only_if_banned` is set to False, regular users will be kicked from
chat upon call of this method on them.
"""
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):
async def restrictChatMember(self, chat_id: Union[int, str], user_id: int,
permissions: Dict[str, bool],
until_date: Union[datetime.datetime, int] = None):
"""Restrict a user in a supergroup.
The bot must be an administrator in the supergroup for this to work
@@ -778,21 +864,26 @@ class TelegramBot:
user.
Return True on success.
See https://core.telegram.org/bots/api#restrictchatmember for details.
until_date must be a Unix timestamp.
"""
if isinstance(until_date, datetime.datetime):
until_date = int(until_date.timestamp())
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):
async def promoteChatMember(self, chat_id: Union[int, str], user_id: int,
is_anonymous: bool = None,
can_change_info: bool = None,
can_post_messages: bool = None,
can_edit_messages: bool = None,
can_delete_messages: bool = None,
can_invite_users: bool = None,
can_restrict_members: bool = None,
can_pin_messages: bool = None,
can_promote_members: bool = 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
@@ -806,7 +897,7 @@ class TelegramBot:
parameters=locals()
)
async def exportChatInviteLink(self, chat_id):
async def exportChatInviteLink(self, chat_id: Union[int, str]):
"""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
@@ -821,7 +912,7 @@ class TelegramBot:
parameters=locals()
)
async def setChatPhoto(self, chat_id, photo):
async def setChatPhoto(self, chat_id: Union[int, str], photo):
"""Set a new profile photo for the chat.
Photos can't be changed for private chats.
@@ -836,7 +927,7 @@ class TelegramBot:
parameters=locals()
)
async def deleteChatPhoto(self, chat_id):
async def deleteChatPhoto(self, chat_id: Union[int, str]):
"""Delete a chat photo.
Photos can't be changed for private chats.
@@ -850,7 +941,7 @@ class TelegramBot:
parameters=locals()
)
async def setChatTitle(self, chat_id, title):
async def setChatTitle(self, chat_id: Union[int, str], title):
"""Change the title of a chat.
Titles can't be changed for private chats.
@@ -864,7 +955,7 @@ class TelegramBot:
parameters=locals()
)
async def setChatDescription(self, chat_id, description):
async def setChatDescription(self, chat_id: Union[int, str], 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
@@ -877,8 +968,8 @@ class TelegramBot:
parameters=locals()
)
async def pinChatMessage(self, chat_id, message_id,
disable_notification=None):
async def pinChatMessage(self, chat_id: Union[int, str], message_id,
disable_notification: bool = 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
@@ -892,7 +983,8 @@ class TelegramBot:
parameters=locals()
)
async def unpinChatMessage(self, chat_id):
async def unpinChatMessage(self, chat_id: Union[int, str],
message_id: int = None):
"""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
@@ -906,7 +998,7 @@ class TelegramBot:
parameters=locals()
)
async def leaveChat(self, chat_id):
async def leaveChat(self, chat_id: Union[int, str]):
"""Make the bot leave a group, supergroup or channel.
Return True on success.
@@ -917,7 +1009,7 @@ class TelegramBot:
parameters=locals()
)
async def getChat(self, chat_id):
async def getChat(self, chat_id: Union[int, str]):
"""Get up to date information about the chat.
Return a Chat object on success.
@@ -928,7 +1020,7 @@ class TelegramBot:
parameters=locals()
)
async def getChatAdministrators(self, chat_id):
async def getChatAdministrators(self, chat_id: Union[int, str]):
"""Get a list of administrators in a chat.
On success, return an Array of ChatMember objects that contains
@@ -944,7 +1036,7 @@ class TelegramBot:
parameters=locals()
)
async def getChatMembersCount(self, chat_id):
async def getChatMembersCount(self, chat_id: Union[int, str]):
"""Get the number of members in a chat.
Returns Int on success.
@@ -955,7 +1047,7 @@ class TelegramBot:
parameters=locals()
)
async def getChatMember(self, chat_id, user_id):
async def getChatMember(self, chat_id: Union[int, str], user_id):
"""Get information about a member of a chat.
Returns a ChatMember object on success.
@@ -966,7 +1058,7 @@ class TelegramBot:
parameters=locals()
)
async def setChatStickerSet(self, chat_id, sticker_set_name):
async def setChatStickerSet(self, chat_id: Union[int, str], 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
@@ -981,7 +1073,7 @@ class TelegramBot:
parameters=locals()
)
async def deleteChatStickerSet(self, chat_id):
async def deleteChatStickerSet(self, chat_id: Union[int, str]):
"""Delete a group sticker set from a supergroup.
The bot must be an administrator in the chat for this to work and must
@@ -1014,11 +1106,13 @@ class TelegramBot:
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,
async def editMessageText(self, text: str,
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: str = None,
parse_mode: str = None,
entities: List[dict] = None,
disable_web_page_preview: bool = None,
reply_markup=None):
"""Edit text and game messages.
@@ -1032,10 +1126,12 @@ class TelegramBot:
)
async def editMessageCaption(self,
chat_id=None, message_id=None,
inline_message_id=None,
caption=None,
parse_mode=None,
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: str = None,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
reply_markup=None):
"""Edit captions of messages.
@@ -1049,8 +1145,9 @@ class TelegramBot:
)
async def editMessageMedia(self,
chat_id=None, message_id=None,
inline_message_id=None,
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: str = None,
media=None,
reply_markup=None):
"""Edit animation, audio, document, photo, or video messages.
@@ -1070,8 +1167,9 @@ class TelegramBot:
)
async def editMessageReplyMarkup(self,
chat_id=None, message_id=None,
inline_message_id=None,
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: str = None,
reply_markup=None):
"""Edit only the reply markup of messages.
@@ -1085,7 +1183,7 @@ class TelegramBot:
parameters=locals()
)
async def stopPoll(self, chat_id, message_id,
async def stopPoll(self, chat_id: Union[int, str], message_id,
reply_markup=None):
"""Stop a poll which was sent by the bot.
@@ -1098,7 +1196,7 @@ class TelegramBot:
parameters=locals()
)
async def deleteMessage(self, chat_id, message_id):
async def deleteMessage(self, chat_id: Union[int, str], message_id):
"""Delete a message, including service messages.
- A message can only be deleted if it was sent less than 48 hours
@@ -1121,19 +1219,28 @@ class TelegramBot:
parameters=locals()
)
async def sendSticker(self, chat_id, sticker,
disable_notification=None,
reply_to_message_id=None,
async def sendSticker(self, chat_id: Union[int, str],
sticker: Union[str, dict, IO],
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = 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(
sticker = self.prepare_file_object(sticker)
if sticker is None:
logging.error("Invalid sticker provided!")
return
result = await self.api_request(
'sendSticker',
parameters=locals()
)
if type(sticker) is dict: # Close sticker file, if it was open
sticker['file'].close()
return result
async def getStickerSet(self, name):
"""Get a sticker set.
@@ -1162,32 +1269,57 @@ class TelegramBot:
parameters=locals()
)
async def createNewStickerSet(self, user_id,
name, title, png_sticker, emojis,
contains_masks=None,
mask_position=None):
async def createNewStickerSet(self, user_id: int, name: str, title: str,
emojis: str,
png_sticker: Union[str, dict, IO] = None,
tgs_sticker: Union[str, dict, IO] = None,
contains_masks: bool = None,
mask_position: dict = 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(
png_sticker = self.prepare_file_object(png_sticker)
tgs_sticker = self.prepare_file_object(tgs_sticker)
if png_sticker is None and tgs_sticker is None:
logging.error("Invalid sticker provided!")
return
result = await self.api_request(
'createNewStickerSet',
parameters=locals()
)
if type(png_sticker) is dict: # Close png_sticker file, if it was open
png_sticker['file'].close()
if type(tgs_sticker) is dict: # Close tgs_sticker file, if it was open
tgs_sticker['file'].close()
return result
async def addStickerToSet(self, user_id, name, png_sticker, emojis,
mask_position=None):
async def addStickerToSet(self, user_id: int, name: str,
emojis: str,
png_sticker: Union[str, dict, IO] = None,
tgs_sticker: Union[str, dict, IO] = None,
mask_position: dict = 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(
png_sticker = self.prepare_file_object(png_sticker)
tgs_sticker = self.prepare_file_object(tgs_sticker)
if png_sticker is None and tgs_sticker is None:
logging.error("Invalid sticker provided!")
return
result = await self.api_request(
'addStickerToSet',
parameters=locals()
)
if type(png_sticker) is dict: # Close png_sticker file, if it was open
png_sticker['file'].close()
if type(tgs_sticker) is dict: # Close tgs_sticker file, if it was open
tgs_sticker['file'].close()
return result
async def setStickerPositionInSet(self, sticker, position):
"""Move a sticker in a set created by the bot to a specific position .
@@ -1231,22 +1363,24 @@ class TelegramBot:
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,
async def sendInvoice(self, chat_id: int, title: str, description: str,
payload: str, provider_token: str,
start_parameter: str, currency: str, prices: List[dict],
provider_data: str = None,
photo_url: str = None,
photo_size: int = None,
photo_width: int = None,
photo_height: int = None,
need_name: bool = None,
need_phone_number: bool = None,
need_email: bool = None,
need_shipping_address: bool = None,
send_phone_number_to_provider: bool = None,
send_email_to_provider: bool = None,
is_flexible: bool = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send an invoice.
@@ -1315,10 +1449,11 @@ class TelegramBot:
parameters=locals()
)
async def sendGame(self, chat_id, game_short_name,
disable_notification=None,
reply_to_message_id=None,
reply_markup=None):
async def sendGame(self, chat_id: Union[int, str], game_short_name,
disable_notification: bool = None,
reply_to_message_id: int = None,
reply_markup=None,
allow_sending_without_reply: bool = None):
"""Send a game.
On success, the sent Message is returned.
@@ -1330,11 +1465,12 @@ class TelegramBot:
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):
async def setGameScore(self, user_id: int, score: int,
force: bool = None,
disable_edit_message: bool = None,
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: str = None):
"""Set the score of the specified user in a game.
On success, if the message was sent by the bot, returns the edited
@@ -1350,8 +1486,9 @@ class TelegramBot:
)
async def getGameHighScores(self, user_id,
chat_id=None, message_id=None,
inline_message_id=None):
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: str = None):
"""Get data for high score tables.
Will return the score of the specified user and several of his
@@ -1372,8 +1509,9 @@ class TelegramBot:
async def sendDice(self,
chat_id: Union[int, str] = None,
emoji: str = None,
disable_notification: bool = False,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None):
"""Send a dice.

150
davtelepot/bot.py
View File

@@ -43,7 +43,7 @@ import re
import sys
from collections import OrderedDict
from typing import Callable, Union, Dict
from typing import Callable, List, Union, Dict
# Third party modules
from aiohttp import web
@@ -1278,16 +1278,19 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
return await method(update=update, *args, **kwargs)
raise Exception("Unsupported keyword arguments for `Bot().reply`.")
async def send_message(self, chat_id=None, text=None,
parse_mode='HTML',
disable_web_page_preview=None,
disable_notification=None,
reply_to_message_id=None,
async def send_message(self, chat_id: Union[int, str] = None,
text: str = None,
entities: List[dict] = None,
parse_mode: str = 'HTML',
disable_web_page_preview: bool = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None,
update=None,
reply_to_update=False,
send_default_keyboard=True,
user_record=None):
update: dict = None,
reply_to_update: bool = False,
send_default_keyboard: bool = True,
user_record: OrderedDict = None):
"""Send text via message(s).
This method wraps lower-level `TelegramBot.sendMessage` method.
@@ -1357,9 +1360,11 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
chat_id=chat_id,
text=text_chunk,
parse_mode=parse_mode,
entities=entities,
disable_web_page_preview=disable_web_page_preview,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=_reply_markup
)
return sent_message_update
@@ -1376,13 +1381,16 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
self.final_tasks.remove(task)
return
async def edit_message_text(self, text,
chat_id=None, message_id=None,
inline_message_id=None,
parse_mode='HTML',
disable_web_page_preview=None,
async def edit_message_text(self, text: str,
chat_id: Union[int, str] = None,
message_id: int = None,
inline_message_id: str = None,
parse_mode: str = 'HTML',
entities: List[dict] = None,
disable_web_page_preview: bool = None,
allow_sending_without_reply: bool = None,
reply_markup=None,
update=None):
update: dict = None):
"""Edit message text, sending new messages if necessary.
This method wraps lower-level `TelegramBot.editMessageText` method.
@@ -1418,6 +1426,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
message_id=message_id,
inline_message_id=inline_message_id,
parse_mode=parse_mode,
entities=entities,
disable_web_page_preview=disable_web_page_preview,
reply_markup=(reply_markup if is_last else None)
)
@@ -1433,7 +1442,9 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
text=text_chunk,
chat_id=chat_id,
parse_mode=parse_mode,
entities=entities,
disable_web_page_preview=disable_web_page_preview,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=(reply_markup if is_last else None),
update=updates[-1],
reply_to_update=True,
@@ -1512,16 +1523,18 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
**message_identifier
)
async def send_photo(self, chat_id=None, photo=None,
caption=None,
parse_mode=None,
disable_notification=None,
reply_to_message_id=None,
async def send_photo(self, chat_id: Union[int, str], photo,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None,
update=None,
reply_to_update=False,
send_default_keyboard=True,
use_stored_file_id=True):
update: dict = None,
reply_to_update: bool = False,
send_default_keyboard: bool = True,
use_stored_file_id: bool = True):
"""Send photos.
This method wraps lower-level `TelegramBot.sendPhoto` method.
@@ -1594,8 +1607,10 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
photo=photo,
caption=caption,
parse_mode=parse_mode,
caption_entities=caption_entities,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup
)
if isinstance(sent_update, Exception):
@@ -1629,20 +1644,22 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
)
return sent_update
async def send_audio(self, chat_id=None, audio=None,
caption=None,
duration=None,
performer=None,
title=None,
async def send_audio(self, chat_id: Union[int, str], audio,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
duration: int = None,
performer: str = None,
title: str = None,
thumb=None,
parse_mode=None,
disable_notification=None,
reply_to_message_id=None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None,
update=None,
reply_to_update=False,
send_default_keyboard=True,
use_stored_file_id=True):
update: dict = None,
reply_to_update: bool = False,
send_default_keyboard: bool = True,
use_stored_file_id: bool = True):
"""Send audio files.
This method wraps lower-level `TelegramBot.sendAudio` method.
@@ -1714,13 +1731,15 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
chat_id=chat_id,
audio=audio,
caption=caption,
parse_mode=parse_mode,
caption_entities=caption_entities,
duration=duration,
performer=performer,
title=title,
thumb=thumb,
parse_mode=parse_mode,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup
)
if isinstance(sent_update, Exception):
@@ -1753,17 +1772,19 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
)
return sent_update
async def send_voice(self, chat_id=None, voice=None,
caption=None,
duration=None,
parse_mode=None,
disable_notification=None,
reply_to_message_id=None,
async def send_voice(self, chat_id: Union[int, str], voice,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
duration: int = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None,
update=None,
reply_to_update=False,
send_default_keyboard=True,
use_stored_file_id=True):
update: dict = None,
reply_to_update: bool = False,
send_default_keyboard: bool = True,
use_stored_file_id: bool = True):
"""Send voice messages.
This method wraps lower-level `TelegramBot.sendVoice` method.
@@ -1835,10 +1856,12 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
chat_id=chat_id,
voice=voice,
caption=caption,
duration=duration,
parse_mode=parse_mode,
caption_entities=caption_entities,
duration=duration,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup
)
if isinstance(sent_update, Exception):
@@ -1871,16 +1894,22 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
)
return sent_update
async def send_document(self, chat_id=None, document=None, thumb=None,
caption=None, parse_mode=None,
disable_notification=None,
reply_to_message_id=None, reply_markup=None,
document_path=None,
document_name=None,
update=None,
reply_to_update=False,
send_default_keyboard=True,
use_stored_file_id=False):
async def send_document(self, chat_id: Union[int, str], document,
thumb=None,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
disable_content_type_detection: bool = None,
disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
reply_markup=None,
document_path: str = None,
document_name: str = None,
update: dict = None,
reply_to_update: bool = False,
send_default_keyboard: bool = True,
use_stored_file_id: bool = False):
"""Send a document.
This method wraps lower-level `TelegramBot.sendDocument` method.
@@ -2014,8 +2043,11 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
thumb=thumb,
caption=caption,
parse_mode=parse_mode,
caption_entities=caption_entities,
disable_content_type_detection=disable_content_type_detection,
disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup
)
if isinstance(sent_update, Exception):