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

(20241204) Savvy Bulk SMS class made.

Browse Source
This commit is contained in:
khushal
2024-12-04 18:27:22 +05:30
parent a6f4986dc3
commit 7d5acfbd0d
2 changed files with 63 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/sms/nimbus/async_nimbus.py
+1 -1
View File
@@ -109,7 +109,7 @@ class AsyncNimbusSMS:
self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self.__printer.disable() if not debug: self.__printer.disable()
# Create an HTTP client to work with: # Accept/create an HTTP client to work with:
if http_client: self.__http_client = http_client if http_client: self.__http_client = http_client
else: self.__http_client = httpx.AsyncClient( else: self.__http_client = httpx.AsyncClient(
limits = httpx.Limits( limits = httpx.Limits(
utils_v2/sms/savvy_bulk_sms/async_savvy_bulk_sms.py
+62 -109
View File
@@ -6,16 +6,15 @@
DATE: DATE:
Monday, 9th Sept., 2024 Wednesday, 4th Dec., 2024
OBJECTIVE: OBJECTIVE:
To be able to send SMSs from Nimbus's API and manage the templates and other things from one place. To be able to send SMSs from Savvy Bulk SMS's API.
REFERENCES: REFERENCES:
01. https://github.com/innovativevijay/SmsHitApiSample N/A
02. https://nimbusit.net/appforms/apimanual.php
DOWNLOADS: DOWNLOADS:
@@ -23,7 +22,7 @@
WEB-PORTAL: WEB-PORTAL:
01. http://nimbusit.net/ N/A
""" """
@@ -74,30 +73,23 @@ from icecream import IceCreamDebugger
# ***************************************************************************************************************** # *****************************************************************************************************************
class AsyncNimbusSMS: class AsyncSavvyBulkSMS:
MESSAGE_TYPE_REGULAR = 0
MESSAGE_TYPE_UNICODE = 1
MESSAGE_TYPE_NOT_FLASH = 0
MESSAGE_TYPE_FLASH = 1
def __init__( def __init__(
self, self,
entity_id, api_key: str,
sender_id, partner_id: str,
user_id, short_code: str,
api_key, http_client: httpx.AsyncClient = None,
debug = True, debug = True,
debug_prefix = "Nimbus SMS | " debug_prefix = "Savvy SMS | "
): ):
""" """
Sets up an instance of the SMS sender through Nimbus IT. Sets up an instance of an SMS sender that sends messages through Savvy Bulk SMS's API.
:param entity_id: The entity id as registered with DLT. :param api_key: The API key received from Savvy Bulk SMS.
:param sender_id: The 6-char code like "HDFCBK", "NSESMS", "ZRODHA" that you see in your SMS inbox. :param partner_id: Your id with Savvy Bulk SMS.
:param user_id: The 6-digit id that Nimbus has assigned to you. :param short_code: Your short code with Savvy Bulk SMS.
: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: 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_prefix: The prefix text to show with the debug string.
""" """
@@ -106,122 +98,88 @@ class AsyncNimbusSMS:
self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self.__printer.disable() if not debug: self.__printer.disable()
# Create an HTTP client to work with: # Accept/create an HTTP client to work with:
limits = httpx.Limits( if http_client: self.__http_client = http_client
max_connections = 5, else: self.__http_client = httpx.AsyncClient(
max_keepalive_connections = 5, limits = httpx.Limits(
keepalive_expiry = 3600 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.
)
) )
self.__http_client = httpx.AsyncClient(limits = limits, timeout = 120)
# Capture the input config: # Capture the input config:
self.__entity_id = entity_id
self.__sender_id = sender_id
self.__user_id = user_id,
self.__api_key = api_key self.__api_key = api_key
self.__partner_id = partner_id
async def get_balance(self): self.__short_code = short_code
"""
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( async def send_sms(
self, self,
template_id,
recipient_number, recipient_number,
message, message
message_type = MESSAGE_TYPE_REGULAR, ) -> dict | None:
flash = MESSAGE_TYPE_NOT_FLASH
):
""" """
Sends one SMS through Nimbus IT's system. The text of the message must match the template that had been Send one SMS from Savvy Bulk SMS's API.
submitted. A mismatch may cause the message to fail at best, and raise troubles in the real-world with :param recipient_number: The mobile no. of the target recipient.
government bodies at worst. Be careful. :param message: The message to send to the target recipient.
:param template_id: The id of the SMS template as registered on Nimbus's portal. :return:
: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.
:param flash: to choose between a regular inbox SMS, or a flash SMS.
: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: # Construct the basic structure of the response of this method:
summary = { summary = {
"success": False, "success": False,
"info": None, "info": None,
"sender": self.__sender_id, "sender": {
"partnerId": self.__partner_id,
"shortCode": self.__short_code
},
"recipient": recipient_number, "recipient": recipient_number,
"message": message, "message": message,
"length": len(message), "length": len(message),
"template_id": template_id,
"raw": None, "raw": None,
"isFlash": True if flash else False
} }
try: try:
# Pre-process the recipient's number:
if not isinstance(recipient_number, (list, set, tuple)): recipient_number = [recipient_number]
# Call the API: # Call the API:
response = await self.__http_client.get( api_response = await self.__http_client.get(
url = r"http://nimbusit.net/api/pushsms", url = r"https://sms.savvybulksms.com/api/services/sendsms/",
params = { params = {
"user": self.__user_id, "apikey": self.__api_key,
"authkey": self.__api_key, "partnerID": self.__partner_id,
"sender": self.__sender_id, "message": message,
"mobile": ",".join([str(num) for num in recipient_number]), "shortcode": self.__short_code,
"text": message, "mobile": recipient_number,
"entityid": self.__entity_id,
"templateid": template_id,
"type": message_type,
"flash": flash
} }
) )
# For a successful API call: # For a successful API call:
if response.status_code == 200: if api_response.status_code in [200]:
response_json = response.json() api_json = api_response.json()
summary["success"] = True if response_json.get("STATUS", "ERROR").lower() in ["ok"] else False first_desc = api_json.get("responses", [{}])[0].get("response-description", "N/A").lower().strip()
summary["info"] = response_json.get("RESPONSE", {}).get("INFO") summary["success"] = True if first_desc == "success" else False
summary["raw"] = { summary["raw"] = {
"http_code": response.status_code, "http_code": api_response.status_code,
"response": response_json, "response": api_json,
} }
# For any other code that indicates some form of failure: # For any other code that indicates some form of failure:
else: summary["raw"] = { else: summary["raw"] = {
"http_code": response.status_code, "http_code": api_response.status_code,
"response": response.content.decode() "response": api_response.content.decode()
} }
# If something goes wrong along the way:
except Exception as exception: except Exception as exception:
self.__printer(exception) self.__printer(exception)
# Done here:
return summary return summary
@@ -238,22 +196,17 @@ if __name__ == "__main__":
async def main(): async def main():
sender = AsyncNimbusSMS( sender = AsyncSavvyBulkSMS(
entity_id = "<your_entity_id>", api_key = "<your_api_key>",
sender_id = "<your_sender_id>", partner_id = "<your_partner_id>",
user_id = "<your_nimbus_user_id>", short_code = "<your_short_code>"
api_key = "<your_nimbus_api_key>"
) )
response = await sender.send_sms( response = await sender.send_sms(
template_id = "<your_sms_template_id>", recipient_number = "<target_recipient>",
recipient_number = "<the_number_you_want_to_send_the_message_to>", message = "<message_for_recipient>"
message = "<your_sms_message>"
) )
print("SMS API RESPONSE:", response) print("SMS API RESPONSE:", response)
my_balance = await sender.get_balance()
print("REMAINING BALANCE:", my_balance)
asyncio.run(main()) asyncio.run(main())