""" AUTHOR: Khushal P Soonderji DATE: Friday, 13th Dec., 2024 OBJECTIVE: To handle all SMS related behaviour from one place. REFERENCES: N/A DOWNLOADS: N/A """ import asyncio # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For Quart: from quart import current_app # My async utils: from utils_v2.string import json from utils_v2.date_time import date_time from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.database.async_mongo_v2 import AsyncMongo, AsyncMongoStorage # Base model: from controllers.base import BaseModel # Data models: from models.core.user import CoreUserInfoModel from models.core.auth_token import CoreAuthTokenModel from models.core.message import CoreMessageModel from models.api.sms.send import ( SMSSendRequestData, NimbusSMSIndiaMessage, SavvyBulkSMSKenyaMessage, SMSSendManyResults ) # SMS Clients: from utils_v2.sms.models.behaviour.nimbus.async_nimbus import AsyncNimbusSMS from utils_v2.sms.models.behaviour.savvy_bulk_sms.async_savvy_bulk_sms import AsyncSavvyBulkSMS from utils_v2.sms.models.data.sms_message import SentSMSMessageModel # To work with MongoDB: from bson import ObjectId from pymongo import InsertOne # To work with datatypes: from typing import Literal, List, Dict, Any # To make API calls: import httpx # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class SMSController: # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ pass # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ pass # ┏┓ ┓ # ┣┫┓┏╋┣┓ # ┛┗┗┻┗┛┗ @staticmethod async def set_token( db_conn: AsyncMySQL, mongo_conn: AsyncMongo, auth_token: CoreAuthTokenModel, session_token: str = None ) -> bool: # Start by assuming failure: success = False # Get a token id: token_id = await current_app.core_auth_token_controller.get_token_id( db_conn = db_conn, mongo_conn = mongo_conn, auth_token = auth_token, token_notes = {}, session_token = session_token ) # Immediately save the details against that token id: success = await current_app.core_auth_token_controller.set_token( db_conn = db_conn, mongo_conn = mongo_conn, token_id = token_id, auth_token = auth_token, token_notes = {}, session_token = session_token ) # Done here: return success @staticmethod async def get_token( mongo_conn: AsyncMongo, token_id: ObjectId | str = None, ) -> CoreAuthTokenModel | None: # Simply call the core model: return await current_app.core_auth_token_controller.get_token( mongo_conn = mongo_conn, token_id = token_id ) # ┏┓ ┓ # ┗┓┏┓┏┓┏┫ # ┗┛┗ ┛┗┗┻ @staticmethod async def __send_from_nimbus_sms_india( http_client: httpx.AsyncClient, token_id: ObjectId | str, auth_token: CoreAuthTokenModel, messages: List[NimbusSMSIndiaMessage], ) -> SMSSendManyResults: # Start with a blank variable: send_results = SMSSendManyResults() # Initialize the third-party client: client = AsyncNimbusSMS( entity_id = auth_token.auth["entityId"], sender_id = auth_token.auth["senderId"], user_id = auth_token.auth["userId"], api_key = auth_token.auth["apiKey"], http_client = http_client ) # Iterate over all the messages you need to send: for message in messages: # Send the SMS and return the response: client_response = await client.send_sms( recipient_number = message.recipientNo, message = message.text, template_id = message.templateId ) # Note down the results: send_results.totalCount += 1 if client_response.success: send_results.successCount += 1 else: send_results.failureCount += 1 send_results.smsMessages.append(CoreMessageModel( ts = client_response.ts, syncTs = date_time.get_current_utc_date_time(as_string = False), tokenId = ObjectId(token_id), serviceType = auth_token.serviceType, client = auth_token.client, clientMessageId = client_response.messageId, clientThreadId = message.recipientNo, isSent = True, isBroadcast = False, sentSuccessfully = client_response.success, sender = None, recipient = message.recipientNo, chat = None, message = client_response.model_dump(), snippet = message.text, aiSnippet = None, tags = ["sms", "nimbusSmsIndia"] )) # Done here: return send_results @staticmethod async def __send_from_savvy_bulk_sms_kenya( http_client: httpx.AsyncClient, token_id: ObjectId | str, auth_token: CoreAuthTokenModel, messages: List[SavvyBulkSMSKenyaMessage], ) -> SMSSendManyResults: # Start with a blank variable: send_results = SMSSendManyResults() # Initialize the third-party client: client = AsyncSavvyBulkSMS( partner_id = auth_token.auth["partnerId"], short_code = auth_token.auth["shortCode"], api_key = auth_token.auth["apiKey"], http_client = http_client ) # Iterate over all the messages you need to send: for message in messages: # Send the SMS and return the response: client_response = await client.send_sms( recipient_number = message.recipientNo, message = message.text ) # Note down the results: send_results.totalCount += 1 if client_response.success: send_results.successCount += 1 else: send_results.failureCount += 1 send_results.smsMessages.append(CoreMessageModel( ts = client_response.ts, syncTs = date_time.get_current_utc_date_time(as_string = False), tokenId = ObjectId(token_id), serviceType = auth_token.serviceType, client = auth_token.client, clientMessageId = client_response.messageId, clientThreadId = message.recipientNo, isSent = True, isBroadcast = False, sentSuccessfully = client_response.success, sender = None, recipient = message.recipientNo, chat = None, message = client_response.model_dump(), snippet = message.text, aiSnippet = None, tags = ["sms", "savvyBulkSmsKenya"] )) # Done here: return send_results async def send( self, mongo_conn: AsyncMongo, http_client: httpx.AsyncClient, token_id: ObjectId | str, messages: List[NimbusSMSIndiaMessage | SavvyBulkSMSKenyaMessage] ) -> SMSSendManyResults: # Start by assuming failure: send_results = SMSSendManyResults() # We first load the authorization tokens: auth_token = await self.get_token( mongo_conn = mongo_conn, token_id = token_id, ) # If we failed to load the authorization tokens: if not auth_token: send_results.message = f"no such token id '{token_id}'" return send_results # Now we route the message to the appropriate client: match auth_token.client: case "nimbusSmsIndia": send_results = await self.__send_from_nimbus_sms_india( http_client = http_client, token_id = token_id, auth_token = auth_token, messages = messages ) case "savvyBulkSmsKenya": send_results = await self.__send_from_savvy_bulk_sms_kenya( http_client = http_client, token_id = token_id, auth_token = auth_token, messages = messages ) case _: send_results.message = f"invalid client {auth_token.client}" # Save the results to MongoDB: tasks = [] for sms in send_results.smsMessages: message_json = sms.model_dump() message_json.pop("_id", None) tasks.append(current_app.core_message_controller.insert( mongo_conn = mongo_conn, message = message_json )) results = await asyncio.gather(*tasks) # Done here: send_results.message = f"{send_results.successCount}/{send_results.totalCount} message(s) sent" return send_results # ┓ • ┏┓ ┏┓ ┳┳┓ # ┃ ┓┏╋ ┣╋ ┃┓┏┓╋ ┃┃┃┏┓┏┏┏┓┏┓┏┓┏ # ┗┛┗┛┗ ┗┻ ┗┛┗ ┗ ┛ ┗┗ ┛┛┗┻┗┫┗ ┛ # ┛ # These are simply for retrieving sms messages. # You need to already have them saved to the database. # @staticmethod # async def list_messages( # mongo_conn: AsyncMongo, # token_ids: List[ObjectId | str], # limit: int = 100, # skip: int = 0, # additional_filter: dict = None # ) -> List[CoreMessageModel] | None: # # # regardless of what additional filter is provided from outside, # # we add a mail-selecting filter here: # if additional_filter is None: additional_filter = {} # additional_filter["serviceType"] = "sms" # # # Simply call the core model: # return await current_app.core_message_controller.get_message( # mongo_conn = mongo_conn, # token_ids = token_ids, # limit = limit, # skip = skip, # additional_filter = additional_filter # ) # # @staticmethod # async def get_one_mail( # mongo_conn: AsyncMongo, # token_id: ObjectId | str, # message_id: ObjectId | str # ) -> CoreMessageModel | None: # # # Simply call the core model: # return await current_app.core_message_controller.get_message( # mongo_conn = mongo_conn, # token_id = token_id, # message_id = message_id # ) # ┳┳ ┓ # ┃┃┏┓┏┫┏┓╋┏┓ # ┗┛┣┛┗┻┗┻┗┗ # ┛ @staticmethod async def update_tags( mongo_conn: AsyncMongo, token_id: ObjectId | str, message_id: ObjectId | str, unset_tags: List[str] = None, set_tags: List[str] = None ) -> bool: # Simply call the core model: return await current_app.core_message_controller.update_tags( mongo_conn = mongo_conn, token_id = token_id, message_id = message_id, unset_tags = unset_tags, set_tags = set_tags ) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass # from utils_v2.string import json # # file_options = [ # r"/home/developer/Downloads/recursive parts parse - 20241210.json", # r"/home/developer/Downloads/recursive parts parse (no attachment) - 20241210.json", # ] # # raw_mail_json = json.from_file(file_options[1]) # print("FROM FILE:", json.to_string(raw_mail_json["payload"])) # print("\n\n---------\n\n") # mail_controller = MailController() # print(json.to_string(mail_controller.drop_attachments(raw_mail_json["payload"])))