khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Squashed 'utils_v2/' content from commit f03179d

Browse Source 
git-subtree-dir: utils_v2
git-subtree-split: f03179d339e69fdcdff2f0cc06e1f352024c55d3
This commit is contained in:
yatmesh
2025-06-12 13:46:57 +05:30
commit bfbc270b65
217 changed files with 149785 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
whatsapp/nimbus/controllers/__init__.py
View File
whatsapp/nimbus/controllers/async_nimbus_whatsapp.py
+236
View File
@@ -0,0 +1,236 @@
"""
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 = None,
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 or " ",
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("RESPONSE:", json.to_string(api_response.json()))
# 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 = "2c6175138e964ee8a6adc2af8faab727") # ... TCAOFF 1
# sender = AsyncNimbusWhatsapp(api_key = "28ee15ef6df14105a855e8840540a791") # ... TCAOFF 2
sender = AsyncNimbusWhatsapp(api_key = "af82dd816b6c42958d052548f40ba5f9") # ... Mr. Gaurav Gupta
# sender = AsyncNimbusWhatsapp(api_key = "2b6aa961ad734611b932a8bd54747403") # ... Mr. Sumeet Shirke
api_result = await sender.send_whatsapp(
recipient_number = "7972314099",
message = f"Hello, Yatmesh ({datetime.datetime.now()})",
# image_0_url = r"https://cdn.britannica.com/39/226539-050-D21D7721/Portrait-of-a-cat-with-whiskers-visible.jpg"
)
print("WHATSAPP API RESULT:", api_result)
print("WHATSAPP API RESULT:", json.to_string(api_result.model_dump(), default = str))
asyncio.run(main())