From be9bdf797a7c37e9be4499accb24292b2964cff2 Mon Sep 17 00:00:00 2001 From: khushal Date: Fri, 17 Jan 2025 15:36:59 +0530 Subject: [PATCH] (20250117) Major revamping in the mail module. Everything revamped. Sending is a pending task. --- .../message/mail/retrieve/get_v2.py | 4 +- .../message/mail/retrieve/list_v2.py | 6 +- api/blueprints/message/mail/sync/sync_v3.py | 72 +++-- api/blueprints/message/mail/tags/update_v2.py | 4 +- api/blueprints/message/sms/tags.py | 4 +- api/main.py | 17 +- controllers_v2/core/message.py | 2 +- controllers_v2/message/mail/all_mail.py | 147 ++++++++- controllers_v2/message/mail/base.py | 143 ++++++++- controllers_v2/message/mail/gmail.py | 295 +++++++++++++++++- controllers_v2/message/sms/base.py | 36 +++ models/core/message.py | 10 +- 12 files changed, 675 insertions(+), 65 deletions(-) diff --git a/api/blueprints/message/mail/retrieve/get_v2.py b/api/blueprints/message/mail/retrieve/get_v2.py index ecd2092..03e55e5 100644 --- a/api/blueprints/message/mail/retrieve/get_v2.py +++ b/api/blueprints/message/mail/retrieve/get_v2.py @@ -6,7 +6,7 @@ DATE: - Tuesday, 3rd Dec., 2024 + Friday, 17th Jan., 2025. OBJECTIVE: @@ -178,7 +178,7 @@ async def get_one_mail( # Get the mail: message = await current_app.mail_controller.get_one_mail( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, message_id = inbound_data.messageId ) diff --git a/api/blueprints/message/mail/retrieve/list_v2.py b/api/blueprints/message/mail/retrieve/list_v2.py index b386915..677bfc9 100644 --- a/api/blueprints/message/mail/retrieve/list_v2.py +++ b/api/blueprints/message/mail/retrieve/list_v2.py @@ -6,7 +6,7 @@ DATE: - Tuesday, 3rd Dec., 2024 + Friday, 17th Jan., 2025. OBJECTIVE: @@ -181,7 +181,7 @@ async def list_mails( # Get the token ids from the token keys: auth_tokens = await current_app.mail_controller.get_tokens_from_keys( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, token_keys = inbound_data.tokenKeys ) token_ids = [t.authTokenId for t in auth_tokens] @@ -193,7 +193,7 @@ async def list_mails( # Get the mails: mails_list = await current_app.mail_controller.list_mails( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, token_ids = token_ids, limit = inbound_data.count, skip = inbound_data.fromCount, diff --git a/api/blueprints/message/mail/sync/sync_v3.py b/api/blueprints/message/mail/sync/sync_v3.py index f1c93d0..af974c6 100644 --- a/api/blueprints/message/mail/sync/sync_v3.py +++ b/api/blueprints/message/mail/sync/sync_v3.py @@ -6,12 +6,12 @@ DATE: - Monday, 2nd Dec., 2024 + Friday, 17th Jan., 2025. OBJECTIVE: To receive requests for synchronising mails from various mail clients to the database. Sync'ing means we pull - the mail from the mail client (like GMail) and store it to our database. The mail is then ready for showing on + the mail from the mail client (like Gmail) and store it to our database. The mail is then ready for showing on the UI at any time. REFERENCES: @@ -129,7 +129,8 @@ def init(blueprint_setup_state): async def sync_mails( user_info: CoreUserInfoModel, inbound_headers: dict, - inbound_data: MailSyncRequestData + inbound_data: MailSyncRequestData, + is_background: bool = True ) -> MailSyncManyResults: """ @@ -138,23 +139,54 @@ async def sync_mails( :param user_info: The information of the user as extracted from the session token. :param inbound_headers: The headers that came in with the request. :param inbound_data: The data that came in with the request. + :param is_background: To know whether, or not, this request is being services in the background. If it is happening + in the foreground, user feedback is not a concern. But, if it is happening in the background, user feedback will + be a concern. This flag will help alter the treatment of sending alerts. :return: The results of the mail-sync'ing attempt. """ - # Try to sync the mails: - return await current_app.mail_controller.sync( - db_conn = current_app.sql_writer, - mongo_conn = current_app.data_mongo, - user_info = user_info, + # Start by assuming failure: + client_controller = None + client_connector = None + sync_results = MailSyncManyResults() + + # Get the auth-token from the key: + auth_token = await current_app.mail_controller.get_token_from_key( + mongo_data_conn = current_app.data_mongo, token_key = inbound_data.tokenKey, - llm = current_app.llm, - force_sync = inbound_data.forceSync, - start_date = inbound_data.startDate, - end_date = inbound_data.endDate, - max_count = inbound_data.maxCount, - session_token = inbound_headers["X-Session-Token"], ) + # Figure out the client connector: + match auth_token.client: + case "gmail": client_controller, client_connector = current_app.gmail_controller, current_app.gmail_client + case _: client_controller, client_connector = None, None + + # Invoke the client: + if client_controller is not None and client_connector is not None: + sync_results = await client_controller.sync_mails( + sql_conn = current_app.sql_writer, + mongo_data_conn = current_app.data_mongo, + mail_client = client_connector, + auth_token = auth_token, + user_info = user_info, + llm = current_app.llm, + force_sync = inbound_data.forceSync, + start_date = inbound_data.startDate, + end_date = inbound_data.endDate, + max_count = inbound_data.maxCount, + session_token = inbound_headers["X-Session-Token"] + ) + + # If the client was not matched: + else: sync_results.message = f"Invalid/unimplemented client '{auth_token.client}'" + + # Send an alert if this is a background process: + if is_background: + pass + + # Done here: + return sync_results + # --------------------------------------------------------------------------------------------------------------------- @@ -206,29 +238,27 @@ async def sync_mail( http_code = HttpCodes.UNAUTHORIZED ) - # Make the variables available in the scope of the current request: - g.inbound_headers = inbound_headers - g.inbound_data = inbound_data - # If we've been asked to sync the mails in the background: if mode in ["background", "bg"]: current_app.add_background_task( sync_mails, user_info = CoreUserInfoModel(**kwargs["session_info"]), inbound_headers = inbound_headers, - inbound_data = inbound_data + inbound_data = inbound_data, + is_background = True ) return ResponseModel( status_code = StatusCodes.OK, http_code = HttpCodes.ACCEPTED, - message = "your mails are being sync'd in the background" + message = "Your mails are being sync'd in the background." ) # Otherwise we process it right here: sync_results = await sync_mails( user_info = CoreUserInfoModel(**kwargs["session_info"]), inbound_headers = inbound_headers, - inbound_data = inbound_data + inbound_data = inbound_data, + is_background = False ) # Response: diff --git a/api/blueprints/message/mail/tags/update_v2.py b/api/blueprints/message/mail/tags/update_v2.py index 55d413c..008fec3 100644 --- a/api/blueprints/message/mail/tags/update_v2.py +++ b/api/blueprints/message/mail/tags/update_v2.py @@ -177,7 +177,7 @@ async def update_mail_tags( # Get the mail: message = await current_app.mail_controller.get_one_mail( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, message_id = inbound_data.messageId ) @@ -204,7 +204,7 @@ async def update_mail_tags( # Update the mail: success = await current_app.mail_controller.update_mail_tags( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, message_id = inbound_data.messageId, unset_tags = inbound_data.unsetTags, set_tags = inbound_data.setTags diff --git a/api/blueprints/message/sms/tags.py b/api/blueprints/message/sms/tags.py index db4283b..803418b 100644 --- a/api/blueprints/message/sms/tags.py +++ b/api/blueprints/message/sms/tags.py @@ -180,7 +180,7 @@ async def update_sms_tags( # Check if the token(s) belong to the user claiming ownership: if not await token_check.is_authorized( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, user_info = CoreUserInfoModel(**kwargs["session_info"]), token_ids = [message.tokenId] ): return ResponseModel( @@ -195,7 +195,7 @@ async def update_sms_tags( # ┛ ┛ # Update the message: - success = await current_app.sms_controller.update_tags( + success = await current_app.sms_controller.update_sms_tags( mongo_data_conn = current_app.data_mongo, message_id = inbound_data.messageId, unset_tags = inbound_data.unsetTags, diff --git a/api/main.py b/api/main.py index 340ae9a..5c0d448 100644 --- a/api/main.py +++ b/api/main.py @@ -80,6 +80,7 @@ from controllers_v2.message.sms.all_sms import AllSMSController from controllers_v2.message.sms.nimbus_sms_india import NimbusSMSIndiaController from controllers_v2.message.sms.savvy_bulk_sms_kenya import SavvyBulkSMSKenyaController # --- +from controllers_v2.message.mail.all_mail import AllMailController from controllers_v2.message.mail.gmail import GmailController # --- from controllers_v2.message.chat.all_chat import AllChatController @@ -102,10 +103,10 @@ from icecream import IceCreamDebugger # Mail Blueprints: from api.blueprints.message.mail.oauth.request_v2 import mail_oauth_request_bp from api.blueprints.message.mail.oauth.callback_v2 import mail_oauth_callback_bp -from api.blueprints.message.mail.sync.sync_v2 import mail_sync_bp -from api.blueprints.message.mail.retrieve.list import mail_list_bp -from api.blueprints.message.mail.retrieve.get import mail_get_bp -from api.blueprints.message.mail.tags.update import mail_tags_update_bp +from api.blueprints.message.mail.sync.sync_v3 import mail_sync_bp +from api.blueprints.message.mail.retrieve.list_v2 import mail_list_bp +from api.blueprints.message.mail.retrieve.get_v2 import mail_get_bp +from api.blueprints.message.mail.tags.update_v2 import mail_tags_update_bp from api.blueprints.message.mail.send.send import mail_send_bp # SMS Blueprints: @@ -424,7 +425,7 @@ async def app_startup(**kwargs): # ┣┫┃┃┃ ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ # ┛┗┣┛┻ ┗┛┗┛┛┗┗┛ ┗┛┗┗┗ ┛ ┛ - current_app.mail_controller = MailController() + # current_app.mail_controller = MailController() # current_app.sms_controller = SMSController() # current_app.payment_controller = PaymentController() @@ -461,6 +462,12 @@ async def app_startup(**kwargs): ) # Messages / Mail Controllers: + current_app.mail_controller = AllMailController( + cache = current_app.module_cache, + http_client = current_app.http_client, + alert_url = current_app.script_data["alerts"]["url"], + debug = enable_debugging + ) current_app.gmail_controller = GmailController( cache = current_app.module_cache, http_client = current_app.http_client, diff --git a/controllers_v2/core/message.py b/controllers_v2/core/message.py index e9d63d0..f2d44fa 100644 --- a/controllers_v2/core/message.py +++ b/controllers_v2/core/message.py @@ -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, diff --git a/controllers_v2/message/mail/all_mail.py b/controllers_v2/message/mail/all_mail.py index 913e712..874d8e0 100644 --- a/controllers_v2/message/mail/all_mail.py +++ b/controllers_v2/message/mail/all_mail.py @@ -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 # ┳┳┓ •┓ ┓ • • # ┃┃┃┏┓┓┃ ┃ ┓┏╋┓┏┓┏┓ diff --git a/controllers_v2/message/mail/base.py b/controllers_v2/message/mail/base.py index a8bfef8..6483d0c 100644 --- a/controllers_v2/message/mail/base.py +++ b/controllers_v2/message/mail/base.py @@ -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 + ) # ***************************************************************************************************************** diff --git a/controllers_v2/message/mail/gmail.py b/controllers_v2/message/mail/gmail.py index 7e6d80d..dd85fec 100644 --- a/controllers_v2/message/mail/gmail.py +++ b/controllers_v2/message/mail/gmail.py @@ -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 # ┳┳┓ •┓ ┓ • • # ┃┃┃┏┓┓┃ ┃ ┓┏╋┓┏┓┏┓ diff --git a/controllers_v2/message/sms/base.py b/controllers_v2/message/sms/base.py index cc29fb4..a9e07fe 100644 --- a/controllers_v2/message/sms/base.py +++ b/controllers_v2/message/sms/base.py @@ -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 + ) + # ***************************************************************************************************************** # ***** **** diff --git a/models/core/message.py b/models/core/message.py index 6d1d3be..b9ff56f 100644 --- a/models/core/message.py +++ b/models/core/message.py @@ -149,12 +149,12 @@ class CoreMessageModel(BaseModel): default = False ) - sender: str | None = Field( + sender: List[str] | None = Field( description = "the name of the sender; null if you are the sender", frozen = True ) - recipient: str | None = Field( + recipient: List[str] | None = Field( description = "the name of the recipient; null if you are the recipient", frozen = True, default = None @@ -260,6 +260,12 @@ class CoreMessageModel(BaseModel): except: pass return value + @field_validator("sender", "recipient", mode = "before") + def validate_participants(cls, value): + if value is None: value = [] + if isinstance(value, str): value = [value] + return value + @field_validator("tags", mode = "before") def validate_tags(cls, value): if value is None: value = []