(20250120) Started integrating the WhatsApp (Nimbus) code from BCT.

2025-01-20 19:06:38 +05:30
parent 54f62bc986
commit 96bb5768da
8 changed files with 686 additions and 0 deletions
@@ -0,0 +1,254 @@
+"""
+
+    AUTHOR:
+
+        Bhushan C Thakkar,
+        Khushal P Soonderji
+
+    DATE:
+
+        Monday, 20th Jan., 2025.
+
+    OBJECTIVE:
+
+        To be able to send Whatsapp from Nimbus Whatsapp`s API.
+
+    REFERENCES:
+
+        01. https://api.wtap.sms4power.com/
+
+    DOWNLOADS:
+
+        N/A
+
+    WEB-PORTAL:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   IMPORT                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+sys.path.append(".")
+sys.path.append("..")
+
+# For system-level activities:
+import os
+import io
+
+# My utils:
+from utils_v2.string import json
+
+# To make API Calls:
+import httpx
+
+# Data models:
+from utils_v2.whatsapp.nimbus.models.send_message_response import NimbusWhatsappSendMessageResponse
+
+# To work with date and time:
+import datetime
+
+# To work with datatypes:
+from typing import List
+
+# For debugging:
+from icecream import IceCreamDebugger
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  CLASSES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class AsyncNimbusWhatsapp:
+
+    # ┏┓
+    # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
+    # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
+
+    def __init__(
+            self,
+            api_key: str,
+            http_client: httpx.AsyncClient = None,
+            debug = True,
+            debug_prefix = "Nimbus (WA) | ",
+            debug_only_errors = True
+    ):
+
+        """
+        Sets up an instance of a message sender that sends messages through Nimbus Whatsapp's API.
+        :param api_key: The API key received from Nimbus Whatsapp.
+        :param http_client: An instance of 'httpx' library's AsyncClient. If not given, one will be instantiated
+            internally. It is recommended that, for multi-auth use cases, you provide a common HTTP client from outside.
+        :param debug: Whether, or not, you would like to show debugging messages (can be changed on the fly).
+        :param debug_prefix: The prefix text to show with the debug string.
+        :param debug_only_errors: Whether you would like to show all debugging messages or just error messages.
+        """
+
+        # Create the debugging tools:
+        self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
+        if not debug: self.__printer.disable()
+        self.__debug_only_errors = debug_only_errors
+
+        # 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 = 2.5,  # ... Time to wait for establishing a connection to the server.
+                write = 10.0,  # .... Time to wait for sending data.
+                read = 2.5  # ....... Time to wait for receiving data.
+            )
+        )
+
+        # Capture the input config:
+        self.__api_key = api_key
+
+    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 send_whatsapp(
+            self,
+            recipient_number: str | List[str],
+            message: str,
+            pdf_url: str = None,
+            image_0_url: str = None,
+            image_1_url: str = None,
+            schedule_on: datetime.datetime = None
+    ) -> NimbusWhatsappSendMessageResponse:
+
+        """
+        To send one message over WhatsApp through Nimbus's API.
+        :param recipient_number: The phone no(s). of the target recipient(s).
+        :param message: The actual content of the message.
+        :param pdf_url: A URL to a PDF file that should be sent to the recipient.
+        :param image_0_url: A URL to an image file that should be sent to the recipient.
+        :param image_1_url: A URL to an image file that should be sent to the recipient.
+        :param schedule_on: The date-time (UTC) to schedule the message delivery at.
+        :return: A structured response.
+        """
+
+        # Start by constructing the response model:
+        send_model = NimbusWhatsappSendMessageResponse(
+            recipientNo = recipient_number,
+            message = message,
+            pdfUrl = pdf_url,
+            image0Url = image_0_url,
+            image1Url = image_1_url,
+            scheduleTs = schedule_on
+        )
+
+        try:
+
+            # Call the API:
+            api_response = await self.__http_client.post(
+                url = r"http://api.wtap.sms4power.com/wapp/api/send/json",
+                headers = {"X-API-KEY":  self.__api_key},
+                json = send_model.api_input_json()
+            )
+
+            # Model the response:
+            send_model = send_model.from_api_response(
+                http_code = api_response.status_code,
+                response_json = api_response.json()
+            )
+
+            # print("HTTP CODE:", api_response.status_code)
+            # print("RAW CONTENT:", api_response.content)
+            # api_json = api_response.json()
+            # print("API JSON:", api_json)
+            # print("API JSON:", json.to_string(api_json, default=str))
+
+            # # For a successful API call:
+            # if api_response.status_code in [200]:
+            #     api_json = api_response.json()
+            #     print(api_json)
+            #     # summary.rawResponse = api_json
+            #     # print (summary.rawResponse)
+            #
+            #     first_response = api_json
+            #     first_desc = first_response.get("status", "N/A").lower().strip()
+            #     summary.success = True if first_desc == "success" else False
+            #     if summary.success:
+            #         summary.messageId = first_response.get("requestid")
+            #     else:
+            #         summary.message = first_response.get("status")
+            #     print (summary)
+            # # For any other code that indicates some form of failure:
+            # else: summary.rawResponse = api_response.content.decode()
+
+        # If something goes wrong along the way:
+        except Exception as exception:
+            self.__printer(exception)
+
+        # Done here:
+        return send_model
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    import asyncio
+
+    async def main():
+
+        sender = AsyncNimbusWhatsapp(api_key = "2c6175138e964ee8a6adc2af8faab726")
+
+        api_result = await sender.send_whatsapp(
+            recipient_number = "919870391155",
+            message = "Test message :)"
+        )
+        print("WHATSAPP API RESULT:", api_result)
+        print("WHATSAPP API RESULT:", json.to_string(api_result.model_dump(), default = str))
+
+
+    asyncio.run(main())