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

(20241209) SMS auth and sending ready.

Browse Source
This commit is contained in:
khushal
2024-12-09 19:39:15 +05:30
parent d10b156c0b
commit 845827a6bc
18 changed files with 367 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/behaviour/ai/llm/open_ai.py
+4 -2
View File
@@ -45,6 +45,7 @@ from models.behaviour.base import BaseModel
# Data Models:
from models.data.api.ai.llm import LLMInput, LLMOutput, LLMUsageTokens
from models.data.core.user_info import CoreUserInfoModel
# To work with LLMs:
from langchain_openai import ChatOpenAI
@@ -140,7 +141,7 @@ class LLMOpenAI(BaseModel):
async def invoke(
self,
mongo_conn: AsyncMongo,
user_info: dict,
user_info: CoreUserInfoModel,
llm_input: LLMInput
) -> LLMOutput:
@@ -167,12 +168,13 @@ class LLMOpenAI(BaseModel):
)
# Store this into MongoDB:
mongo_document = {"user": user_info}
mongo_document = {"user": user_info.model_dump()}
for k, v in llm_response.model_dump().items(): mongo_document[k] = v
inserted_id = await mongo_conn.insert_one(
collection = self.AI_USAGE_COLLECTION,
document = mongo_document
)
if inserted_id: llm_response.invocationId = str(inserted_id)
# Done here:
return llm_response
models/behaviour/mail/oauth_v3.py
+1 -1
View File
@@ -125,7 +125,7 @@ class MailOAuthModel(BaseModel):
mongo_json = await mongo_conn.find_one_and_update(
collection = MailOAuthModel.AUTH_COLLECTION,
filter = mongo_conn.dict_to_dot_notation({
"serviceType": "email",
"serviceType": auth_token.serviceType,
"user": {
"entityId": auth_token.user.entityId,
"billingAccountId": auth_token.user.billingAccountId
models/behaviour/mail/sync_v3.py
+90 -69
View File
@@ -6,11 +6,12 @@
DATE:
tuesday, 3rd Dec., 2024
ORIGINAL: Tuesday, 3rd Dec., 2024
UPGRADED: Monday, 9th Dec., 2024
OBJECTIVE:
From here we sync all mails between the mail client's server and TheCAOffice's database.
From here we sync all mails between the mail client's server and our internal database.
REFERENCES:
@@ -31,6 +32,10 @@
# To make sibling directories accessible for imports:
import sys
from models.data.core.auth_token import CoreAuthTokenModel
from models.data.core.message import CoreMessageModel
sys.path.append(".")
sys.path.append("..")
@@ -52,6 +57,7 @@ from models.behaviour.base import BaseModel
# Data models:
from models.data.api.mail.sync import MailSyncOneResult, MailSyncManyResults
from models.data.core.user_info import CoreUserInfoModel
# To work with MongoDB:
from bson import ObjectId
@@ -193,6 +199,10 @@ class MailSyncModel(BaseModel):
attachment_copy["url"] = api_data["url"]
break
# If the upload failed:
await asyncio.sleep(retry_delay)
retry_delay = retry_delay * backoff_multiplier
# Done here:
return attachment_copy
@@ -246,10 +256,12 @@ class MailSyncModel(BaseModel):
async def __sync_one_gmail(
self,
session_token: str,
user_info: dict,
user_info: CoreUserInfoModel,
mongo_conn: AsyncMongo,
token_id: ObjectId,
auth_token: CoreAuthTokenModel,
mail_client: AsyncGMailClient,
tokens: GoogleAuthTokens,
google_tokens: GoogleAuthTokens,
message_id: str,
llm: LLMOpenAI = None,
force_sync: bool = False
@@ -257,11 +269,9 @@ class MailSyncModel(BaseModel):
"""
Sync on mail from GMail.
:param session_token: The session token of the uer who is trying to upload this file.
:param user_info: The information of the user (derived from his session token).
:param mongo_conn: The instance of the connection to the database to use.
:param mail_client: The instance of the mail client to use to perform the action.
:param tokens: The tokens to use to fetch the mails.
:param google_tokens: The tokens to use to fetch the mails.
:param message_id: The id that Google uses to identify this mail. This will be received in the 'list_messages'
method.
:param llm: The instance of the LLM to use to summarize the mail's content.
@@ -278,19 +288,17 @@ class MailSyncModel(BaseModel):
if not force_sync:
mail_record = await mongo_conn.find_one(
collection = self.MAIL_COLLECTION,
filter = mongo_conn.dict_to_dot_notation({
"payload": {
"messageId": message_id
},
"user_info": {
"entityId": user_info["entityId"],
"billingAccountId": user_info["billingAccountId"]
}
}),
filter = {
"tokenId": ObjectId(token_id),
"serviceType": auth_token.serviceType,
"client": auth_token.client,
"clientMessageId": message_id
},
projection = {
"_id": False,
"readTs": "payload.readTs"
}
"readTs": True
},
raise_exception = True
)
if mail_record:
sync_result.success = True
@@ -299,7 +307,7 @@ class MailSyncModel(BaseModel):
# Now that we know that we have to fetch the mail from GMail:
client_response = await mail_client.get_message(
tokens = tokens,
tokens = google_tokens,
message_id = message_id,
return_raw = False
)
@@ -314,18 +322,18 @@ class MailSyncModel(BaseModel):
session_token = session_token,
attachments = client_response.data["attachments"],
attachment_tags = [
"email",
"gmail",
auth_token.serviceType,
auth_token.client,
client_response.data["from"][0]["name"],
client_response.data["from"][0]["email"],
tokens.email,
google_tokens.email,
],
attachment_metadata = {
"project": "tcaoff",
"serviceType": "email",
"client": "gmail",
"serviceType": auth_token.serviceType,
"client": auth_token.client,
"from": client_response.data["from"][0]["email"],
"to": tokens.email
"to": google_tokens.email
},
retry_count = 3
)
@@ -333,7 +341,7 @@ class MailSyncModel(BaseModel):
# Give a quick indicator of whether this mail is an inbox mail or sent mail:
all_recipients = []
for field in ["to", "cc", "bcc"]: all_recipients += [item["email"] for item in client_response.data[field]]
if tokens.email in all_recipients: client_response.data["isInbox"] = True
if google_tokens.email in all_recipients: client_response.data["isInbox"] = True
else: client_response.data["isInbox"] = False
# If an LLM is given,
@@ -342,14 +350,17 @@ class MailSyncModel(BaseModel):
if llm:
# Invoke the LLM:
llm_response = response = await llm.invoke(
llm_response = await llm.invoke(
mongo_conn = mongo_conn,
user_info = user_info,
llm_input = LLMInput(
messages = self.PROMPT_TEMPLATE + [
{
"role": "human",
"content": f"Please summarize this mail: \"\"\"{client_response.data['unformattedText']}\"\"\""
"content": (
"Please summarize this mail: "
f"\"\"\"{client_response.data['unformattedText']}\"\"\""
)
}
]
)
@@ -365,19 +376,30 @@ class MailSyncModel(BaseModel):
# Add the LLM's response to the main data:
client_response.data["aiSnippet"] = llm_json
# Fit the mail message into the model:
sync_result.mailMessage = CoreMessageModel(
ts = client_response.data["ts"],
readTs = date_time.get_current_utc_date_time(as_string = False),
tokenId = token_id,
serviceType = auth_token.serviceType,
client = auth_token.client,
clientMessageId = message_id,
clientThreadId = client_response.data["threadId"],
payload = client_response.data
)
# Done here:
sync_result.success = True
sync_result.mailMessage = client_response.data
return sync_result
async def __sync_many_gmail(
self,
session_token: str,
user_info: dict,
user_info: CoreUserInfoModel,
mongo_conn: AsyncMongo,
token_id: ObjectId,
auth_token: CoreAuthTokenModel,
mail_client: AsyncGMailClient,
tokens: GoogleAuthTokens,
llm: LLMOpenAI = None,
force_sync: bool = False,
start_date: datetime.datetime = None,
@@ -387,13 +409,10 @@ class MailSyncModel(BaseModel):
"""
Sync many mails from GMail in one shot.
:param session_token: The session token of the uer who is trying to upload this file.
:param user_info: The information of the user (derived from his session token).
:param mongo_conn: The instance of the connection to the database to use.
:param token_id: The id of the document in the database that holds the tokens to access the account.
Needed only for refreshing the tokens and saving them.
:param mail_client: The instance of the mail client to use to perform the action.
:param tokens: The tokens to use to fetch the mails.
:param llm: The instance of the LLM to use to summarize the mail's content.
:param force_sync: Whether you would like to forcefully re-sync the mail even if it is already present in the
database.
@@ -406,20 +425,24 @@ class MailSyncModel(BaseModel):
# Start by assuming failure:
sync_results = MailSyncManyResults()
# Extract the client's tokens from the full token payload given by the database:
google_tokens = GoogleAuthTokens(**auth_token.token)
# Refresh the tokens (if needed):
tokens_refreshed = await tokens.arefresh(
tokens_refreshed = await google_tokens.arefresh(
http_client = current_app.http_client,
client_id = mail_client.client_id,
client_secret = mail_client.client_secret
)
if tokens_refreshed: await current_app.mail_oauth_model.set_token(
db_conn = current_app.sql_writer,
mongo_conn = mongo_conn,
token_id = token_id,
client_user_id = tokens.client_user_id,
token = tokens,
session_token = session_token
)
if tokens_refreshed:
auth_token.token = google_tokens.model_dump()
auth_token.lastRefreshTs = date_time.get_current_utc_date_time(as_string = True)
await current_app.mail_oauth_model.set_token(
db_conn = current_app.sql_writer,
mongo_conn = mongo_conn,
token_id = token_id,
auth_token = auth_token
)
# Let's build the query:
sub_queries = []
@@ -429,7 +452,7 @@ class MailSyncModel(BaseModel):
# Let's enlist all the mails that fall in the date range:
client_response = await mail_client.list_messages(
tokens = tokens,
tokens = google_tokens,
max_count = max_count,
query = query_string
)
@@ -444,8 +467,10 @@ class MailSyncModel(BaseModel):
session_token = session_token,
user_info = user_info,
mongo_conn = mongo_conn,
token_id = token_id,
auth_token = auth_token,
mail_client = mail_client,
tokens = tokens,
google_tokens = google_tokens,
message_id = v["id"],
llm = llm,
force_sync = force_sync
@@ -462,21 +487,19 @@ class MailSyncModel(BaseModel):
else: sync_results.failureCount += 1
if result.mailMessage: mongo_operations.append(ReplaceOne(
filter = {
"serviceType": "email",
"$or": [
{
"client": "gmail",
"payload.messageId": result.mailMessage["messageId"]
}
]
},
replacement = {
"version": "1.0.0",
"tokenId": ObjectId(token_id),
"serviceType": "email",
"client": "gmail",
"payload": result.mailMessage
"tokenId": token_id,
"serviceType": auth_token.serviceType,
"client": auth_token.client,
"clientMessageId": result.mailMessage.clientMessageId
# "serviceType": auth_token.serviceType,
# "$or": [
# {
# "client": auth_token.client,
# "messageId": result.mailMessage.clientMessageId
# }
# ]
},
replacement = result.mailMessage.model_dump(),
upsert = True
))
@@ -490,9 +513,9 @@ class MailSyncModel(BaseModel):
# Apply the labels to the read messages:
try:
client_response = await mail_client.modify_messages(
tokens = tokens,
tokens = google_tokens,
message_ids = [v["id"] for v in messages_list.values()],
add_label_ids = [tokens.labels.get("TCAOFF", {}).get("id")]
add_label_ids = [google_tokens.labels.get("TCAOFF", {}).get("id")]
)
except Exception as exception:
self._printer(exception)
@@ -508,7 +531,7 @@ class MailSyncModel(BaseModel):
async def sync(
self,
session_token: str,
user_info: dict,
user_info: CoreUserInfoModel,
mongo_conn: AsyncMongo,
token_id: ObjectId,
llm: LLMOpenAI = None,
@@ -521,8 +544,6 @@ class MailSyncModel(BaseModel):
"""
Sync many mails at once from many types of clients. Use this as a common entry point after which you internally
route the request to the appropriate clients.
:param session_token: The session token of the uer who is trying to upload this file.
:param user_info: The information of the user (derived from his session token).
:param mongo_conn: The instance of the connection to the database to use.
:param token_id: The id of the document in the database that holds the tokens to access the account.
Needed only for refreshing the tokens and saving them.
@@ -543,13 +564,13 @@ class MailSyncModel(BaseModel):
# ┻ ┗ ┗┗┛┗ ┻ ┗┛┛┗┗ ┛┗┛
# We first load the authorization tokens:
auth_json = await current_app.mail_oauth_model.get_token(
auth_token = await current_app.mail_oauth_model.get_token(
mongo_conn = mongo_conn,
token_id = token_id,
)
# If we failed to load the authorization tokens:
if not auth_json:
if not auth_token:
sync_results.message = f"no such token id '{token_id}'"
return sync_results
@@ -557,14 +578,14 @@ class MailSyncModel(BaseModel):
# ┣ ┏┓┏┓ ┃┓┃┃┃┏┓┓┃
# ┻ ┗┛┛ ┗┛┛ ┗┗┻┗┗
if auth_json["client"] == "gmail":
if auth_token.client == "gmail":
return await self.__sync_many_gmail(
session_token = session_token,
user_info = user_info,
mongo_conn = mongo_conn,
token_id = token_id,
auth_token = auth_token,
mail_client = current_app.gmail_client,
tokens = GoogleAuthTokens(**auth_json["token"]),
llm = llm,
force_sync = force_sync,
start_date = start_date,
@@ -577,7 +598,7 @@ class MailSyncModel(BaseModel):
# ┻┛┗┗┛┗┻┗┗┗┻ ┗┛┗┗┗ ┛┗┗
# If we haven't been able to sync mail due to not entering any 'if' condition:
sync_results.message = f"no such mail client '{auth_json['client']}'"
sync_results.message = f"no such mail client '{auth_token.client}'"
return sync_results
models/behaviour/sms/auth_v2.py
+1 -1
View File
@@ -123,7 +123,7 @@ class SMSAuthModel(BaseModel):
mongo_json = await mongo_conn.find_one_and_update(
collection = self.AUTH_COLLECTION,
filter = mongo_conn.dict_to_dot_notation({
"serviceType": "email",
"serviceType": auth_token.serviceType,
"user": {
"entityId": auth_token.user.entityId,
"billingAccountId": auth_token.user.billingAccountId
models/behaviour/sms/send.py
+89 -102
View File
@@ -6,12 +6,11 @@
DATE:
ORIGINAL: Thursday, 5th Dec., 2024
UPGRADED: Monday, 9th Dec., 2024
Monday, 9th Dec., 2024
OBJECTIVE:
To work with auth details of SMS clients like Nimbus SMS (India) and Savvy Bulk SMS (Kenya).
To send SMS from clients like Nimbus SMS (India) and Savvy Bulk SMS (Kenya).
REFERENCES:
@@ -41,11 +40,23 @@ 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
# SMS-related utils:
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
# Base model:
from models.behaviour.base import BaseModel
# Data models:
from models.data.core.auth_token import CoreAuthTokenModel
from models.data.core.message import CoreMessageModel
from models.data.api.sms.send import (
SMSSendRequestHeaders,
SMSSendRequestData,
NimbusSMSIndiaMessage,
SavvyBulkSMSKenyaMessage
)
from utils_v2.sms.models.data.sms_message import SentSMSMessageModel
# To work with MongoDB:
from bson import ObjectId
@@ -94,125 +105,101 @@ import copy
# *****************************************************************************************************************
class SMSAuthModel(BaseModel):
class SMSSendModel(BaseModel):
AUTH_COLLECTION = "_authTokens"
MESSAGES_COLLECTION = "_messages"
async def set(
async def send_sms(
self,
db_conn: AsyncMySQL,
mongo_conn: AsyncMongo,
token_id: ObjectId | str,
auth_token: CoreAuthTokenModel,
inbound_data: SMSSendRequestData,
session_token: str = None
) -> ObjectId | None:
) -> SentSMSMessageModel:
"""
To store auth/tokens for a particular service to the database.
:param db_conn: The database connection (MariaDB) to use to perform the action.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param auth_token: An instance of the core auth-token model that holds data in the database.
:param session_token: The session token of the user who requested this service.
:return: An ObjectId to later store the granted tokens.
"""
# Note down the timestamp at which this event occurred:
request_ts = date_time.get_current_utc_date_time(as_string = False)
# Basic prep:
event_ts = date_time.get_current_utc_date_time(as_string = False)
client_response = None
message_id = None
sms_sent = None
# Get the identifier from the database:
# BE CAREFUL WITH THE KEYS HERE, THEY SHOULD MATCH THE FIELDS OF THE CORE AUTH-TOKEN MODEL:
mongo_json = await mongo_conn.find_one_and_update(
collection = self.AUTH_COLLECTION,
filter = mongo_conn.dict_to_dot_notation({
"serviceType": auth_token.serviceType,
"user": {
"entityId": auth_token.user.entityId,
"billingAccountId": auth_token.user.billingAccountId
},
"clientUserId": auth_token.clientUserId
}),
update = {
"$set": {
"lastRequestTs": auth_token.lastRequestTs,
"status": auth_token.status,
"syncFreq": auth_token.syncFreq
},
"$setOnInsert": {
"version": auth_token.version,
"serviceType": auth_token.serviceType,
"client": auth_token.client,
"authType": auth_token.authType,
"user": auth_token.user.model_dump(),
"clientUserId": auth_token.clientUserId,
"auth": auth_token.auth,
"token": auth_token.token,
"firstRefreshTs": auth_token.firstRefreshTs,
"lastRefreshTs": auth_token.lastRefreshTs,
"firstRequestTs": auth_token.firstRequestTs or request_ts
}
},
projection = {
"_id": True
},
upsert = True,
return_updated = True
)
# ┏┓ ┳┓• ┓ ┏┓┳┳┓┏┓ ┳ ┓•
# ┣ ┏┓┏┓ ┃┃┓┏┳┓┣┓┓┏┏ ┗┓┃┃┃┗┓ ┃┏┓┏┫┓┏┓
# ┻ ┗┛┛ ┛┗┗┛┗┗┗┛┗┻┛ ┗┛┛ ┗┗┛ ┻┛┗┗┻┗┗┻
# Tell MariaDB that an authorization request was initiated:
db_json = {}
if mongo_json is not None:
token_notes = auth_token.clientUserId
db_json = await self.call_procedure(
db_conn = db_conn,
proc_name = "entity_integration_save",
proc_args = (
auth_token.user.entityId, # ..................................... 'p_entity_id'
auth_token.client, # ............................................ 'p_provider'
auth_token.status, # ............................................ 'p_current_status'
"Auth Details Accepted", # ...................................... 'p_last_action'
None, # ......................................................... 'p_display_name'
None, # ......................................................... 'p_display_picture'
str(mongo_json["_id"]), # ....................................... 'p_token_id'
json.to_string(python_data = token_notes, no_space = True), # ... 'p_notes'
auth_token.user.userId # ........................................ 'p_created_by'
),
session_token = session_token
if isinstance(inbound_data.message, NimbusSMSIndiaMessage):
# Prepare the client:
sms_client = AsyncNimbusSMS(
entity_id = auth_token.auth.get("entityId"),
sender_id = auth_token.auth.get("senderId"),
user_id = auth_token.auth.get("userId"),
api_key = auth_token.auth.get("apiKey"),
http_client = self._http_client
)
# Send the SMS:
client_response = await sms_client.send_sms(
recipient_number = inbound_data.message.recipientNo,
message = inbound_data.message.text,
template_id = inbound_data.message.templateId
)
# ┏┓ ┏┓ ┳┓ ┓┓ ┏┓┳┳┓┏┓ ┓┏┓
# ┣ ┏┓┏┓ ┗┓┏┓┓┏┓┏┓┏ ┣┫┓┏┃┃┏ ┗┓┃┃┃┗┓ ┃┫ ┏┓┏┓┓┏┏┓
# ┻ ┗┛┛ ┗┛┗┻┗┛┗┛┗┫ ┻┛┗┻┗┛┗ ┗┛┛ ┗┗┛ ┛┗┛┗ ┛┗┗┫┗┻
# ┛ ┛
elif isinstance(inbound_data.message, SavvyBulkSMSKenyaMessage):
# Prepare the client:
sms_client = AsyncSavvyBulkSMS(
api_key = auth_token.auth.get("apiKey"),
partner_id = auth_token.auth.get("partnerId"),
short_code = auth_token.auth.get("shortCode"),
http_client = self._http_client
)
# Send the SMS:
client_response = await sms_client.send_sms(
recipient_number = inbound_data.message.recipientNo,
message = inbound_data.message.text
)
# ┏┓ ┏┳┓┓ ┳┳┓
# ┗┓┏┓┓┏┏┓ ┃ ┣┓┏┓ ┃┃┃┏┓┏┏┏┓┏┓┏┓
# ┗┛┗┻┗┛┗ ┻ ┛┗┗ ┛ ┗┗ ┛┛┗┻┗┫┗
# ┛
# Save the message:
if client_response:
message_id = await mongo_conn.insert_one(
collection = self.MESSAGES_COLLECTION,
document = CoreMessageModel(
ts = event_ts,
readTs = event_ts,
tokenId = ObjectId(token_id),
serviceType = auth_token.serviceType,
client = auth_token.client,
clientMessageId = client_response.messageId,
clientThreadId = None,
isInward = False,
sentSuccessfully = client_response.success,
payload = client_response.model_dump()
).model_dump()
)
# Done here:
return mongo_json["_id"] if mongo_json and db_json.get("status") == 1 else None
async def get(
self,
mongo_conn: AsyncMongo,
token_id: ObjectId | str = None,
**kwargs
) -> dict | None:
"""
To retrieve stored auth/tokens from the database.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param token_id: The identifier granted providing auth details for the first time in 'set_token'.
:param kwargs: Any set of key-value pairs to build custom search criteria. This could be things like the user
info, the client, the type of authentication used, or even the kind of service.
:return: The retrieved record that has the token, and information about the service and client if found, else
None when there is no matching record.
"""
# Build the filter:
filter_json = {k: v for k, v in kwargs.items()}
if token_id: filter_json["_id"] = ObjectId(token_id)
# If there is no search criteria, we exit with failure:
if not filter_json: return None
# If there is some filtering possible, we fetch the token:
token = await mongo_conn.find_one(
collection = self.AUTH_COLLECTION,
filter = filter_json,
)
# Done here:
return CoreAuthTokenModel(**token) if token else None
return client_response
# *****************************************************************************************************************