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

(20250117) Major revamping in the mail module. Everything revamped. Sending is a pending task.

Browse Source
This commit is contained in:
khushal
2025-01-17 15:36:59 +05:30
parent 32c243f3d3
commit be9bdf797a
12 changed files with 675 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controllers_v2/core/message.py
+1 -1
View File
@@ -403,7 +403,7 @@ class CoreMessageController(CoreAuthTokenController):
# We don't support updating messages themselves,
# but we will allow updating fields like tags, marking as read or unread, etc.
async def update_tags(
async def update_message_tags(
self,
mongo_data_conn: AsyncMongo,
message_id: ObjectId | str,
controllers_v2/message/mail/all_mail.py
+132 -15
View File
@@ -36,34 +36,39 @@ sys.path.append(".")
sys.path.append("..")
# My async utils:
from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
# Controllers:
from controllers_v2.message.mail.base import MailController
from controllers.core.ai.llm import CoreLLMController
# Models:
from models.core.user import CoreUserInfoModel
from models.core.auth_token import CoreAuthTokenModel
from models.api.sms.send import (
NimbusSMSIndiaMessage,
SavvyBulkSMSKenyaMessage,
SMSSendOneResult,
SMSSendManyResults
from models.api.message.mail.oauth import (
OAuthMailAuthorizationRequestHeaders,
OAuthMailAuthorizationRequestData
)
from models.message.mail.oauth import OAuthMailGetAuthorizationURLResponse, OAuthMailHandleCallbackResponse
from models.core.message import CoreMessageModel
from models.core.ai.llm import LLMInput, LLMOutput, LLMInputMessage
from models.message.mail.sync import MailSyncOneResult, MailSyncManyResults
from models.message.mail.send import MailSendOneResult
# SMS clients:
from utils_v2.sms.india.nimbus.controllers.async_nimbus import AsyncNimbusSMS
from utils_v2.sms.kenya.savvy_bulk_sms.controllers.async_savvy_bulk_sms import AsyncSavvyBulkSMS
# Mail Client(s):
from utils_v2.goog.controllers.gmail.gmail_client import AsyncGMailClient
# To work with datatypes:
from typing import List, Any
# To work with date and time:
import datetime
# To make HTTP requests:
import httpx
# To make abstract classes:
from abc import ABC, abstractmethod
# *****************************************************************************************************************
# ***** ****
@@ -115,7 +120,7 @@ class AllMailController(MailController):
alert_url: str = None,
base_filter: dict = None,
debug: bool = True,
debug_prefix: str = "Mail (C) | ",
debug_prefix: str = "All Mail (C) | ",
debug_only_errors: bool = True
):
@@ -138,7 +143,7 @@ class AllMailController(MailController):
sms_filter["serviceType"] = "sms"
# Invoke the parent's constructor:
CoreMessageController.__init__(
MailController.__init__(
self,
cache = cache,
alert_url = alert_url,
@@ -149,11 +154,90 @@ class AllMailController(MailController):
debug_only_errors = debug_only_errors
)
# ┓┏ ┓
# ┣┫┏┓┃┏┓┏┓┏┓┏
# ┛┗┗ ┗┣┛┗ ┛ ┛
# ┛
pass
# ┏┓┏┓ ┓ ┏┓ ┏┓
# ┃┃┣┫┓┏╋┣┓┏┛ ┃┫
# ┗┛┛┗┗┻┗┛┗┗━•┗┛
pass
async def get_authorization_url(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
mail_client: AsyncGMailClient,
user_info: CoreUserInfoModel,
inbound_data: OAuthMailAuthorizationRequestData,
session_token: str
) -> OAuthMailGetAuthorizationURLResponse:
"""
To accept an incoming request for mail integration and provide a URL that the user can use to authorize your
service to access his mail inbox.
:param sql_conn: The database connection to use to perform this task.
:param mongo_data_conn: The database connection to use to perform this task.
:param mail_client: The instance of the third-party mail client that will be used to get the URL.
:param user_info: The information about your user who is trying to use this system.
:param inbound_data: The data that came in with the request (API call).
:param session_token: The session token of the user.
:return: A structure response with details about the URL generation process.
"""
raise NotImplementedError
async def handle_authorization_callback(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
mail_client: AsyncGMailClient,
request_url: str,
inbound_data: dict,
session_token: str = None
) -> OAuthMailHandleCallbackResponse:
"""
To handle the authorization callback for the mail client. The user may grant or deny authorization.
:param sql_conn: The database connection to use to perform this task.
:param mongo_data_conn: The database connection to use to perform this task.
:param mail_client: The instance of the third-party mail client that will be used to get the URL.
:param request_url: The full callback URL invoked by the third-party client.
:param inbound_data: The data that came in with the request (API call).
:param session_token: The session token of the user. It is expected that this will be null in all cases.
:return: A structured response of the process of handling the mail callback.
"""
raise NotImplementedError
async def refresh_authorization(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
mail_client: AsyncGMailClient,
http_client: httpx.AsyncClient,
auth_token: CoreAuthTokenModel,
force_refresh: bool = False,
session_token: str = None
) -> CoreAuthTokenModel:
"""
To refresh the third-party client's access/authorization token(s) before use.
:param sql_conn: The database connection to use when storing the refreshed tokens.
:param mongo_data_conn: The database connection to use when storing the refreshed tokens.
:param mail_client: The connection of the third-party mail client.
:param http_client: The HTTP client to use to make the token refresh request.
:param auth_token: The auth-token model of the existing integration. This may get updated if a refresh is needed
(or forced).
:param force_refresh: Whether, or not, you would like to force a refresh even if the token hasn't expired yet.
:param session_token: The session token of the user. This will be null if this method is invoked by a cron
script in the background. Needed only to identify the user in case of a failure to send a timely alert.
:return: The same auth-token model instance, but maybe with updated tokens.
"""
raise NotImplementedError
# ┳┳┓ •┓ ┏┓ • •
# ┃┃┃┏┓┓┃ ┗┓┓┏┏┳┓┏┳┓┏┓┏┓┓┓┏┓╋┓┏┓┏┓
@@ -169,7 +253,40 @@ class AllMailController(MailController):
# To synchronize the mails on the third-party client's server and your server. You are effectively making a copy of
# the mail on your database.
pass
async def sync_mails(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
mail_client: AsyncGMailClient,
auth_token: CoreAuthTokenModel,
user_info: CoreUserInfoModel | None,
llm: CoreLLMController = None,
force_sync: bool = False,
start_date: datetime.datetime = None,
end_date: datetime.datetime = None,
max_count: int = 100,
session_token: str = None
) -> MailSyncManyResults:
"""
To fetch mails from the third-party client and store them to your database.
:param sql_conn: The database connection to use to perform this task.
:param mongo_data_conn: The database connection to use to perform this task.
:param mail_client: The instance of the third-party mail client that will be used to get the URL.
:param auth_token: The credentials to the account with the third-party client.
:param user_info: The information about your user who is trying to use this system. Needed to note LLM token
usage in the process of mail summarization.
:param llm: The instance of the LLm that can be used to summarize the contents of the mail.
:param force_sync: To forcefully sync a mail even if it already exists in the database.
:param start_date: The starting date (inclusive) from which mails must be sync'd.
:param end_date: The ending date (inclusive) till which mails must be sync'd.
:param max_count: The max. no. of mails to sync.
:param session_token: TO identify a user session. This will be null if a cron script invokes this method, else
it will be received from the inputs of the API call.
:return:
"""
raise NotImplementedError
# ┳┳┓ •┓ ┓ • •
# ┃┃┃┏┓┓┃ ┃ ┓┏╋┓┏┓┏┓
controllers_v2/message/mail/base.py
+136 -7
View File
@@ -66,12 +66,12 @@ from typing import List, Any
# To make HTTP requests:
import httpx
# To work with MongoDB:
from bson.objectid import ObjectId
# To parse the HTML content in the mail:
from bs4 import BeautifulSoup
# To work with MongoDB:
from bson import ObjectId
# To work with date and time:
import datetime
@@ -264,6 +264,33 @@ class MailController(CoreMessageController, ABC):
# Done here:
return text_parts
def drop_attachments(
self,
payload: dict
) -> dict:
"""
At the time of creating the system, we don't have a mechanism to save, organize and serve files in a
satisfactory way. This is a simple way to ignore all attachments till then by dropping them entirely.
:param payload: The mail's full payload.
:return: The same payload, but with all attachment data wiped clean.
"""
# If the part is some sort of file:
if payload["contentMainType"] not in ["multipart", "text"]:
payload["payload"] = None
payload["payloadId"] = None
payload["payloadUrl"] = None
# If the payload is of multipart type,
# we use recursion to look inside it:
elif payload["contentMainType"] == "multipart":
for part in payload["payload"]:
self.drop_attachments(part)
# Done here:
return payload
# ┏┓┏┓ ┓ ┏┓ ┏┓
# ┃┃┣┫┓┏╋┣┓┏┛ ┃┫
# ┗┛┛┗┗┻┗┛┗┗━•┗┛
@@ -317,6 +344,34 @@ class MailController(CoreMessageController, ABC):
pass
@abstractmethod
async def refresh_authorization(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
mail_client: AsyncGMailClient,
http_client: httpx.AsyncClient,
auth_token: CoreAuthTokenModel,
force_refresh: bool = False,
session_token: str = None
) -> CoreAuthTokenModel:
"""
To refresh the third-party client's access/authorization token(s) before use.
:param sql_conn: The database connection to use when storing the refreshed tokens.
:param mongo_data_conn: The database connection to use when storing the refreshed tokens.
:param mail_client: The connection of the third-party mail client.
:param http_client: The HTTP client to use to make the token refresh request.
:param auth_token: The auth-token model of the existing integration. This may get updated if a refresh is needed
(or forced).
:param force_refresh: Whether, or not, you would like to force a refresh even if the token hasn't expired yet.
:param session_token: The session token of the user. This will be null if this method is invoked by a cron
script in the background. Needed only to identify the user in case of a failure to send a timely alert.
:return: The same auth-token model instance, but maybe with updated tokens.
"""
pass
# ┳┳┓ •┓ ┏┓ • •
# ┃┃┃┏┓┓┃ ┗┓┓┏┏┳┓┏┳┓┏┓┏┓┓┓┏┓╋┓┏┓┏┓
# ┛ ┗┗┻┗┗ ┗┛┗┻┛┗┗┛┗┗┗┻┛ ┗┗┗┻┗┗┗┛┛┗
@@ -372,7 +427,7 @@ class MailController(CoreMessageController, ABC):
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
mail_client: AsyncGMailClient,
token_key: ObjectId | str,
auth_token: CoreAuthTokenModel,
user_info: CoreUserInfoModel | None,
llm: CoreLLMController = None,
force_sync: bool = False,
@@ -387,7 +442,7 @@ class MailController(CoreMessageController, ABC):
:param sql_conn: The database connection to use to perform this task.
:param mongo_data_conn: The database connection to use to perform this task.
:param mail_client: The instance of the third-party mail client that will be used to get the URL.
:param token_key: The key by which the auth-tokens to this account are identified.
:param auth_token: The credentials to the account with the third-party client.
:param user_info: The information about your user who is trying to use this system. Needed to note LLM token
usage in the process of mail summarization.
:param llm: The instance of the LLm that can be used to summarize the contents of the mail.
@@ -410,7 +465,58 @@ class MailController(CoreMessageController, ABC):
# Use these to show your users their mails once the mails are on your server. This would include activities like
# listing mails, showing full mails, showing mail trails, etc.
pass
async def list_mails(
self,
mongo_data_conn: AsyncMongo,
token_ids: List[ObjectId | str],
limit: int = 100,
skip: int = 0,
additional_filter: dict = None
) -> List[CoreMessageModel] | None:
"""
To list mails (just previews, not full payloads).
:param mongo_data_conn: The database connection to use to fetch the data.
:param token_ids: The ids by which the mails will be identified. These are the auth-token ids of the accounts to
which the mails belong.
:param limit: The max. no. of mails to fetch. Good for pagination.
:param skip: The no. of initial mails to skip before picking mails to show. Good for pagination.
:param additional_filter: Any additional constraints.
:return: A list of message models that describe the contents of the mails.
"""
# 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"] = "email"
# Simply call the core model:
return await self.get_message_previews(
mongo_data_conn = mongo_data_conn,
token_ids = token_ids,
limit = limit,
skip = skip,
additional_filter = additional_filter
)
async def get_one_mail(
self,
mongo_data_conn: AsyncMongo,
message_id: ObjectId | str
) -> CoreMessageModel | None:
"""
To get one full mail (the full payload, not just the preview).
:param mongo_data_conn: The database connection to use to fetch the data.
:param message_id: The ObjectId of the document that holds the mail.
:return: The message model containing the full payload of the mail.
"""
# Simply call the core model:
return await self.get_message(
mongo_data_conn = mongo_data_conn,
message_id = message_id
)
# ┳┳┓ •┓ ┏┓ ┓•
# ┃┃┃┏┓┓┃ ┗┓┏┓┏┓┏┫┓┏┓┏┓
@@ -427,7 +533,30 @@ class MailController(CoreMessageController, ABC):
# We cannot modify the mails themselves, but we can set/unset tags on them for internal referencing and filtering.
# This will help the users organize their inboxes well.
pass
async def update_mail_tags(
self,
mongo_data_conn: AsyncMongo,
message_id: ObjectId | str,
unset_tags: List[str] = None,
set_tags: List[str] = None
) -> bool:
"""
To set and unset tags on a mail.
:param mongo_data_conn: The database connection to use to perform this action.
:param message_id: The ObjectId of the document in MongoDb that holds the message.
:param unset_tags: The list of tags to unset (done before setting new tags).
:param set_tags: The list of tags to set (done after unsetting old tags).
:return: True if successful, else False.
"""
# Simply call the core model:
return await self.update_message_tags(
mongo_data_conn = mongo_data_conn,
message_id = message_id,
unset_tags = unset_tags,
set_tags = set_tags
)
# *****************************************************************************************************************
controllers_v2/message/mail/gmail.py
+290 -5
View File
@@ -58,12 +58,14 @@ from models.message.mail.send import MailSendOneResult
# Mail Client(s):
from utils_v2.goog.controllers.gmail.gmail_client import AsyncGMailClient, SCOPES_GMAIL_MAIL_MANAGEMENT
from utils_v2.goog.models.auth_tokens import GoogleAuthTokens
# To work with datatypes:
from typing import List, Any
# To work with MongoDB:
from bson import ObjectId
from pymongo import InsertOne, UpdateOne, ReplaceOne
# To work with LLMs:
from controllers.core.ai.llm import CoreLLMController
@@ -362,6 +364,74 @@ class GmailController(MailController):
# Done here:
return response
async def refresh_authorization(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
mail_client: AsyncGMailClient,
http_client: httpx.AsyncClient,
auth_token: CoreAuthTokenModel,
force_refresh: bool = False,
session_token: str = None
) -> CoreAuthTokenModel:
"""
To refresh the third-party client's access/authorization token(s) before use.
:param sql_conn: The database connection to use when storing the refreshed tokens.
:param mongo_data_conn: The database connection to use when storing the refreshed tokens.
:param mail_client: The connection of the third-party mail client.
:param http_client: The HTTP client to use to make the token refresh request.
:param auth_token: The auth-token model of the existing integration. This may get updated if a refresh is needed
(or forced).
:param force_refresh: Whether, or not, you would like to force a refresh even if the token hasn't expired yet.
:param session_token: The session token of the user. This will be null if this method is invoked by a cron
script in the background. Needed only to identify the user in case of a failure to send a timely alert.
:return: The same auth-token model instance, but maybe with updated tokens.
"""
# Extract the client's tokens from the full token payload given by the database:
google_tokens = GoogleAuthTokens(**auth_token.token)
# Refresh the tokens (if/as needed):
tokens_refreshed = await google_tokens.arefresh(
http_client = http_client,
client_id = mail_client.client_id,
client_secret = mail_client.client_secret,
force_refresh = force_refresh
)
# If the tokens were refreshed:
if tokens_refreshed:
# Try getting the user's profile from Gmail:
user_profile = await mail_client.get_user_profile(tokens = google_tokens)
if user_profile.success:
google_tokens.email = user_profile.data["emailAddress"]
google_tokens.displayName = user_profile.data["displayName"]
google_tokens.displayPictureUrl = user_profile.data["displayPictureUrl"]
# Update the existing auth-token model:
auth_token.token = google_tokens.model_dump()
auth_token.lastRefreshTs = date_time.get_current_utc_date_time(as_string = True)
# Try to update the record in the database:
await self.set_token(
sql_conn = sql_conn,
mongo_data_conn = mongo_data_conn,
token_key = auth_token.key,
auth_token = auth_token,
token_notes = {
"email": auth_token.clientUserId.get("email"),
"client": auth_token.client
},
display_name = google_tokens.email,
display_picture = google_tokens.displayPictureUrl,
session_token = session_token,
)
# Whether refreshed, or not, return the auth-token model:
return auth_token
# ┳┳┓ •┓ ┏┓ • •
# ┃┃┃┏┓┓┃ ┗┓┓┏┏┳┓┏┳┓┏┓┏┓┓┓┏┓╋┓┏┓┏┓
# ┛ ┗┗┻┗┗ ┗┛┗┻┛┗┗┛┗┗┗┻┛ ┗┗┗┻┗┗┗┛┛┗
@@ -376,12 +446,123 @@ class GmailController(MailController):
# To synchronize the mails on the third-party client's server and your server. You are effectively making a copy of
# the mail on your database.
async def __sync_one_mail(
self,
mongo_data_conn: AsyncMongo,
user_info: CoreUserInfoModel,
auth_token: CoreAuthTokenModel,
mail_client: AsyncGMailClient,
google_tokens: GoogleAuthTokens,
message_id: str,
llm: CoreLLMController = None,
force_sync: bool = False
) -> MailSyncOneResult:
"""
To fetch one mail from the third-party client and store it in your database.
:param mongo_data_conn: The database connection to use to store the mail's payload.
:param user_info: The information about the user to whom this mail belongs.
:param auth_token: The credentials to use to get the mail from the third-party client.
:param mail_client: The third-party client's connection object.
:param google_tokens: The mail client's tokens the way they have to be used in their connection.
:param message_id: The way the third-party client recognizes the mail.
:param llm: To summarize the mail.
:param force_sync: If you'd like to forcefully re-sync the mail if its record already exists in your database.
:return: The structured response to express how the mail fetching went.
"""
# Start by assuming failure:
sync_result = MailSyncOneResult()
# If we've not been forced to re-sync the mail message,
# we first check if the mail already exists in our database:
if not force_sync:
mail_records = await self.get_message_previews(
mongo_data_conn = mongo_data_conn,
token_ids = [ObjectId(auth_token.authTokenId)],
limit = 1,
skip = 0,
additional_filter = {
"tokenId": ObjectId(auth_token.authTokenId),
"serviceType": auth_token.serviceType,
"client": auth_token.client,
"clientMessageId": message_id
}
)
if mail_records:
sync_result.success = True
sync_result.message = (
f"Gmail message '{message_id}' already "
f"sync'd on {mail_records[0].syncTs} (UTC)."
)
return sync_result
# Now that we know that we have to fetch the mail from GMail:
client_response = await mail_client.get_message(
tokens = google_tokens,
message_id = message_id,
return_raw = False
)
# If we didn't get the mail from Gmail;
if not client_response.success:
sync_result.message = f"Gmail (messageId: '{message_id}'): {client_response.message}"
return sync_result
# HANDLE ATTACHMENTS HERE:
client_response.data["payload"] = self.drop_attachments(client_response.data["payload"])
# Now we structure the message into the model:
all_recipients = []
for field in ["to", "cc", "bcc"]: all_recipients += [item["email"] for item in client_response.data[field]]
is_sent = False if google_tokens.email in all_recipients else True
mail_message = CoreMessageModel(
ts = client_response.data["ts"],
syncTs = date_time.get_current_utc_date_time(as_string = False),
tokenId = auth_token.authTokenId,
serviceType = auth_token.serviceType,
client = auth_token.client,
clientMessageId = message_id,
clientThreadId = client_response.data["threadId"],
isSent = True if is_sent else False,
isBroadcast = False,
sentSuccessfully = True if is_sent else False,
sender = [client_response.data["from"][0]["name"]],
recipient = all_recipients,
chat = None,
message = client_response.data,
snippet = client_response.data["subject"],
aiSnippet = None,
tags = ["Email", "Gmail"]
)
# Invoke the LLM:
try:
ai_snippet = await self.summarize_mail_with_ai(
mongo_data_conn = mongo_data_conn,
user_info = user_info,
llm = llm,
message = mail_message,
prompt_template = self.SENT_MAIL_SUMMARIZATION_PROMPT_TEMPLATE if is_sent else self.RECEIVED_MAIL_SUMMARIZATION_PROMPT_TEMPLATE
)
ai_json = ai_snippet.json
mail_message.aiSnippet = ai_snippet.summary
mail_message.aiSnippet["output"] = ai_json["summary"]
if ai_json["senderType"] is not None: mail_message.tags.append(ai_json["senderType"])
except Exception as exception:
self._printer(exception)
# Done here:
sync_result.success = True
sync_result.mailMessage = mail_message
return sync_result
async def sync_mails(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
mail_client: AsyncGMailClient,
token_key: ObjectId | str,
auth_token: CoreAuthTokenModel,
user_info: CoreUserInfoModel | None,
llm: CoreLLMController = None,
force_sync: bool = False,
@@ -393,10 +574,11 @@ class GmailController(MailController):
"""
To fetch mails from the third-party client and store them to your database.
:param sql_conn: The database connection to use to perform this task.
:param sql_conn: The database connection to use to perform this task. Needed if the auth tokens need to be
refreshed or updated.
:param mongo_data_conn: The database connection to use to perform this task.
:param mail_client: The instance of the third-party mail client that will be used to get the URL.
:param token_key: The key by which the auth-tokens to this account are identified.
:param auth_token: The credentials to the account with the third-party client.
:param user_info: The information about your user who is trying to use this system. Needed to note LLM token
usage in the process of mail summarization.
:param llm: The instance of the LLm that can be used to summarize the contents of the mail.
@@ -406,10 +588,113 @@ class GmailController(MailController):
:param max_count: The max. no. of mails to sync.
:param session_token: TO identify a user session. This will be null if a cron script invokes this method, else
it will be received from the inputs of the API call.
:return:
:return: The structured response to express how the mail fetching went.
"""
pass
# Start by assuming failure:
sync_results = MailSyncManyResults()
# Refresh the access token(s) if needed:
auth_token = await self.refresh_authorization(
sql_conn = sql_conn,
mongo_data_conn = mongo_data_conn,
mail_client = mail_client,
http_client = mail_client.http_client,
auth_token = auth_token,
force_refresh = False,
session_token = session_token
)
# If the user info was not given, take it from the token model:
if user_info is None: user_info = auth_token.user
# Extract the client's tokens from the full token model,
# and check if they are valid (not expired):
google_tokens = GoogleAuthTokens(**auth_token.token)
if google_tokens.expired:
sync_results.message = "Gmail token(s) have expired."
# Let's build the query to send to Google:
sub_queries = []
if start_date: sub_queries.append(start_date.strftime("after:%Y/%m/%d"))
if end_date: sub_queries.append((end_date + datetime.timedelta(days = 1)).strftime("before:%Y/%m/%d"))
query_string = " ".join(sub_queries)
# Let's enlist all the mails that fall in the date range:
client_response = await mail_client.list_messages(
tokens = google_tokens,
max_count = max_count,
query = query_string
)
# If the mail listing fails:
if not client_response.success:
self._printer(
client_response.success,
client_response.message,
client_response.data,
client_response.exception
)
sync_results.message = f"Gmail: {client_response.message}"
return sync_results
# Now, for every mail in the list, we fetch the mail and note the results:
messages_list = client_response.data["messages"]
tasks = [
self.__sync_one_mail(
mongo_data_conn = mongo_data_conn,
user_info = user_info,
auth_token = auth_token,
mail_client = mail_client,
google_tokens = google_tokens,
message_id = v["id"],
llm = llm,
force_sync = force_sync
) for v in messages_list.values()
]
individual_sync_results = await asyncio.gather(*tasks)
# Now we create operations for each mail,
# and maintain success/failure counters:
sync_results.totalCount = len(individual_sync_results)
mongo_operations = []
for result in individual_sync_results:
if result.success: sync_results.successCount += 1
else: sync_results.failureCount += 1
if result.mailMessage:
replacement_json = result.mailMessage.model_dump()
replacement_json.pop("_id", None)
mongo_operations.append(
ReplaceOne(
filter = {
"tokenId": ObjectId(auth_token.authTokenId),
"serviceType": auth_token.serviceType,
"client": auth_token.client,
"clientMessageId": result.mailMessage.clientMessageId
},
replacement = replacement_json,
upsert = True
)
)
# Make the bulk insert operation:
if mongo_operations:
sync_count = await self.bulk_operate_messages(
mongo_data_conn = mongo_data_conn,
mongo_operations = mongo_operations
)
# Apply the labels to the read messages:
try: client_response = await mail_client.modify_messages(
tokens = google_tokens,
message_ids = [v["id"] for v in messages_list.values()],
add_label_ids = [google_tokens.labels.get("TCAOFF", {}).get("id")]
)
except Exception as exception: pass
# Done here:
sync_results.message = f"{sync_results.successCount}/{sync_results.totalCount} mail(s) sync'd from Gmail."
return sync_results
# ┳┳┓ •┓ ┓ • •
# ┃┃┃┏┓┓┃ ┃ ┓┏╋┓┏┓┏┓
controllers_v2/message/sms/base.py
+36
View File
@@ -61,6 +61,9 @@ from typing import List, Any
# To make HTTP requests:
import httpx
# to work with MongoDB:
from bson.objectid import ObjectId
# To make abstract classes:
from abc import ABC, abstractmethod
@@ -197,6 +200,39 @@ class SMSController(CoreMessageController, ABC):
pass
# ┏┓┳┳┓┏┓ ┳┳ ┓ •
# ┗┓┃┃┃┗┓ ┃┃┏┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┛ ┗┗┛ ┗┛┣┛┗┻┗┻┗┗┛┗┗┫
# ┛ ┛
# We cannot modify the SMS messages themselves, but we can set/unset tags on them for internal referencing and
# filtering. This will help the users organize their inboxes well.
async def update_sms_tags(
self,
mongo_data_conn: AsyncMongo,
message_id: ObjectId | str,
unset_tags: List[str] = None,
set_tags: List[str] = None
) -> bool:
"""
To set and unset tags on an SMS message.
:param mongo_data_conn: The database connection to use to perform this action.
:param message_id: The ObjectId of the document in MongoDb that holds the message.
:param unset_tags: The list of tags to unset (done before setting new tags).
:param set_tags: The list of tags to set (done after unsetting old tags).
:return: True if successful, else False.
"""
# Simply call the core model:
return await self.update_message_tags(
mongo_data_conn = mongo_data_conn,
message_id = message_id,
unset_tags = unset_tags,
set_tags = set_tags
)
# *****************************************************************************************************************
# ***** ****