diff --git a/utils_v2/telegram/models/behaviour/telegram_bot.py b/utils_v2/telegram/models/behaviour/telegram_bot.py index 33ea305..51c4bb6 100644 --- a/utils_v2/telegram/models/behaviour/telegram_bot.py +++ b/utils_v2/telegram/models/behaviour/telegram_bot.py @@ -14,7 +14,7 @@ REFERENCES: - N/A + 1. Official API Documentation: https://core.telegram.org/bots/api DOWNLOADS: @@ -37,9 +37,11 @@ sys.path.append("..") # System-level activities: import io +import os # My utils: from utils_v2.string import json +from utils_v2.system import files from utils_v2.date_time import date_time # Data models: @@ -210,7 +212,9 @@ class AsyncTelegramBot: headers: dict = None, json: dict = None, data: dict = None, - content: str | bytes = None + params: dict = None, + content: str | bytes = None, + files: dict = None ) -> TelegramApiResponse: """ @@ -219,7 +223,9 @@ class AsyncTelegramBot: :param headers: The headers to pass. :param json: The params to send in the JSON body. :param data: The params to send in the form-data in the body. + :param params: The params to send in the query string itself. :param content: The raw content to be sent in the body (typically as an octet-stream). + :param files: Any file that you may want to send. :return: A structured response that includes the raw response, the exception (if any), and so on. """ @@ -238,7 +244,9 @@ class AsyncTelegramBot: headers = headers, json = json, data = data, - content = content + params = params, + content = content, + files = files ) # Note down the results: @@ -347,9 +355,9 @@ class AsyncTelegramBot: # Done here: return api_response - # ┳┳┓ ┓┓ ┏┓ ┓┓ ┳┓ - # ┃┃┃┏┓┏┓┓┏┏┓┃┃┓┏ ┃┃┓┏┃┃ ┃┃┏┓╋┏┓ - # ┛ ┗┗┻┛┗┗┻┗┻┗┗┗┫ ┣┛┗┻┗┗ ┻┛┗┻┗┗┻ + # ┳┓ • • + # ┣┫┏┓┏┏┓┓┓┏┓┏┓┏┓ + # ┛┗┗ ┗┗ ┗┗┛┗┛┗┗┫ # ┛ async def get_updates( @@ -383,7 +391,6 @@ class AsyncTelegramBot: # If the call succeeds: if api_response.httpCode in [200, 400]: api_json = await api_response.get_json() - # api_response.message = api_json["description"] api_response.data = api_json if api_json["ok"]: api_response.success = True @@ -392,6 +399,218 @@ class AsyncTelegramBot: # Done here: return api_response + async def get_file( + self, + file_id: str, + ) -> TelegramApiResponse: + + """ + To download a file from Telegram. + NOTE: THIS DOES NOT WORK WHILE A WEBHOOK IS SET UP. YOU CAN EITHER USE THIS OR A WEBHOOK, NOT BOTH + SIMULTANEOUSLY. + :param file_id: The 'file_id' of this file. Not to be confused with 'file_unique_id'. + :return: A standard response structure with the file's content in the 'data' field. + """ + + # Make the API call: + api_response = await self.__get( + url = f"https://api.telegram.org/bot{self.__bot_token}/getFile", + params = {"file_id": file_id} + ) + + # If the call failed: + if api_response.httpCode == 400: + api_json = await api_response.get_json() + api_response.message = api_json["description"] + + # If the call succeeds: + if api_response.httpCode == 200: + api_json = await api_response.get_json() + file_path = api_json["result"]["file_path"] + api_response = await self.__get(url = f"https://api.telegram.org/file/bot{self.__bot_token}/{file_path}") + if api_response.httpCode == 200: + api_response.success = True + api_response.data = await api_response.get_content() + + # Done here: + return api_response + + # ┏┓ ┓• + # ┗┓┏┓┏┓┏┫┓┏┓┏┓ + # ┗┛┗ ┛┗┗┻┗┛┗┗┫ + # ┛ + + @staticmethod + def escape_special_chars( + text: str, + parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None + ) -> str: + + r""" + Escapes certain special chars for each type of parsing that is available. There's not much use of this function, + though, TBH. This can be useful when trying to parse more complex thing like links. + USE CASE: Suppose you want to send a message "Equation: 5 * 2 = 10" such that the actual equation part + ("5 * 2 = 10") is in bold, and you want this through, say, MarkdownV2 formatting. You could achieve this by + first passing the part that you want to format through this function and then inserting it in the rest. Maybe + like this: + >> escaped_message = f"Equation: *{escape_special_chars(text = '5 * 2 = 10', parse_mode = 'MarkdownV2')}*" + Or you could just manually achieve the same by typing: + >> escaped_message = r"Equation: *5 \* 2 \= 10*" + :param text: The raw string. + :param parse_mode: The kind of parsing that is desired. + :return: The escaped string. + """ + + # For MarkdownV2: + if parse_mode == "Markdown" or parse_mode == "MarkdownV2": + for k, v in { + "*": "\\*", + "`": "\\`", + "[": "\\[", + "]": "\\]", + "(": "\\(", + ")": "\\)", + "~": "\\~", + ">": "\\>", + "#": "\\#", + "+": "\\+", + "-": "\\-", + ".": "\\.", + "!": "\\!", + "=": "\\=", + }.items(): text = text.replace(k, v) + + # For HTML: + elif parse_mode == "HTML": + for k, v in { + "&": "&", + "<": "<", + ">": ">", + "\"": """, + "'": "'" + }.items(): text = text.replace(k, v) + + # Done here: + return text + + async def send_message( + self, + chat_id: int | str, + message: str, + parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None, + disable_notification: bool = False, + protect_content: bool = False + ) -> TelegramApiResponse: + + """ + To send one message to one recipient using a specified parse mode. + :param chat_id: The id of the recipient user/group/channel. + :param message: The actual message that you want to send. + :param parse_mode: The parsing mode for adding visual formatting. + :param disable_notification: Use this when you want to send an alert without causing the recipient's device to + show a notification alert. + :param protect_content: Use this to prevent recipient participants from copying/forwarding messages. + :return: A standard response structure with an instance of 'TelegramUpdate' in the 'data' field. + """ + + # Make the API call: + api_response = await self.__get( + url = f"https://api.telegram.org/bot{self.__bot_token}/sendMessage", + params = { + "chat_id": chat_id, + "text": message, + "parse_mode": parse_mode, + "disable_notification": disable_notification, + "protect_content": protect_content + } + ) + + # If the call failed: + if api_response.httpCode == 400: + api_json = await api_response.get_json() + api_response.message = api_json["description"] + + # If the call succeeds: + if api_response.httpCode == 200: + api_json = await api_response.get_json() + if api_json["ok"]: + api_response.success = True + api_response.data = TelegramUpdate(message = api_json["result"]) + + # Done here: + return api_response + + async def send_photo( + self, + chat_id: int | str, + file: str | io.BytesIO, + file_name: str = None, + caption: str = None, + parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None, + hide: bool = False, + disable_notification: bool = False, + protect_content: bool = False + ) -> TelegramApiResponse: + + """ + To send one photo from the bot to one recipient. A caption with a specified parsing mode may also be added. + Note that you must send at least one of the three file options. + :param chat_id: The id of the recipient user/group/channel. + :param file: Could be either a file on the disk, or a file held in a buffer in RAM or a 'file_id' of a file + already on Telegram's server, or a URL of a file hosted on the internet. + :param file_name: The name of the file. + :param caption: The text to write under the file. + :param parse_mode: The parsing mode for adding visual formatting to the caption. + :param hide: Whether, or not, you would like to apply a spoiler on top of the file. + :param disable_notification: Use this when you want to send an alert without causing the recipient's device to + show a notification alert. + :param protect_content: Use this to prevent recipient participants from copying/forwarding messages. + :return: A standard response structure with an instance of 'TelegramUpdate' in the 'data' field. + """ + + # Start with placeholder params: + request_params = { + "chat_id": chat_id, + "disable_notification": disable_notification, + "protect_content": protect_content + } + if caption: request_params["caption"] = caption + if parse_mode: request_params["parse_mode"] = parse_mode + if hide: request_params["has_spoiler"] = True + request_files = None + + # If the file is given to you as a buffer: + if isinstance(file, io.BytesIO): + file.seek(0) + request_files = {"photo": (file_name, file)} + + # If the file is given to you as string: + elif isinstance(file, str): + if os.path.exists(file): request_files = {"photo": (file_name, open(file, "rb"))} + else: request_params["photo"] = file + + # Make the API call: + api_response = await self.__post( + url = f"https://api.telegram.org/bot{self.__bot_token}/sendPhoto", + params = request_params, + files = request_files + ) + + # If the call failed: + if api_response.httpCode == 400: + api_json = await api_response.get_json() + api_response.message = api_json["description"] + + # If the call succeeds: + if api_response.httpCode == 200: + api_json = await api_response.get_json() + if api_json["ok"]: + api_response.success = True + api_response.data = TelegramUpdate(message = api_json["result"]) + + # Done here: + return api_response + # ***************************************************************************************************************** # ***** **** @@ -403,18 +622,33 @@ class AsyncTelegramBot: if __name__ == "__main__": import asyncio + from PIL import Image async def main(): # Create the instance and define needed variables: my_tg = AsyncTelegramBot(bot_token = r"7003670393:AAH9qF6XGqa-u2_TM2JCUkP0Fp48kMw8Ka8") - recipient_chat_id = "" + recipient_chat_id = 1275560043 # Test out the services: # tg_response = await my_tg.bot_info() # tg_response = await my_tg.set_webhook(webhook_url = r"https://api.thecaoffice.com/converse/chat/webhook/123") # tg_response = await my_tg.delete_webhook() - tg_response = await my_tg.get_updates(update_id = 872199521, limit = 5) + # tg_response = await my_tg.get_updates(update_id = 872199521, limit = 10) + # tg_response = await my_tg.get_file(file_id = r"BQACAgUAAxkBAAIHZWdT4NNG6WwSYLsHeT-0RccCGUVaAAKeFAAC5T-hVoKttePhObnbNgQ") + # tg_response = await my_tg.send_message( + # chat_id = recipient_chat_id, + # message = f"Hello, World\! \N{WAVING HAND SIGN}\n[ASCII Art]({my_tg.escape_special_chars(text = r'https://patorjk.com/software/taag/#p=display&f=Tmplr&t=Type%20Something%20', parse_mode = 'MarkdownV2')})", + # parse_mode = "MarkdownV2" + # ) + tg_response = await my_tg.send_photo( + chat_id = recipient_chat_id, + file_name = "cat.jpg", + file = r"https://images.unsplash.com/photo-1543852786-1cf6624b9987", + caption = f"[Image URL]({my_tg.escape_special_chars(text = r'https://images.unsplash.com/photo-1543852786-1cf6624b9987', parse_mode = 'MarkdownV2')})", + parse_mode = "MarkdownV2", + hide = False + ) # Show the response: print("SUMMARY:", tg_response.to_markdown(), "\n\n---\n\n") diff --git a/utils_v2/telegram/models/data/update.py b/utils_v2/telegram/models/data/update.py index 25db35e..423bf2c 100644 --- a/utils_v2/telegram/models/data/update.py +++ b/utils_v2/telegram/models/data/update.py @@ -86,10 +86,11 @@ class TelegramParticipant(BaseModel): frozen = True ) - firstName: str = Field( + firstName: str | None = Field( description = "the first name of this participant", alias = "first_name", - frozen = True + frozen = True, + default = None ) lastName: str | None = Field( @@ -133,10 +134,11 @@ class TelegramChat(BaseModel): frozen = True ) - firstName: str = Field( + firstName: str | None = Field( description = "the first name of this chat", alias = "first_name", - frozen = True + frozen = True, + default = None ) lastName: str | None = Field( @@ -661,8 +663,10 @@ class TelegramMessage(BaseModel): @field_validator("photo") def sort_photo_sizes(cls, value): - value = sorted(value, key = lambda x: x.width, reverse = True) - for index, res in enumerate(["original", "large", "medium", "small"]): value[index].resolution = res + value = sorted(value, key = lambda x: x.width, reverse = False) + # res_list = ["original", "large", "medium", "small"] + res_list = ["small", "medium", "large", "max"] + for index, res in enumerate(res_list[:len(value)]): value[index].resolution = res return value @@ -671,10 +675,11 @@ class TelegramMessage(BaseModel): class TelegramUpdate(BaseModel): - updateId: int = Field( + updateId: int | None = Field( description = "an update from telegram; typically a message", alias = "update_id", - frozen = True + frozen = True, + default = None ) message: TelegramMessage | None = Field(