Resetting utils subtree.
This commit is contained in:
yatmesh
@@ -1,236 +0,0 @@
|
||||
"""
|
||||
|
||||
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())
|
||||
@@ -1,287 +0,0 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Bhushan C Thakkar
|
||||
|
||||
DATE:
|
||||
|
||||
Monday, 20th Jan., 2025.
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To give a structure to how Nimbus's API will respond when we try to send messages.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
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
|
||||
|
||||
# For making data behaviour_models:
|
||||
from pydantic import BaseModel, Field, field_validator, PastDatetime, AwareDatetime
|
||||
from typing import Optional, Literal, Union, Any, List
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import regex
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
# To work with MongoDB:
|
||||
from bson.objectid import ObjectId
|
||||
|
||||
# To work with date and time:
|
||||
import datetime
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class NimbusWhatsappSendMessageResponse(BaseModel):
|
||||
|
||||
ts: AwareDatetime = Field(
|
||||
description = "The time (UTC) at which this message was sent by the sender.",
|
||||
frozen = True,
|
||||
default_factory = lambda: date_time.get_current_utc_date_time(as_string = False)
|
||||
)
|
||||
|
||||
success: bool = Field(
|
||||
description = "Whether, or not, the request was successful.",
|
||||
frozen = True,
|
||||
default = False
|
||||
)
|
||||
|
||||
recipientNo: List[str] = Field(
|
||||
description = "One or more target nos. to send the WhatsApp message to.",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
message: str | None = Field(
|
||||
description = "The actual message to send to the recipient(s).",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
pdfUrl: str | None = Field(
|
||||
description = "A PDF file to send with your message.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
image0Url: str | None = Field(
|
||||
description = "An image file to send with your message.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
image1Url: str | None = Field(
|
||||
description = "An image file to send with your message.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
scheduleTs: AwareDatetime | None = Field(
|
||||
description = "The time (UTC) at which the message needs to be sent. Null for immediate delivery.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
httpCode: str | int | None = Field(
|
||||
description = "The HTTP code of the API call.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
status: str | None = Field(
|
||||
description = "To indicate success or failure.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
statusCode: str | int | None = Field(
|
||||
description = "The response code as defined by Nimbus.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
apiMessage: str = Field(
|
||||
description = "A message about any error, or a general acknowledgement message.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
requestId: str | int | None = Field(
|
||||
description = "How Nimbus identifies your message.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
messageCount: int | None = Field(
|
||||
description = "The no. of messages sent.",
|
||||
frozen = True,
|
||||
default = 0
|
||||
)
|
||||
|
||||
messageCost: float | int | None = Field(
|
||||
description = "The cost (in nimbus' 'credits') of messages sent.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
balance: int | float | None = Field(
|
||||
description = "The balance remaining in the account.",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "allow"
|
||||
|
||||
# ┓┏ ┓• ┓ •
|
||||
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
|
||||
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
|
||||
|
||||
@field_validator("ts", "scheduleTs", mode = "before")
|
||||
def parse_date_time(cls, value):
|
||||
return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC)
|
||||
|
||||
@field_validator("recipientNo", mode = "before")
|
||||
def validate_recipients(cls, value):
|
||||
if isinstance(value, str): value = [value]
|
||||
return value
|
||||
|
||||
# ┏┓ ┏┓
|
||||
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
|
||||
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
|
||||
|
||||
def api_input_json(self) -> dict:
|
||||
|
||||
# Start with the bare minimum:
|
||||
api_input = {
|
||||
"mobile": ",".join(self.recipientNo),
|
||||
"msg": self.message or " "
|
||||
}
|
||||
|
||||
# Add the optional media elements:
|
||||
if self.pdfUrl: api_input["pdf"] = self.pdfUrl
|
||||
if self.image0Url: api_input["img1"] = self.image0Url
|
||||
if self.image1Url: api_input["img2"] = self.image1Url
|
||||
|
||||
# Add the scheduled delivery time:
|
||||
if self.scheduleTs:
|
||||
api_input["scheduleon"] = date_time.to_timezone(
|
||||
datetime_object = self.scheduleTs,
|
||||
timezone = date_time.TIMEZONE_IST
|
||||
).strftime("%Y%m%d%H%M")
|
||||
|
||||
# Done here:
|
||||
return api_input
|
||||
|
||||
def from_api_response(
|
||||
self,
|
||||
http_code: int,
|
||||
response_json: dict = None
|
||||
) -> "NimbusWhatsappSendMessageResponse":
|
||||
|
||||
# Ensure that we are working with a dictionary:
|
||||
response_json = response_json or {}
|
||||
|
||||
# Create and return the model:
|
||||
return NimbusWhatsappSendMessageResponse(
|
||||
ts = self.ts,
|
||||
recipientNo = self.recipientNo,
|
||||
message = self.message,
|
||||
pdfUrl = self.pdfUrl,
|
||||
image0Url = self.image0Url,
|
||||
image1Url = self.image1Url,
|
||||
scheduleTs = self.scheduleTs,
|
||||
success = True if response_json.get("statuscode") == 200 or response_json.get("status").strip().lower() == "success" else False,
|
||||
httpCode = http_code,
|
||||
status = response_json.get("status"),
|
||||
statusCode = response_json.get("statuscode"),
|
||||
apiMessage = response_json.get("msg"),
|
||||
requestId = response_json.get("requestId"),
|
||||
messageCount = response_json.get("msgcount"),
|
||||
messageCost = response_json.get("msgcost"),
|
||||
balance = response_json.get("balance")
|
||||
)
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
sample_response_json = {
|
||||
"status": "success",
|
||||
"msg": "submitted successfully",
|
||||
"statuscode": 200,
|
||||
"requestid": "ES5OmOpIPaVSmT",
|
||||
"msgcount": "1",
|
||||
"msgcost": 1,
|
||||
"balance": "0.1"
|
||||
}
|
||||
|
||||
send_result = NimbusWhatsappSendMessageResponse(
|
||||
recipientNo = ["9870391155"],
|
||||
message = "Hello, Bhopli!"
|
||||
)
|
||||
|
||||
send_result = send_result.from_api_response(
|
||||
http_code = 200,
|
||||
response_json = sample_response_json
|
||||
)
|
||||
|
||||
print(send_result)
|
||||
Reference in New Issue
Block a user