khushalps
22abf16a50
git-subtree-dir: utils_v2 git-subtree-split: ac9e5c331d9da4be516c6149cbdc9b5ebc19aa64
253 lines
9.6 KiB
Python
253 lines
9.6 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Monday, 9th Sept., 2024
|
|
|
|
OBJECTIVE:
|
|
|
|
To be able to send SMSs from Nimbus's API and manage the templates and other things from one place.
|
|
|
|
REFERENCES:
|
|
|
|
01. https://github.com/innovativevijay/SmsHitApiSample
|
|
02. https://nimbusit.net/appforms/apimanual.php
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
WEB-PORTAL:
|
|
|
|
01. http://nimbusit.net/
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# To make API Calls:
|
|
import httpx
|
|
|
|
# For debugging:
|
|
from icecream import IceCreamDebugger
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
class AsyncNimbusSMS:
|
|
|
|
MESSAGE_TYPE_REGULAR = 0
|
|
MESSAGE_TYPE_UNICODE = 1
|
|
|
|
def __init__(
|
|
self,
|
|
entity_id,
|
|
sender_id,
|
|
user_id,
|
|
api_key,
|
|
debug = True,
|
|
debug_prefix = "Nimbus SMS | "
|
|
):
|
|
|
|
"""
|
|
Sets up an instance of the SMS sender through Nimbus IT.
|
|
:param entity_id: The entity id as registered with DLT.
|
|
:param sender_id: The 6-char code like "HDFCBK", "NSESMS", "ZRODHA" that you see in your SMS inbox.
|
|
:param user_id: The 6-digit id that Nimbus has assigned to you.
|
|
:param api_key: The API key generated through Nimbus's portal.
|
|
: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.
|
|
"""
|
|
|
|
# Create the debugging tools:
|
|
self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
|
|
if not debug: self.__printer.disable()
|
|
|
|
# Create an HTTP client to work with:
|
|
limits = httpx.Limits(
|
|
max_connections = 5,
|
|
max_keepalive_connections = 5,
|
|
keepalive_expiry = 3600
|
|
)
|
|
self.__http_client = httpx.AsyncClient(limits = limits, timeout = 120)
|
|
|
|
# Capture the input config:
|
|
self.__entity_id = entity_id
|
|
self.__sender_id = sender_id
|
|
self.__user_id = user_id,
|
|
self.__api_key = api_key
|
|
|
|
async def get_balance(self):
|
|
|
|
"""
|
|
Checks the balance in the Nimbus wallet.
|
|
:return: The balance (float) if the request was successful, else None.
|
|
"""
|
|
|
|
balance = None
|
|
|
|
try:
|
|
|
|
# Call the API:
|
|
response = await self.__http_client.get(
|
|
url = r"http://nimbusit.net/api/balance",
|
|
params = {"user": self.__user_id, "authkey": self.__api_key}
|
|
)
|
|
|
|
# The response of a successful API call looks like "BALANCE:599". We need just the number:
|
|
if response.status_code in [200]: balance = float(response.content.decode().split(":")[-1].strip())
|
|
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
|
|
return balance
|
|
|
|
async def send_sms(
|
|
self,
|
|
template_id,
|
|
recipient_number,
|
|
message,
|
|
message_type = MESSAGE_TYPE_REGULAR
|
|
):
|
|
|
|
"""
|
|
Sends one SMS through Nimbus IT's system. The text of the message must match the template that had been
|
|
submitted. A mismatch may cause the message to fail at best, and raise troubles in the real-world with
|
|
government bodies at worst. Be careful.
|
|
:param template_id: The id of the SMS template as registered on Nimbus's portal.
|
|
:param recipient_number: The phone number of the recipient. You can send an array of numbers, too, BUT IT IS
|
|
STRONGLY RECOMMENDED TO NOT DO THAT TO AVOID BEING BLOCKED BY DLT.
|
|
:param message: The message to send to the recipient. Should match the template that is being sent.
|
|
:param message_type: Choose between 'AsyncNimbusSMS.MESSAGE_TYPE_REGULAR' (default) and
|
|
'AsyncNimbusSMS.MESSAGE_TYPE_UNICODE' based on the type of characters being sent. Both are class variables.
|
|
:return: The dict of all the details of the message that was sent including whether, or not, it was successfully
|
|
sent. Other details depend on the service provider (Nimbus IT in this case).
|
|
"""
|
|
|
|
# Construct the basic structure of the response of this method:
|
|
summary = {
|
|
"success": False,
|
|
"info": None,
|
|
"sender": self.__sender_id,
|
|
"recipient": recipient_number,
|
|
"message": message,
|
|
"length": len(message),
|
|
"template_id": template_id,
|
|
"raw": None
|
|
}
|
|
|
|
try:
|
|
|
|
# Pre-process the recipient's number:
|
|
if not isinstance(recipient_number, (list, set, tuple)): recipient_number = [recipient_number]
|
|
|
|
# Call the API:
|
|
response = await self.__http_client.get(
|
|
url = r"http://nimbusit.net/api/pushsms",
|
|
params = {
|
|
"user": self.__user_id,
|
|
"authkey": self.__api_key,
|
|
"sender": self.__sender_id,
|
|
"mobile": ",".join([str(num) for num in recipient_number]),
|
|
"text": message,
|
|
"entityid": self.__entity_id,
|
|
"templateid": template_id,
|
|
"type": message_type
|
|
}
|
|
)
|
|
|
|
# For a successful API call:
|
|
if response.status_code == 200:
|
|
response_json = response.json()
|
|
summary["success"] = True if response_json.get("STATUS", "ERROR").lower() in ["ok"] else False
|
|
summary["info"] = response_json.get("RESPONSE", {}).get("INFO")
|
|
summary["raw"] = {
|
|
"http_code": response.status_code,
|
|
"response": response_json,
|
|
}
|
|
|
|
# For any other code that indicates some form of failure:
|
|
else: summary["raw"] = {
|
|
"http_code": response.status_code,
|
|
"response": response.content.decode()
|
|
}
|
|
|
|
except Exception as exception:
|
|
self.__printer(exception)
|
|
|
|
return summary
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
import asyncio
|
|
|
|
async def main():
|
|
|
|
sender = AsyncNimbusSMS(
|
|
entity_id = "<your_entity_id>",
|
|
sender_id = "<your_sender_id>",
|
|
user_id = "<your_nimbus_user_id>",
|
|
api_key = "<your_nimbus_api_key>"
|
|
)
|
|
|
|
response = await sender.send_sms(
|
|
template_id = "<your_sms_template_id>",
|
|
recipient_number = "<the_number_you_want_to_send_the_message_to>",
|
|
message = "<your_sms_message>"
|
|
)
|
|
print("SMS API RESPONSE:", response)
|
|
|
|
my_balance = await sender.get_balance()
|
|
print("REMAINING BALANCE:", my_balance)
|
|
|
|
|
|
asyncio.run(main())
|