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

Compliance with bot API 7.1

Browse Source 
- Added the class ReplyParameters and replaced parameters reply_to_message_id and allow_sending_without_reply in the methods copyMessage, sendMessage, sendPhoto, sendVideo, sendAnimation, sendAudio, sendDocument, sendSticker, sendVideoNote, sendVoice, sendLocation, sendVenue, sendContact, sendPoll, sendDice, sendInvoice, sendGame, and sendMediaGroup with the field reply_parameters of type ReplyParameters.
- Added the class LinkPreviewOptions and replaced the parameter disable_web_page_preview with link_preview_options in the methods sendMessage and editMessageText.
This commit is contained in:
Davte 2024-03-10 20:22:05 +01:00
parent a3b28bc1d6
commit a343e095e8
Signed by: Davte
GPG Key ID: 70336F92E6814706
3 changed files with 415 additions and 120 deletions
Show Stats 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" __email__ = "davide@davte.it"
__credits__ = ["Marco Origlia", "Nick Lee @Nickoala"] __credits__ = ["Marco Origlia", "Nick Lee @Nickoala"]
__license__ = "GNU General Public License v3.0" __license__ = "GNU General Public License v3.0"
__version__ = "2.9.12" __version__ = "2.10.1"
__maintainer__ = "Davide Testa" __maintainer__ = "Davide Testa"
__contact__ = "t.me/davte" __contact__ = "t.me/davte"

406
davtelepot/api.py
View File

@ -7,6 +7,7 @@ A simple aiohttp asynchronous web client is used to make requests.
# Standard library modules # Standard library modules
import asyncio import asyncio
import datetime import datetime
import inspect
import io import io
import json import json
import logging import logging
@ -349,6 +350,104 @@ class InlineQueryResultsButton(dict):
return return
class DictToDump(dict):
def dumps(self):
parameters = {key: value for key, value in self.items() if value}
return json.dumps(parameters, separators=(',', ':'))
class ReplyParameters(DictToDump):
def __init__(self, message_id: int,
chat_id: Union[int, str] = None,
allow_sending_without_reply: bool = None,
quote: str = None,
quote_parse_mode: str = None,
quote_entities: list = None,
quote_position: int = None):
super().__init__(self)
self['message_id'] = message_id
self['chat_id'] = chat_id
self['allow_sending_without_reply'] = allow_sending_without_reply
self['quote'] = quote
self['quote_parse_mode'] = quote_parse_mode
self['quote_entities'] = quote_entities
self['quote_position'] = quote_position
class LinkPreviewOptions(DictToDump):
def __init__(self,
is_disabled: bool = None,
url: str = None,
prefer_small_media: bool = None,
prefer_large_media: bool = None,
show_above_text: bool = None):
super().__init__(self)
self['is_disabled'] = is_disabled
self['url'] = url
self['prefer_small_media'] = prefer_small_media
self['prefer_large_media'] = prefer_large_media
self['show_above_text'] = show_above_text
class ReactionType(DictToDump):
def __init__(self,
type_: str,
emoji: str = None,
custom_emoji_id: str = None):
super().__init__(self)
if type_ not in ('emoji', 'custom_emoji'):
raise TypeError(
f"ReactionType must be `emoji` or `custom_emoji`.\n"
f"Unknown type {type_}"
)
self['type'] = type_
if emoji and custom_emoji_id:
raise TypeError(
"One and only one of the two fields `emoji` or `custom_emoji` "
"may be not None."
)
elif emoji:
self['emoji'] = emoji
elif custom_emoji_id:
self['custom_emoji_id'] = custom_emoji_id
else:
raise TypeError(
"At least one of the two fields `emoji` or `custom_emoji` "
"must be provided and not None."
)
def handle_deprecated_disable_web_page_preview(parameters: dict,
kwargs: dict):
if 'disable_web_page_preview' in kwargs:
if parameters['link_preview_options'] is None:
parameters['link_preview_options'] = LinkPreviewOptions()
parameters['link_preview_options']['is_disabled'] = True
logging.error("DEPRECATION WARNING: `disable_web_page_preview` "
f"parameter of function `{inspect.stack()[2][3]}` has been "
"deprecated since Bot API 7.0. "
"Use `link_preview_options` instead.")
return parameters
def handle_deprecated_reply_parameters(parameters: dict,
kwargs: dict):
if 'reply_to_message_id' in kwargs and kwargs['reply_to_message_id']:
if parameters['reply_parameters'] is None:
parameters['reply_parameters'] = ReplyParameters(
message_id=kwargs['reply_to_message_id']
)
parameters['reply_parameters']['message_id'] = kwargs['reply_to_message_id']
if 'allow_sending_without_reply' in kwargs:
parameters['reply_parameters'][
'allow_sending_without_reply'
] = kwargs['allow_sending_without_reply']
logging.error(f"DEPRECATION WARNING: `reply_to_message_id` and "
f"`allow_sending_without_reply` parameters of function "
f"`{inspect.stack()[2][3]}` have been deprecated since "
f"Bot API 7.0. Use `reply_parameters` instead.")
return parameters
# This class needs to mirror Telegram API, so camelCase method are needed # This class needs to mirror Telegram API, so camelCase method are needed
# noinspection PyPep8Naming # noinspection PyPep8Naming
@ -491,6 +590,8 @@ class TelegramBot:
if (type(value) in (int, list,) if (type(value) in (int, list,)
or (type(value) is dict and 'file' not in value)): or (type(value) is dict and 'file' not in value)):
value = json.dumps(value, separators=(',', ':')) value = json.dumps(value, separators=(',', ':'))
elif isinstance(value, DictToDump):
value = value.dumps()
data.add_field(key, value) data.add_field(key, value)
return data return data
@ -757,19 +858,27 @@ class TelegramBot:
message_thread_id: int = None, message_thread_id: int = None,
parse_mode: str = None, parse_mode: str = None,
entities: List[dict] = None, entities: List[dict] = None,
disable_web_page_preview: bool = None, link_preview_options: LinkPreviewOptions = None,
disable_notification: bool = None, disable_notification: bool = None,
protect_content: bool = None, protect_content: bool = None,
reply_to_message_id: int = None, reply_parameters: ReplyParameters = None,
allow_sending_without_reply: bool = None, reply_markup=None,
reply_markup=None): **kwargs):
"""Send a text message. On success, return it. """Send a text message. On success, return it.
See https://core.telegram.org/bots/api#sendmessage for details. See https://core.telegram.org/bots/api#sendmessage for details.
""" """
parameters = handle_deprecated_disable_web_page_preview(
parameters=locals().copy(),
kwargs=kwargs
)
parameters = handle_deprecated_reply_parameters(
parameters=parameters,
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendMessage', 'sendMessage',
parameters=locals() parameters=parameters
) )
async def forwardMessage(self, chat_id: Union[int, str], async def forwardMessage(self, chat_id: Union[int, str],
@ -794,17 +903,21 @@ class TelegramBot:
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
has_spoiler: bool = None, has_spoiler: bool = None,
reply_markup=None): reply_parameters: ReplyParameters = None,
reply_markup=None,
**kwargs):
"""Send a photo from file_id, HTTP url or file. """Send a photo from file_id, HTTP url or file.
See https://core.telegram.org/bots/api#sendphoto for details. See https://core.telegram.org/bots/api#sendphoto for details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendPhoto', 'sendPhoto',
parameters=locals() parameters=parameters
) )
async def sendAudio(self, chat_id: Union[int, str], audio, async def sendAudio(self, chat_id: Union[int, str], audio,
@ -816,10 +929,9 @@ class TelegramBot:
title: str = None, title: str = None,
thumbnail=None, thumbnail=None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_parameters: ReplyParameters = None,
reply_markup=None, reply_markup=None,
**kwargs): **kwargs):
"""Send an audio file from file_id, HTTP url or file. """Send an audio file from file_id, HTTP url or file.
@ -831,9 +943,13 @@ class TelegramBot:
logging.error("DEPRECATION WARNING: `thumb` parameter of function" logging.error("DEPRECATION WARNING: `thumb` parameter of function"
"`sendAudio` has been deprecated since Bot API 6.6. " "`sendAudio` has been deprecated since Bot API 6.6. "
"Use `thumbnail` instead.") "Use `thumbnail` instead.")
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendAudio', 'sendAudio',
parameters=locals() parameters=parameters
) )
async def sendDocument(self, chat_id: Union[int, str], document, async def sendDocument(self, chat_id: Union[int, str], document,
@ -843,10 +959,9 @@ class TelegramBot:
caption_entities: List[dict] = None, caption_entities: List[dict] = None,
disable_content_type_detection: bool = None, disable_content_type_detection: bool = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_parameters: ReplyParameters = None,
reply_markup=None, reply_markup=None,
**kwargs): **kwargs):
"""Send a document from file_id, HTTP url or file. """Send a document from file_id, HTTP url or file.
@ -858,9 +973,13 @@ class TelegramBot:
logging.error("DEPRECATION WARNING: `thumb` parameter of function" logging.error("DEPRECATION WARNING: `thumb` parameter of function"
"`sendDocument` has been deprecated since Bot API 6.6. " "`sendDocument` has been deprecated since Bot API 6.6. "
"Use `thumbnail` instead.") "Use `thumbnail` instead.")
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendDocument', 'sendDocument',
parameters=locals() parameters=parameters
) )
async def sendVideo(self, chat_id: Union[int, str], video, async def sendVideo(self, chat_id: Union[int, str], video,
@ -873,11 +992,10 @@ class TelegramBot:
caption_entities: List[dict] = None, caption_entities: List[dict] = None,
supports_streaming: bool = None, supports_streaming: bool = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
has_spoiler: bool = None, has_spoiler: bool = None,
reply_parameters: ReplyParameters = None,
reply_markup=None, reply_markup=None,
**kwargs): **kwargs):
"""Send a video from file_id, HTTP url or file. """Send a video from file_id, HTTP url or file.
@ -889,9 +1007,13 @@ class TelegramBot:
logging.error("DEPRECATION WARNING: `thumb` parameter of function" logging.error("DEPRECATION WARNING: `thumb` parameter of function"
"`sendVideo` has been deprecated since Bot API 6.6. " "`sendVideo` has been deprecated since Bot API 6.6. "
"Use `thumbnail` instead.") "Use `thumbnail` instead.")
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendVideo', 'sendVideo',
parameters=locals() parameters=parameters
) )
async def sendAnimation(self, chat_id: Union[int, str], animation, async def sendAnimation(self, chat_id: Union[int, str], animation,
@ -903,11 +1025,10 @@ class TelegramBot:
parse_mode: str = None, parse_mode: str = None,
caption_entities: List[dict] = None, caption_entities: List[dict] = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
has_spoiler: bool = None, has_spoiler: bool = None,
reply_parameters: ReplyParameters = None,
reply_markup=None, reply_markup=None,
**kwargs): **kwargs):
"""Send animation files (GIF or H.264/MPEG-4 AVC video without sound). """Send animation files (GIF or H.264/MPEG-4 AVC video without sound).
@ -919,9 +1040,13 @@ class TelegramBot:
logging.error("DEPRECATION WARNING: `thumb` parameter of function" logging.error("DEPRECATION WARNING: `thumb` parameter of function"
"`sendAnimation` has been deprecated since Bot API 6.6. " "`sendAnimation` has been deprecated since Bot API 6.6. "
"Use `thumbnail` instead.") "Use `thumbnail` instead.")
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendAnimation', 'sendAnimation',
parameters=locals() parameters=parameters
) )
async def sendVoice(self, chat_id: Union[int, str], voice, async def sendVoice(self, chat_id: Union[int, str], voice,
@ -930,19 +1055,23 @@ class TelegramBot:
caption_entities: List[dict] = None, caption_entities: List[dict] = None,
duration: int = None, duration: int = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_markup=None): reply_parameters: ReplyParameters = None,
reply_markup=None,
**kwargs):
"""Send an audio file to be displayed as playable voice message. """Send an audio file to be displayed as playable voice message.
`voice` must be in an .ogg file encoded with OPUS. `voice` must be in an .ogg file encoded with OPUS.
See https://core.telegram.org/bots/api#sendvoice for details. See https://core.telegram.org/bots/api#sendvoice for details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendVoice', 'sendVoice',
parameters=locals() parameters=parameters
) )
async def sendVideoNote(self, chat_id: Union[int, str], video_note, async def sendVideoNote(self, chat_id: Union[int, str], video_note,
@ -950,10 +1079,9 @@ class TelegramBot:
length: int = None, length: int = None,
thumbnail=None, thumbnail=None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_parameters: ReplyParameters = None,
reply_markup=None, reply_markup=None,
**kwargs): **kwargs):
"""Send a rounded square mp4 video message of up to 1 minute long. """Send a rounded square mp4 video message of up to 1 minute long.
@ -965,26 +1093,34 @@ class TelegramBot:
logging.error("DEPRECATION WARNING: `thumb` parameter of function" logging.error("DEPRECATION WARNING: `thumb` parameter of function"
"`sendVideoNote` has been deprecated since Bot API 6.6. " "`sendVideoNote` has been deprecated since Bot API 6.6. "
"Use `thumbnail` instead.") "Use `thumbnail` instead.")
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendVideoNote', 'sendVideoNote',
parameters=locals() parameters=parameters
) )
async def sendMediaGroup(self, chat_id: Union[int, str], media: list, async def sendMediaGroup(self, chat_id: Union[int, str], media: list,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
allow_sending_without_reply: bool = None): reply_parameters: ReplyParameters = None,
**kwargs):
"""Send a group of photos or videos as an album. """Send a group of photos or videos as an album.
`media` must be a list of `InputMediaPhoto` and/or `InputMediaVideo` `media` must be a list of `InputMediaPhoto` and/or `InputMediaVideo`
objects. objects.
See https://core.telegram.org/bots/api#sendmediagroup for details. See https://core.telegram.org/bots/api#sendmediagroup for details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendMediaGroup', 'sendMediaGroup',
parameters=locals() parameters=parameters
) )
async def sendLocation(self, chat_id: Union[int, str], async def sendLocation(self, chat_id: Union[int, str],
@ -994,11 +1130,11 @@ class TelegramBot:
heading: int = None, heading: int = None,
proximity_alert_radius: int = None, proximity_alert_radius: int = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_markup=None): reply_parameters: ReplyParameters = None,
reply_markup=None,
**kwargs):
"""Send a point on the map. May be kept updated for a `live_period`. """Send a point on the map. May be kept updated for a `live_period`.
See https://core.telegram.org/bots/api#sendlocation for details. See https://core.telegram.org/bots/api#sendlocation for details.
@ -1011,9 +1147,13 @@ class TelegramBot:
heading = max(1, min(heading, 360)) heading = max(1, min(heading, 360))
if proximity_alert_radius: # Distance 1-100000 m if proximity_alert_radius: # Distance 1-100000 m
proximity_alert_radius = max(1, min(proximity_alert_radius, 100000)) proximity_alert_radius = max(1, min(proximity_alert_radius, 100000))
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendLocation', 'sendLocation',
parameters=locals() parameters=parameters
) )
async def editMessageLiveLocation(self, latitude: float, longitude: float, async def editMessageLiveLocation(self, latitude: float, longitude: float,
@ -1072,19 +1212,23 @@ class TelegramBot:
google_place_id: str = None, google_place_id: str = None,
google_place_type: str = None, google_place_type: str = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_markup=None): reply_parameters: ReplyParameters = None,
reply_markup=None,
**kwargs):
"""Send information about a venue. """Send information about a venue.
Integrated with FourSquare. Integrated with FourSquare.
See https://core.telegram.org/bots/api#sendvenue for details. See https://core.telegram.org/bots/api#sendvenue for details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendVenue', 'sendVenue',
parameters=locals() parameters=parameters
) )
async def sendContact(self, chat_id: Union[int, str], async def sendContact(self, chat_id: Union[int, str],
@ -1093,18 +1237,22 @@ class TelegramBot:
last_name: str = None, last_name: str = None,
vcard: str = None, vcard: str = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_markup=None): reply_parameters: ReplyParameters = None,
reply_markup=None,
**kwargs):
"""Send a phone contact. """Send a phone contact.
See https://core.telegram.org/bots/api#sendcontact for details. See https://core.telegram.org/bots/api#sendcontact for details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendContact', 'sendContact',
parameters=locals() parameters=parameters
) )
async def sendPoll(self, async def sendPoll(self,
@ -1122,11 +1270,11 @@ class TelegramBot:
close_date: Union[int, datetime.datetime] = None, close_date: Union[int, datetime.datetime] = None,
is_closed: bool = None, is_closed: bool = None,
disable_notification: bool = None, disable_notification: bool = None,
allow_sending_without_reply: bool = None,
reply_to_message_id: int = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_markup=None): reply_parameters: ReplyParameters = None,
reply_markup=None,
**kwargs):
"""Send a native poll in a group, a supergroup or channel. """Send a native poll in a group, a supergroup or channel.
See https://core.telegram.org/bots/api#sendpoll for details. See https://core.telegram.org/bots/api#sendpoll for details.
@ -1150,6 +1298,10 @@ class TelegramBot:
parameters = locals().copy() parameters = locals().copy()
parameters['type'] = parameters['type_'] parameters['type'] = parameters['type_']
del parameters['type_'] del parameters['type_']
parameters = handle_deprecated_reply_parameters(
parameters=parameters,
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendPoll', 'sendPoll',
parameters=parameters parameters=parameters
@ -1497,17 +1649,22 @@ class TelegramBot:
inline_message_id: str = None, inline_message_id: str = None,
parse_mode: str = None, parse_mode: str = None,
entities: List[dict] = None, entities: List[dict] = None,
disable_web_page_preview: bool = None, link_preview_options: LinkPreviewOptions = None,
reply_markup=None): reply_markup=None,
**kwargs):
"""Edit text and game messages. """Edit text and game messages.
On success, if edited message is sent by the bot, the edited Message On success, if edited message is sent by the bot, the edited Message
is returned, otherwise True is returned. is returned, otherwise True is returned.
See https://core.telegram.org/bots/api#editmessagetext for details. See https://core.telegram.org/bots/api#editmessagetext for details.
""" """
parameters = handle_deprecated_disable_web_page_preview(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'editMessageText', 'editMessageText',
parameters=locals() parameters=parameters
) )
async def editMessageCaption(self, async def editMessageCaption(self,
@ -1604,15 +1761,28 @@ class TelegramBot:
parameters=locals() parameters=locals()
) )
async def deleteMessages(self, chat_id: Union[int, str],
message_ids: List[int]):
"""Delete multiple messages simultaneously.
If some of the specified messages can't be found, they are skipped.
Returns True on success.
See https://core.telegram.org/bots/api#deletemessages for details.
"""
return await self.api_request(
'deleteMessages',
parameters=locals()
)
async def sendSticker(self, chat_id: Union[int, str], async def sendSticker(self, chat_id: Union[int, str],
sticker: Union[str, dict, IO], sticker: Union[str, dict, IO],
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
emoji: str = None, emoji: str = None,
reply_markup=None): reply_parameters: ReplyParameters = None,
reply_markup=None,
**kwargs):
"""Send `.webp` stickers. """Send `.webp` stickers.
`sticker` must be a file path, a URL, a file handle or a dict `sticker` must be a file path, a URL, a file handle or a dict
@ -1624,9 +1794,13 @@ class TelegramBot:
if sticker is None: if sticker is None:
logging.error("Invalid sticker provided!") logging.error("Invalid sticker provided!")
return return
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
result = await self.api_request( result = await self.api_request(
'sendSticker', 'sendSticker',
parameters=locals() parameters=parameters
) )
if type(sticker) is dict: # Close sticker file, if it was open if type(sticker) is dict: # Close sticker file, if it was open
sticker['file'].close() sticker['file'].close()
@ -1715,7 +1889,7 @@ class TelegramBot:
raise TypeError(f"Unknown sticker type `{sticker_type}`.") raise TypeError(f"Unknown sticker type `{sticker_type}`.")
result = await self.api_request( result = await self.api_request(
'createNewStickerSet', 'createNewStickerSet',
parameters=locals(), parameters=locals().copy(),
exclude=['old_sticker_format'] exclude=['old_sticker_format']
) )
return result return result
@ -1749,7 +1923,7 @@ class TelegramBot:
return return
result = await self.api_request( result = await self.api_request(
'addStickerToSet', 'addStickerToSet',
parameters=locals(), parameters=locals().copy(),
exclude=['old_sticker_format'] exclude=['old_sticker_format']
) )
return result return result
@ -1821,17 +1995,21 @@ class TelegramBot:
send_email_to_provider: bool = None, send_email_to_provider: bool = None,
is_flexible: bool = None, is_flexible: bool = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None, reply_parameters: ReplyParameters = None,
allow_sending_without_reply: bool = None, reply_markup=None,
reply_markup=None): **kwargs):
"""Send an invoice. """Send an invoice.
On success, the sent Message is returned. On success, the sent Message is returned.
See https://core.telegram.org/bots/api#sendinvoice for details. See https://core.telegram.org/bots/api#sendinvoice for details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendInvoice', 'sendInvoice',
parameters=locals() parameters=parameters
) )
async def answerShippingQuery(self, shipping_query_id, ok, async def answerShippingQuery(self, shipping_query_id, ok,
@ -1895,18 +2073,22 @@ class TelegramBot:
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None, reply_parameters: ReplyParameters = None,
reply_markup=None, reply_markup=None,
allow_sending_without_reply: bool = None): **kwargs):
"""Send a game. """Send a game.
On success, the sent Message is returned. On success, the sent Message is returned.
See https://core.telegram.org/bots/api#sendgame for See https://core.telegram.org/bots/api#sendgame for
details. details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendGame', 'sendGame',
parameters=locals() parameters=parameters
) )
async def setGameScore(self, user_id: int, score: int, async def setGameScore(self, user_id: int, score: int,
@ -1954,11 +2136,11 @@ class TelegramBot:
chat_id: Union[int, str] = None, chat_id: Union[int, str] = None,
emoji: str = None, emoji: str = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None,
allow_sending_without_reply: bool = None,
message_thread_id: int = None, message_thread_id: int = None,
protect_content: bool = None, protect_content: bool = None,
reply_markup=None): reply_parameters: ReplyParameters = None,
reply_markup=None,
**kwargs):
"""Send a dice. """Send a dice.
Use this method to send a dice, which will have a random value from 1 Use this method to send a dice, which will have a random value from 1
@ -1969,9 +2151,13 @@ class TelegramBot:
See https://core.telegram.org/bots/api#senddice for See https://core.telegram.org/bots/api#senddice for
details. details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'sendDice', 'sendDice',
parameters=locals() parameters=parameters
) )
async def setChatAdministratorCustomTitle(self, async def setChatAdministratorCustomTitle(self,
@ -2100,9 +2286,9 @@ class TelegramBot:
parse_mode: str = None, parse_mode: str = None,
caption_entities: list = None, caption_entities: list = None,
disable_notification: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None, reply_parameters: ReplyParameters = None,
allow_sending_without_reply: bool = None, reply_markup=None,
reply_markup=None): **kwargs):
"""Use this method to copy messages of any kind. """Use this method to copy messages of any kind.
The method is analogous to the method forwardMessages, but the copied The method is analogous to the method forwardMessages, but the copied
@ -2110,9 +2296,13 @@ class TelegramBot:
Returns the MessageId of the sent message on success. Returns the MessageId of the sent message on success.
See https://core.telegram.org/bots/api#copymessage for details. See https://core.telegram.org/bots/api#copymessage for details.
""" """
parameters = handle_deprecated_reply_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request( return await self.api_request(
'copyMessage', 'copyMessage',
parameters=locals() parameters=parameters
) )
async def unpinAllChatMessages(self, chat_id: Union[int, str]): async def unpinAllChatMessages(self, chat_id: Union[int, str]):
@ -2700,3 +2890,79 @@ class TelegramBot:
'unpinAllGeneralForumTopicMessages', 'unpinAllGeneralForumTopicMessages',
parameters=locals() parameters=locals()
) )
async def getUserChatBoosts(self, chat_id: Union[int, str], user_id: int):
"""Get the list of boosts added to a chat by a user.
Requires administrator rights in the chat.
Returns a UserChatBoosts object.
See https://core.telegram.org/bots/api#getuserchatboosts for details.
"""
return await self.api_request(
'getUserChatBoosts',
parameters=locals()
)
async def forwardMessages(self, chat_id: Union[int, str],
from_chat_id: Union[int, str],
message_ids: List[int],
message_thread_id: int = None,
disable_notification: bool = None,
protect_content: bool = None):
"""Forward multiple messages of any kind.
If some of the specified messages can't be found or forwarded, they are
skipped.
Service messages and messages with protected content can't be
forwarded.
Album grouping is kept for forwarded messages.
On success, an array of MessageId of the sent messages is returned.
See https://core.telegram.org/bots/api#forwardmessages for details.
"""
return await self.api_request(
'forwardMessages',
parameters=locals()
)
async def copyMessages(self, chat_id: Union[int, str],
from_chat_id: Union[int, str],
message_ids: List[int],
message_thread_id: int = None,
disable_notification: bool = None,
protect_content: bool = None,
remove_caption: bool = None):
"""Copy messages of any kind.
If some of the specified messages can't be found or copied, they are
skipped.
Service messages, giveaway messages, giveaway winners messages, and
invoice messages can't be copied.
A quiz poll can be copied only if the value of the field
correct_option_id is known to the bot.
The method is analogous to the method forwardMessages, but the copied
messages don't have a link to the original message.
Album grouping is kept for copied messages.
On success, an array of MessageId of the sent messages is returned.
See https://core.telegram.org/bots/api#copymessages for details.
"""
return await self.api_request(
'copyMessages',
parameters=locals()
)
async def setMessageReaction(self, chat_id: Union[int, str],
message_id: int,
reaction: List[ReactionType] = None,
is_big: bool = None):
"""Change the chosen reactions on a message.
Service messages can't be reacted to.
Automatically forwarded messages from a channel to its discussion group
have the same available reactions as messages in the channel.
Returns True on success.
See https://core.telegram.org/bots/api#setmessagereaction for details.
"""
return await self.api_request(
'setMessageReaction',
parameters=locals()
)

127
davtelepot/bot.py
View File

@ -2,34 +2,6 @@
camelCase methods mirror API directly, while snake_case ones act as middleware camelCase methods mirror API directly, while snake_case ones act as middleware
someway. someway.
Usage
```
import sys
from davtelepot.bot import Bot
from data.passwords import my_token, my_other_token
long_polling_bot = Bot(token=my_token, database_url='my_db')
webhook_bot = Bot(token=my_other_token, hostname='example.com',
certificate='path/to/certificate.pem',
database_url='my_other_db')
@long_polling_bot.command('/foo')
async def foo_command(bot, update, user_record, language):
return "Bar!"
@webhook_bot.command('/bar')
async def bar_command(bot, update, user_record, language):
return "Foo!"
exit_state = Bot.run(
local_host='127.0.0.5',
port=8552
)
sys.exit(exit_state)
```
""" """
# Standard library modules # Standard library modules
@ -49,7 +21,9 @@ from typing import Callable, List, Union, Dict
import aiohttp.web import aiohttp.web
# Project modules # Project modules
from davtelepot.api import TelegramBot, TelegramError from davtelepot.api import (
LinkPreviewOptions, ReplyParameters, TelegramBot, TelegramError
)
from davtelepot.database import ObjectWithDatabase from davtelepot.database import ObjectWithDatabase
from davtelepot.languages import MultiLanguageObject from davtelepot.languages import MultiLanguageObject
from davtelepot.messages import davtelepot_messages from davtelepot.messages import davtelepot_messages
@ -1319,19 +1293,21 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
async def send_message(self, chat_id: Union[int, str] = None, async def send_message(self, chat_id: Union[int, str] = None,
text: str = None, text: str = None,
message_thread_id: int = None,
entities: List[dict] = None, entities: List[dict] = None,
parse_mode: str = 'HTML', parse_mode: str = 'HTML',
message_thread_id: int = None, link_preview_options: LinkPreviewOptions = None,
disable_notification: bool = None,
protect_content: bool = None, protect_content: bool = None,
disable_web_page_preview: bool = None, disable_web_page_preview: bool = None,
disable_notification: bool = None,
reply_to_message_id: int = None, reply_to_message_id: int = None,
allow_sending_without_reply: bool = None, allow_sending_without_reply: bool = None,
reply_markup=None,
update: dict = None, update: dict = None,
reply_to_update: bool = False, reply_to_update: bool = False,
send_default_keyboard: bool = True, send_default_keyboard: bool = True,
user_record: OrderedDict = None): user_record: OrderedDict = None,
reply_parameters: ReplyParameters = None,
reply_markup=None):
"""Send text via message(s). """Send text via message(s).
This method wraps lower-level `TelegramBot.sendMessage` method. This method wraps lower-level `TelegramBot.sendMessage` method.
@ -1352,6 +1328,10 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
user_record = self.db['users'].find_one(telegram_id=chat_id) user_record = self.db['users'].find_one(telegram_id=chat_id)
if reply_to_update and 'message_id' in update: if reply_to_update and 'message_id' in update:
reply_to_message_id = update['message_id'] reply_to_message_id = update['message_id']
if disable_web_page_preview:
if link_preview_options is None:
link_preview_options = LinkPreviewOptions()
link_preview_options['is_disabled'] = True
if ( if (
send_default_keyboard send_default_keyboard
and reply_markup is None and reply_markup is None
@ -1395,19 +1375,27 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
limit=self.__class__.TELEGRAM_MESSAGES_MAX_LEN - 100, limit=self.__class__.TELEGRAM_MESSAGES_MAX_LEN - 100,
parse_mode=parse_mode parse_mode=parse_mode
) )
if reply_to_message_id:
if reply_parameters is None:
reply_parameters = ReplyParameters(message_id=reply_to_message_id)
reply_parameters['message_id'] = reply_to_message_id
if allow_sending_without_reply:
reply_parameters['allow_sending_without_reply'] = allow_sending_without_reply
if reply_to_update and 'chat' in update and 'id' in update['chat']:
if update['chat']['id'] != chat_id:
reply_parameters['chat_id'] = update['chat']['id']
for text_chunk, is_last in text_chunks: for text_chunk, is_last in text_chunks:
_reply_markup = (reply_markup if is_last else None) _reply_markup = (reply_markup if is_last else None)
sent_message_update = await self.sendMessage( sent_message_update = await self.sendMessage(
chat_id=chat_id, chat_id=chat_id,
text=text_chunk, text=text_chunk,
message_thread_id=message_thread_id,
parse_mode=parse_mode, parse_mode=parse_mode,
entities=entities, entities=entities,
message_thread_id=message_thread_id, link_preview_options=link_preview_options,
protect_content=protect_content,
disable_web_page_preview=disable_web_page_preview,
disable_notification=disable_notification, disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id, protect_content=protect_content,
allow_sending_without_reply=allow_sending_without_reply, reply_parameters=reply_parameters,
reply_markup=_reply_markup reply_markup=_reply_markup
) )
return sent_message_update return sent_message_update
@ -1431,6 +1419,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
parse_mode: str = 'HTML', parse_mode: str = 'HTML',
entities: List[dict] = None, entities: List[dict] = None,
disable_web_page_preview: bool = None, disable_web_page_preview: bool = None,
link_preview_options: LinkPreviewOptions = None,
allow_sending_without_reply: bool = None, allow_sending_without_reply: bool = None,
reply_markup=None, reply_markup=None,
update: dict = None): update: dict = None):
@ -1463,6 +1452,10 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
) )
): ):
if i == 0: if i == 0:
if disable_web_page_preview:
if link_preview_options is None:
link_preview_options = LinkPreviewOptions()
link_preview_options['is_disabled'] = True
edited_message = await self.editMessageText( edited_message = await self.editMessageText(
text=text_chunk, text=text_chunk,
chat_id=chat_id, chat_id=chat_id,
@ -1470,7 +1463,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
inline_message_id=inline_message_id, inline_message_id=inline_message_id,
parse_mode=parse_mode, parse_mode=parse_mode,
entities=entities, entities=entities,
disable_web_page_preview=disable_web_page_preview, link_preview_options=link_preview_options,
reply_markup=(reply_markup if is_last else None) reply_markup=(reply_markup if is_last else None)
) )
if chat_id is None: if chat_id is None:
@ -1576,6 +1569,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
reply_markup=None, reply_markup=None,
update: dict = None, update: dict = None,
reply_to_update: bool = False, reply_to_update: bool = False,
reply_parameters: ReplyParameters = None,
send_default_keyboard: bool = True, send_default_keyboard: bool = True,
use_stored_file_id: bool = True): use_stored_file_id: bool = True):
"""Send photos. """Send photos.
@ -1599,6 +1593,15 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
chat_id = self.get_chat_id(update) chat_id = self.get_chat_id(update)
if reply_to_update and 'message_id' in update: if reply_to_update and 'message_id' in update:
reply_to_message_id = update['message_id'] reply_to_message_id = update['message_id']
if reply_to_message_id:
if reply_parameters is None:
reply_parameters = ReplyParameters(message_id=reply_to_message_id)
reply_parameters['message_id'] = reply_to_message_id
if allow_sending_without_reply:
reply_parameters['allow_sending_without_reply'] = allow_sending_without_reply
if reply_to_update and 'chat' in update and 'id' in update['chat']:
if update['chat']['id'] != chat_id:
reply_parameters['chat_id'] = update['chat']['id']
if ( if (
send_default_keyboard send_default_keyboard
and reply_markup is None and reply_markup is None
@ -1652,8 +1655,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
parse_mode=parse_mode, parse_mode=parse_mode,
caption_entities=caption_entities, caption_entities=caption_entities,
disable_notification=disable_notification, disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id, reply_parameters=reply_parameters,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup reply_markup=reply_markup
) )
if isinstance(sent_update, Exception): if isinstance(sent_update, Exception):
@ -1701,6 +1703,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
reply_markup=None, reply_markup=None,
update: dict = None, update: dict = None,
reply_to_update: bool = False, reply_to_update: bool = False,
reply_parameters: ReplyParameters = None,
send_default_keyboard: bool = True, send_default_keyboard: bool = True,
use_stored_file_id: bool = True): use_stored_file_id: bool = True):
"""Send audio files. """Send audio files.
@ -1724,6 +1727,15 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
chat_id = self.get_chat_id(update) chat_id = self.get_chat_id(update)
if reply_to_update and 'message_id' in update: if reply_to_update and 'message_id' in update:
reply_to_message_id = update['message_id'] reply_to_message_id = update['message_id']
if reply_to_message_id:
if reply_parameters is None:
reply_parameters = ReplyParameters(message_id=reply_to_message_id)
reply_parameters['message_id'] = reply_to_message_id
if allow_sending_without_reply:
reply_parameters['allow_sending_without_reply'] = allow_sending_without_reply
if reply_to_update and 'chat' in update and 'id' in update['chat']:
if update['chat']['id'] != chat_id:
reply_parameters['chat_id'] = update['chat']['id']
if ( if (
send_default_keyboard send_default_keyboard
and reply_markup is None and reply_markup is None
@ -1781,8 +1793,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
title=title, title=title,
thumbnail=thumbnail, thumbnail=thumbnail,
disable_notification=disable_notification, disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id, reply_parameters=reply_parameters,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup reply_markup=reply_markup
) )
if isinstance(sent_update, Exception): if isinstance(sent_update, Exception):
@ -1826,6 +1837,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
reply_markup=None, reply_markup=None,
update: dict = None, update: dict = None,
reply_to_update: bool = False, reply_to_update: bool = False,
reply_parameters: ReplyParameters = None,
send_default_keyboard: bool = True, send_default_keyboard: bool = True,
use_stored_file_id: bool = True): use_stored_file_id: bool = True):
"""Send voice messages. """Send voice messages.
@ -1849,6 +1861,15 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
chat_id = self.get_chat_id(update) chat_id = self.get_chat_id(update)
if reply_to_update and 'message_id' in update: if reply_to_update and 'message_id' in update:
reply_to_message_id = update['message_id'] reply_to_message_id = update['message_id']
if reply_to_message_id:
if reply_parameters is None:
reply_parameters = ReplyParameters(message_id=reply_to_message_id)
reply_parameters['message_id'] = reply_to_message_id
if allow_sending_without_reply:
reply_parameters['allow_sending_without_reply'] = allow_sending_without_reply
if reply_to_update and 'chat' in update and 'id' in update['chat']:
if update['chat']['id'] != chat_id:
reply_parameters['chat_id'] = update['chat']['id']
if ( if (
send_default_keyboard send_default_keyboard
and reply_markup is None and reply_markup is None
@ -1903,8 +1924,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
caption_entities=caption_entities, caption_entities=caption_entities,
duration=duration, duration=duration,
disable_notification=disable_notification, disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id, reply_parameters=reply_parameters,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup reply_markup=reply_markup
) )
if isinstance(sent_update, Exception): if isinstance(sent_update, Exception):
@ -1951,6 +1971,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
document_name: str = None, document_name: str = None,
update: dict = None, update: dict = None,
reply_to_update: bool = False, reply_to_update: bool = False,
reply_parameters: ReplyParameters = None,
send_default_keyboard: bool = True, send_default_keyboard: bool = True,
use_stored_file_id: bool = False): use_stored_file_id: bool = False):
"""Send a document. """Send a document.
@ -1983,6 +2004,15 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
return return
if reply_to_update and 'message_id' in update: if reply_to_update and 'message_id' in update:
reply_to_message_id = update['message_id'] reply_to_message_id = update['message_id']
if reply_to_message_id:
if reply_parameters is None:
reply_parameters = ReplyParameters(message_id=reply_to_message_id)
reply_parameters['message_id'] = reply_to_message_id
if allow_sending_without_reply:
reply_parameters['allow_sending_without_reply'] = allow_sending_without_reply
if reply_to_update and 'chat' in update and 'id' in update['chat']:
if update['chat']['id'] != chat_id:
reply_parameters['chat_id'] = update['chat']['id']
if chat_id > 0: if chat_id > 0:
user_record = self.db['users'].find_one(telegram_id=chat_id) user_record = self.db['users'].find_one(telegram_id=chat_id)
language = self.get_language(update=update, user_record=user_record) language = self.get_language(update=update, user_record=user_record)
@ -2061,7 +2091,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
caption=caption, caption=caption,
parse_mode=parse_mode, parse_mode=parse_mode,
disable_notification=disable_notification, disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id, reply_parameters=reply_parameters,
reply_markup=reply_markup, reply_markup=reply_markup,
update=update, update=update,
reply_to_update=reply_to_update, reply_to_update=reply_to_update,
@ -2092,8 +2122,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
caption_entities=caption_entities, caption_entities=caption_entities,
disable_content_type_detection=disable_content_type_detection, disable_content_type_detection=disable_content_type_detection,
disable_notification=disable_notification, disable_notification=disable_notification,
reply_to_message_id=reply_to_message_id, reply_parameters=reply_parameters,
allow_sending_without_reply=allow_sending_without_reply,
reply_markup=reply_markup reply_markup=reply_markup
) )
if isinstance(sent_update, Exception): if isinstance(sent_update, Exception):
@ -3505,7 +3534,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
Each bot will receive updates via long polling or webhook according to Each bot will receive updates via long polling or webhook according to
its initialization parameters. its initialization parameters.
A single aiohttp.web.Application instance will be run (cls.app) on A single aiohttp.web.Application instance will be run (cls.app) on
local_host:port and it may serve custom-defined routes as well. local_host:port, and it may serve custom-defined routes as well.
""" """
if local_host is not None: if local_host is not None:
cls.local_host = local_host cls.local_host = local_host