(20241207) More progress in the telegram wrapper.

2024-12-07 13:41:40 +05:30
parent 3f13ff8517
commit 8b26baa29a
2 changed files with 256 additions and 17 deletions
@@ -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 {
+                "&": "&amp;",
+                "<": "&lt;",
+                ">": "&gt;",
+                "\"": "&quot;",
+                "'": "&apos;"
+            }.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")
@@ -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(