diff --git a/pyrogram/client.py b/pyrogram/client.py index 6907ee05f..5d9277fdd 100644 --- a/pyrogram/client.py +++ b/pyrogram/client.py @@ -362,6 +362,8 @@ def __init__( self.message_cache = Cache(self.max_message_cache_size) self.business_user_connection_cache = Cache(self.max_business_user_connection_cache_size) + self.app_constant = Constant(client_instance=self) + # Sometimes, for some reason, the server will stop sending updates and will only respond to pings. # This watchdog will invoke updates.GetState in order to wake up the server and enable it sending updates again # after some idle time has been detected. @@ -1242,3 +1244,47 @@ def __setitem__(self, key, value): if len(self.store) > self.capacity: for _ in range(self.capacity // 2 + 1): del self.store[next(iter(self.store))] + + +class Constant: + """Tuple of min, max, premium max lengths""" + + # Text of the message to be sent, 1-4096 characters + MAX_TEXT_LENGTH = (1, 4096, 0) + + # Caption for the animation, audio, document, photo, video or voice, 0-1024 characters + MAX_CAPTION_LENGTH = (0, 1024, 0) + + MAX_FIRST_NAME_LENGTH = (1, 64, 0) + MAX_LAST_NAME_LENGTH = (0, 64, 0) + + # USER's BIO + MAX_BIO_LENGTH = (0, 70, 140) + + # ADMIN TITLE + MAX_TITLE_LENGTH = (0, 16, 0) + + # Use the InlineQuery.answer() method. No more than 50 results per query are allowed. + MAX_RESULTS_LENGTH = (0, 50, 0) + + def __init__(self, client_instance: Client): + self.is_premium = client_instance.me.is_premium + + def check_valid_length(self, text: Union[List, str], arg_type: str, ): + if not isinstance(text, (str, list)): + raise ValueError(f"Argument {arg_type} must be a str | list") + + text_length = len(text) + + attr_str = f"max_{arg_type}_length".upper() + + # MIN, MAX, PREM-MAX + lengths: tuple[int, int, int] = getattr(self, attr_str, (0, 0, 0)) + + min_length = lengths[0] + # Check if client is premium and the premium value isn't 0 + max_length = lengths[2] if self.is_premium and lengths[2] != 0 else lengths[1] + + error_info = f"\nInvalid length of {text_length} for arg {arg_type}\nValid Lengths: {min_length}-{max_length}" + + assert bool(min_length < text_length <= max_length), error_info diff --git a/pyrogram/methods/bots/answer_inline_query.py b/pyrogram/methods/bots/answer_inline_query.py index c3a450a01..31f524555 100644 --- a/pyrogram/methods/bots/answer_inline_query.py +++ b/pyrogram/methods/bots/answer_inline_query.py @@ -96,6 +96,8 @@ async def answer_inline_query( InputTextMessageContent("Message content"))]) """ + self.app_constant.check_valid_length(text=results, arg_type="results") + return await self.invoke( raw.functions.messages.SetInlineBotResults( query_id=int(inline_query_id), diff --git a/pyrogram/methods/chats/set_administrator_title.py b/pyrogram/methods/chats/set_administrator_title.py index 2c77066ed..767f6537d 100644 --- a/pyrogram/methods/chats/set_administrator_title.py +++ b/pyrogram/methods/chats/set_administrator_title.py @@ -16,7 +16,7 @@ # You should have received a copy of the GNU Lesser General Public License # along with Pyrogram. If not, see . -from typing import Union +from typing import Optional, Union import pyrogram from pyrogram import raw @@ -27,7 +27,7 @@ async def set_administrator_title( self: "pyrogram.Client", chat_id: Union[int, str], user_id: Union[int, str], - title: str, + title: Optional[str] = None, ) -> bool: """Set a custom title (rank) to an administrator of a supergroup. @@ -45,6 +45,7 @@ async def set_administrator_title( For a contact that exists in your Telegram address book you can use his phone number (str). title (``str``, *optional*): + New custom title for the administrator; 0-16 characters, emoji are not allowed. A custom title that will be shown to all members instead of "Owner" or "Admin". Pass None or "" (empty string) to remove the custom title. @@ -56,6 +57,10 @@ async def set_administrator_title( await app.set_administrator_title(chat_id, user_id, "Admin Title") """ + + if title: + self.app_constant.check_valid_length(text=title, arg_type="title") + chat_id = await self.resolve_peer(chat_id) user_id = await self.resolve_peer(user_id) diff --git a/pyrogram/methods/messages/edit_message_text.py b/pyrogram/methods/messages/edit_message_text.py index 85ca2cb2f..e2d4d6239 100644 --- a/pyrogram/methods/messages/edit_message_text.py +++ b/pyrogram/methods/messages/edit_message_text.py @@ -94,6 +94,7 @@ async def edit_message_text( ) ) """ + if disable_web_page_preview and link_preview_options: raise ValueError( "Parameters `disable_web_page_preview` and `link_preview_options` are mutually " diff --git a/pyrogram/methods/messages/send_message.py b/pyrogram/methods/messages/send_message.py index d4cc8342d..ef9a7ea9b 100644 --- a/pyrogram/methods/messages/send_message.py +++ b/pyrogram/methods/messages/send_message.py @@ -157,6 +157,7 @@ async def send_message( ] ])) """ + if disable_web_page_preview and link_preview_options: raise ValueError( "Parameters `disable_web_page_preview` and `link_preview_options` are mutually " @@ -192,6 +193,9 @@ async def send_message( ) message, entities = (await utils.parse_text_entities(self, text, parse_mode, entities)).values() + if message: + self.app_constant.check_valid_length(text=message, arg_type="text") + session = None business_connection = None if business_connection_id: diff --git a/pyrogram/methods/users/update_profile.py b/pyrogram/methods/users/update_profile.py index 6c44ffe50..dfd6987d6 100644 --- a/pyrogram/methods/users/update_profile.py +++ b/pyrogram/methods/users/update_profile.py @@ -63,6 +63,10 @@ async def update_profile( await app.update_profile(last_name="") """ + for var_name in ("first_name", "last_name", "bio"): + if var_value := locals()[var_name]: + self.app_constant.check_valid_length(text=var_value, arg_type=var_name) + return bool( await self.invoke( raw.functions.account.UpdateProfile( diff --git a/pyrogram/session/session.py b/pyrogram/session/session.py index 5df270bda..1e8425536 100644 --- a/pyrogram/session/session.py +++ b/pyrogram/session/session.py @@ -125,7 +125,7 @@ async def start(self): await self.send(raw.functions.Ping(ping_id=0), timeout=self.START_TIMEOUT) if not self.is_cdn: - await self.send( + cfg = await self.send( raw.functions.InvokeWithLayer( layer=layer, query=raw.functions.InitConnection( @@ -146,6 +146,9 @@ async def start(self): ), timeout=self.START_TIMEOUT ) + if isinstance(cfg, raw.types.Config): # TODO + self.client.app_constant.MAX_TEXT_LENGTH = (1, cfg.message_length_max, cfg.message_length_max) + self.client.app_constant.MAX_CAPTION_LENGTH = (0, cfg.caption_length_max, cfg.caption_length_max) self.ping_task = self.loop.create_task(self.ping_worker())