Improve validate_token function, add type annotations

examples/asynchronous_telebot/continue_handling.py

@@ -15,12 +15,12 @@ async def start2(message):
     This handler comes after the first one, but it will never be called.
     But you can call it by returning ContinueHandling() in the first handler.
 
-    If you return ContinueHandling() in the first handler, the next 
+    If you return ContinueHandling() in the first handler, the next
     registered handler with appropriate filters will be called.
     """
     await bot.send_message(message.chat.id, 'Hello World2!')
 
-import asyncio 
+import asyncio
 asyncio.run(bot.polling()) # just a reminder that infinity polling
 # wraps polling into try/except block just as sync version,
 # but you can use any of them because neither of them stops if you

examples/asynchronous_telebot/mini_app.py

@@ -1,16 +1,16 @@
 # The source of the "https://pytelegrambotminiapp.vercel.app" can be found in https://github.com/eternnoir/pyTelegramBotAPI/tree/master/examples/mini_app_web
 
 import asyncio
-from telebot.async_telebot import AsyncTeleBot 
+from telebot.async_telebot import AsyncTeleBot
 from telebot.types import (
-    ReplyKeyboardMarkup, 
-    KeyboardButton, 
+    ReplyKeyboardMarkup,
+    KeyboardButton,
     WebAppInfo,
     InlineKeyboardMarkup,
     InlineKeyboardButton
 )
 
-BOT_TOKEN = "" 
+BOT_TOKEN = ""
 WEB_URL = "https://pytelegrambotminiapp.vercel.app"
 
 bot = AsyncTeleBot(BOT_TOKEN)

examples/asynchronous_telebot/timer_bot_async.py

@@ -1,6 +1,6 @@
 #!/usr/bin/python3
 
-# This is a simple bot with schedule timer 
+# This is a simple bot with schedule timer
 # https://github.com/ibrb/python-aioschedule
 # https://schedule.readthedocs.io
 

examples/continue_handling.py

@@ -15,7 +15,7 @@ def start2(message):
     This handler comes after the first one, but it will never be called.
     But you can call it by returning ContinueHandling() in the first handler.
 
-    If you return ContinueHandling() in the first handler, the next 
+    If you return ContinueHandling() in the first handler, the next
     registered handler with appropriate filters will be called.
     """
     bot.send_message(message.chat.id, 'Hello World2!')

examples/inline_example.py

@@ -42,7 +42,7 @@ def query_video(inline_query):
     try:
         r = types.InlineQueryResultVideo('1',
                                          'https://github.com/eternnoir/pyTelegramBotAPI/blob/master/tests/test_data/test_video.mp4?raw=true',
-                                         'video/mp4', 
+                                         'video/mp4',
                                          'https://raw.githubusercontent.com/eternnoir/pyTelegramBotAPI/master/examples/detailed_example/rooster.jpg',
                                          'Title'
                                          )

examples/message_reaction_example.py

@@ -3,11 +3,11 @@
 # This is a simple bot using message reactions (emoji)
 # https://core.telegram.org/bots/api#reactiontype
 # https://core.telegram.org/bots/api#update
-# allowed_updates: Specify an empty list to receive all update types except, chat_member, message_reaction, and message_reaction_count. 
-# If you want to receive message_reaction events, you cannot simply leave the allowed_updates=None default value. 
-# The default list of events does not include chat_member, message_reaction, and message_reaction_count events. 
-# You must explicitly specify all the events you wish to receive and add message_reaction to this list. 
-# It’s also important to note that after using allowed_updates=[...], in the future, using allowed_updates=None will mean 
+# allowed_updates: Specify an empty list to receive all update types except, chat_member, message_reaction, and message_reaction_count.
+# If you want to receive message_reaction events, you cannot simply leave the allowed_updates=None default value.
+# The default list of events does not include chat_member, message_reaction, and message_reaction_count events.
+# You must explicitly specify all the events you wish to receive and add message_reaction to this list.
+# It’s also important to note that after using allowed_updates=[...], in the future, using allowed_updates=None will mean
 # that the list of events you will receive will consist of the events you last explicitly specified.
 
 import random

examples/mini_app.py

@@ -2,15 +2,15 @@
 
 from telebot import TeleBot
 from telebot.types import (
-    ReplyKeyboardMarkup, 
-    KeyboardButton, 
+    ReplyKeyboardMarkup,
+    KeyboardButton,
     WebAppInfo,
     InlineKeyboardMarkup,
     InlineKeyboardButton
 )
 
-BOT_TOKEN = "" 
-WEB_URL = "https://pytelegrambotminiapp.vercel.app" 
+BOT_TOKEN = ""
+WEB_URL = "https://pytelegrambotminiapp.vercel.app"
 
 bot = TeleBot(BOT_TOKEN)
 

examples/stars_payment.py

@@ -1,6 +1,6 @@
 # Import required libraries
-import telebot 
-from telebot import types  
+import telebot
+from telebot import types
 
 bot = telebot.TeleBot('TOKEN')
 

telebot/__init__.py

@@ -168,7 +168,7 @@ def __init__(
             next_step_backend: Optional[HandlerBackend]=None, reply_backend: Optional[HandlerBackend]=None,
             exception_handler: Optional[ExceptionHandler]=None, last_update_id: Optional[int]=0,
             suppress_middleware_excepions: Optional[bool]=False, state_storage: Optional[StateStorageBase]=StateMemoryStorage(),
-            use_class_middlewares: Optional[bool]=False, 
+            use_class_middlewares: Optional[bool]=False,
             disable_web_page_preview: Optional[bool]=None,
             disable_notification: Optional[bool]=None,
             protect_content: Optional[bool]=None,
@@ -1571,7 +1571,7 @@ def get_chat_administrators(self, chat_id: Union[int, str]) -> List[types.ChatMe
         On success, returns an Array of ChatMember objects that contains
         information about all chat administrators except other bots.
 
-        Telegram documentation: https://core.telegram.org/bots/api#getchatadministrators    
+        Telegram documentation: https://core.telegram.org/bots/api#getchatadministrators
 
         :param chat_id: Unique identifier for the target chat or username
             of the target supergroup or channel (in the format @channelusername)
@@ -7051,7 +7051,7 @@ def create_new_sticker_set(
     def add_sticker_to_set(
             self, user_id: int, name: str, emojis: Union[List[str], str],
             png_sticker: Optional[Union[Any, str]]=None, 
-            tgs_sticker: Optional[Union[Any, str]]=None,  
+            tgs_sticker: Optional[Union[Any, str]]=None,
             webm_sticker: Optional[Union[Any, str]]=None,
             mask_position: Optional[types.MaskPosition]=None,
             sticker: Optional[types.InputSticker]=None) -> bool:

telebot/apihelper.py

@@ -190,7 +190,7 @@ def _check_result(method_name, result):
             raise ApiHTTPException(method_name, result)
         else:
             raise ApiInvalidJSONException(method_name, result)
-    else:    
+    else:
         if not result_json['ok']:
             raise ApiTelegramException(method_name, result, result_json)
 
@@ -2496,9 +2496,9 @@ def convert_input_media_array(array):
             if 'thumbnail' in media_dict:
                 thumbnail = media_dict['thumbnail']
                 if isinstance(thumbnail, types.InputFile):
-                    thumbnail_key = 'thumbnail_' + key  
-                    files[thumbnail_key] = thumbnail    
-                    media_dict['thumbnail'] = 'attach://' + thumbnail_key   
+                    thumbnail_key = 'thumbnail_' + key
+                    files[thumbnail_key] = thumbnail
+                    media_dict['thumbnail'] = 'attach://' + thumbnail_key
             media.append(media_dict)
     return json.dumps(media), files
 

telebot/async_telebot.py

@@ -237,7 +237,7 @@ async def get_updates(self, offset: Optional[int]=None, limit: Optional[int]=Non
 
         :return: An Array of Update objects is returned.
         :rtype: :obj:`list` of :class:`telebot.types.Update`
-        """       
+        """
         json_updates = await asyncio_helper.get_updates(self.token, offset, limit, timeout, allowed_updates, request_timeout)
         return [types.Update.de_json(ju) for ju in json_updates]
 
@@ -3045,7 +3045,7 @@ async def get_chat_administrators(self, chat_id: Union[int, str]) -> List[types.
         On success, returns an Array of ChatMember objects that contains
         information about all chat administrators except other bots.
 
-        Telegram documentation: https://core.telegram.org/bots/api#getchatadministrators    
+        Telegram documentation: https://core.telegram.org/bots/api#getchatadministrators
 
         :param chat_id: Unique identifier for the target chat or username
             of the target supergroup or channel (in the format @channelusername)
@@ -8474,8 +8474,8 @@ async def create_new_sticker_set(
 
     async def add_sticker_to_set(
             self, user_id: int, name: str, emojis: Union[List[str], str]=None,
-            png_sticker: Optional[Union[Any, str]]=None, 
-            tgs_sticker: Optional[Union[Any, str]]=None,  
+            png_sticker: Optional[Union[Any, str]]=None,
+            tgs_sticker: Optional[Union[Any, str]]=None,
             webm_sticker: Optional[Union[Any, str]]=None,
             mask_position: Optional[types.MaskPosition]=None,
             sticker: Optional[types.InputSticker]=None) -> bool:

telebot/asyncio_helper.py

@@ -271,7 +271,7 @@ async def _check_result(method_name, result: aiohttp.ClientResponse):
             raise ApiHTTPException(method_name, result)
         else:
             raise ApiInvalidJSONException(method_name, result)
-    else:    
+    else:
         if not result_json['ok']:
             raise ApiTelegramException(method_name, result, result_json)
 
@@ -446,7 +446,7 @@ async def forward_message(
 
 
 async def copy_message(token, chat_id, from_chat_id, message_id, caption=None, parse_mode=None, caption_entities=None,
-                 disable_notification=None,  
+                 disable_notification=None,
                  reply_markup=None, timeout=None, protect_content=None, message_thread_id=None, reply_parameters=None, show_caption_above_media=None,
                  allow_paid_broadcast=None, video_start_timestamp=None):
     method_url = r'copyMessage'
@@ -623,8 +623,8 @@ async def send_media_group(
 
 async def send_location(
         token, chat_id, latitude, longitude,
-        live_period=None,  
-        reply_markup=None, disable_notification=None, 
+        live_period=None,
+        reply_markup=None, disable_notification=None,
         timeout=None, horizontal_accuracy=None, heading=None,
         proximity_alert_radius=None,  protect_content=None, message_thread_id=None,reply_parameters=None, business_connection_id=None,
         message_effect_id=None, allow_paid_broadcast=None):
@@ -2258,7 +2258,7 @@ async def send_poll(
         token, chat_id, question, options,
         is_anonymous = None, type = None, allows_multiple_answers = None, correct_option_id = None,
         explanation = None, explanation_parse_mode=None, open_period = None, close_date = None, is_closed = None,
-        disable_notification=False,  
+        disable_notification=False,
         reply_markup=None, timeout=None, explanation_entities=None, protect_content=None, message_thread_id=None,
         reply_parameters=None,business_connection_id=None, question_parse_mode=None, question_entities=None, message_effect_id=None,
         allow_paid_broadcast=None):
@@ -2460,9 +2460,9 @@ async def convert_input_media_array(array):
             if 'thumbnail' in media_dict:
                 thumbnail = media_dict['thumbnail']
                 if isinstance(thumbnail, types.InputFile):
-                    thumbnail_key = 'thumbnail_' + key  
-                    files[thumbnail_key] = thumbnail    
-                    media_dict['thumbnail'] = 'attach://' + thumbnail_key     
+                    thumbnail_key = 'thumbnail_' + key
+                    files[thumbnail_key] = thumbnail
+                    media_dict['thumbnail'] = 'attach://' + thumbnail_key
             media.append(media_dict)
     return json.dumps(media), files
 

telebot/types.py

@@ -3272,15 +3272,15 @@ def __init__(self, user, status, custom_title=None, is_anonymous=None, can_be_ed
                  can_send_photos=None, can_send_videos=None, can_send_video_notes=None,
                  can_send_voice_notes=None,
                  can_send_polls=None,
-                 can_send_other_messages=None, can_add_web_page_previews=None,  
-                 can_manage_chat=None, can_manage_video_chats=None, 
-                 until_date=None, can_manage_topics=None, 
+                 can_send_other_messages=None, can_add_web_page_previews=None,
+                 can_manage_chat=None, can_manage_video_chats=None,
+                 until_date=None, can_manage_topics=None,
                  can_post_stories=None, can_edit_stories=None, can_delete_stories=None,
                  **kwargs):
         self.user: User = user
         self.status: str = status
         self.custom_title: str = custom_title
-        self.is_anonymous: bool = is_anonymous 
+        self.is_anonymous: bool = is_anonymous
         self.can_be_edited: bool = can_be_edited
         self.can_post_messages: bool = can_post_messages
         self.can_edit_messages: bool = can_edit_messages
@@ -3597,7 +3597,7 @@ class ChatPermissions(JsonDeserializable, JsonSerializable, Dictionaryable):
 
     :param can_manage_topics: Optional. True, if the user is allowed to create forum topics. If omitted defaults to the
         value of can_pin_messages
-    :type can_manage_topics: :obj:`bool`    
+    :type can_manage_topics: :obj:`bool`
 
     :param can_send_media_messages: deprecated.
     :type can_send_media_messages: :obj:`bool`
@@ -4310,7 +4310,7 @@ def to_dict(self):
         if self.need_shipping_address is not None:
             json_dict['need_shipping_address'] = self.need_shipping_address 
         if self.send_phone_number_to_provider is not None:
-            json_dict['send_phone_number_to_provider'] = self.send_phone_number_to_provider      
+            json_dict['send_phone_number_to_provider'] = self.send_phone_number_to_provider
         if self.send_email_to_provider is not None:
             json_dict['send_email_to_provider'] = self.send_email_to_provider 
         if self.is_flexible is not None:
@@ -7664,7 +7664,7 @@ class MessageAutoDeleteTimerChanged(JsonDeserializable):
 
     :return: Instance of the class
     :rtype: :class:`telebot.types.MessageAutoDeleteTimerChanged`
-    """   
+    """
     @classmethod
     def de_json(cls, json_string):
         if json_string is None: return None
@@ -9349,7 +9349,7 @@ def de_json(cls, json_string):
         obj = cls.check_json(json_string)
         if 'quote_entities' in obj:
             obj['quote_entities'] = [MessageEntity.de_json(entity) for entity in obj['quote_entities']]
-        return cls(**obj)    
+        return cls(**obj)
 
     def __init__(self, message_id: int, chat_id: Optional[Union[int, str]] = None,
                  allow_sending_without_reply: Optional[bool] = None, quote: Optional[str] = None,
@@ -9442,7 +9442,7 @@ class ChatBoostUpdated(JsonDeserializable):
     @classmethod
     def de_json(cls, json_string):
         if json_string is None: return None
-        obj = cls.check_json(json_string)        
+        obj = cls.check_json(json_string)
         obj['chat'] = Chat.de_json(obj['chat'])
         obj['boost'] = ChatBoost.de_json(obj['boost'])
         return cls(**obj)
@@ -9479,7 +9479,7 @@ def de_json(cls, json_string):
         if json_string is None: return None
         obj = cls.check_json(json_string)
         obj['chat'] = Chat.de_json(obj['chat'])
-        obj['source'] = ChatBoostSource.de_json(obj['source'])        
+        obj['source'] = ChatBoostSource.de_json(obj['source'])
         return cls(**obj)
 
     def __init__(self, chat, boost_id, remove_date, source, **kwargs):

telebot/util.py

@@ -463,7 +463,7 @@ def quick_markup(values: Dict[str, Dict[str, Any]], row_width: int = 2) -> types
     return markup
 
 
-# CREDITS TO http://stackoverflow.com/questions/12317940#answer-12320352
+# CREDITS TO https://stackoverflow.com/questions/12317940#answer-12320352
 def or_set(self):
     """
     :meta private:
@@ -519,7 +519,7 @@ def busy_wait():
     return or_event
 
 
-def per_thread(key, construct_value, reset=False):
+def per_thread(key: str, construct_value, reset=False):
     """
     :meta private:
     """
@@ -660,7 +660,7 @@ def parse_web_app_data(token: str, raw_init_data: str):
     return result
 
 
-def validate_web_app_data(token: str, raw_init_data: str):
+def validate_web_app_data(token: str, raw_init_data: str) -> bool:
     """
     Validates web app data.
 
@@ -670,7 +670,8 @@ def validate_web_app_data(token: str, raw_init_data: str):
     :param raw_init_data: The raw init data
     :type raw_init_data: :obj:`str`
 
-    :return: The parsed init data
+    :return: Whether the web app data is valid or not
+    :rtype: :obj:`bool`
     """
     try:
         parsed_data = dict(parse_qsl(raw_init_data))
@@ -686,19 +687,40 @@ def validate_web_app_data(token: str, raw_init_data: str):
     return hmac.new(secret_key.digest(), data_check_string.encode(), sha256).hexdigest() == init_data_hash
 
 
-def validate_token(token) -> bool:        
+def validate_token(token: str) -> bool:
+    """
+    Validates bot token.
+
+    :param token: The bot token
+    :type token: :obj:`str`
+
+    :return: Whether the bot token is valid or not
+    :rtype: :obj:`bool`
+    """
+
     if any(char.isspace() for char in token):
         raise ValueError('Token must not contain spaces')
-    
+
     if ':' not in token:
         raise ValueError('Token must contain a colon')
-
-    if len(token.split(':')) != 2:
+
+    parts = tuple(token.split(':'))
+
+    if len(parts) != 2:
         raise ValueError('Token must contain exactly 2 parts separated by a colon')
-
+
+    bot_id, alphanumeric_part = parts
+
+    if not bot_id.isdigit():
+        raise ValueError('First part must contain only numbers')
+
+    if len(alphanumeric_part) != 35:
+        raise ValueError('Second part must be exactly 35 characters long')
+
     return True
 
-def extract_bot_id(token) -> Union[int, None]:
+
+def extract_bot_id(token: str) -> Union[int, None]:
     try:
         validate_token(token)
     except ValueError: