From 795f7fff7fb41e6cc74b88f1229fee56032d8872 Mon Sep 17 00:00:00 2001
From: SwissCorePy <51398037+SwissCorePy@users.noreply.github.com>
Date: Sat, 19 Jun 2021 17:59:55 +0200
Subject: [PATCH] Some small changes
* Fixed type warnings in some editors by changing `var: Type = None` to `var: Union[Type, None] = None`
* changed some args from `obj['arg']` to `obj.get('arg')` if arg is optional
* better PEP-8 compliance for less weak warnings
* added tests for the new type `ChatInviteLink`
---
telebot/__init__.py | 157 +++++++++++++++++++++++-------------
telebot/types.py | 180 +++++++++++++++++++++---------------------
telebot/util.py | 35 +++++---
tests/test_telebot.py | 18 +++++
tests/test_types.py | 11 +++
5 files changed, 245 insertions(+), 156 deletions(-)
diff --git a/telebot/__init__.py b/telebot/__init__.py
index cb582dc..d5d2e1a 100644
--- a/telebot/__init__.py
+++ b/telebot/__init__.py
@@ -9,6 +9,7 @@ import time
import traceback
from typing import List, Union
+# this imports are used to avoid circular import error
import telebot.util
import telebot.types
@@ -23,7 +24,7 @@ logger.addHandler(console_output_handler)
logger.setLevel(logging.ERROR)
-from telebot import apihelper, types, util
+from telebot import apihelper, util, types
from telebot.handler_backends import MemoryHandlerBackend, FileHandlerBackend
"""
@@ -252,19 +253,30 @@ class TeleBot:
drop_pending_updates = None, timeout=None):
"""
Use this method to specify a url and receive incoming updates via an outgoing webhook. Whenever there is an
- update for the bot, we will send an HTTPS POST request to the specified url, containing a JSON-serialized Update.
- In case of an unsuccessful request, we will give up after a reasonable amount of attempts. Returns True on success.
+ update for the bot, we will send an HTTPS POST request to the specified url,
+ containing a JSON-serialized Update.
+ In case of an unsuccessful request, we will give up after a reasonable amount of attempts.
+ Returns True on success.
:param url: HTTPS url to send updates to. Use an empty string to remove webhook integration
- :param certificate: Upload your public key certificate so that the root certificate in use can be checked. See our self-signed guide for details.
- :param max_connections: Maximum allowed number of simultaneous HTTPS connections to the webhook for update delivery, 1-100. Defaults to 40. Use lower values to limit the load on your bot's server, and higher values to increase your bot's throughput.
- :param allowed_updates: A JSON-serialized list of the update types you want your bot to receive. For example, specify [“message”, “edited_channel_post”, “callback_query”] to only receive updates of these types. See Update for a complete list of available update types. Specify an empty list to receive all updates regardless of type (default). If not specified, the previous setting will be used.
- :param ip_address: The fixed IP address which will be used to send webhook requests instead of the IP address resolved through DNS
+ :param certificate: Upload your public key certificate so that the root certificate in use can be checked.
+ See our self-signed guide for details.
+ :param max_connections: Maximum allowed number of simultaneous HTTPS connections to the webhook
+ for update delivery, 1-100. Defaults to 40. Use lower values to limit the load on your bot's server,
+ and higher values to increase your bot's throughput.
+ :param allowed_updates: A JSON-serialized list of the update types you want your bot to receive.
+ For example, specify [“message”, “edited_channel_post”, “callback_query”] to only receive updates
+ of these types. See Update for a complete list of available update types.
+ Specify an empty list to receive all updates regardless of type (default).
+ If not specified, the previous setting will be used.
+ :param ip_address: The fixed IP address which will be used to send webhook requests instead of the IP address
+ resolved through DNS
:param drop_pending_updates: Pass True to drop all pending updates
:param timeout: Integer. Request connection timeout
:return:
"""
- return apihelper.set_webhook(self.token, url, certificate, max_connections, allowed_updates, ip_address, drop_pending_updates, timeout)
+ return apihelper.set_webhook(self.token, url, certificate, max_connections, allowed_updates, ip_address,
+ drop_pending_updates, timeout)
def delete_webhook(self, drop_pending_updates=None, timeout=None):
"""
@@ -330,14 +342,14 @@ class TeleBot:
if self.skip_pending:
logger.debug('Skipped {0} pending messages'.format(self.__skip_updates()))
self.skip_pending = False
- updates = self.get_updates(offset=(self.last_update_id + 1), timeout=timeout, long_polling_timeout = long_polling_timeout)
+ updates = self.get_updates(offset=(self.last_update_id + 1),
+ timeout=timeout, long_polling_timeout=long_polling_timeout)
self.process_new_updates(updates)
def process_new_updates(self, updates):
upd_count = len(updates)
logger.debug('Received {0} new updates'.format(upd_count))
- if (upd_count == 0):
- return
+ if upd_count == 0: return
new_messages = None
new_edited_messages = None
@@ -488,11 +500,13 @@ class TeleBot:
:param timeout: Request connection timeout
:param long_polling_timeout: Timeout in seconds for long polling (see API docs)
- :param logger_level: Custom logging level for infinity_polling logging. Use logger levels from logging as a value. None/NOTSET = no error logging
+ :param logger_level: Custom logging level for infinity_polling logging.
+ Use logger levels from logging as a value. None/NOTSET = no error logging
"""
while not self.__stop_polling.is_set():
try:
- self.polling(none_stop=True, timeout=timeout, long_polling_timeout=long_polling_timeout, *args, **kwargs)
+ self.polling(none_stop=True, timeout=timeout, long_polling_timeout=long_polling_timeout,
+ *args, **kwargs)
except Exception as e:
if logger_level and logger_level >= logging.ERROR:
logger.error("Infinity polling exception: %s", str(e))
@@ -590,7 +604,7 @@ class TeleBot:
self.worker_pool.clear_exceptions() #*
logger.info('Stopped polling.')
- def __non_threaded_polling(self, non_stop=False, interval=0, timeout = None, long_polling_timeout = None):
+ def __non_threaded_polling(self, non_stop=False, interval=0, timeout=None, long_polling_timeout=None):
logger.info('Started polling.')
self.__stop_polling.clear()
error_interval = 0.25
@@ -675,7 +689,8 @@ class TeleBot:
def log_out(self) -> bool:
"""
Use this method to log out from the cloud Bot API server before launching the bot locally.
- You MUST log out the bot before running it locally, otherwise there is no guarantee that the bot will receive updates.
+ You MUST log out the bot before running it locally, otherwise there is no guarantee
+ that the bot will receive updates.
After a successful call, you can immediately log in on a local server,
but will not be able to log in back to the cloud Bot API server for 10 minutes.
Returns True on success.
@@ -685,13 +700,13 @@ class TeleBot:
def close(self) -> bool:
"""
Use this method to close the bot instance before moving it from one local server to another.
- You need to delete the webhook before calling this method to ensure that the bot isn't launched again after server restart.
+ You need to delete the webhook before calling this method to ensure that the bot isn't launched again
+ after server restart.
The method will return error 429 in the first 10 minutes after the bot is launched.
Returns True on success.
"""
return apihelper.close(self.token)
-
def get_user_profile_photos(self, user_id, offset=None, limit=None) -> types.UserProfilePhotos:
"""
Retrieves the user profile photos of the person with 'user_id'
@@ -807,7 +822,8 @@ class TeleBot:
apihelper.send_message(self.token, chat_id, text, disable_web_page_preview, reply_to_message_id,
reply_markup, parse_mode, disable_notification, timeout))
- def forward_message(self, chat_id, from_chat_id, message_id, disable_notification=None, timeout=None) -> types.Message:
+ def forward_message(self, chat_id, from_chat_id, message_id,
+ disable_notification=None, timeout=None) -> types.Message:
"""
Use this method to forward messages of any kind.
:param disable_notification:
@@ -897,7 +913,8 @@ class TeleBot:
reply_to_message_id=None, reply_markup=None, parse_mode=None, disable_notification=None,
timeout=None, thumb=None) -> types.Message:
"""
- Use this method to send audio files, if you want Telegram clients to display them in the music player. Your audio must be in the .mp3 format.
+ Use this method to send audio files, if you want Telegram clients to display them in the music player.
+ Your audio must be in the .mp3 format.
:param chat_id:Unique identifier for the message recipient
:param audio:Audio file to send.
:param caption:
@@ -909,7 +926,7 @@ class TeleBot:
:param parse_mode
:param disable_notification:
:param timeout:
- :param thumb:
+ :param thumb:
:return: Message
"""
parse_mode = self.parse_mode if (parse_mode is None) else parse_mode
@@ -921,7 +938,8 @@ class TeleBot:
def send_voice(self, chat_id, voice, caption=None, duration=None, reply_to_message_id=None, reply_markup=None,
parse_mode=None, disable_notification=None, timeout=None) -> types.Message:
"""
- Use this method to send audio files, if you want Telegram clients to display the file as a playable voice message.
+ Use this method to send audio files, if you want Telegram clients to display the file
+ as a playable voice message.
:param chat_id:Unique identifier for the message recipient.
:param voice:
:param caption:
@@ -984,7 +1002,8 @@ class TeleBot:
"""
Use this method to send video files, Telegram clients support mp4 videos.
:param chat_id: Integer : Unique identifier for the message recipient — User or GroupChat id
- :param data: InputFile or String : Video to send. You can either pass a file_id as String to resend a video that is already on the Telegram server
+ :param data: InputFile or String : Video to send. You can either pass a file_id as String to resend
+ a video that is already on the Telegram server
:param duration: Integer : Duration of sent video in seconds
:param caption: String : Video caption (may also be used when resending videos by file_id).
:param parse_mode:
@@ -993,7 +1012,7 @@ class TeleBot:
:param reply_markup:
:param disable_notification:
:param timeout:
- :param thumb: InputFile or String : Thumbnail of the file sent
+ :param thumb: InputFile or String : Thumbnail of the file sent
:param width:
:param height:
:return:
@@ -1011,7 +1030,8 @@ class TeleBot:
"""
Use this method to send animation files (GIF or H.264/MPEG-4 AVC video without sound).
:param chat_id: Integer : Unique identifier for the message recipient — User or GroupChat id
- :param animation: InputFile or String : Animation to send. You can either pass a file_id as String to resend an animation that is already on the Telegram server
+ :param animation: InputFile or String : Animation to send. You can either pass a file_id as String to resend an
+ animation that is already on the Telegram server
:param duration: Integer : Duration of sent video in seconds
:param caption: String : Animation caption (may also be used when resending animation by file_id).
:param parse_mode:
@@ -1025,16 +1045,18 @@ class TeleBot:
parse_mode = self.parse_mode if (parse_mode is None) else parse_mode
return types.Message.de_json(
- apihelper.send_animation(self.token, chat_id, animation, duration, caption, reply_to_message_id, reply_markup,
- parse_mode, disable_notification, timeout, thumb))
+ apihelper.send_animation(self.token, chat_id, animation, duration, caption, reply_to_message_id,
+ reply_markup, parse_mode, disable_notification, timeout, thumb))
def send_video_note(self, chat_id, data, duration=None, length=None,
reply_to_message_id=None, reply_markup=None,
disable_notification=None, timeout=None, thumb=None) -> types.Message:
"""
- As of v.4.0, Telegram clients support rounded square mp4 videos of up to 1 minute long. Use this method to send video messages.
+ As of v.4.0, Telegram clients support rounded square mp4 videos of up to 1 minute long. Use this method to send
+ video messages.
:param chat_id: Integer : Unique identifier for the message recipient — User or GroupChat id
- :param data: InputFile or String : Video note to send. You can either pass a file_id as String to resend a video that is already on the Telegram server
+ :param data: InputFile or String : Video note to send. You can either pass a file_id as String to resend
+ a video that is already on the Telegram server
:param duration: Integer : Duration of sent video in seconds
:param length: Integer : Video width and height, Can't be None and should be in range of (0, 640)
:param reply_to_message_id:
@@ -1133,7 +1155,8 @@ class TeleBot:
:param title: String : Name of the venue
:param address: String : Address of the venue
:param foursquare_id: String : Foursquare identifier of the venue
- :param foursquare_type: Foursquare type of the venue, if known. (For example, “arts_entertainment/default”, “arts_entertainment/aquarium” or “food/icecream”.)
+ :param foursquare_type: Foursquare type of the venue, if known. (For example, “arts_entertainment/default”,
+ “arts_entertainment/aquarium” or “food/icecream”.)
:param disable_notification:
:param reply_to_message_id:
:param reply_markup:
@@ -1162,7 +1185,8 @@ class TeleBot:
its typing status).
:param chat_id:
:param action: One of the following strings: 'typing', 'upload_photo', 'record_video', 'upload_video',
- 'record_audio', 'upload_audio', 'upload_document', 'find_location', 'record_video_note', 'upload_video_note'.
+ 'record_audio', 'upload_audio', 'upload_document', 'find_location', 'record_video_note',
+ 'upload_video_note'.
:param timeout:
:return: API reply. :type: boolean
"""
@@ -1187,7 +1211,8 @@ class TeleBot:
the user is not a member of the chat, but will be able to join it. So if the user is a member of the chat
they will also be removed from the chat. If you don't want this, use the parameter only_if_banned.
- :param chat_id: Unique identifier for the target group or username of the target supergroup or channel (in the format @username)
+ :param chat_id: Unique identifier for the target group or username of the target supergroup or channel
+ (in the format @username)
:param user_id: Unique identifier of the target user
:param only_if_banned: Do nothing if the user is not banned
:return: True on success
@@ -1219,9 +1244,10 @@ class TeleBot:
use inline bots, implies can_send_media_messages
:param can_add_web_page_previews: Pass True, if the user may add web page previews to their messages,
implies can_send_media_messages
- :param can_change_info: Pass True, if the user is allowed to change the chat title, photo and other settings. Ignored in public supergroups
- :param can_invite_users: Pass True, if the user is allowed to invite new users to the chat,
- implies can_invite_users
+ :param can_change_info: Pass True, if the user is allowed to change the chat title, photo and other settings.
+ Ignored in public supergroups
+ :param can_invite_users: Pass True, if the user is allowed to invite new users to the chat,
+ implies can_invite_users
:param can_pin_messages: Pass True, if the user is allowed to pin messages. Ignored in public supergroups
:return: True on success
"""
@@ -1463,9 +1489,13 @@ class TeleBot:
return result
return types.Message.de_json(result)
- def edit_message_media(self, media, chat_id=None, message_id=None, inline_message_id=None, reply_markup=None) -> Union[types.Message, bool]:
+ def edit_message_media(self, media, chat_id=None, message_id=None,
+ inline_message_id=None, reply_markup=None) -> Union[types.Message, bool]:
"""
- Use this method to edit animation, audio, document, photo, or video messages. If a message is a part of a message album, then it can be edited only to a photo or a video. Otherwise, message type can be changed arbitrarily. When inline message is edited, new file can't be uploaded. Use previously uploaded file via its file_id or specify a URL.
+ Use this method to edit animation, audio, document, photo, or video messages.
+ If a message is a part of a message album, then it can be edited only to a photo or a video.
+ Otherwise, message type can be changed arbitrarily. When inline message is edited, new file can't be uploaded.
+ Use previously uploaded file via its file_id or specify a URL.
:param media:
:param chat_id:
:param message_id:
@@ -1478,7 +1508,8 @@ class TeleBot:
return result
return types.Message.de_json(result)
- def edit_message_reply_markup(self, chat_id=None, message_id=None, inline_message_id=None, reply_markup=None) -> Union[types.Message, bool]:
+ def edit_message_reply_markup(self, chat_id=None, message_id=None,
+ inline_message_id=None, reply_markup=None) -> Union[types.Message, bool]:
"""
Use this method to edit only the reply markup of messages.
:param chat_id:
@@ -1523,13 +1554,14 @@ class TeleBot:
:param disable_edit_message:
:return:
"""
- result = apihelper.set_game_score(self.token, user_id, score, force, disable_edit_message, chat_id, message_id,
- inline_message_id)
+ result = apihelper.set_game_score(self.token, user_id, score, force, disable_edit_message, chat_id,
+ message_id, inline_message_id)
if type(result) == bool:
return result
return types.Message.de_json(result)
- def get_game_high_scores(self, user_id, chat_id=None, message_id=None, inline_message_id=None) -> List[types.GameHighScore]:
+ def get_game_high_scores(self, user_id, chat_id=None,
+ message_id=None, inline_message_id=None) -> List[types.GameHighScore]:
"""
Gets top points and game play
:param user_id:
@@ -1555,12 +1587,17 @@ class TeleBot:
:param chat_id: Unique identifier for the target private chat
:param title: Product name
:param description: Product description
- :param invoice_payload: Bot-defined invoice payload, 1-128 bytes. This will not be displayed to the user, use for your internal processes.
+ :param invoice_payload: Bot-defined invoice payload, 1-128 bytes. This will not be displayed to the user,
+ use for your internal processes.
:param provider_token: Payments provider token, obtained via @Botfather
- :param currency: Three-letter ISO 4217 currency code, see https://core.telegram.org/bots/payments#supported-currencies
- :param prices: Price breakdown, a list of components (e.g. product price, tax, discount, delivery cost, delivery tax, bonus, etc.)
- :param start_parameter: Unique deep-linking parameter that can be used to generate this invoice when used as a start parameter
- :param photo_url: URL of the product photo for the invoice. Can be a photo of the goods or a marketing image for a service. People like it better when they see what they are paying for.
+ :param currency: Three-letter ISO 4217 currency code,
+ see https://core.telegram.org/bots/payments#supported-currencies
+ :param prices: Price breakdown, a list of components
+ (e.g. product price, tax, discount, delivery cost, delivery tax, bonus, etc.)
+ :param start_parameter: Unique deep-linking parameter that can be used to generate this invoice
+ when used as a start parameter
+ :param photo_url: URL of the product photo for the invoice. Can be a photo of the goods
+ or a marketing image for a service. People like it better when they see what they are paying for.
:param photo_size: Photo size
:param photo_width: Photo width
:param photo_height: Photo height
@@ -1573,8 +1610,10 @@ class TeleBot:
:param send_email_to_provider: Pass True, if user's email address should be sent to provider
:param disable_notification: Sends the message silently. Users will receive a notification with no sound.
:param reply_to_message_id: If the message is a reply, ID of the original message
- :param reply_markup: A JSON-serialized object for an inline keyboard. If empty, one 'Pay total price' button will be shown. If not empty, the first button must be a Pay button
- :param provider_data: A JSON-serialized data about the invoice, which will be shared with the payment provider. A detailed description of required fields should be provided by the payment provider.
+ :param reply_markup: A JSON-serialized object for an inline keyboard. If empty,
+ one 'Pay total price' button will be shown. If not empty, the first button must be a Pay button
+ :param provider_data: A JSON-serialized data about the invoice, which will be shared with the payment provider.
+ A detailed description of required fields should be provided by the payment provider.
:param timeout:
:return:
"""
@@ -1651,7 +1690,7 @@ class TeleBot:
:param ok:
:param error_message:
:return:
- """
+ """
return apihelper.answer_pre_checkout_query(self.token, pre_checkout_query_id, ok, error_message)
def edit_message_caption(self, caption, chat_id=None, message_id=None, inline_message_id=None,
@@ -1677,7 +1716,7 @@ class TeleBot:
def reply_to(self, message, text, **kwargs) -> types.Message:
"""
Convenience function for `send_message(message.chat.id, text, reply_to_message_id=message.message_id, **kwargs)`
- :param message:
+ :param message:
:param text:
:param kwargs:
:return:
@@ -1691,11 +1730,14 @@ class TeleBot:
No more than 50 results per query are allowed.
:param inline_query_id: Unique identifier for the answered query
:param results: Array of results for the inline query
- :param cache_time: The maximum amount of time in seconds that the result of the inline query may be cached on the server.
- :param is_personal: Pass True, if results may be cached on the server side only for the user that sent the query.
- :param next_offset: Pass the offset that a client should send in the next query with the same text to receive more results.
+ :param cache_time: The maximum amount of time in seconds that the result of the inline query
+ may be cached on the server.
+ :param is_personal: Pass True, if results may be cached on the server side only for
+ the user that sent the query.
+ :param next_offset: Pass the offset that a client should send in the next query with the same text
+ to receive more results.
:param switch_pm_parameter: If passed, clients will display a button with specified text that switches the user
- to a private chat with the bot and sends the bot a start message with the parameter switch_pm_parameter
+ to a private chat with the bot and sends the bot a start message with the parameter switch_pm_parameter
:param switch_pm_text: Parameter for the start message sent to the bot when user presses the switch button
:return: True means success.
"""
@@ -1973,18 +2015,21 @@ class TeleBot:
bot.send_message(message.chat.id, 'Did someone call for help?')
# Handle all sent documents of type 'text/plain'.
- @bot.message_handler(func=lambda message: message.document.mime_type == 'text/plain', content_types=['document'])
+ @bot.message_handler(func=lambda message: message.document.mime_type == 'text/plain',
+ content_types=['document'])
def command_handle_document(message):
bot.send_message(message.chat.id, 'Document received, sir!')
# Handle all other messages.
- @bot.message_handler(func=lambda message: True, content_types=['audio', 'photo', 'voice', 'video', 'document', 'text', 'location', 'contact', 'sticker'])
+ @bot.message_handler(func=lambda message: True, content_types=['audio', 'photo', 'voice', 'video', 'document',
+ 'text', 'location', 'contact', 'sticker'])
def default_command(message):
bot.send_message(message.chat.id, "This is the default command handler.")
:param commands: Optional list of strings (commands to handle).
:param regexp: Optional regular expression.
- :param func: Optional lambda function. The lambda receives the message to test as the first parameter. It must return True if the command should handle the message.
+ :param func: Optional lambda function. The lambda receives the message to test as the first parameter.
+ It must return True if the command should handle the message.
:param content_types: Supported message content types. Must be a list. Defaults to ['text'].
"""
diff --git a/telebot/types.py b/telebot/types.py
index 6ecb75d..9139374 100644
--- a/telebot/types.py
+++ b/telebot/types.py
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
import logging
-from typing import Dict, List
+from typing import Dict, List, Union
try:
import ujson as json
@@ -92,7 +92,7 @@ class JsonDeserializable(object):
class Update(JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
update_id = obj['update_id']
message = Message.de_json(obj.get('message'))
@@ -128,7 +128,7 @@ class Update(JsonDeserializable):
class WebhookInfo(JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
url = obj['url']
has_custom_certificate = obj['has_custom_certificate']
@@ -156,7 +156,7 @@ class WebhookInfo(JsonDeserializable):
class User(JsonDeserializable, Dictionaryable, JsonSerializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
return cls(**obj)
@@ -197,7 +197,7 @@ class User(JsonDeserializable, Dictionaryable, JsonSerializable):
class GroupChat(JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
return cls(**obj)
@@ -249,8 +249,7 @@ class Chat(JsonDeserializable):
class MessageID(JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if(json_string is None):
- return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
message_id = obj['message_id']
return cls(message_id)
@@ -262,7 +261,7 @@ class MessageID(JsonDeserializable):
class Message(JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
message_id = obj['message_id']
from_user = User.de_json(obj.get('from'))
@@ -306,7 +305,8 @@ class Message(JsonDeserializable):
opts['document'] = Document.de_json(obj['document'])
content_type = 'document'
if 'animation' in obj:
- # Document content type accompanies "animation", so "animation" should be checked below "document" to override it
+ # Document content type accompanies "animation",
+ # so "animation" should be checked below "document" to override it
opts['animation'] = Animation.de_json(obj['animation'])
content_type = 'animation'
if 'game' in obj:
@@ -424,49 +424,49 @@ class Message(JsonDeserializable):
self.from_user: User = from_user
self.date: int = date
self.chat: Chat = chat
- self.forward_from: User = None
- self.forward_from_chat: Chat = None
- self.forward_from_message_id: int = None
- self.forward_signature: str = None
- self.forward_sender_name: str = None
- self.forward_date: int = None
- self.reply_to_message: Message = None
- self.via_bot: User = None
- self.edit_date: int = None
- self.media_group_id: str = None
- self.author_signature: str = None
- self.text: str = None
- self.entities: List[MessageEntity] = None
- self.caption_entities: List[MessageEntity] = None
- self.audio: Audio = None
- self.document: Document = None
- self.photo: List[PhotoSize] = None
- self.sticker: Sticker = None
- self.video: Video = None
- self.video_note: VideoNote = None
- self.voice: Voice = None
- self.caption: str = None
- self.contact: Contact = None
- self.location: Location = None
- self.venue: Venue = None
- self.animation: Animation = None
- self.dice: Dice = None
- self.new_chat_member: User = None # Deprecated since Bot API 3.0. Not processed anymore
- self.new_chat_members: List[User] = None
- self.left_chat_member: User = None
- self.new_chat_title: str = None
- self.new_chat_photo: List[PhotoSize] = None
- self.delete_chat_photo: bool = None
- self.group_chat_created: bool = None
- self.supergroup_chat_created: bool = None
- self.channel_chat_created: bool = None
- self.migrate_to_chat_id: int = None
- self.migrate_from_chat_id: int = None
- self.pinned_message: Message = None
- self.invoice: Invoice = None
- self.successful_payment: SuccessfulPayment = None
- self.connected_website: str = None
- self.reply_markup: InlineKeyboardMarkup = None
+ self.forward_from: Union[User, None] = None
+ self.forward_from_chat: Union[Chat, None] = None
+ self.forward_from_message_id: Union[int, None] = None
+ self.forward_signature: Union[str, None] = None
+ self.forward_sender_name: Union[str, None] = None
+ self.forward_date: Union[int, None] = None
+ self.reply_to_message: Union[Message, None] = None
+ self.via_bot: Union[User, None] = None
+ self.edit_date: Union[int, None] = None
+ self.media_group_id: Union[str, None] = None
+ self.author_signature: Union[str, None] = None
+ self.text: Union[str, None] = None
+ self.entities: Union[List[MessageEntity], None] = None
+ self.caption_entities: Union[List[MessageEntity], None] = None
+ self.audio: Union[Audio, None] = None
+ self.document: Union[Document, None] = None
+ self.photo: Union[List[PhotoSize], None] = None
+ self.sticker: Union[Sticker, None] = None
+ self.video: Union[Video, None] = None
+ self.video_note: Union[VideoNote, None] = None
+ self.voice: Union[Voice, None] = None
+ self.caption: Union[str, None] = None
+ self.contact: Union[Contact, None] = None
+ self.location: Union[Location, None] = None
+ self.venue: Union[Venue, None] = None
+ self.animation: Union[Animation, None] = None
+ self.dice: Union[Dice, None] = None
+ self.new_chat_member: Union[User, None] = None # Deprecated since Bot API 3.0. Not processed anymore
+ self.new_chat_members: Union[List[User], None] = None
+ self.left_chat_member: Union[User, None] = None
+ self.new_chat_title: Union[str, None] = None
+ self.new_chat_photo: Union[List[PhotoSize], None] = None
+ self.delete_chat_photo: Union[bool, None] = None
+ self.group_chat_created: Union[bool, None] = None
+ self.supergroup_chat_created: Union[bool, None] = None
+ self.channel_chat_created: Union[bool, None] = None
+ self.migrate_to_chat_id: Union[int, None] = None
+ self.migrate_from_chat_id: Union[int, None] = None
+ self.pinned_message: Union[Message, None] = None
+ self.invoice: Union[Invoice, None] = None
+ self.successful_payment: Union[SuccessfulPayment, None] = None
+ self.connected_website: Union[str, None] = None
+ self.reply_markup: Union[InlineKeyboardMarkup, None] = None
for key in options:
setattr(self, key, options[key])
self.json = json_string
@@ -481,7 +481,7 @@ class Message(JsonDeserializable):
message.html_text
>> "Test parse formatting, url, text_mention and mention @username"
- Cusom subs:
+ Custom subs:
You can customize the substitutes. By default, there is no substitute for the entities: hashtag, bot_command, email. You can add or modify substitute an existing entity.
Example:
message.custom_subs = {"bold": "{text}", "italic": "{text}", "mention": "{text}"}
@@ -493,15 +493,15 @@ class Message(JsonDeserializable):
return text
_subs = {
- "bold" : "{text}",
- "italic" : "{text}",
- "pre" : "{text}",
- "code" : "{text}",
- #"url" : "{text}", # @badiboy plain URLs have no text and do not need tags
+ "bold": "{text}",
+ "italic": "{text}",
+ "pre": "{text}",
+ "code": "{text}",
+ # "url": "{text}", # @badiboy plain URLs have no text and do not need tags
"text_link": "{text}",
"strikethrough": "{text}",
"underline": "{text}"
- }
+ }
if hasattr(self, "custom_subs"):
for key, value in self.custom_subs.items():
@@ -551,11 +551,12 @@ class Message(JsonDeserializable):
class MessageEntity(Dictionaryable, JsonSerializable, JsonDeserializable):
@staticmethod
- def to_list_of_dicts(entity_list) -> List[Dict]:
+ def to_list_of_dicts(entity_list) -> Union[List[Dict], None]:
res = []
for e in entity_list:
res.append(MessageEntity.to_dict(e))
return res or None
+
@classmethod
def de_json(cls, json_string):
if json_string is None: return None
@@ -587,7 +588,7 @@ class MessageEntity(Dictionaryable, JsonSerializable, JsonDeserializable):
class Dice(JsonSerializable, Dictionaryable, JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
return cls(**obj)
@@ -606,7 +607,7 @@ class Dice(JsonSerializable, Dictionaryable, JsonDeserializable):
class PhotoSize(JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
return cls(**obj)
@@ -621,7 +622,7 @@ class PhotoSize(JsonDeserializable):
class Audio(JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
if 'thumb' in obj and 'file_id' in obj['thumb']:
obj['thumb'] = PhotoSize.de_json(obj['thumb'])
@@ -645,7 +646,7 @@ class Audio(JsonDeserializable):
class Voice(JsonDeserializable):
@classmethod
def de_json(cls, json_string):
- if (json_string is None): return None
+ if json_string is None: return None
obj = cls.check_json(json_string)
return cls(**obj)
@@ -682,7 +683,7 @@ class Video(JsonDeserializable):
def de_json(cls, json_string):
if json_string is None: return None
obj = cls.check_json(json_string)
- if ('thumb' in obj and 'file_id' in obj['thumb']):
+ if 'thumb' in obj and 'file_id' in obj['thumb']:
obj['thumb'] = PhotoSize.de_json(obj['thumb'])
return cls(**obj)
@@ -703,7 +704,7 @@ class VideoNote(JsonDeserializable):
def de_json(cls, json_string):
if json_string is None: return None
obj = cls.check_json(json_string)
- if ('thumb' in obj and 'file_id' in obj['thumb']):
+ if 'thumb' in obj and 'file_id' in obj['thumb']:
obj['thumb'] = PhotoSize.de_json(obj['thumb'])
return cls(**obj)
@@ -738,8 +739,8 @@ class Location(JsonDeserializable, JsonSerializable, Dictionaryable):
obj = cls.check_json(json_string)
return cls(**obj)
- def __init__(self, longitude: float, latitude: float, horizontal_accuracy:float=None,
- live_period: int=None, heading: int=None, proximity_alert_radius: int=None, **kwargs):
+ def __init__(self, longitude, latitude, horizontal_accuracy=None,
+ live_period=None, heading=None, proximity_alert_radius=None, **kwargs):
self.longitude: float = longitude
self.latitude: float = latitude
self.horizontal_accuracy: float = horizontal_accuracy
@@ -766,7 +767,7 @@ class Venue(JsonDeserializable):
def de_json(cls, json_string):
if json_string is None: return None
obj = cls.check_json(json_string)
- obj['location'] = Location.de_json(obj['location'])
+ obj['location'] = Location.de_json(obj.get('location'))
return cls(**obj)
def __init__(self, location, title, address, foursquare_id=None, foursquare_type=None,
@@ -785,11 +786,12 @@ class UserProfilePhotos(JsonDeserializable):
def de_json(cls, json_string):
if json_string is None: return None
obj = cls.check_json(json_string)
- photos = [[PhotoSize.de_json(y) for y in x] for x in obj['photos']]
- obj['photos'] = photos
+ if 'photos' in obj:
+ photos = [[PhotoSize.de_json(y) for y in x] for x in obj['photos']]
+ obj['photos'] = photos
return cls(**obj)
- def __init__(self, total_count, photos, **kwargs):
+ def __init__(self, total_count, photos=None, **kwargs):
self.total_count: int = total_count
self.photos: List[PhotoSize] = photos
@@ -859,8 +861,7 @@ class ReplyKeyboardMarkup(JsonSerializable):
"""
if row_width is None:
row_width = self.row_width
-
-
+
if row_width > self.max_row_keys:
# Todo: Will be replaced with Exception in future releases
if not DISABLE_KEYLEN_ERROR:
@@ -946,7 +947,7 @@ class InlineKeyboardMarkup(Dictionaryable, JsonSerializable, JsonDeserializable)
keyboard = [[InlineKeyboardButton.de_json(button) for button in row] for row in obj['inline_keyboard']]
return cls(keyboard)
- def __init__(self, keyboard=None ,row_width=3):
+ def __init__(self, keyboard=None, row_width=3):
"""
This object represents an inline keyboard that appears
right next to the message it belongs to.
@@ -993,7 +994,7 @@ class InlineKeyboardMarkup(Dictionaryable, JsonSerializable, JsonDeserializable)
def row(self, *args):
"""
Adds a list of InlineKeyboardButton to the keyboard.
- This metod does not consider row_width.
+ This method does not consider row_width.
InlineKeyboardMarkup.row("A").row("B", "C").to_json() outputs:
'{keyboard: [["A"], ["B", "C"]]}'
@@ -1255,8 +1256,6 @@ class InlineQuery(JsonDeserializable):
self.offset: str = offset
self.chat_type: str = chat_type
self.location: Location = location
-
-
class InputTextMessageContent(Dictionaryable):
@@ -1337,7 +1336,7 @@ class InputContactMessageContent(Dictionaryable):
self.vcard: str = vcard
def to_dict(self):
- json_dict = {'phone_numbe': self.phone_number, 'first_name': self.first_name}
+ json_dict = {'phone_number': self.phone_number, 'first_name': self.first_name}
if self.last_name:
json_dict['last_name'] = self.last_name
if self.vcard:
@@ -2043,7 +2042,10 @@ class Animation(JsonDeserializable):
def de_json(cls, json_string):
if (json_string is None): return None
obj = cls.check_json(json_string)
- obj["thumb"] = PhotoSize.de_json(obj['thumb'])
+ if 'thumb' in obj and 'file_id' in obj['thumb']:
+ obj["thumb"] = PhotoSize.de_json(obj['thumb'])
+ else:
+ obj['thumb'] = None
return cls(**obj)
def __init__(self, file_id, file_unique_id, width=None, height=None, duration=None,
@@ -2122,10 +2124,10 @@ class OrderInfo(JsonDeserializable):
def de_json(cls, json_string):
if (json_string is None): return None
obj = cls.check_json(json_string)
- obj['shipping_address'] = ShippingAddress.de_json(obj['shipping_address'])
+ obj['shipping_address'] = ShippingAddress.de_json(obj.get('shipping_address'))
return cls(**obj)
- def __init__(self, name, phone_number, email, shipping_address, **kwargs):
+ def __init__(self, name=None, phone_number=None, email=None, shipping_address=None, **kwargs):
self.name: str = name
self.phone_number: str = phone_number
self.email: str = email
@@ -2160,8 +2162,7 @@ class SuccessfulPayment(JsonDeserializable):
def de_json(cls, json_string):
if (json_string is None): return None
obj = cls.check_json(json_string)
- if 'order_info' in obj:
- obj['order_info'] = OrderInfo.de_json(obj['order_info'])
+ obj['order_info'] = OrderInfo.de_json(obj.get('order_info'))
return cls(**obj)
def __init__(self, currency, total_amount, invoice_payload, shipping_option_id=None, order_info=None,
@@ -2197,8 +2198,7 @@ class PreCheckoutQuery(JsonDeserializable):
if (json_string is None): return None
obj = cls.check_json(json_string)
obj['from_user'] = User.de_json(obj.pop('from'))
- if 'order_info' in obj:
- obj['order_info'] = OrderInfo.de_json(obj['order_info'])
+ obj['order_info'] = OrderInfo.de_json(obj.get('order_info'))
return cls(**obj)
def __init__(self, id, from_user, currency, total_amount, invoice_payload, shipping_option_id=None, order_info=None, **kwargs):
@@ -2473,6 +2473,7 @@ class PollAnswer(JsonSerializable, JsonDeserializable, Dictionaryable):
if (json_string is None): return None
obj = cls.check_json(json_string)
obj['user'] = User.de_json(obj['user'])
+ # Strange name, i think it should be `option_ids` not `options_ids` maybe replace that
obj['options_ids'] = obj.pop('option_ids')
return cls(**obj)
@@ -2487,6 +2488,7 @@ class PollAnswer(JsonSerializable, JsonDeserializable, Dictionaryable):
def to_dict(self):
return {'poll_id': self.poll_id,
'user': self.user.to_dict(),
+ #should be `option_ids` not `options_ids` could cause problems here
'options_ids': self.options_ids}
@@ -2498,7 +2500,7 @@ class ChatLocation(JsonSerializable, JsonDeserializable, Dictionaryable):
obj['location'] = Location.de_json(obj['location'])
return cls(**obj)
- def __init__(self, location: Location, address: str, **kwargs):
+ def __init__(self, location, address, **kwargs):
self.location: Location = location
self.address: str = address
@@ -2520,8 +2522,8 @@ class ChatInviteLink(JsonSerializable, JsonDeserializable, Dictionaryable):
obj['creator'] = User.de_json(obj['creator'])
return cls(**obj)
- def __init__(self, invite_link: str, creator: User, is_primary: bool, is_revoked: bool,
- expire_date: int=None, member_limit: int=None, **kwargs):
+ def __init__(self, invite_link, creator, is_primary, is_revoked,
+ expire_date=None, member_limit=None, **kwargs):
self.invite_link: str = invite_link
self.creator: User = creator
self.is_primary: bool = is_primary
diff --git a/telebot/util.py b/telebot/util.py
index beb9e90..7b87959 100644
--- a/telebot/util.py
+++ b/telebot/util.py
@@ -6,7 +6,7 @@ import threading
import traceback
import warnings
import functools
-from typing import Any, List, Dict
+from typing import Any, List, Dict, Union
import queue as Queue
import logging
@@ -36,6 +36,7 @@ content_type_service = [
'supergroup_chat_created', 'channel_chat_created', 'migrate_to_chat_id', 'migrate_from_chat_id', 'pinned_message'
]
+
class WorkerThread(threading.Thread):
count = 0
@@ -170,15 +171,19 @@ def async_dec():
def is_string(var):
return isinstance(var, str)
+
def is_dict(var):
return isinstance(var, dict)
+
def is_bytes(var):
return isinstance(var, bytes)
+
def is_pil_image(var):
return pil_imported and isinstance(var, Image.Image)
+
def pil_image_to_file(image, extension='JPEG', quality='web_low'):
if pil_imported:
photoBuffer = BytesIO()
@@ -189,17 +194,18 @@ def pil_image_to_file(image, extension='JPEG', quality='web_low'):
else:
raise RuntimeError('PIL module is not imported')
+
def is_command(text: str) -> bool:
"""
Checks if `text` is a command. Telegram chat commands start with the '/' character.
:param text: Text to check.
:return: True if `text` is a command, else False.
"""
- if (text is None): return False
+ if text is None: return False
return text.startswith('/')
-def extract_command(text: str) -> str:
+def extract_command(text: str) -> Union[str, None]:
"""
Extracts the command from `text` (minus the '/') if `text` is a command (see is_command).
If `text` is not a command, this function returns None.
@@ -213,7 +219,7 @@ def extract_command(text: str) -> str:
:param text: String to extract the command from
:return: the command if `text` is a command (according to is_command), else None.
"""
- if (text is None): return None
+ if text is None: return None
return text.split()[0].split('@')[0][1:] if is_command(text) else None
@@ -229,7 +235,7 @@ def extract_arguments(text: str) -> str:
:param text: String to extract the arguments from a command
:return: the arguments if `text` is a command (according to is_command), else None.
"""
- regexp = re.compile(r"/\w*(@\w*)*\s*([\s\S]*)",re.IGNORECASE)
+ regexp = re.compile(r"/\w*(@\w*)*\s*([\s\S]*)", re.IGNORECASE)
result = regexp.match(text)
return result.group(2) if is_command(text) else None
@@ -247,16 +253,17 @@ def split_string(text: str, chars_per_string: int) -> List[str]:
def smart_split(text: str, chars_per_string: int=MAX_MESSAGE_LENGTH) -> List[str]:
- f"""
+ """
Splits one string into multiple strings, with a maximum amount of `chars_per_string` characters per string.
This is very useful for splitting one giant message into multiples.
- If `chars_per_string` > {MAX_MESSAGE_LENGTH}: `chars_per_string` = {MAX_MESSAGE_LENGTH}.
+ If `chars_per_string` > 4096: `chars_per_string` = 4096.
Splits by '\n', '. ' or ' ' in exactly this priority.
:param text: The text to split
:param chars_per_string: The number of maximum characters per part the text is split to.
:return: The splitted text as a list of strings.
"""
+
def _text_before_last(substr: str) -> str:
return substr.join(part.split(substr)[:-1]) + substr
@@ -270,9 +277,9 @@ def smart_split(text: str, chars_per_string: int=MAX_MESSAGE_LENGTH) -> List[str
part = text[:chars_per_string]
- if ("\n" in part): part = _text_before_last("\n")
- elif (". " in part): part = _text_before_last(". ")
- elif (" " in part): part = _text_before_last(" ")
+ if "\n" in part: part = _text_before_last("\n")
+ elif ". " in part: part = _text_before_last(". ")
+ elif " " in part: part = _text_before_last(" ")
parts.append(part)
text = text[len(part):]
@@ -296,7 +303,7 @@ def user_link(user: types.User, include_id: bool=False) -> str:
Attention: Don't forget to set parse_mode to 'HTML'!
Example:
- bot.send_message(your_user_id, user_link(message.from_user) + ' startet the bot!', parse_mode='HTML')
+ bot.send_message(your_user_id, user_link(message.from_user) + ' started the bot!', parse_mode='HTML')
:param user: the user (not the user_id)
:param include_id: include the user_id
@@ -333,6 +340,7 @@ def quick_markup(values: Dict[str, Dict[str, Any]], row_width: int=2) -> types.I
}
:param values: a dict containing all buttons to create in this format: {text: kwargs} {str:}
+ :param row_width: int row width
:return: InlineKeyboardMarkup
"""
markup = types.InlineKeyboardMarkup(row_width=row_width)
@@ -363,8 +371,10 @@ def orify(e, changed_callback):
e.set = lambda: or_set(e)
e.clear = lambda: or_clear(e)
+
def OrEvent(*events):
or_event = threading.Event()
+
def changed():
bools = [ev.is_set() for ev in events]
if any(bools):
@@ -391,15 +401,18 @@ def per_thread(key, construct_value, reset=False):
return getattr(thread_local, key)
+
def chunks(lst, n):
"""Yield successive n-sized chunks from lst."""
# https://stackoverflow.com/a/312464/9935473
for i in range(0, len(lst), n):
yield lst[i:i + n]
+
def generate_random_token():
return ''.join(random.sample(string.ascii_letters, 16))
+
def deprecated(func):
"""This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emitted
diff --git a/tests/test_telebot.py b/tests/test_telebot.py
index a70911e..e8f57a9 100644
--- a/tests/test_telebot.py
+++ b/tests/test_telebot.py
@@ -6,6 +6,7 @@ sys.path.append('../')
import time
import pytest
import os
+from datetime import datetime, timedelta
import telebot
from telebot import types
@@ -407,6 +408,23 @@ class TestTeleBot:
cn = tb.get_chat_members_count(GROUP_ID)
assert cn > 1
+ def test_export_chat_invite_link(self):
+ tb = telebot.TeleBot(TOKEN)
+ il = tb.export_chat_invite_link(GROUP_ID)
+ assert isinstance(il, str)
+
+ def test_create_revoke_detailed_chat_invite_link(self):
+ tb = telebot.TeleBot(TOKEN)
+ cil = tb.create_chat_invite_link(GROUP_ID,
+ (datetime.now() + timedelta(minutes=1)).timestamp(), member_limit=5)
+ assert isinstance(cil.invite_link, str)
+ assert cil.creator.id == tb.get_me().id
+ assert isinstance(cil.expire_date, (float, int))
+ assert cil.member_limit == 5
+ assert not cil.is_revoked
+ rcil = tb.revoke_chat_invite_link(GROUP_ID, cil.invite_link)
+ assert rcil.is_revoked
+
def test_edit_markup(self):
text = 'CI Test Message'
tb = telebot.TeleBot(TOKEN)
diff --git a/tests/test_types.py b/tests/test_types.py
index 355f690..9ca73c9 100644
--- a/tests/test_types.py
+++ b/tests/test_types.py
@@ -219,3 +219,14 @@ def test_KeyboardButtonPollType():
json_str = markup.to_json()
assert 'request_poll' in json_str
assert 'quiz' in json_str
+
+
+def test_json_chat_invite_link():
+ json_string = r'{"invite_link": "https://t.me/joinchat/z-abCdEFghijKlMn", "creator": {"id": 329343347, "is_bot": false, "first_name": "Test", "username": "test_user", "last_name": "User", "language_code": "en"}, "is_primary": false, "is_revoked": false, "expire_date": 1624119999, "member_limit": 10}'
+ invite_link = types.ChatInviteLink.de_json(json_string)
+ assert invite_link.invite_link == 'https://t.me/joinchat/z-abCdEFghijKlMn'
+ assert isinstance(invite_link.creator, types.User)
+ assert not invite_link.is_primary
+ assert not invite_link.is_revoked
+ assert invite_link.expire_date == 1624119999
+ assert invite_link.member_limit == 10
\ No newline at end of file