""" AUTHOR: Khushal P Soonderji DATE: Friday, 6th Dec., 2024 OBJECTIVE: To provide a way to send and receive messages and media via Telegram asynchronously. REFERENCES: 1. Official API Documentation: https://core.telegram.org/bots/api DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") 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: from utils_v2.telegram.models.api_call import TelegramApiResponse from utils_v2.telegram.models.update import TelegramUpdate # To make API calls: import httpx # To work with date and time: import datetime # For working with datatypes: from typing import Literal, List, Any # For debugging: from icecream import IceCreamDebugger import inspect # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncTelegramBot: def __init__( self, bot_token: str, http_client: httpx.AsyncClient = None, debug = True, debug_prefix = "TGram | ", debug_only_errors = True ): """ To initialize the instance of this Telegram messenger. :param bot_token: The token granted by BotFather. :param http_client: An instance of 'httpx' library's AsyncClient. If not given, one will be instantiated internally. It is recommended that, for multi-bot use cases, you provide a common HTTP client from outside. :param debug: Whether, or not, you would like to show debugging messages on the terminal. :param debug_prefix: The prefix string to identify the debugging messages. :param debug_only_errors: Whether you would like to show all debugging messages or just error messages. """ # Prepare the debugging utility: self._debug_prefix = debug_prefix self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) if not debug: self._printer.disable() self._debug_only_errors = debug_only_errors # Accept the input configuration: self.__bot_token = bot_token # Accept/create an HTTP client to work with: if http_client: self.__http_client = http_client else: self.__http_client = httpx.AsyncClient( limits = httpx.Limits( max_connections = 100, # ............ Maximum number of connections allowed in the pool. max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. ), timeout = httpx.Timeout( pool = 120.0, # .... Time to wait for a free connection from the pool. connect = 5.0, # ... Time to wait for establishing a connection to the server. write = 10.0, # .... Time to wait for sending data. read = 120.0 # ..... Time to wait for receiving data. ) ) def enable_debug(self): self._printer.enable() def disable_debug(self): self._printer.disable() def debug_only_errors(self): self._debug_only_errors = True def debug_everything(self): self._debug_only_errors = False # ┏┓┏┓┳ ┏┓ ┓┓• # ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓ # ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫ # ┛ async def __get( self, url: str, headers: dict = None, params: dict = None ) -> TelegramApiResponse: """ To call an API using the GET method. :param url: The URL to call. :param headers: The headers to pass. :param params: The params to send in the query string itself. :return: A structured response that includes the raw response, the exception (if any), and so on. """ # Prepare the structure of the response: api_response = TelegramApiResponse( action = inspect.stack()[1].function, url = url, method = "GET" ) try: # Make the API call: response = await self.__http_client.get( url = url, headers = headers, params = params ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = response.reason_phrase # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self._printer(exception, api_response.url, api_response.method, headers, params) # Done here: return api_response async def __post( self, url: str, headers: dict = None, json: dict = None, data: dict = None, params: dict = None, content: str | bytes = None, files: dict = None ) -> TelegramApiResponse: """ To call an API using the POST method. :param url: The URL to call. :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. """ # Prepare the structure of the response: api_response = TelegramApiResponse( action = inspect.stack()[1].function, url = url, method = "POST" ) try: # Make the API call: response = await self.__http_client.post( url = url, headers = headers, json = json, data = data, params = params, content = content, files = files ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = response.reason_phrase # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self._printer(exception, api_response.url, api_response.method, headers, json, data) # Done here: return api_response # ┳┓ ┳ ┏ # ┣┫┏┓╋ ┃┏┓╋┏┓ # ┻┛┗┛┗ ┻┛┗┛┗┛ async def bot_info(self) -> TelegramApiResponse: """ To get a bot's basic info. :return: The structured response with the JSON response from Telegram in the 'data' field. """ # Make the API call: api_response = await self.__get( url = f"https://api.telegram.org/bot{self.__bot_token}/getMe", ) # If the call succeeds: if api_response.httpCode in [200]: api_json = await api_response.get_json() if api_json["ok"]: api_result = api_json["result"] api_response.success = True api_response.data = { "id": api_result["id"], "displayName": api_result["first_name"], "username": api_result["username"], } # Done here: return api_response # ┓ ┏ ┓ ┓ ┓ # ┃┃┃┏┓┣┓┣┓┏┓┏┓┃┏ # ┗┻┛┗ ┗┛┛┗┗┛┗┛┛┗ async def set_webhook( self, webhook_url: str ) -> TelegramApiResponse: """ To set up a webhook to start receiving updates from Telegram. Use this same method to update an existing webhook without having to call the 'delete_webhook' method first. NOTE: WHEN A WEBHOOK IS SET, YOU CANNOT MANUALLY PULL UPDATES USING THE 'get_updates' METHOD. :param webhook_url: The URL on which Telegram must send you updates for this bot. :return: A standard response structure. """ # Make the API call: api_response = await self.__post( url = f"https://api.telegram.org/bot{self.__bot_token}/setWebhook", data = {"url": webhook_url} ) # 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 # Done here: return api_response async def delete_webhook( self, ) -> TelegramApiResponse: """ To stop receiving updates for this bot via the webhook. You can call this at any time even if no webhook is active, and you just want to be sure. NOTE: WHEN A WEBHOOK IS SET, YOU CANNOT MANUALLY PULL UPDATES USING THE 'get_updates' METHOD. :return: A standard response structure. """ # Make the API call: api_response = await self.__get( url = f"https://api.telegram.org/bot{self.__bot_token}/deleteWebhook", ) # 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 # Done here: return api_response # ┳┓ • • # ┣┫┏┓┏┏┓┓┓┏┓┏┓┏┓ # ┛┗┗ ┗┗ ┗┗┛┗┛┗┗┫ # ┛ async def get_updates( self, update_id: int = None, limit: int = 50, timeout: int | float = 30.0 ) -> TelegramApiResponse: """ To manually pull updates 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 update_id: The starting update id from where you want to start fetching updates. Leave this as None to fetch all available updates. Useful for when the webhook goes down and updates get missed. :param limit: The no. of updates to fetch. :param timeout: The max. timeout in seconds for long-polling. :return: A standard response structure with a list of 'TelegramUpdate's in the 'data' field. """ # Make the API call: api_response = await self.__get( url = f"https://api.telegram.org/bot{self.__bot_token}/getUpdates", params = { "offset": update_id, "limit": limit, "timeout": int(timeout) } ) # If the call succeeds: if api_response.httpCode in [200, 400]: api_json = await api_response.get_json() api_response.data = api_json if api_json["ok"]: api_response.success = True api_response.data = [TelegramUpdate(**r) for r in api_json["result"]] # Done here: return api_response async def get_file( self, file_id: str, ) -> TelegramApiResponse: """ To downloads 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: Any, 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. """ # Null safety: if not isinstance(text, str): return text if text is None: return "null" # 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 # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** 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 = 1275560043 recipient_chat_id = 7501974519 # 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 = 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 = True ) # Show the response: print("SUMMARY:", tg_response.to_markdown(), "\n\n---\n\n") if tg_response.success: print("DATA:", json.to_string(tg_response.data, default = str)) else: print("JSON:", json.to_string(await tg_response.get_json(), default = str)) asyncio.run(main())