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

Compliance with bot API 7.2

Browse Source 
- Included python version in `/version` command
- Added the parameter `business_connection_id` to the methods sendMessage, sendPhoto, sendVideo, sendAnimation, sendAudio, sendDocument, sendSticker, sendVideoNote, sendVoice, sendLocation, sendVenue, sendContact, sendPoll, sendDice, sendGame, and sendMediaGroup, sendChatAction.
- Moved the parameter `format` to Sticker instead of StickerSet (and related methods)
This commit is contained in:
Davte 2024-04-13 11:15:42 +02:00
parent a343e095e8
commit 6b489363de
Signed by: Davte
GPG Key ID: 70336F92E6814706
3 changed files with 156 additions and 82 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"
__credits__ = ["Marco Origlia", "Nick Lee @Nickoala"]
__license__ = "GNU General Public License v3.0"
__version__ = "2.10.1"
__version__ = "2.10.2"
__maintainer__ = "Davide Testa"
__contact__ = "t.me/davte"

150
davtelepot/administration_tools.py
View File

@ -13,6 +13,7 @@ import asyncio
import datetime
import json
import logging
import platform
import re
import types
@ -223,9 +224,9 @@ def get_talk_panel(bot: Bot,
return text, reply_markup
async def _talk_command(bot: Bot,
update,
user_record):
async def talk_command(bot: Bot,
update,
user_record):
text = get_cleaned_text(
update,
bot,
@ -343,17 +344,17 @@ async def end_session(bot: Bot,
return
async def _talk_button(bot: Bot,
update,
user_record,
data):
async def talk_button(bot: Bot,
update,
user_record,
data):
telegram_id = user_record['telegram_id']
command, *arguments = data
result, text, reply_markup = '', '', None
if command == 'search':
bot.set_individual_text_message_handler(
await async_wrapper(
_talk_command,
talk_command,
),
update
)
@ -426,7 +427,7 @@ async def _talk_button(bot: Bot,
return result
async def _restart_command(bot: Bot,
async def restart_command(bot: Bot,
update,
user_record):
with bot.db as db:
@ -453,7 +454,7 @@ async def _restart_command(bot: Bot,
return
async def _stop_command(bot: Bot,
async def stop_command(bot: Bot,
update,
user_record):
text = bot.get_message(
@ -495,10 +496,10 @@ async def stop_bots(bot: Bot):
return
async def _stop_button(bot: Bot,
update,
user_record,
data: List[Union[int, str]]):
async def stop_button(bot: Bot,
update,
user_record,
data: List[Union[int, str]]):
result, text, reply_markup = '', '', None
telegram_id = user_record['telegram_id']
command = data[0] if len(data) > 0 else 'None'
@ -535,7 +536,7 @@ async def _stop_button(bot: Bot,
return result
async def _send_bot_database(bot: Bot, user_record: OrderedDict, language: str):
async def send_bot_database(bot: Bot, user_record: OrderedDict, language: str):
if not all(
[
bot.db_url.endswith('.db'),
@ -562,7 +563,7 @@ async def _send_bot_database(bot: Bot, user_record: OrderedDict, language: str):
)
async def _query_command(bot, update, user_record):
async def query_command(bot, update, user_record):
query = get_cleaned_text(
update,
bot,
@ -640,7 +641,7 @@ async def _query_command(bot, update, user_record):
)
async def _query_button(bot, update, user_record, data):
async def query_button(bot, update, user_record, data):
result, text, reply_markup = '', '', None
command = data[0] if len(data) else 'default'
error_message = bot.get_message(
@ -677,7 +678,7 @@ async def _query_button(bot, update, user_record, data):
return result
async def _log_command(bot, update, user_record):
async def log_command(bot, update, user_record):
if bot.log_file_path is None:
return bot.get_message(
'admin', 'log_command', 'no_log',
@ -730,7 +731,7 @@ async def _log_command(bot, update, user_record):
return
async def _errors_command(bot, update, user_record):
async def errors_command(bot, update, user_record):
# Always send errors log file in private chat
chat_id = update['from']['id']
if bot.errors_file_path is None:
@ -774,7 +775,7 @@ async def _errors_command(bot, update, user_record):
return
async def _maintenance_command(bot, update, user_record):
async def maintenance_command(bot, update, user_record):
maintenance_message = get_cleaned_text(update, bot, ['maintenance'])
if maintenance_message.startswith('{'):
maintenance_message = json.loads(maintenance_message)
@ -883,14 +884,15 @@ async def get_new_versions(bot: Bot,
return news
async def _version_command(bot: Bot, update: dict,
user_record: OrderedDict, language: str):
async def version_command(bot: Bot, update: dict,
user_record: OrderedDict, language: str):
last_commit = await get_last_commit()
text = bot.get_message(
'admin', 'version_command', 'header',
last_commit=last_commit,
update=update, user_record=user_record
) + '\n\n'
text += f'<b>Python: </b> <code>{platform.python_version()}</code>\n'
text += '\n'.join(
f"<b>{package.__name__}</b>: "
f"<code>{package.__version__}</code>"
@ -1032,7 +1034,7 @@ async def get_package_updates(bot: Bot,
await asyncio.sleep(monitoring_interval)
async def _send_start_messages(bot: Bot):
async def send_start_messages(bot: Bot):
"""Send restart messages at restart."""
for restart_message in bot.db['restart_messages'].find(sent=None):
asyncio.ensure_future(
@ -1060,7 +1062,7 @@ async def _send_start_messages(bot: Bot):
return
async def _load_talking_sessions(bot: Bot):
async def load_talking_sessions(bot: Bot):
sessions = []
for session in bot.db.query(
"""SELECT *
@ -1139,7 +1141,7 @@ def get_custom_commands(bot: Bot, language: str = None) -> List[dict]:
)
async def _father_command(bot, language):
async def father_command(bot, language):
modes = [
{
key: (
@ -1443,12 +1445,12 @@ async def edit_bot_father_settings_via_message(bot: Bot,
return result, text, reply_markup
async def _father_button(bot: Bot, user_record: OrderedDict,
async def father_button(bot: Bot, user_record: OrderedDict,
language: str, data: list):
"""Handle BotFather button.
Operational modes
- main: back to main page (see _father_command)
- main: back to main page (see `father_command`)
- get: show commands stored by @BotFather
- set: edit commands stored by @BotFather
"""
@ -1542,7 +1544,7 @@ async def _father_button(bot: Bot, user_record: OrderedDict,
elif command == 'main':
return dict(
text='',
edit=(await _father_command(bot=bot, language=language))
edit=(await father_command(bot=bot, language=language))
)
elif command == 'set':
stored_commands = await bot.getMyCommands()
@ -1810,8 +1812,8 @@ async def _father_button(bot: Bot, user_record: OrderedDict,
return result
async def _config_command(bot: Bot, update: dict,
user_record: dict, language: str):
async def config_command(bot: Bot, update: dict,
user_record: dict, language: str):
text = get_cleaned_text(
update,
bot,
@ -1907,16 +1909,16 @@ def init(telegram_bot: Bot,
# Tasks to complete before starting bot
@telegram_bot.additional_task(when='BEFORE')
async def load_talking_sessions():
return await _load_talking_sessions(bot=telegram_bot)
async def _load_talking_sessions():
return await load_talking_sessions(bot=telegram_bot)
@telegram_bot.additional_task(when='BEFORE', bot=telegram_bot)
async def notify_version(bot):
return await notify_new_version(bot=bot)
@telegram_bot.additional_task('BEFORE')
async def send_restart_messages():
return await _send_start_messages(bot=telegram_bot)
async def _send_start_messages():
return await send_start_messages(bot=telegram_bot)
# Administration commands
@telegram_bot.command(command='/db',
@ -1925,10 +1927,10 @@ def init(telegram_bot: Bot,
description=admin_messages[
'db_command']['description'],
authorization_level='admin')
async def send_bot_database(bot, user_record, language):
return await _send_bot_database(bot=bot,
user_record=user_record,
language=language)
async def _send_bot_database(bot, user_record, language):
return await send_bot_database(bot=bot,
user_record=user_record,
language=language)
@telegram_bot.command(command='/errors',
aliases=[],
@ -1936,8 +1938,8 @@ def init(telegram_bot: Bot,
description=admin_messages[
'errors_command']['description'],
authorization_level='admin')
async def errors_command(bot, update, user_record):
return await _errors_command(bot, update, user_record)
async def _errors_command(bot, update, user_record):
return await errors_command(bot, update, user_record)
@telegram_bot.command(command='/father',
aliases=[],
@ -1948,14 +1950,14 @@ def init(telegram_bot: Bot,
if key in ('description', )
},
authorization_level='admin')
async def father_command(bot, language):
return await _father_command(bot=bot, language=language)
async def _father_command(bot, language):
return await father_command(bot=bot, language=language)
@telegram_bot.button(prefix='father:///',
separator='|',
authorization_level='admin')
async def query_button(bot, user_record, language, data):
return await _father_button(bot=bot,
async def _father_button(bot, user_record, language, data):
return await father_button(bot=bot,
user_record=user_record,
language=language,
data=data)
@ -1966,16 +1968,16 @@ def init(telegram_bot: Bot,
description=admin_messages[
'log_command']['description'],
authorization_level='admin')
async def log_command(bot, update, user_record):
return await _log_command(bot, update, user_record)
async def _log_command(bot, update, user_record):
return await log_command(bot, update, user_record)
@telegram_bot.command(command='/maintenance', aliases=[],
show_in_keyboard=False,
description=admin_messages[
'maintenance_command']['description'],
authorization_level='admin')
async def maintenance_command(bot, update, user_record):
return await _maintenance_command(bot, update, user_record)
async def _maintenance_command(bot, update, user_record):
return await maintenance_command(bot, update, user_record)
@telegram_bot.command(command='/query',
aliases=[],
@ -1983,16 +1985,16 @@ def init(telegram_bot: Bot,
description=admin_messages[
'query_command']['description'],
authorization_level='admin')
async def query_command(bot, update, user_record):
return await _query_command(bot, update, user_record)
async def _query_command(bot, update, user_record):
return await query_command(bot, update, user_record)
@telegram_bot.button(prefix='db_query:///',
separator='|',
description=admin_messages[
'query_command']['description'],
authorization_level='admin')
async def query_button(bot, update, user_record, data):
return await _query_button(bot, update, user_record, data)
async def _query_button(bot, update, user_record, data):
return await query_button(bot, update, user_record, data)
@telegram_bot.command(command='/restart',
aliases=[],
@ -2000,8 +2002,8 @@ def init(telegram_bot: Bot,
description=admin_messages[
'restart_command']['description'],
authorization_level='admin')
async def restart_command(bot, update, user_record):
return await _restart_command(bot, update, user_record)
async def _restart_command(bot, update, user_record):
return await restart_command(bot, update, user_record)
@telegram_bot.command(command='/select',
aliases=[],
@ -2009,8 +2011,8 @@ def init(telegram_bot: Bot,
description=admin_messages[
'select_command']['description'],
authorization_level='admin')
async def select_command(bot, update, user_record):
return await _query_command(bot, update, user_record)
async def _select_command(bot, update, user_record):
return await query_command(bot, update, user_record)
@telegram_bot.command(command='/stop',
aliases=[],
@ -2018,16 +2020,16 @@ def init(telegram_bot: Bot,
description=admin_messages[
'stop_command']['description'],
authorization_level='admin')
async def stop_command(bot, update, user_record):
return await _stop_command(bot, update, user_record)
async def _stop_command(bot, update, user_record):
return await stop_command(bot, update, user_record)
@telegram_bot.button(prefix='stop:///',
separator='|',
description=admin_messages[
'stop_command']['description'],
authorization_level='admin')
async def stop_button(bot, update, user_record, data):
return await _stop_button(bot, update, user_record, data)
async def _stop_button(bot, update, user_record, data):
return await stop_button(bot, update, user_record, data)
@telegram_bot.command(command='/talk',
aliases=[],
@ -2035,14 +2037,14 @@ def init(telegram_bot: Bot,
description=admin_messages[
'talk_command']['description'],
authorization_level='admin')
async def talk_command(bot, update, user_record):
return await _talk_command(bot, update, user_record)
async def _talk_command(bot, update, user_record):
return await talk_command(bot, update, user_record)
@telegram_bot.button(prefix='talk:///',
separator='|',
authorization_level='admin')
async def talk_button(bot, update, user_record, data):
return await _talk_button(bot, update, user_record, data)
async def _talk_button(bot, update, user_record, data):
return await talk_button(bot, update, user_record, data)
@telegram_bot.command(command='/version',
aliases=[],
@ -2053,11 +2055,11 @@ def init(telegram_bot: Bot,
},
show_in_keyboard=False,
authorization_level='admin')
async def version_command(bot, update, user_record, language):
return await _version_command(bot=bot,
update=update,
user_record=user_record,
language=language)
async def _version_command(bot, update, user_record, language):
return await version_command(bot=bot,
update=update,
user_record=user_record,
language=language)
@telegram_bot.command(command='/config',
aliases=[],
@ -2068,8 +2070,8 @@ def init(telegram_bot: Bot,
},
show_in_keyboard=False,
authorization_level='admin')
async def config_command(bot, update, user_record, language):
return await _config_command(bot=bot,
update=update,
user_record=user_record,
language=language)
async def _config_command(bot, update, user_record, language):
return await config_command(bot=bot,
update=update,
user_record=user_record,
language=language)

86
davtelepot/api.py
View File

@ -305,7 +305,8 @@ class MaskPosition(dict):
class InputSticker(dict):
"""This object describes a sticker to be added to a sticker set."""
def __init__(self, sticker: Union[str, dict, IO], emoji_list: List[str],
def __init__(self, sticker: Union[str, dict, IO], format_: str,
emoji_list: List[str],
mask_position: Union['MaskPosition', None] = None,
keywords: Union[List[str], None] = None):
"""This object describes a sticker to be added to a sticker set.
@ -318,6 +319,9 @@ class InputSticker(dict):
multipart/form-data under <file_attach_name> name.
Animated and video stickers can't be uploaded via HTTP URL.
More information on Sending Files: https://core.telegram.org/bots/api#sending-files
@param format_: Format of the added sticker, must be one of “static”
for a .WEBP or .PNG image, “animated” for a .TGS animation,
“video” for a WEBM video
@param emoji_list: List of 1-20 emoji associated with the sticker
@param mask_position: Optional. Position where the mask should be
placed on faces. For “mask” stickers only.
@ -327,6 +331,10 @@ class InputSticker(dict):
"""
super().__init__(self)
self['sticker'] = sticker
if format_ not in ("static", "animated", "video"):
logging.error(f"Invalid format `{format_}")
else:
self['format'] = format_
self['emoji_list'] = emoji_list
self['mask_position'] = mask_position
self['keywords'] = keywords
@ -449,6 +457,16 @@ def handle_deprecated_reply_parameters(parameters: dict,
return parameters
def handle_forbidden_names_for_parameters(parameters: dict,
kwargs: dict):
if 'format' in kwargs:
parameters['format'] = kwargs['format']
if 'format_' in parameters:
parameters['format'] = parameters['format_']
del parameters['format_']
return parameters
# This class needs to mirror Telegram API, so camelCase method are needed
# noinspection PyPep8Naming
class TelegramBot:
@ -646,7 +664,8 @@ class TelegramBot:
sticker: Union[dict, str, IO],
emoji_list: Union[List[str], str],
mask_position: Union[MaskPosition, None] = None,
keywords: Union[List[str], None] = None) -> InputSticker:
keywords: Union[List[str], None] = None,
format_: str = 'static') -> InputSticker:
if isinstance(emoji_list, str):
emoji_list = [c for c in emoji_list]
if isinstance(keywords, str):
@ -654,7 +673,8 @@ class TelegramBot:
if isinstance(sticker, str) and os.path.isfile(sticker):
sticker = self.prepare_file_object(sticker)
return InputSticker(sticker=sticker, emoji_list=emoji_list,
mask_position=mask_position, keywords=keywords)
mask_position=mask_position, keywords=keywords,
format_=format_)
async def prevent_flooding(self, chat_id):
"""Await until request may be sent safely.
@ -855,6 +875,7 @@ class TelegramBot:
async def sendMessage(self, chat_id: Union[int, str],
text: str,
business_connection_id: str = None,
message_thread_id: int = None,
parse_mode: str = None,
entities: List[dict] = None,
@ -897,6 +918,7 @@ class TelegramBot:
)
async def sendPhoto(self, chat_id: Union[int, str], photo,
business_connection_id: str = None,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
@ -921,6 +943,7 @@ class TelegramBot:
)
async def sendAudio(self, chat_id: Union[int, str], audio,
business_connection_id: str = None,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
@ -953,6 +976,7 @@ class TelegramBot:
)
async def sendDocument(self, chat_id: Union[int, str], document,
business_connection_id: str = None,
thumbnail=None,
caption: str = None,
parse_mode: str = None,
@ -983,6 +1007,7 @@ class TelegramBot:
)
async def sendVideo(self, chat_id: Union[int, str], video,
business_connection_id: str = None,
duration: int = None,
width: int = None,
height: int = None,
@ -1017,6 +1042,7 @@ class TelegramBot:
)
async def sendAnimation(self, chat_id: Union[int, str], animation,
business_connection_id: str = None,
duration: int = None,
width: int = None,
height: int = None,
@ -1050,6 +1076,7 @@ class TelegramBot:
)
async def sendVoice(self, chat_id: Union[int, str], voice,
business_connection_id: str = None,
caption: str = None,
parse_mode: str = None,
caption_entities: List[dict] = None,
@ -1075,6 +1102,7 @@ class TelegramBot:
)
async def sendVideoNote(self, chat_id: Union[int, str], video_note,
business_connection_id: str = None,
duration: int = None,
length: int = None,
thumbnail=None,
@ -1103,6 +1131,7 @@ class TelegramBot:
)
async def sendMediaGroup(self, chat_id: Union[int, str], media: list,
business_connection_id: str = None,
disable_notification: bool = None,
message_thread_id: int = None,
protect_content: bool = None,
@ -1125,6 +1154,7 @@ class TelegramBot:
async def sendLocation(self, chat_id: Union[int, str],
latitude: float, longitude: float,
business_connection_id: str = None,
horizontal_accuracy: float = None,
live_period=None,
heading: int = None,
@ -1207,6 +1237,7 @@ class TelegramBot:
async def sendVenue(self, chat_id: Union[int, str],
latitude: float, longitude: float,
title: str, address: str,
business_connection_id: str = None,
foursquare_id: str = None,
foursquare_type: str = None,
google_place_id: str = None,
@ -1234,6 +1265,7 @@ class TelegramBot:
async def sendContact(self, chat_id: Union[int, str],
phone_number: str,
first_name: str,
business_connection_id: str = None,
last_name: str = None,
vcard: str = None,
disable_notification: bool = None,
@ -1259,6 +1291,7 @@ class TelegramBot:
chat_id: Union[int, str],
question: str,
options: List[str],
business_connection_id: str = None,
is_anonymous: bool = True,
type_: str = 'regular',
allows_multiple_answers: bool = False,
@ -1308,6 +1341,7 @@ class TelegramBot:
)
async def sendChatAction(self, chat_id: Union[int, str], action,
business_connection_id: str = None,
message_thread_id: int = None):
"""Fake a typing status or similar.
@ -1776,6 +1810,7 @@ class TelegramBot:
async def sendSticker(self, chat_id: Union[int, str],
sticker: Union[str, dict, IO],
business_connection_id: str = None,
disable_notification: bool = None,
message_thread_id: int = None,
protect_content: bool = None,
@ -1850,7 +1885,6 @@ class TelegramBot:
async def createNewStickerSet(self, user_id: int, name: str, title: str,
stickers: List['InputSticker'],
sticker_format: str = 'static',
sticker_type: str = 'regular',
needs_repainting: bool = False,
**kwargs):
@ -1862,6 +1896,10 @@ class TelegramBot:
"""
if stickers is None:
stickers = []
if 'sticker_format' in kwargs:
logging.error("Parameter `sticker_format` of method "
"`createNewStickerSet` has been deprecated. "
"Use `format` parameter of class `InputSticker` instead.")
if 'contains_masks' in kwargs:
logging.error("Parameter `contains_masks` of method "
"`createNewStickerSet` has been deprecated. "
@ -2070,6 +2108,7 @@ class TelegramBot:
)
async def sendGame(self, chat_id: Union[int, str], game_short_name,
business_connection_id: str = None,
message_thread_id: int = None,
protect_content: bool = None,
disable_notification: bool = None,
@ -2133,7 +2172,8 @@ class TelegramBot:
)
async def sendDice(self,
chat_id: Union[int, str] = None,
chat_id: Union[int, str],
business_connection_id: str = None,
emoji: str = None,
disable_notification: bool = None,
message_thread_id: int = None,
@ -2843,7 +2883,10 @@ class TelegramBot:
parameters=locals()
)
async def setStickerSetThumbnail(self, name: str, user_id: int, thumbnail: 'InputFile or String'):
async def setStickerSetThumbnail(self, name: str, user_id: int,
format_: str,
thumbnail: 'InputFile or String',
**kwargs):
"""Set the thumbnail of a regular or mask sticker set.
The format of the thumbnail file must match the format of the stickers
@ -2851,9 +2894,13 @@ class TelegramBot:
Returns True on success.
See https://core.telegram.org/bots/api#setstickersetthumbnail for details.
"""
parameters = handle_forbidden_names_for_parameters(
parameters=locals().copy(),
kwargs=kwargs
)
return await self.api_request(
'setStickerSetThumbnail',
parameters=locals()
parameters=parameters
)
async def setCustomEmojiStickerSetThumbnail(self, name: str, custom_emoji_id: str):
@ -2966,3 +3013,28 @@ class TelegramBot:
'setMessageReaction',
parameters=locals()
)
async def getBusinessConnection(self, business_connection_id: str):
"""Get information about the connection of the bot with a business account.
Returns a BusinessConnection object on success.
See https://core.telegram.org/bots/api#getbusinessconnection for details.
"""
return await self.api_request(
'getBusinessConnection',
parameters=locals()
)
async def replaceStickerInSet(self, user_id: int, name: str,
old_sticker: str, sticker: 'InputSticker'):
"""Replace an existing sticker in a sticker set with a new one.
The method is equivalent to calling deleteStickerFromSet, then
addStickerToSet, then setStickerPositionInSet.
Returns True on success.
See https://core.telegram.org/bots/api#replacestickerinset for details.
"""
return await self.api_request(
'replaceStickerInSet',
parameters=locals()
)