diff --git a/api/blueprints/mail/list.py b/api/blueprints/mail/list.py new file mode 100644 index 0000000..e8d8712 --- /dev/null +++ b/api/blueprints/mail/list.py @@ -0,0 +1,255 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Tuesday, 3rd Dec., 2024 + + OBJECTIVE: + + To enlist multiple e-mails for a given user at a time. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + + NOTES: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For using Quart: +from quart import Blueprint, current_app, g, request + +# My utils: +from utils_v2.string import json +from utils_v2.database.async_mongo_v2 import AsyncMongo +from utils_v2.api.codes import StatusCodes, HttpCodes +from utils_v2.api.response import ResponseModel +from utils_v2.api.async_quart import ( + set_api_version, + read_input, + get_session_info, + log_request_to_mongo, + log_chain_to_mongo, + should_not_be_under_maintenance, + only_whitelisted_ips, + limit_rate, + validate_input, + handle_cancelled_request +) + +# GMail-related utils: +from utils_v2.goog.gmail.gmail_client import SCOPES_GMAIL_MAIL_MANAGEMENT +from utils_v2.goog.models.data.auth_tokens import GoogleAuthTokens + +# Common: +from shared import constants + +# Data Models: +from models.data.mail.sync import MailSyncRequestHeaders, MailSyncRequestData +from models.data.mail.sync import MailSyncOneResult, MailSyncManyResults + +# To work with datatypes: +from typing import Literal + +# For asynchronous activities: +import asyncio + +# To work with LLMs: +from langchain_openai import ChatOpenAI + +# To work with date and time: +import datetime + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# Related to Quart: +mail_sync_bp = Blueprint("mail_sync", __name__) + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +@mail_sync_bp.record_once +def init(blueprint_setup_state): + + # This gets called when the blueprint is registered. + # Consider this to be a one-time setup for the whole blueprint: + pass + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def sync_mails( + mongo_conn: AsyncMongo, + llm: ChatOpenAI, + inbound_headers: dict, + inbound_data: MailSyncRequestData +) -> MailSyncManyResults: + + """ + A very simple function, but kept separate so that we get the option to switch between running it in the foreground + and running it in the background. + :param mongo_conn: The instance of the database connector to use to sync the mails. + :param llm: The instance of the LLM to use to summarize the mails. + :param inbound_headers: The headers that came in with the request. + :param inbound_data: The data that came in with the request. + :return: The results of the mail-sync'ing attempt. + """ + + # Try to sync the mails: + return await current_app.mail_sync_model.sync( + session_token = inbound_headers["X-Session-Token"], + mongo_conn = mongo_conn, + account_identifier = inbound_data.accountId, + llm = llm, + force_sync = inbound_data.forceSync, + start_date = inbound_data.startDate, + end_date = inbound_data.endDate, + max_count = inbound_data.maxCount + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +@mail_sync_bp.route("/sync", methods = ["POST"]) +@mail_sync_bp.route("/sync/", methods = ["POST"]) +@set_api_version(api_version = "1.0.0") +@read_input(sanitize_headers = False, sanitize_data = False) +@get_session_info(key = "X-Session-Token", session_coro = "get_session") +@log_request_to_mongo( + attr_name = "logs_mongo", + project = constants.PROJECT_NAME, + log_type = constants.MODULE_NAME, + operation = "mailOAuthUrlReqApi", + log_input = True, + log_output = True, + sensitive_keys = ["sessionToken", "X-Session-Token"] +) +@log_chain_to_mongo(attr_name = "logs_mongo") +@should_not_be_under_maintenance(attr_name = "is_under_maintenance") +@validate_input( + header_validator = lambda x: MailSyncRequestHeaders(**x).model_dump(), + data_validator = lambda x: MailSyncRequestData(**x) +) +@handle_cancelled_request() +async def sync_mail( + mode: Literal["background", "bg"] = None, + inbound_headers: dict | MailSyncRequestHeaders = None, + inbound_data: dict | MailSyncRequestData = None, + inbound_files: dict = None, + **kwargs +): + + """ + Use this when the user wants to pull old mails from some mail client (like GMail) and save it to the database for + ready access on the UI. + :param mode: Set it to one of the specified options to make the sync'ing process go to the background. + :param inbound_headers: auto-extracted by the decorators. + :param inbound_data: auto-extracted by the decorators. + :param inbound_files: auto-extracted by the decorators. + :param kwargs: Any number of extra inputs supplied by the decorators. + :return: A standard response structure. + """ + + # If the session token is invalid/expired: + if kwargs.get("session_info") is None: + return ResponseModel( + status_code = StatusCodes.FAILED, + 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, + mongo_conn = current_app.data_mongo, + llm = current_app.llm, + inbound_headers = inbound_headers, + inbound_data = inbound_data + ) + return ResponseModel( + status_code = StatusCodes.OK, + http_code = HttpCodes.ACCEPTED, + message = "your mails are being sync'd in the background" + ) + + # Otherwise we process it right here: + sync_results = await sync_mails( + mongo_conn = current_app.data_mongo, + llm = current_app.llm, + inbound_headers = inbound_headers, + inbound_data = inbound_data + ) + + # Response: + return ResponseModel( + status_code = StatusCodes.FAILED if sync_results.failureCount > 0 else StatusCodes.OK, + http_code = HttpCodes.INTERNAL_SERVER_ERROR if sync_results.failureCount > 0 else HttpCodes.SUCCESS, + message = sync_results.message, + data = { + "totalCount": sync_results.totalCount, + "successCount": sync_results.successCount, + "failureCount": sync_results.failureCount + } + ) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/api/blueprints/mail/oauth_callback.py b/api/blueprints/mail/oauth_callback.py index 3abbccb..8759b46 100644 --- a/api/blueprints/mail/oauth_callback.py +++ b/api/blueprints/mail/oauth_callback.py @@ -108,12 +108,12 @@ def init(blueprint_setup_state): # --------------------------------------------------------------------------------------------------------------------- -async def handle_gmail_callback(): +async def handle_gmail_callback() -> render_template: """ To handle the callbacks from GMail specifically. Refer to the individual comments to check what hap[pens at each step of the process. - :return: A rendered template of the final status of the authorization. + :return: A rendered template (HTML) of the final status of the authorization. """ # Start by assuming failure: diff --git a/api/blueprints/mail/oauth_request.py b/api/blueprints/mail/oauth_request.py index 1465061..59854a8 100644 --- a/api/blueprints/mail/oauth_request.py +++ b/api/blueprints/mail/oauth_request.py @@ -126,7 +126,7 @@ def init(blueprint_setup_state): operation = "mailOAuthUrlReqApi", log_input = True, log_output = True, - sensitive_keys = ["sessionToken"] + sensitive_keys = ["sessionToken", "X-Session-Token"] ) @log_chain_to_mongo(attr_name = "logs_mongo") @should_not_be_under_maintenance(attr_name = "is_under_maintenance") diff --git a/api/blueprints/mail/sync.py b/api/blueprints/mail/sync.py index f4321c2..b0778a6 100644 --- a/api/blueprints/mail/sync.py +++ b/api/blueprints/mail/sync.py @@ -42,10 +42,11 @@ sys.path.append(".") sys.path.append("..") # For using Quart: -from quart import Blueprint, current_app, request +from quart import Blueprint, current_app, g, request # My utils: from utils_v2.string import json +from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.api.codes import StatusCodes, HttpCodes from utils_v2.api.response import ResponseModel from utils_v2.api.async_quart import ( @@ -70,10 +71,20 @@ from shared import constants # Data Models: from models.data.mail.sync import MailSyncRequestHeaders, MailSyncRequestData +from models.data.mail.sync import MailSyncOneResult, MailSyncManyResults + +# To work with datatypes: +from typing import Literal # For asynchronous activities: import asyncio +# To work with LLMs: +from langchain_openai import ChatOpenAI + +# To work with date and time: +import datetime + # ***************************************************************************************************************** # ***** **** @@ -114,7 +125,41 @@ def init(blueprint_setup_state): # --------------------------------------------------------------------------------------------------------------------- +async def sync_mails( + mongo_conn: AsyncMongo, + llm: ChatOpenAI, + inbound_headers: dict, + inbound_data: MailSyncRequestData +) -> MailSyncManyResults: + + """ + A very simple function, but kept separate so that we get the option to switch between running it in the foreground + and running it in the background. + :param mongo_conn: The instance of the database connector to use to sync the mails. + :param llm: The instance of the LLM to use to summarize the mails. + :param inbound_headers: The headers that came in with the request. + :param inbound_data: The data that came in with the request. + :return: The results of the mail-sync'ing attempt. + """ + + # Try to sync the mails: + return await current_app.mail_sync_model.sync( + session_token = inbound_headers["X-Session-Token"], + mongo_conn = mongo_conn, + account_identifier = inbound_data.accountId, + llm = llm, + force_sync = inbound_data.forceSync, + start_date = inbound_data.startDate, + end_date = inbound_data.endDate, + max_count = inbound_data.maxCount + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + @mail_sync_bp.route("/sync", methods = ["POST"]) +@mail_sync_bp.route("/sync/", methods = ["POST"]) @set_api_version(api_version = "1.0.0") @read_input(sanitize_headers = False, sanitize_data = False) @get_session_info(key = "X-Session-Token", session_coro = "get_session") @@ -125,7 +170,7 @@ def init(blueprint_setup_state): operation = "mailOAuthUrlReqApi", log_input = True, log_output = True, - sensitive_keys = ["sessionToken"] + sensitive_keys = ["sessionToken", "X-Session-Token"] ) @log_chain_to_mongo(attr_name = "logs_mongo") @should_not_be_under_maintenance(attr_name = "is_under_maintenance") @@ -135,6 +180,7 @@ def init(blueprint_setup_state): ) @handle_cancelled_request() async def sync_mail( + mode: Literal["background", "bg"] = None, inbound_headers: dict | MailSyncRequestHeaders = None, inbound_data: dict | MailSyncRequestData = None, inbound_files: dict = None, @@ -144,6 +190,7 @@ async def sync_mail( """ Use this when the user wants to pull old mails from some mail client (like GMail) and save it to the database for ready access on the UI. + :param mode: Set it to one of the specified options to make the sync'ing process go to the background. :param inbound_headers: auto-extracted by the decorators. :param inbound_data: auto-extracted by the decorators. :param inbound_files: auto-extracted by the decorators. @@ -151,11 +198,6 @@ async def sync_mail( :return: A standard response structure. """ - # ┏┓ - # ┃┃┏┓┏┓┏┓┏┓┏┓┏┏┓┏┏ - # ┣┛┛ ┗ ┣┛┛ ┗┛┗┗ ┛┛ - # ┛ - # If the session token is invalid/expired: if kwargs.get("session_info") is None: return ResponseModel( @@ -163,76 +205,43 @@ async def sync_mail( http_code = HttpCodes.UNAUTHORIZED ) - # Start by assuming failure: - mails_count = 0 + # 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, + mongo_conn = current_app.data_mongo, + llm = current_app.llm, + inbound_headers = inbound_headers, + inbound_data = inbound_data + ) + return ResponseModel( + status_code = StatusCodes.OK, + http_code = HttpCodes.ACCEPTED, + message = "your mails are being sync'd in the background" + ) - # Make a user identifier from the session info: - user_auth = await current_app.mail_oauth_model.get_token( + # Otherwise we process it right here: + sync_results = await sync_mails( mongo_conn = current_app.data_mongo, - serviceType = "email", - user = kwargs["session_info"], + llm = current_app.llm, + inbound_headers = inbound_headers, + inbound_data = inbound_data ) - print("MAIL TOKEN(S):", json.to_string(user_auth, default = str)) - - if not user_auth: return ResponseModel( - status_code = StatusCodes.FAILED, - http_code = HttpCodes.NOT_FOUND, - message = "user's email not connected" - ) - - # ┏┓┳┳┓ •┓ - # ┃┓┃┃┃┏┓┓┃ - # ┗┛┛ ┗┗┻┗┗ - - if user_auth["client"] == "gmail": - - # Load the tokens into an object: - user_tokens = GoogleAuthTokens(**user_auth["token"]) - - # Try refreshing the tokens: - tokens_refreshed = await user_tokens.arefresh( - http_client = current_app.http_client, - client_id = current_app.gmail_client.client_id, - client_secret = current_app.gmail_client.client_secret, - force_refresh = False - ) - - # Update the token in the database if needed: - if tokens_refreshed: await current_app.mail_oauth_model.set_token( - db_conn = current_app.sql_writer, - mongo_conn = current_app.data_mongo, - user_identifier = user_auth["_id"], - token = user_tokens.model_dump(), - session_token = inbound_headers["X-Session-Token"] - ) - - # Now we try to sync the mails: - mails_count = await current_app.mail_sync_model.sync( - mongo_conn = current_app.data_mongo, - user_info = kwargs["session_info"], - mail_client = current_app.gmail_client, - tokens = user_tokens, - llm = None, - force_sync = False, - start_date = inbound_data.startDate, - end_date = inbound_data.endDate, - max_count = inbound_data.maxCount - ) - - # ┳┓ - # ┣┫┏┓┏┏┓┏┓┏┓┏┏┓ - # ┛┗┗ ┛┣┛┗┛┛┗┛┗ - # ┛ - - # Done here: + # Response: return ResponseModel( - status_code = StatusCodes.OK if mails_count else StatusCodes.FAILED, - message = f"{mails_count} mail(s) sync'd" + status_code = StatusCodes.FAILED if sync_results.failureCount > 0 else StatusCodes.OK, + http_code = HttpCodes.INTERNAL_SERVER_ERROR if sync_results.failureCount > 0 else HttpCodes.SUCCESS, + message = sync_results.message, + data = { + "totalCount": sync_results.totalCount, + "successCount": sync_results.successCount, + "failureCount": sync_results.failureCount + } ) diff --git a/api/main.py b/api/main.py index 45c11a4..78aa166 100644 --- a/api/main.py +++ b/api/main.py @@ -52,7 +52,7 @@ from shared import constants from utils_v2.string import json from utils_v2.api import async_quart from utils_v2.date_time import date_time -from utils_v2.database.async_mongo_v2 import AsyncMongo +from utils_v2.database.async_mongo_v2 import AsyncMongo, AsyncMongoStorage from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache from utils_v2.serialization.json_serializer import JSONSerializer @@ -72,7 +72,7 @@ from utils_v2.goog.gmail.gmail_client import AsyncGMailClient # Behaviour Models: from models.behaviour.mail.oauth_v2 import MailOAuthModel -from models.behaviour.mail.sync import MailSyncModel +from models.behaviour.mail.sync_v2 import MailSyncModel # To make REST API calls: import httpx @@ -90,6 +90,9 @@ from api.blueprints.test.callback import test_callback_bp # All the helpers: from api.helpers.user import session +# To work with LLMs: +from langchain_openai import ChatOpenAI + # ***************************************************************************************************************** # ***** **** @@ -278,6 +281,13 @@ async def app_startup(**kwargs): debug = enable_debugging ) await current_app.data_mongo.connect() + current_app.files_mongo = AsyncMongo( + connection_string = script_cred["mongoDb"]["files"]["connectionString"], + database_name = script_cred["mongoDb"]["files"]["dbName"], + max_connections = script_cred["mongoDb"]["files"]["poolSize"], + debug = enable_debugging + ) + await current_app.data_mongo.connect() # ┳ ┓ ┳┳┓ ┓ ┓ # ┃┏┓╋┏┓┏┓┏┓┏┓┃ ┃┃┃┏┓┏┫┏┓┃┏ @@ -315,6 +325,12 @@ async def app_startup(**kwargs): debug_only_errors = False ) + # For LLMs: + current_app.llm = ChatOpenAI( + model = script_cred["openAi"]["model"], + openai_api_key = script_cred["openAi"]["openai_api_key"] + ) + # Pick the important stuff: current_app.whitelisted_ips = current_app.script_data["whitelistedIps"] diff --git a/models/behaviour/mail/sync_v2.py b/models/behaviour/mail/sync_v2.py new file mode 100644 index 0000000..591b458 --- /dev/null +++ b/models/behaviour/mail/sync_v2.py @@ -0,0 +1,542 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + tuesday, 3rd Dec., 2024 + + OBJECTIVE: + + From here we sync all mails between the mail client's server and TheCAOffice's database. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + +# ***************************************************************************************************************** +# ***** **** +# *** 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 + +# Mail Clients: +from utils_v2.goog.gmail.gmail_client import AsyncGMailClient +from utils_v2.goog.models.data.auth_tokens import GoogleAuthTokens + +# Base model: +from models.behaviour.base import BaseModel + +# Data models: +from models.data.mail.sync import MailSyncOneResult, MailSyncManyResults + +# To work with MongoDB: +from bson import ObjectId +from pymongo import InsertOne, UpdateOne, ReplaceOne + +# To work with LLMs: +from langchain_openai import ChatOpenAI +from langchain_core.prompts import ChatPromptTemplate + +# To work with datatypes: +from typing import Literal, List, Dict, Any + +# To make deep-copies: +import copy + +# To work with base-64 encoding: +import base64 + +# To work with date and time: +import datetime + +# For asynchronous activities: +import asyncio + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class MailSyncModel(BaseModel): + + # For MongoDB: + AUTH_COLLECTION = "_authTokens" + MAIL_COLLECTION = "_messages" + + # For AI Magic through LLMs: + prompt_template = ChatPromptTemplate.from_messages([ + ( + "system", + "You're a mail summary expert that summarizes mails in 150 chars or less. HIDE SENSITIVE INFO (LIKE OTPs) FROM THE SUMMARY." + ), + ( + "user", + "Please summarize this mail: \"\"\"{mail}\"\"\"" + ) + ]) + + @staticmethod + async def __save_one_attachment( + self, + session_token: str, + attachment: Dict[str, Any], + attachment_tags: List[str], + attachment_metadata: dict, + retry_count: int = 1, + retry_delay: int = 1, + backoff_multiplier: float = 1.1 + ) -> Dict[str, Any]: + + """ + Saves one attachment and generates a URL that can be later used to retrieve it. + :param session_token: The session token of the uer who is trying to upload this file. + :param attachment: The JSON that describes the attachment. + :param attachment_tags: Any tags to put on the file for easy search later. + :param attachment_metadata: Any metadata to put on the file for easy search later. + :param retry_count: How many max. retries to do in case of failure. + :param retry_delay: The interval between the delays. + :param backoff_multiplier: By what rate the delay between 2 attempts must change. + :return: The JSON that describes the same attachment, except that the payload's data is replaced by the id and + url of where to find the attachment. + """ + + # Make a deep-copy of the attachment JSON, + # and process the payload in advance: + attachment_copy = copy.deepcopy(attachment) + attachment_payload = attachment_copy.pop("payload").encode() + if attachment_copy.pop("contentTransferEncoding", "?").strip().lower() == "base64": + attachment_payload = base64.b64decode(attachment_payload) + + # Start by assuming failure, + # and retry as many times as asked: + attachment_copy["id"] = None + attachment_copy["url"] = None + for _ in range(retry_count): + + # Make the upload: + api_response = await current_app.http_client.post( + url = current_app.script_data["fileUpload"]["url"], + headers = { + "X-Session-Token": session_token, + "X-File-Name": attachment["filename"], + "X-File-Private": "false", + "X-File-Tags": json.to_string(attachment_tags, no_space = True), + "X-File-Metadata": json.to_string(attachment_metadata, no_space = True) + }, + data = attachment_payload + ) + + # If the upload was successful: + if api_response.status_code in [200]: + api_data = api_response.json()["data"] + attachment_copy["id"] = api_data["id"] + attachment_copy["url"] = api_data["url"] + break + + # Done here: + return attachment_copy + + async def __save_many_attachments( + self, + session_token: str, + attachments: List[Dict[str, Any]], + attachment_tags: List[str], + attachment_metadata: dict, + retry_count: int = 1, + retry_delay:int = 1, + backoff_multiplier: float = 1.1 + ) -> List[Dict[str, Any]]: + + """ + Saves all the attachments received in the mail (whether inline or otherwise) and makes them available through + simple download URLs. + :param session_token: The session token of the uer who is trying to upload this file. + :param attachments: The JSON that describes the attachments. + :param attachment_tags: Any tags to put on the file for easy search later. + :param attachment_metadata: Any metadata to put on the file for easy search later. + :param retry_count: How many max. retries to do in case of failure. + :param retry_delay: The interval between the delays. + :param backoff_multiplier: By what rate the delay between 2 attempts must change. + :return: The JSON that describes the same attachments, except that the payload's data is replaced by the id and + url of where to find each attachment. + """ + + # Create and fire all the tasks + # needed to save the files: + tasks = [ + self.__save_one_attachment( + session_token = session_token, + attachment = attachment, + attachment_tags = attachment_tags, + attachment_metadata = attachment_metadata, + retry_count = retry_count, + retry_delay = retry_delay, + backoff_multiplier = backoff_multiplier + ) for attachment in attachments + ] + uploaded_attachments = await asyncio.gather(*tasks) + + # Done here: + return uploaded_attachments + + async def __sync_one_gmail( + self, + session_token: str, + mongo_conn: AsyncMongo, + mail_client: AsyncGMailClient, + tokens: GoogleAuthTokens, + message_id: str, + llm: ChatOpenAI = None, + force_sync: bool = False + ) -> MailSyncOneResult: + + """ + Sync on mail from GMail. + :param session_token: The session token of the uer who is trying to upload this file. + :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 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. + :param force_sync: Whether you would like to forcefully re-sync the mail even if it is already present in the + database. + :return: + """ + + # 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_record = await mongo_conn.find_one( + collection = self.MAIL_COLLECTION, + filter = mongo_conn.dict_to_dot_notation({ + "payload": { + "messageId": message_id + } + }), + projection = { + "_id": False, + "readTs": "payload.readTs" + } + ) + if mail_record: + sync_result.success = True + sync_result.message = f"gmail message '{message_id}' already sync'd on '{mail_record['readTs']} (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 = tokens, + message_id = message_id, + return_raw = False + ) + if not client_response.success: + sync_result.message = f"gmail (messageId: '{message_id}'): {client_response.message}" + + # We upload the attachments: + client_response.data["attachments"] = await self.__save_many_attachments( + session_token = session_token, + attachments = client_response.data["attachments"], + attachment_tags = [ + "email", + "gmail", + client_response.data["from"][0]["name"], + client_response.data["from"][0]["email"], + tokens.email, + ], + attachment_metadata = { + "project": "tcaoff", + "serviceType": "email", + "client": "gmail", + "from": client_response.data["from"][0]["email"], + "to": tokens.email + }, + retry_count = 3 + ) + + # 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 + else: client_response.data["isInbox"] = False + + # If an LLM is given, we add an AI summary: + llm_json = None + if llm: + llm_response = response = await llm.ainvoke( + self.prompt_template.invoke({ + "mail": client_response.data["unformattedText"] + }) + ) + llm_json = { + "snippet": llm_response.content, + "usage": llm_response.usage_metadata + } + client_response.data["aiSnippet"] = llm_json + + # Done here: + sync_result.success = True + sync_result.mailMessage = client_response.data + return sync_result + + async def __sync_many_gmail( + self, + session_token: str, + mongo_conn: AsyncMongo, + account_identifier: ObjectId, + mail_client: AsyncGMailClient, + tokens: GoogleAuthTokens, + llm: ChatOpenAI = None, + force_sync: bool = False, + start_date: datetime.datetime = None, + end_date: datetime.datetime = None, + max_count: int = 100 + ) -> MailSyncManyResults: + + """ + 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 mongo_conn: The instance of the connection to the database to use. + :param account_identifier: 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. + :param start_date: The starting date (inclusive) from when to sync the mails. + :param end_date: The ending date (inclusive) from when to sync the mails. + :param max_count: The max. no. of mails to sync. + :return: The result of the sync'ing. + """ + + # Start by assuming failure: + sync_results = MailSyncManyResults() + + # Refresh the tokens (if needed): + tokens_refreshed = await 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, + account_identifier = account_identifier, + email_id = tokens.email, + token = tokens, + session_token = session_token + ) + + # Let's build the query: + 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 = tokens, + max_count = max_count, + query = query_string + ) + if not client_response.success: + sync_results["message"] = f"gmail: {client_response.message}" + return sync_results + messages_list = client_response.data["messages"] + + # Now, for every mail in the list, we fetch the mail and note the results: + tasks = [ + self.__sync_one_gmail( + session_token = session_token, + mongo_conn = mongo_conn, + mail_client = mail_client, + tokens = 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: mongo_operations.append(ReplaceOne( + filter = { + "serviceType": "email", + "$or": [ + { + "client": "gmail", + "payload.messageId": result.mailMessage["messageId"] + } + ] + }, + replacement = { + "version": "1.0.0", + "accountId": ObjectId(account_identifier), + "serviceType": "email", + "client": "gmail", + "payload": result.mailMessage + }, + upsert = True + )) + + # Make the bulk write: + if mongo_operations: + mongo_count = await mongo_conn.bulk_write( + collection = self.MAIL_COLLECTION, + requests = mongo_operations + ) + + # Done here: + sync_results.message = f"{sync_results.successCount}/{sync_results.totalCount} mail(s) sync'd from gmail" + return sync_results + + async def sync( + self, + session_token: str, + mongo_conn: AsyncMongo, + account_identifier: ObjectId, + llm: ChatOpenAI = None, + force_sync: bool = False, + start_date: datetime.datetime = None, + end_date: datetime.datetime = None, + max_count: int = 100 + ) -> MailSyncManyResults: + + """ + 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 mongo_conn: The instance of the connection to the database to use. + :param account_identifier: 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 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. + :param start_date: The starting date (inclusive) from when to sync the mails. + :param end_date: The ending date (inclusive) from when to sync the mails. + :param max_count: The max. no. of mails to sync. + :return: The result of the sync'ing. + """ + + # Start by assuming failure: + sync_results = MailSyncManyResults() + + # ┏┓ ┓ ┏┳┓ ┓ + # ┣ ┏┓╋┏┣┓ ┃ ┏┓┃┏┏┓┏┓┏ + # ┻ ┗ ┗┗┛┗ ┻ ┗┛┛┗┗ ┛┗┛ + + # We first load the authorization tokens: + auth_json = await current_app.mail_oauth_model.get_token( + mongo_conn = mongo_conn, + account_identifier = account_identifier, + ) + + # If we failed to load the authorization tokens: + if not auth_json: + sync_results.message = f"no such account identifier '{account_identifier}'" + return sync_results + + # ┏┓ ┏┓┳┳┓ •┓ + # ┣ ┏┓┏┓ ┃┓┃┃┃┏┓┓┃ + # ┻ ┗┛┛ ┗┛┛ ┗┗┻┗┗ + + if auth_json["client"] == "gmail": + return await self.__sync_many_gmail( + session_token = session_token, + mongo_conn = mongo_conn, + account_identifier = account_identifier, + mail_client = current_app.gmail_client, + tokens = GoogleAuthTokens(**auth_json["token"]), + llm = llm, + force_sync = force_sync, + start_date = start_date, + end_date = end_date, + max_count = max_count + ) + + # ┳ ┓• ┓ ┏┓┓• + # ┃┏┓┓┏┏┓┃┓┏┫ ┃ ┃┓┏┓┏┓╋ + # ┻┛┗┗┛┗┻┗┗┗┻ ┗┛┗┗┗ ┛┗┗ + + # 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']}'" + return sync_results + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/models/data/mail/sync.py b/models/data/mail/sync.py index 6bad524..ee3965c 100644 --- a/models/data/mail/sync.py +++ b/models/data/mail/sync.py @@ -101,9 +101,16 @@ class MailSyncRequestHeaders(BaseModel): class MailSyncRequestData(BaseModel): + accountId: str = Field( + description = "the account identifier (Mongo ObjectId) granted by 'MailOAuthModel.get_account_identifier'", + frozen = True + ) + maxCount: int = Field( description = "the max. no. of e-mails to sync at a given time", default = 100, + ge = 1, + le = 100, frozen = True ) @@ -119,6 +126,11 @@ class MailSyncRequestData(BaseModel): frozen = True ) + forceSync: bool = Field( + description = "use this to forcefully re-sync mails when you need to overwrite existing data in mongodb", + default = False + ) + # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ @@ -141,6 +153,69 @@ class MailSyncRequestData(BaseModel): return value +# --------------------------------------------------------------------------------------------------------------------- + + +class MailSyncOneResult(BaseModel): + + success: bool = Field( + description = "whether, or not, the mail was successfully sync'd", + default = False + ) + + message: str | None = Field( + description = "a brief message to summarize the result of the process", + default = None + ) + + mailMessage: dict | None = Field( + description = "the actual data of the mail; can be null in a successful process if the mail is already sync'd", + default = None + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + + +# --------------------------------------------------------------------------------------------------------------------- + + +class MailSyncManyResults(BaseModel): + + totalCount: int = Field( + description = "the total no. of mails that were to be sync'd", + default = 0 + ) + + successCount: int = Field( + description = "the no. of mails that were successfully sync'd", + default = 0 + ) + + failureCount: int = Field( + description = "the no. of mails that were successfully sync'd", + default = 0 + ) + + message: str = Field( + description = "a brief message to summarize the results of the process", + default = None + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + + # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** diff --git a/utils_v2/mail/mail_parser.py b/utils_v2/mail/mail_parser.py index 2f01cec..8bda8a3 100644 --- a/utils_v2/mail/mail_parser.py +++ b/utils_v2/mail/mail_parser.py @@ -156,16 +156,19 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]: # Put everything together: return { "ts": date_time.to_timezone(parsed_mail.date, timezone = date_time.TIMEZONE_UTC), + "readTs": date_time.get_current_utc_date_time(as_string = False), "headers": parsed_mail.headers, "from": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers["From"]], "to": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers["To"]], "cc": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers.get("Cc", [])], "bcc": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers.get("Bcc", [])], + "subject": parsed_mail.headers["Subject"], "text": parsed_mail.text_plain, "html": parsed_mail.text_html, "parts": parts, "unformattedText": "\n".join(unformatted_text), "attachments": message_attachments, + "isInbox": None }