(20241203) Mail sync'ing started. You can now hit a dedicated API endpoint to trigger a mail-sync process (in the background if you want) and save the mails to your database.

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/<mode>", 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

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:

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")

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/<mode>", 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
+        }
     )
 
 

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"]
 

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

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                                                    ***

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
     }