diff --git a/api/blueprints/ai/llm/invoke.py b/api/blueprints/ai/llm/invoke.py index 7c8d1d6..3f8b57d 100644 --- a/api/blueprints/ai/llm/invoke.py +++ b/api/blueprints/ai/llm/invoke.py @@ -63,8 +63,9 @@ from utils_v2.api.async_quart import ( from shared import constants # Data Models: -from models.data.api.ai.llm import LLMRequestHeaders, LLMInput -from models.data.core.user import CoreUserInfoModel +from models.api.ai.llm import LLMRequestHeaders +from models.core.ai.llm import LLMInput +from models.core.user import CoreUserInfoModel # For asynchronous activities: import asyncio diff --git a/api/blueprints/mail/oauth_callback.py b/api/blueprints/mail/oauth_callback.py index 8780935..7958726 100644 --- a/api/blueprints/mail/oauth_callback.py +++ b/api/blueprints/mail/oauth_callback.py @@ -63,6 +63,9 @@ from utils_v2.api.async_quart import ( # GMail-related utils: from utils_v2.goog.gmail.gmail_client import SCOPES_GMAIL_MAIL_MANAGEMENT +# Data Models: +from models.core.auth_token import CoreAuthTokenModel + # Common: from shared import constants @@ -113,7 +116,7 @@ def init(blueprint_setup_state): api_version = "1.0.0", project = constants.PROJECT_NAME, log_type = constants.MODULE_NAME, - operation = "gmailCllbk", + operation = "gmailClbk", log_input = 2, log_output = 1, sensitive_keys = ["sessionToken", "X-Session-Token"] @@ -156,7 +159,7 @@ async def handle_gmail_callback() -> render_template: # Here's where we do the checking of the e-mails, # if they don't match, we reject the authorization: - auth_token = await current_app.mail_oauth_model.get_token( + auth_token = await current_app.mail_controller.get_token( mongo_conn = current_app.data_mongo, token_id = g.inbound_data["state"] ) @@ -211,7 +214,7 @@ async def handle_gmail_callback() -> render_template: auth_token.clientUserId = google_tokens.client_user_id auth_token.token = google_tokens.model_dump() auth_token.status = "active" - tokens_saved = await current_app.mail_oauth_model.set_token( + tokens_saved = await current_app.mail_controller.set_token( db_conn = current_app.sql_writer, mongo_conn = current_app.data_mongo, session_token = g.inbound_headers.get("X-Session-Token"), diff --git a/api/blueprints/mail/oauth_request.py b/api/blueprints/mail/oauth_request.py index 287696f..058471c 100644 --- a/api/blueprints/mail/oauth_request.py +++ b/api/blueprints/mail/oauth_request.py @@ -67,11 +67,11 @@ from utils_v2.goog.gmail.gmail_client import SCOPES_GMAIL_MAIL_MANAGEMENT from shared import constants # Data Models: -from models.data.api.mail.oauth import ( +from models.api.mail.oauth import ( OAuthMailAuthorizationRequestHeaders, OAuthMailAuthorizationRequestData ) -from models.data.core.auth_token import CoreAuthTokenModel +from models.core.auth_token import CoreAuthTokenModel # For asynchronous activities: import asyncio @@ -175,7 +175,7 @@ async def request_oauth_authorization_url( # ┗┛┗ ┛┗┗ ┛ ┗┻┗┗ ┻ ┗┛┛┗┗ ┛┗ ┻┗┻ # Make a user identifier from the session info: - token_id = await current_app.mail_oauth_model.get_token_id( + token_id = await current_app.mail_controller.get_token_id( db_conn = current_app.sql_writer, mongo_conn = current_app.data_mongo, auth_token = CoreAuthTokenModel( diff --git a/api/blueprints/mail/sync.py b/api/blueprints/mail/sync_old.py similarity index 100% rename from api/blueprints/mail/sync.py rename to api/blueprints/mail/sync_old.py diff --git a/api/blueprints/mail/sync_v2.py b/api/blueprints/mail/sync_v2.py new file mode 100644 index 0000000..0363155 --- /dev/null +++ b/api/blueprints/mail/sync_v2.py @@ -0,0 +1,256 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Monday, 2nd Dec., 2024 + + 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 UI at any 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.core.user import CoreUserInfoModel +from models.api.mail.sync import MailSyncRequestHeaders, MailSyncRequestData +from models.api.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( + user_info: CoreUserInfoModel, + 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 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. + :return: The results of the mail-sync'ing attempt. + """ + + # Try to sync the mails: + return await current_app.mail_api_model.sync( + db_conn = current_app.sql_writer, + mongo_conn = current_app.data_mongo, + user_info = user_info, + token_id = inbound_data.tokenId, + 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"], + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +@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 = "mailSyncApi", + 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, + user_info = CoreUserInfoModel(**kwargs["session_info"]), + 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( + user_info = CoreUserInfoModel(**kwargs["session_info"]), + 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/tech/chat_alerts.py b/api/blueprints/tech/chat_alerts.py index 649cde3..ba3d35d 100644 --- a/api/blueprints/tech/chat_alerts.py +++ b/api/blueprints/tech/chat_alerts.py @@ -63,7 +63,7 @@ from utils_v2.api.async_quart import ( ) # Data models: -from models.data.api.tech.alerts import ( +from models.api.tech.alerts import ( ChatAlertRequestHeaders, ChatAlertRequestData ) diff --git a/api/helpers/user/session.py b/api/helpers/user/session.py index 2a88a61..c93243d 100644 --- a/api/helpers/user/session.py +++ b/api/helpers/user/session.py @@ -38,7 +38,7 @@ sys.path.append("..") from quart import current_app # The data model: -from models.data.core.user import CoreUserInfoModel +from models.core.user import CoreUserInfoModel # ***************************************************************************************************************** diff --git a/api/main.py b/api/main.py index 3a00e6e..a1b1544 100644 --- a/api/main.py +++ b/api/main.py @@ -70,13 +70,20 @@ from utils_v2.api.async_quart import ( # GMail-related utils: from utils_v2.goog.gmail.gmail_client import AsyncGMailClient -# Behaviour Models: -from models.behaviour.mail.oauth_v3 import MailOAuthModel -from models.behaviour.mail.sync_v3 import MailSyncModel -from models.behaviour.mail.retrieve import MailRetrieveModel -from models.behaviour.sms.auth_v2 import SMSAuthModel -from models.behaviour.sms.send import SMSSendModel -from models.behaviour.ai.llm.open_ai import LLMOpenAI +# Core Controller Models: +from controllers.core.message import MessageController +from controllers.core.auth_token import AuthTokenController +from controllers.core.ai.llm import LLMController + +# API Controller Models: +from controllers.api.mail import MailController + +# # Old Behaviour Models: +# from controllers.mail.oauth_v3 import MailOAuthModel +# from controllers.mail.sync_v3 import MailSyncModel +# from controllers.mail.retrieve import MailRetrieveModel +# from controllers.sms.auth_v2 import SMSAuthModel +# from controllers.sms.send import SMSSendModel # To make REST API calls: import httpx @@ -87,13 +94,13 @@ from icecream import IceCreamDebugger # All the blueprints: from api.blueprints.mail.oauth_request import mail_oauth_bp from api.blueprints.mail.oauth_callback import mail_callback_bp -from api.blueprints.mail.sync import mail_sync_bp -from api.blueprints.mail.list import mail_list_bp -from api.blueprints.mail.retrieve import mail_retrieve_bp -from api.blueprints.sms.auth import sms_auth_bp -from api.blueprints.sms.send import sms_send_bp -from api.blueprints.chat.auth import chat_auth_bp -from api.blueprints.chat.webhook import chat_webhook_bp +# from api.blueprints.mail.sync_v2 import mail_sync_bp +# from api.blueprints.mail.list import mail_list_bp +# from api.blueprints.mail.retrieve import mail_retrieve_bp +# from api.blueprints.sms.auth import sms_auth_bp +# from api.blueprints.sms.send import sms_send_bp +# from api.blueprints.chat.auth import chat_auth_bp +# from api.blueprints.chat.webhook import chat_webhook_bp from api.blueprints.tech.chat_alerts import tech_chat_alert_bp from api.blueprints.test.callback import test_callback_bp from api.blueprints.ai.llm.invoke import llm_invoke_bp @@ -126,13 +133,13 @@ app = Quart(__name__, template_folder = r"../views") app = cors(app) app.register_blueprint(mail_oauth_bp, url_prefix = f"/{MODULE_BASE}/mail") app.register_blueprint(mail_callback_bp, url_prefix = f"/{MODULE_BASE}/mail") -app.register_blueprint(mail_sync_bp, url_prefix = f"/{MODULE_BASE}/mail") -app.register_blueprint(mail_list_bp, url_prefix = f"/{MODULE_BASE}/mail") -app.register_blueprint(mail_retrieve_bp, url_prefix = f"/{MODULE_BASE}/mail") -app.register_blueprint(sms_auth_bp, url_prefix = f"/{MODULE_BASE}/sms") -app.register_blueprint(sms_send_bp, url_prefix = f"/{MODULE_BASE}/sms") -app.register_blueprint(chat_auth_bp, url_prefix = f"/{MODULE_BASE}/chat") -app.register_blueprint(chat_webhook_bp, url_prefix = f"/{MODULE_BASE}/chat") +# app.register_blueprint(mail_sync_bp, url_prefix = f"/{MODULE_BASE}/mail") +# app.register_blueprint(mail_list_bp, url_prefix = f"/{MODULE_BASE}/mail") +# app.register_blueprint(mail_retrieve_bp, url_prefix = f"/{MODULE_BASE}/mail") +# app.register_blueprint(sms_auth_bp, url_prefix = f"/{MODULE_BASE}/sms") +# app.register_blueprint(sms_send_bp, url_prefix = f"/{MODULE_BASE}/sms") +# app.register_blueprint(chat_auth_bp, url_prefix = f"/{MODULE_BASE}/chat") +# app.register_blueprint(chat_webhook_bp, url_prefix = f"/{MODULE_BASE}/chat") app.register_blueprint(tech_chat_alert_bp, url_prefix = f"/{MODULE_BASE}/tech/alert") app.register_blueprint(test_callback_bp, url_prefix = f"/{MODULE_BASE}/test") app.register_blueprint(llm_invoke_bp, url_prefix = f"/{MODULE_BASE}/ai") @@ -318,53 +325,32 @@ async def app_startup(**kwargs): current_app.printer("MongoDB ready.") - # ┳ ┓ ┳┳┓ ┓ ┓ - # ┃┏┓╋┏┓┏┓┏┓┏┓┃ ┃┃┃┏┓┏┫┏┓┃┏ - # ┻┛┗┗┗ ┛ ┛┗┗┻┗ ┛ ┗┗┛┗┻┗ ┗┛ + # ┏┓ ┳┳┓ ┓ ┓ + # ┃ ┏┓┏┓┏┓ ┃┃┃┏┓┏┫┏┓┃┏ + # ┗┛┗┛┛ ┗ ┛ ┗┗┛┗┻┗ ┗┛ - current_app.mail_oauth_model = MailOAuthModel( + current_app.core_message_controller = MessageController( cache = current_app.module_cache, alert_url = current_app.script_data["alerts"]["url"], http_client = current_app.http_client, debug = enable_debugging, - debug_prefix = "Mail-OAuth | ", + debug_prefix = "Message (CM) | ", debug_only_errors = True ) - current_app.mail_sync_model = MailSyncModel( + current_app.core_auth_token_controller = AuthTokenController( cache = current_app.module_cache, alert_url = current_app.script_data["alerts"]["url"], http_client = current_app.http_client, debug = enable_debugging, - debug_prefix = "Mail-Sync | ", - debug_only_errors = True - ) - current_app.mail_retrieve_model = MailRetrieveModel( - cache = current_app.module_cache, - alert_url = current_app.script_data["alerts"]["url"], - http_client = current_app.http_client, - debug = enable_debugging, - debug_prefix = "Mail-Retr | ", + debug_prefix = "Message (CM) | ", debug_only_errors = True ) - current_app.sms_auth_model = SMSAuthModel( - cache = current_app.module_cache, - alert_url = current_app.script_data["alerts"]["url"], - http_client = current_app.http_client, - debug = enable_debugging, - debug_prefix = "SMS-Auth | ", - debug_only_errors = True - ) - current_app.sms_send_model = SMSSendModel( - cache = current_app.module_cache, - alert_url = current_app.script_data["alerts"]["url"], - http_client = current_app.http_client, - debug = enable_debugging, - debug_prefix = "SMS-Send | ", - debug_only_errors = True - ) + # ┏┓┏┓┳ ┏┓ ┓┓ + # ┣┫┃┃┃ ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ + # ┛┗┣┛┻ ┗┛┗┛┛┗┗┛ ┗┛┗┗┗ ┛ ┛ - current_app.printer("Internal models ready.") + current_app.mail_controller = MailController() # ┏┓ ┓ ┏┓┓• # ┃ ┏┓┏┓┏┓┏┓┏╋┏┓┏┓┏ ┏┓┏┓┏┫ ┃ ┃┓┏┓┏┓╋┏ @@ -389,7 +375,7 @@ async def app_startup(**kwargs): # ┛ # For LLMs: - current_app.llm = LLMOpenAI( + current_app.llm = LLMController( llm_creds = { "model": script_cred["openAi"]["model"], "openai_api_key": script_cred["openAi"]["openai_api_key"] diff --git a/models/behaviour/__init__.py b/controllers/__init__.py similarity index 100% rename from models/behaviour/__init__.py rename to controllers/__init__.py diff --git a/models/behaviour/ai/__init__.py b/controllers/api/__init__.py similarity index 100% rename from models/behaviour/ai/__init__.py rename to controllers/api/__init__.py diff --git a/models/behaviour/mail/sync_v3.py b/controllers/api/mail.py similarity index 53% rename from models/behaviour/mail/sync_v3.py rename to controllers/api/mail.py index bbaf130..cf486c2 100644 --- a/models/behaviour/mail/sync_v3.py +++ b/controllers/api/mail.py @@ -6,12 +6,11 @@ DATE: - ORIGINAL: Tuesday, 3rd Dec., 2024 - UPGRADED: Monday, 9th Dec., 2024 + Thursday, 12th Dec., 2024 OBJECTIVE: - From here we sync all mails between the mail client's server and our internal database. + To handle all auth-tokens from one place. REFERENCES: @@ -32,10 +31,6 @@ # To make sibling directories accessible for imports: import sys - -from models.data.core.auth_token import CoreAuthTokenModel -from models.data.core.message import CoreMessageModel - sys.path.append(".") sys.path.append("..") @@ -48,33 +43,32 @@ 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 +# Base model: +from controllers.base import BaseModel + +# Data models: +from models.core.user import CoreUserInfoModel +from models.core.auth_token import CoreAuthTokenModel +from models.core.message import CoreMessageModel +from models.api.mail.sync import MailSyncOneResult, MailSyncManyResults + # 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.api.mail.sync import MailSyncOneResult, MailSyncManyResults -from models.data.core.user import CoreUserInfoModel - # To work with MongoDB: from bson import ObjectId from pymongo import InsertOne, UpdateOne, ReplaceOne # To work with LLMs: -from models.behaviour.ai.llm.open_ai import LLMOpenAI -from models.data.api.ai.llm import LLMInput +from controllers.core.ai.llm import LLMController +from models.core.ai.llm import LLMInput, LLMOutput # 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 parse the HTML content in the mail: +from bs4 import BeautifulSoup # To work with date and time: import datetime @@ -120,11 +114,11 @@ import asyncio # ***************************************************************************************************************** -class MailSyncModel(BaseModel): +class MailController: - # For MongoDB: - AUTH_COLLECTION = "_authTokens" - MAIL_COLLECTION = "_messages" + # ┏┓┓ ┓┏ + # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ + # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ # For AI Magic through LLMs: PROMPT_TEMPLATE = [ @@ -138,167 +132,185 @@ class MailSyncModel(BaseModel): } ] - # ┏┓ ┓ - # ┣┫╋╋┏┓┏┣┓┏┳┓┏┓┏┓╋┏ - # ┛┗┗┗┗┻┗┛┗┛┗┗┗ ┛┗┗┛ + # ┓┏ ┓ + # ┣┫┏┓┃┏┓┏┓┏┓┏ + # ┛┗┗ ┗┣┛┗ ┛ ┛ + # ┛ + + def extract_plaintext_parts( + self, + payload: dict + ) -> List[str]: + + # Start with just a holder: + text_parts = [] + + # If a direct text/plain part occurs, + # we just add it to the list: + if ( + payload["contentMainType"] == "text" and + payload["contentSubType"] == "plain" + ): + text_parts.append(payload["payload"]) + + # If a direct text/html part occurs, + # we just add it to the list: + if ( + payload["contentMainType"] == "text" and + payload["contentSubType"] == "html" + ): + html_parser = BeautifulSoup(payload["payload"], "html.parser") + text_parts.append(html_parser.get_text()) + + # If a multipart/alternative part occurs, + # we pick just the ready plaintext part: + if ( + payload["contentMainType"] == "multipart" and + payload["contentSubType"] == "alternative" + ): + for part in payload["payload"]: + if part["contentSubType"] == "plain": + text_parts.append(part["payload"]) + + # If a multipart/mixed or multipart/related part occurs, + # we use recursion to look for plaintext parts nested inside: + if ( + payload["contentMainType"] == "multipart" and + ( + payload["contentSubType"] == "mixed" or + payload["contentSubType"] == "related" + ) + ): + for part in payload["payload"]: + text_parts += self.extract_plaintext_parts(payload = part) + + # Done here: + return text_parts + + async def summarize_mail_with_ai( + self, + mongo_conn: AsyncMongo, + user_info: CoreUserInfoModel, + llm: LLMController, + message: CoreMessageModel + ) -> LLMOutput: + + # Extract the text from the message here: + text_parts = self.extract_plaintext_parts(payload = message.message["payload"]) + text = "\n".join(text_parts) + + # Invoke the LLM and return the response: + return await llm.invoke( + mongo_conn = mongo_conn, + user_info = user_info, + llm_input = LLMInput( + messages = self.PROMPT_TEMPLATE + [ + { + "role": "human", + "content": f"Please summarize this mail: \"\"\"{text}\"\"\"" + } + ] + ) + ) + + # ┏┓┏┓ ┓ ┏┓ ┏┓ + # ┃┃┣┫┓┏╋┣┓┏┛ ┃┫ + # ┗┛┛┗┗┻┗┛┗┗━•┗┛ @staticmethod - async def __save_one_attachment( - 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]: + async def get_token_id( + db_conn: AsyncMySQL, + mongo_conn: AsyncMongo, + auth_token: CoreAuthTokenModel, + session_token: str = None + ) -> ObjectId: - """ - 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. - """ + # Simply call the core model: + return await current_app.core_auth_token_controller.get_token_id( + db_conn = db_conn, + mongo_conn = mongo_conn, + auth_token = auth_token, + token_notes = { + "email": None + }, + session_token = session_token, + ) - # 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) + @staticmethod + async def set_token( + db_conn: AsyncMySQL, + mongo_conn: AsyncMongo, + token_id: ObjectId | str, + auth_token: CoreAuthTokenModel, + session_token: str = None + ) -> bool: - # Start by assuming failure, - # and retry as many times as asked: - attachment_copy["id"] = None - attachment_copy["url"] = None - for _ in range(retry_count): + # Simply call the core model: + return await current_app.core_auth_token_controller.set_token( + db_conn = db_conn, + mongo_conn = mongo_conn, + token_id = token_id, + auth_token = auth_token, + token_notes = { + "email": auth_token.token["email"], + "displayName": auth_token.token.get("displayName"), + "displayPictureUrl": auth_token.token.get("displayPictureUrl"), + }, + session_token = session_token, + ) - # 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 - ) + @staticmethod + async def get_token( + mongo_conn: AsyncMongo, + token_id: ObjectId | str = None, + **kwargs + ) -> CoreAuthTokenModel | None: - # 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 + # Simply call the core model: + return await current_app.core_auth_token_controller.get_token( + mongo_conn = mongo_conn, + token_id = token_id, + kwargs = kwargs + ) - # If the upload failed: - await asyncio.sleep(retry_delay) - retry_delay = retry_delay * backoff_multiplier + # ┏┓ ┳┳┓ + # ┗┓┓┏┏┓┏ ┃┃┃┏┓┏┏┏┓┏┓┏┓┏ + # ┗┛┗┫┛┗┗ ┛ ┗┗ ┛┛┗┻┗┫┗ ┛ + # ┛ ┛ - # 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 - - # ┏┓ ┏┓┳┳┓ •┓ - # ┣ ┏┓┏┓ ┃┓┃┃┃┏┓┓┃ - # ┻ ┗┛┛ ┗┛┛ ┗┗┻┗┗ + # In this section, we pull mails from the third-party clients (like GMail), and store them on our server. This makes + # those mails available on the platform. async def __sync_one_gmail( self, - session_token: str, - user_info: CoreUserInfoModel, mongo_conn: AsyncMongo, + user_info: CoreUserInfoModel, token_id: ObjectId, auth_token: CoreAuthTokenModel, mail_client: AsyncGMailClient, google_tokens: GoogleAuthTokens, message_id: str, - llm: LLMOpenAI = None, + llm: LLMController = None, force_sync: bool = False ) -> MailSyncOneResult: - """ - Sync on mail from GMail. - :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 google_tokens: The tokens to use to fetch the mails. - :param message_id: The id that Google uses to identify this mail. This will be received in the 'list_messages' - method. - :param llm: The instance of the LLM to use to summarize the mail's content. - :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 = { + mail_record = await current_app.core_message_controller.get_previews( + mongo_conn = mongo_conn, + token_ids = [ObjectId(token_id)], + limit = 1, + skip = 0, + additional_filter = { "tokenId": ObjectId(token_id), "serviceType": auth_token.serviceType, "client": auth_token.client, "clientMessageId": message_id - }, - projection = { - "_id": False, - "readTs": True - }, - raise_exception = True + } ) if mail_record: sync_result.success = True @@ -317,69 +329,13 @@ class MailSyncModel(BaseModel): sync_result.message = f"gmail (messageId: '{message_id}'): {client_response.message}" return sync_result - # We upload the attachments: - client_response.data["attachments"] = await self.__save_many_attachments( - session_token = session_token, - attachments = client_response.data["attachments"], - attachment_tags = [ - auth_token.serviceType, - auth_token.client, - client_response.data["from"][0]["name"], - client_response.data["from"][0]["email"], - google_tokens.email, - ], - attachment_metadata = { - "project": "tcaoff", - "serviceType": auth_token.serviceType, - "client": auth_token.client, - "from": client_response.data["from"][0]["email"], - "to": google_tokens.email - }, - retry_count = 3 - ) + # HANDLE ATTACHMENTS HERE: + pass - # 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 google_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: - - # Invoke the LLM: - llm_response = await llm.invoke( - mongo_conn = mongo_conn, - user_info = user_info, - llm_input = LLMInput( - messages = self.PROMPT_TEMPLATE + [ - { - "role": "human", - "content": ( - "Please summarize this mail: " - f"\"\"\"{client_response.data['unformattedText']}\"\"\"" - ) - } - ] - ) - ) - - # Format the response: - llm_json = { - "ts": llm_response.ts, - "snippet": llm_response.output, - "tokens": llm_response.tokens.model_dump() - } - - # Add the LLM's response to the main data: - client_response.data["aiSnippet"] = llm_json - - # Fit the mail message into the model: - sync_result.mailMessage = CoreMessageModel( + # Now we structure the message into the model: + mail_message = CoreMessageModel( ts = client_response.data["ts"], - readTs = date_time.get_current_utc_date_time(as_string = False), + syncTs = date_time.get_current_utc_date_time(as_string = False), tokenId = token_id, serviceType = auth_token.serviceType, client = auth_token.client, @@ -388,40 +344,41 @@ class MailSyncModel(BaseModel): payload = client_response.data ) + # 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 google_tokens.email in all_recipients: mail_message.isSent = False + else: mail_message.isSent = True + + # Invoke the LLM: + mail_message.aiSnippet = await self.summarize_mail_with_ai( + mongo_conn = mongo_conn, + user_info = user_info, + llm = llm, + message = mail_message + ) + # Done here: + print("ONE MAIL:", json.to_string(mail_message.model_dump(), default = str)) sync_result.success = True return sync_result async def __sync_many_gmail( self, - session_token: str, - user_info: CoreUserInfoModel, + db_conn: AsyncMySQL, mongo_conn: AsyncMongo, + user_info: CoreUserInfoModel, token_id: ObjectId, auth_token: CoreAuthTokenModel, mail_client: AsyncGMailClient, - llm: LLMOpenAI = None, + llm: LLMController = None, force_sync: bool = False, start_date: datetime.datetime = None, end_date: datetime.datetime = None, - max_count: int = 100 + max_count: int = 100, + session_token: str = None ) -> MailSyncManyResults: - """ - Sync many mails from GMail in one shot. - :param mongo_conn: The instance of the connection to the database to use. - :param token_id: The id of the document in the database that holds the tokens to access the account. - Needed only for refreshing the tokens and saving them. - :param mail_client: The instance of the mail client to use to perform the action. - :param 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() @@ -437,14 +394,15 @@ class MailSyncModel(BaseModel): if tokens_refreshed: auth_token.token = google_tokens.model_dump() auth_token.lastRefreshTs = date_time.get_current_utc_date_time(as_string = True) - await current_app.mail_oauth_model.set_token( - db_conn = current_app.sql_writer, + await self.set_token( + db_conn = db_conn, mongo_conn = mongo_conn, token_id = token_id, - auth_token = auth_token + auth_token = auth_token, + session_token = session_token ) - # Let's build the query: + # 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")) @@ -464,9 +422,8 @@ class MailSyncModel(BaseModel): # Now, for every mail in the list, we fetch the mail and note the results: tasks = [ self.__sync_one_gmail( - session_token = session_token, - user_info = user_info, mongo_conn = mongo_conn, + user_info = user_info, token_id = token_id, auth_token = auth_token, mail_client = mail_client, @@ -491,22 +448,15 @@ class MailSyncModel(BaseModel): "serviceType": auth_token.serviceType, "client": auth_token.client, "clientMessageId": result.mailMessage.clientMessageId - # "serviceType": auth_token.serviceType, - # "$or": [ - # { - # "client": auth_token.client, - # "messageId": result.mailMessage.clientMessageId - # } - # ] }, replacement = result.mailMessage.model_dump(), upsert = True )) - # Make the bulk write: + # Make the bulk insert operation: if mongo_operations: - mongo_count = await mongo_conn.bulk_write( - collection = self.MAIL_COLLECTION, + sync_count = await current_app.core_message_controller.bulk_write( + mongo_conn = mongo_conn, requests = mongo_operations ) @@ -518,44 +468,26 @@ class MailSyncModel(BaseModel): add_label_ids = [google_tokens.labels.get("TCAOFF", {}).get("id")] ) except Exception as exception: - self._printer(exception) + pass # 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, - user_info: CoreUserInfoModel, + db_conn: AsyncMySQL, mongo_conn: AsyncMongo, - token_id: ObjectId, - llm: LLMOpenAI = None, + user_info: CoreUserInfoModel, + token_id: ObjectId | str, + llm: LLMController = None, force_sync: bool = False, start_date: datetime.datetime = None, end_date: datetime.datetime = None, - max_count: int = 100 + max_count: int = 100, + session_token: str = None ) -> 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 mongo_conn: The instance of the connection to the database to use. - :param token_id: The id of the document in the database that holds the tokens to access the account. - Needed only for refreshing the tokens and saving them. - :param 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() @@ -580,9 +512,9 @@ class MailSyncModel(BaseModel): if auth_token.client == "gmail": return await self.__sync_many_gmail( - session_token = session_token, - user_info = user_info, + db_conn = db_conn, mongo_conn = mongo_conn, + user_info = user_info, token_id = token_id, auth_token = auth_token, mail_client = current_app.gmail_client, @@ -590,7 +522,8 @@ class MailSyncModel(BaseModel): force_sync = force_sync, start_date = start_date, end_date = end_date, - max_count = max_count + max_count = max_count, + session_token = session_token, ) # ┳ ┓• ┓ ┏┓┓• @@ -612,3 +545,16 @@ class MailSyncModel(BaseModel): if __name__ == "__main__": pass + + # from utils_v2.string import json + # + # file_options = [ + # r"/home/developer/Downloads/recursive parts parse - 20241210.json", + # r"/home/developer/Downloads/recursive parts parse (no attachment) - 20241210.json", + # ] + # + # raw_mail_json = json.from_file(file_options[1]) + # print("FROM FILE:", json.to_string(raw_mail_json["payload"])) + # print("\n\n---------\n\n") + # mail_model = MailAPIModel() + # print(mail_model.extract_plaintext_parts(raw_mail_json["payload"])) diff --git a/models/behaviour/base.py b/controllers/base.py similarity index 100% rename from models/behaviour/base.py rename to controllers/base.py diff --git a/models/behaviour/ai/llm/__init__.py b/controllers/core/__init__.py similarity index 100% rename from models/behaviour/ai/llm/__init__.py rename to controllers/core/__init__.py diff --git a/models/behaviour/file/__init__.py b/controllers/core/ai/__init__.py similarity index 100% rename from models/behaviour/file/__init__.py rename to controllers/core/ai/__init__.py diff --git a/models/behaviour/ai/llm/open_ai.py b/controllers/core/ai/llm.py similarity index 83% rename from models/behaviour/ai/llm/open_ai.py rename to controllers/core/ai/llm.py index 4fb2ca8..8fd772d 100644 --- a/models/behaviour/ai/llm/open_ai.py +++ b/controllers/core/ai/llm.py @@ -35,30 +35,18 @@ sys.path.append(".") sys.path.append("..") # 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 # Base model: -from models.behaviour.base import BaseModel +from controllers.base import BaseModel # Data Models: -from models.data.api.ai.llm import LLMInput, LLMOutput, LLMUsageTokens -from models.data.core.user import CoreUserInfoModel +from models.core.user import CoreUserInfoModel +from models.core.ai.llm import LLMInput, LLMOutput, LLMUsageTokens # To work with LLMs: from langchain_openai import ChatOpenAI -# To work with MongoDB: -from bson import ObjectId - -# To work with datatypes: -from typing import Literal - -# To make deep-copies: -import copy - # ***************************************************************************************************************** # ***** **** @@ -97,7 +85,7 @@ import copy # ***************************************************************************************************************** -class LLMOpenAI(BaseModel): +class LLMController(BaseModel): AI_USAGE_COLLECTION = "_aiUsage" @@ -155,6 +143,8 @@ class LLMOpenAI(BaseModel): # Invoke the AI, and format the response: llm_response = await self.__llm.ainvoke(prompt) + print("LLM RESPONSE:", llm_response) + print("INPUT MESSAGES:", llm_input.messages) llm_response = LLMOutput( messages = llm_input.messages, output = llm_response.content, @@ -190,34 +180,3 @@ class LLMOpenAI(BaseModel): if __name__ == "__main__": pass - - # import asyncio - # - # llm_messages = [ - # { - # "role": "system", - # "content": "You are an office assistant." - # }, - # { - # "role": "ai", - # "content": "Hello, sir. How may I help you today?" - # }, - # { - # "role": "human", - # "content": "Please summarize this mail for me..." - # } - # ] - # - # my_llm = LLMOpenAI( - # llm_creds = { - # "model": "gpt-4o-mini", - # "openai_api_key": "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" - # } - # ) - # - # async def main(): - # - # llm_response = await my_llm.invoke(llm_input = LLMInput(messages = llm_messages)) - # print("LLM RESPONSE:", llm_response.model_dump_json(indent = 4)) - # - # asyncio.run(main()) diff --git a/models/behaviour/mail/oauth_v3.py b/controllers/core/auth_token.py similarity index 90% rename from models/behaviour/mail/oauth_v3.py rename to controllers/core/auth_token.py index 5a8af3d..50517c7 100644 --- a/models/behaviour/mail/oauth_v3.py +++ b/controllers/core/auth_token.py @@ -6,13 +6,11 @@ DATE: - ORIGINAL: Monday, 2nd Dec., 2024 - UPGRADED: Monday, 9th Dec., 2024 + Thursday, 12th Dec., 2024 OBJECTIVE: - To define the interaction between the UI layer and the database connectivity in one place. Here we shall handle - all the activities for OAuth2.0 authorization requests for all the users of our service. + To handle all auth-tokens from one place. REFERENCES: @@ -36,6 +34,9 @@ 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 @@ -43,20 +44,14 @@ from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.database.async_mongo_v2 import AsyncMongo # Base model: -from models.behaviour.base import BaseModel +from controllers.base import BaseModel # Data models: -from models.data.core.auth_token import CoreAuthTokenModel +from models.core.auth_token import CoreAuthTokenModel # To work with MongoDB: from bson import ObjectId -# To work with datatypes: -from typing import Literal - -# To make deep-copies: -import copy - # ***************************************************************************************************************** # ***** **** @@ -95,8 +90,13 @@ import copy # ***************************************************************************************************************** -class MailOAuthModel(BaseModel): +class AuthTokenController(BaseModel): + # ┏┓┓ ┓┏ + # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ + # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ + + # For MongoDB: AUTH_COLLECTION = "_authTokens" async def get_token_id( @@ -104,7 +104,8 @@ class MailOAuthModel(BaseModel): db_conn: AsyncMySQL, mongo_conn: AsyncMongo, auth_token: CoreAuthTokenModel, - session_token: str = None + token_notes: dict, + session_token: str = None, ) -> ObjectId: """ @@ -113,6 +114,7 @@ class MailOAuthModel(BaseModel): :param db_conn: The database connection (MariaDB) to use to perform the action. :param mongo_conn: The database connection (MongoDB) to use to perform the action. :param auth_token: An instance of the core auth-token model that holds data in the database. + :param token_notes: Any notes to feed into MariaDB with the token identifier. :param session_token: The session token of the user who requested this service. :return: An ObjectId to later store the granted tokens. """ @@ -120,10 +122,10 @@ class MailOAuthModel(BaseModel): # Note down the timestamp at which this event occurred: request_ts = date_time.get_current_utc_date_time(as_string = False) - # Get the identifier from the database. + # Get the identifier from the database if it already exists, else create one. # BE CAREFUL WITH THE KEYS HERE, THEY SHOULD MATCH THE FIELDS OF THE CORE AUTH-TOKEN MODEL: mongo_json = await mongo_conn.find_one_and_update( - collection = MailOAuthModel.AUTH_COLLECTION, + collection = self.AUTH_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "serviceType": auth_token.serviceType, "user": { @@ -165,15 +167,15 @@ class MailOAuthModel(BaseModel): db_conn = db_conn, proc_name = "entity_integration_save", proc_args = ( - auth_token.user.entityId, # ......................................... 'p_entity_id' - auth_token.client, # ................................................ 'p_provider' - auth_token.status, # ................................................ 'p_current_status' - "Auth Requested", # ................................................. 'p_last_action' - None, # ............................................................. 'p_display_name' - None, # ............................................................. 'p_display_picture' - str(mongo_json["_id"]), # ........................................... 'p_token_id' - json.to_string(python_data = {"email": None}, no_space = True), # ... 'p_notes' - auth_token.user.userId # ............................................ 'p_created_by' + auth_token.user.entityId, # ..................................... 'p_entity_id' + auth_token.client, # ............................................ 'p_provider' + auth_token.status, # ............................................ 'p_current_status' + "Auth Requested", # ............................................. 'p_last_action' + None, # ......................................................... 'p_display_name' + None, # ......................................................... 'p_display_picture' + str(mongo_json["_id"]), # ....................................... 'p_token_id' + json.to_string(python_data = token_notes, no_space = True), # ... 'p_notes' + auth_token.user.userId # ........................................ 'p_created_by' ), session_token = session_token ) @@ -187,6 +189,7 @@ class MailOAuthModel(BaseModel): mongo_conn: AsyncMongo, token_id: ObjectId | str, auth_token: CoreAuthTokenModel, + token_notes: dict, session_token: str = None ) -> bool: @@ -198,6 +201,7 @@ class MailOAuthModel(BaseModel): :param mongo_conn: The database connection (MongoDB) to use to perform the action. :param token_id: The identifier granted by the 'get_token_id' method. :param auth_token: The actual auth/token data to be saved to the database. + :param token_notes: Any notes to feed into MariaDB with the token identifier. :param session_token: The session token of the user who requested this service. :return: True if saved, False if failed. """ @@ -211,13 +215,14 @@ class MailOAuthModel(BaseModel): # Save the token to MongoDB. # BE CAREFUL WITH THE KEYS HERE, THEY SHOULD MATCH THE FIELDS OF THE CORE AUTH-TOKEN MODEL: mongo_json = await mongo_conn.find_one_and_update( - collection = MailOAuthModel.AUTH_COLLECTION, + collection = self.AUTH_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "_id": ObjectId(token_id), "clientUserId": auth_token.clientUserId }), update = [{ "$set": { + "auth": auth_token.auth, "token": auth_token.token, "status": auth_token.status, "lastRefreshTs": request_ts, @@ -242,11 +247,6 @@ class MailOAuthModel(BaseModel): # Tell MariaDB that the token was saved: if mongo_json is not None: - token_notes = { - "email": auth_token.token["email"], - "displayName": auth_token.token.get("displayName"), - "displayPictureUrl": auth_token.token.get("displayPictureUrl"), - } db_json = await self.call_procedure( db_conn = db_conn, proc_name = "entity_integration_save", diff --git a/controllers/core/message.py b/controllers/core/message.py new file mode 100644 index 0000000..dfa9151 --- /dev/null +++ b/controllers/core/message.py @@ -0,0 +1,389 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Thursday, 12th Dec., 2024 + + OBJECTIVE: + + To handle all messages from one place. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +from pyexpat.errors import messages + +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 + +# Base model: +from controllers.base import BaseModel + +# Data models: +from models.core.auth_token import CoreAuthTokenModel +from models.core.message import CoreMessageModel +from models.core.user import CoreUserInfoModel + +# To work with MongoDB: +from bson import ObjectId +from pymongo import InsertOne, UpdateOne, ReplaceOne + +# 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 MessageController(BaseModel): + + # ┏┓┓ ┓┏ + # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ + # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ + + # For MongoDB: + MESSAGES_COLLECTION = "_messages" + + # ┏┓┳┓┳┳┳┓ ┏┓ + # ┃ ┣┫┃┃┃┃ ━━ ┃ ┏┓┏┓┏┓╋┏┓ + # ┗┛┛┗┗┛┻┛ ┗┛┛ ┗ ┗┻┗┗ + + async def insert( + self, + mongo_conn: AsyncMongo, + message: CoreMessageModel + ) -> ObjectId: + + # Simply insert the document: + return await mongo_conn.insert_one( + collection = self.MESSAGES_COLLECTION, + document = message, + raise_exception = True + ) + + async def bulk_write( + self, + mongo_conn: AsyncMongo, + mongo_operations + ) -> int: + + return await mongo_conn.bulk_write( + collection = self.MESSAGES_COLLECTION, + requests = mongo_operations + ) + + # ┏┓┳┓┳┳┳┓ ┳┓ • + # ┃ ┣┫┃┃┃┃ ━━ ┣┫┏┓╋┏┓┓┏┓┓┏┏┓ + # ┗┛┛┗┗┛┻┛ ┛┗┗ ┗┛ ┗┗ ┗┛┗ + + async def count_messages( + self, + mongo_conn: AsyncMongo, + token_ids: List[ObjectId | str], + additional_filter: dict = None + ) -> int: + + """ + Just counts the no. of messages that match a given set of conditions. + :param mongo_conn: The instance of the database connector to use for the operation. + :param token_ids: The token ids of the accounts from which these messages must be fetched. + :param additional_filter: Any addition filters to use. + :return: The no. of messages that match the given conditions. + """ + + # Prepare the filter: + if not isinstance(token_ids, list): token_ids = [token_ids] + token_ids = [ObjectId(t) for t in token_ids] + filter_json = {"tokenId": {"$in": token_ids}} + if additional_filter: + for k, v in additional_filter.items(): + filter_json[k] = v + + # Get the count of the documents that match the criteria: + count = await mongo_conn.count( + collection = self.MESSAGES_COLLECTION, + filter = filter_json, + raise_exception = True + ) + + # Done here: + return count + + async def get_previews( + self, + mongo_conn: AsyncMongo, + token_ids: List[ObjectId | str], + limit: int = 100, + skip: int = 0, + additional_filter: dict = None + ) -> List[CoreMessageModel] | None: + + """ + Fetches many messages in one call, but leaves out the full payloads. This does not mark messages as read. + :param mongo_conn: The instance of the database connector to use for the operation. + :param token_ids: The token ids of the accounts from which these messages must be fetched. + :param limit: The max. no. of messages to retrieve in this call. + :param skip: The no. of initial messages to skip. Useful for pagination. + :param additional_filter: Any addition filters to use. + :return: The list of messages (as the message model). This list can be empty. + """ + + # Prepare the filter: + if not isinstance(token_ids, list): token_ids = [token_ids] + token_ids = [ObjectId(t) for t in token_ids] + filter_json = {"tokenId": {"$in": token_ids}} + if additional_filter: + for k, v in additional_filter.items(): + filter_json[k] = v + + # We fetch the messages that are identified by a specific token id, + # with the specified fetching limits, while enforcing the sorting condition: + records = await mongo_conn.find_many( + collection = self.MESSAGES_COLLECTION, + filter = filter_json, + limit = limit, + skip = skip, + sort = {"ts": -1}, + projection = { + "_id": True, + "ts": True, + "markedAsUnread": True, + "serviceType": True, + "client": True, + "clientMessageId": True, + "clientThreadId": True, + "isSent": True, + "isBroadcast": True, + "sentSuccessfully": True, + "aiSnippet": True, + "preview": True, + "message": {}, + "tags": True, + "usedAi": True + }, + raise_exception = True + ) + + # Convert the fetched records to instances of the data model and return: + return [CoreMessageModel(**record) for record in records] + + async def get_messages( + self, + mongo_conn: AsyncMongo, + token_ids: List[ObjectId | str], + limit: int = 100, + skip: int = 0, + additional_filter: dict = None + ) -> List[CoreMessageModel] | None: + + """ + Fetches many full messages in one call. + :param mongo_conn: The instance of the database connector to use for the operation. + :param token_ids: The token ids of the accounts from which these messages must be fetched. + :param limit: The max. no. of messages to retrieve in this call. + :param skip: The no. of initial messages to skip. Useful for pagination. + :param additional_filter: Any addition filters to use. + :return: The list of messages (as the message model). This list can be empty. + """ + + # Note down the timestamp at which this event occurred: + request_ts = date_time.get_current_utc_date_time(as_string = False) + + # Prepare the filter: + if not isinstance(token_ids, list): token_ids = [token_ids] + token_ids = [ObjectId(t) for t in token_ids] + filter_json = {"tokenId": {"$in": token_ids}} + if additional_filter: + for k, v in additional_filter.items(): + filter_json[k] = v + + # We fetch the messages that are identified by a specific token id, + # with the specified fetching limits, while enforcing the sorting condition: + records = await mongo_conn.find_many( + collection = self.MESSAGES_COLLECTION, + filter = filter_json, + limit = limit, + skip = skip, + sort = {"ts": -1}, + raise_exception = True + ) + + # We now mark these fetched messages as read through a bulk-write operation: + operations = [ + UpdateOne( + filter = {"_id": record["_id"]}, + update = [{ + "$set": { + "readTs": { + "$cond": { + "if": { + "$or": [ + {"$eq": ["$readTs", None]}, + {"$eq": [{"$type": "$readTs"}, "missing"]} + ] + }, + "then": request_ts, + "else": "$readTs" + } + } + } + }], + upsert = False + ) for record in records + ] + updated_count = await mongo_conn.bulk_write( + collection = self.MESSAGES_COLLECTION, + requests = operations, + raise_exception = True + ) + + # Convert the fetched records to instances of the data model and return: + return [CoreMessageModel(**record) for record in records] + + async def get_message( + self, + mongo_conn: AsyncMongo, + message_id: ObjectId | str, + ) -> CoreMessageModel | None: + + """ + Gets one message if you know its message id. Marks that message as read. + :param mongo_conn: + :param message_id: + :return: + """ + + # Note down the timestamp at which this event occurred: + request_ts = date_time.get_current_utc_date_time(as_string = False) + + # We fetch the whole payload of that one message + # while also marking it as read if not already marked: + record = await mongo_conn.find_one_and_update( + collection = self.MESSAGES_COLLECTION, + filter = {"_id": ObjectId(message_id)}, + update = [{ + "$set": { + "readTs": { + "$cond": { + "if": { + "$or": [ + {"$eq": ["$readTs", None]}, + {"$eq": [{"$type": "$readTs"}, "missing"]} + ] + }, + "then": request_ts, + "else": "$readTs" + } + } + } + }], + raise_exception = True + ) + + # If no such message was found: + if record is None: return None + + # If a record was found, + # we return it as our data model: + return CoreMessageModel(**record) + + # ┏┓┳┓┳┳┳┓ ┳┳ ┓ + # ┃ ┣┫┃┃┃┃ ━━ ┃┃┏┓┏┫┏┓╋┏┓ + # ┗┛┛┗┗┛┻┛ ┗┛┣┛┗┻┗┻┗┗ + # ┛ + + # We don't support updating messages themselves, + # but we will allow updating fields like tags, marking as read or unread, etc. + + # ┏┓┳┓┳┳┳┓ ┳┓ ┓ + # ┃ ┣┫┃┃┃┃ ━━ ┃┃┏┓┃┏┓╋┏┓ + # ┗┛┛┗┗┛┻┛ ┻┛┗ ┗┗ ┗┗ + + # No support whatsoever for deleting messages. + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/models/behaviour/mail/__init__.py b/models/api/__init__.py similarity index 100% rename from models/behaviour/mail/__init__.py rename to models/api/__init__.py diff --git a/models/behaviour/sms/__init__.py b/models/api/ai/__init__.py similarity index 100% rename from models/behaviour/sms/__init__.py rename to models/api/ai/__init__.py diff --git a/models/data/api/ai/llm.py b/models/api/ai/llm.py similarity index 100% rename from models/data/api/ai/llm.py rename to models/api/ai/llm.py diff --git a/models/data/__init__.py b/models/api/chat/__init__.py similarity index 100% rename from models/data/__init__.py rename to models/api/chat/__init__.py diff --git a/models/data/api/chat/auth.py b/models/api/chat/auth.py similarity index 100% rename from models/data/api/chat/auth.py rename to models/api/chat/auth.py diff --git a/models/data/api/__init__.py b/models/api/mail/__init__.py similarity index 100% rename from models/data/api/__init__.py rename to models/api/mail/__init__.py diff --git a/models/data/api/mail/list.py b/models/api/mail/list.py similarity index 100% rename from models/data/api/mail/list.py rename to models/api/mail/list.py diff --git a/models/data/api/mail/oauth.py b/models/api/mail/oauth.py similarity index 100% rename from models/data/api/mail/oauth.py rename to models/api/mail/oauth.py diff --git a/models/data/api/mail/retrieve.py b/models/api/mail/retrieve.py similarity index 100% rename from models/data/api/mail/retrieve.py rename to models/api/mail/retrieve.py diff --git a/models/data/api/mail/sync.py b/models/api/mail/sync.py similarity index 99% rename from models/data/api/mail/sync.py rename to models/api/mail/sync.py index f9b370d..7f26a6c 100644 --- a/models/data/api/mail/sync.py +++ b/models/api/mail/sync.py @@ -44,7 +44,7 @@ from utils_v2.string import regex from utils_v2.date_time import date_time # Data models: -from models.data.core.message import CoreMessageModel +from models.core.message import CoreMessageModel # To work with date and time: import datetime diff --git a/models/data/api/ai/__init__.py b/models/api/sms/__init__.py similarity index 100% rename from models/data/api/ai/__init__.py rename to models/api/sms/__init__.py diff --git a/models/data/api/sms/auth.py b/models/api/sms/auth.py similarity index 100% rename from models/data/api/sms/auth.py rename to models/api/sms/auth.py diff --git a/models/data/api/sms/send.py b/models/api/sms/send.py similarity index 100% rename from models/data/api/sms/send.py rename to models/api/sms/send.py diff --git a/models/data/api/chat/__init__.py b/models/api/tech/__init__.py similarity index 100% rename from models/data/api/chat/__init__.py rename to models/api/tech/__init__.py diff --git a/models/data/api/tech/alerts.py b/models/api/tech/alerts.py similarity index 100% rename from models/data/api/tech/alerts.py rename to models/api/tech/alerts.py diff --git a/models/behaviour/file/file.py b/models/behaviour/file/file.py deleted file mode 100644 index 3d54849..0000000 --- a/models/behaviour/file/file.py +++ /dev/null @@ -1,444 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - Thursday, 12th Dec., 2024 - - OBJECTIVE: - - To define all file-management activities in one place. - - REFERENCES: - - N/A - - DOWNLOADS: - - N/A - -""" - - -# ***************************************************************************************************************** -# ***** **** -# *** IMPORT *** -# ***** **** -# ***************************************************************************************************************** - - -# To make sibling directories accessible for imports: -import sys -sys.path.append(".") -sys.path.append("..") - -# System-level: -import io -# My async utils: -from utils_v2.string import json -from utils_v2.date_time import date_time -from utils_v2.system import files -from utils_v2.security.hash import Hasher -from utils_v2.database.async_mysql_v2 import AsyncMySQL -from utils_v2.database.async_mongo_v2 import AsyncMongo, AsyncMongoStorage - -# Base model: -from models.behaviour.base import BaseModel - -# Data models: -from models.data.core.user import CoreUserInfoModel -from models.data.core.file import CoreFileInfoModel, CoreFileAccessResponseModel - -# To work with MongoDB: -from bson import ObjectId - -# To work with datatypes: -from typing import Any, Literal, List - -# To make deep copies: -import copy - -# For debugging: -from icecream import IceCreamDebugger - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** CLASSES *** -# ***** **** -# ***************************************************************************************************************** - - -class FileManagementModel: - - def __init__( - self, - debug = True, - debug_prefix = "File Objs. | ", - ): - - # Debugging: - self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) - if not debug: self._printer.disable() - - def enable_debug(self): - self._printer.enable() - - def disable_debug(self): - self._printer.disable() - - # ┏┓ • ┓ ┏┓ • - # ┃┃┓┏┓┏┃┏ ┃┃┓┏┏┓┏┓┓┏┓┏ - # ┗┻┗┻┗┗┛┗ ┗┻┗┻┗ ┛ ┗┗ ┛ - - async def exists( - self, - mongo_conn: AsyncMongoStorage, - file_id: ObjectId | str - ) -> CoreFileAccessResponseModel: - - """ - To check whether, or not, a particular file's record exists in the database. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_id: The id of the file to check. - :return: A structured response where the existence of the file is noted in the 'result' field. - """ - - # Create a response: - response = CoreFileAccessResponseModel() - - try: - - # Run the query: - record = await mongo_conn.find_one_file( - filter = {"_id": ObjectId(file_id)}, - projection = {"_id": True, "user": True}, - raise_exception = True - ) - - # Note down the result: - if record: - response.data = True - response.success = True - response.message = "ok" - - except Exception as exception: - self._printer(exception) - response.exception = exception - response.message = str(exception) - response.success = False - response.data = None - - # Done here: - return response - - async def info( - self, - mongo_conn: AsyncMongoStorage, - file_id: ObjectId | str - ) -> CoreFileAccessResponseModel: - - """ - To get the information about this file. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_id: The id of the file to check. - :return: A structured response where the info of the file is noted in the 'result' field. - """ - - # Create a response: - response = CoreFileAccessResponseModel() - - try: - - # Run the query: - record = await mongo_conn.find_one_file( - filter = {"_id": ObjectId(file_id)}, - raise_exception = True - ) - - # Note down the result: - response.success = True - if record: - response.data = CoreFileInfoModel(**record["metadata"]) - response.message = "ok" - else: - response.message = "no such file object" - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - response.exception = exception - response.message = str(exception) - response.success = False - response.data = None - - # Done here: - return response - - async def is_private( - self, - mongo_conn: AsyncMongoStorage, - file_id: ObjectId | str - ) -> bool | None: - - """ - To check whether, or not, a particular file is publicly readable. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_id: The id of the file to check. - :return: True if private, else False. None if it doesn't exist at all. - """ - - # Create a response: - response = CoreFileAccessResponseModel() - - try: - - # Run the query: - record = await mongo_conn.find_one_file( - filter = {"_id": ObjectId(file_id)}, - projection = {"_id": False, "isPrivate": True}, - raise_exception = True - ) - - # Note down the result: - if record: - response.data = record["metadata"]["isPrivate"] - response.success = True - response.message = "ok" - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - response.exception = exception - response.message = str(exception) - response.success = False - response.data = None - - # Done here: - return response - - # ┓ • • - # ┃ ┓┏╋┓┏┓┏┓ - # ┗┛┗┛┗┗┛┗┗┫ - # ┛ - - pass - - # ┳┓ ┓• - # ┣┫┏┓┏┓┏┫┓┏┓┏┓ - # ┛┗┗ ┗┻┗┻┗┛┗┗┫ - # ┛ - - async def download_file( - self, - mongo_conn: AsyncMongoStorage, - file_id: ObjectId | str - ) -> CoreFileAccessResponseModel: - - """ - To quickly download small files. Do not use this for larger files because the file will be held in RAM first - and any large file will end up filling RAM fast. It's okay for smaller files that won't block up the memory. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_id: The id of the file to fetch. - :return: The file in a BytesIO buffer in the 'data' field of the structured response. - """ - - # Create a response: - response = CoreFileAccessResponseModel() - - try: - - # Get the file from the database: - buffer = io.BytesIO() - response.success = await mongo_conn.easy_download( - destination = buffer, - file_id = ObjectId(file_id), - raise_exception = True - ) - buffer.seek(0) - - # Note down the results: - response.message = ( - "file fetched successfully" if response.success - else "file fetching failed" - ) - response.data = buffer if response.success else None - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - response.exception = exception - response.message = str(exception) - response.success = False - response.data = None - - # Done here: - return response - - @staticmethod - async def get_file_download_stream( - mongo_conn: AsyncMongoStorage, - file_id: ObjectId - ) -> Any: - - """ - To download any file as a stream. Better than the simple 'download' method because it doesn't block RAM. Once - the stream is created you can read from it, and terminate it like this: - READ: await stream.read(chunk_size) - CLOSE (without awaiting): stream.close() - :param mongo_conn: The instance of the database connection to perform this action. - :param file_id: The id of the file whose stream you would like to fetch. - :return: Returns the stream object that will allow more efficient downloads of files on the user's end. - """ - - return await mongo_conn.get_download_stream(file_id = file_id) - - # ┓ ┏ • • - # ┃┃┃┏┓┓╋┓┏┓┏┓ - # ┗┻┛┛ ┗┗┗┛┗┗┫ - # ┛ - - async def upload_file( - self, - mongo_conn: AsyncMongoStorage, - file_info: CoreFileInfoModel, - file_data: io.BytesIO | str, - chunk_size: int = None - ): - - # Create a response: - response = CoreFileAccessResponseModel() - - try: - - # Read the file's data into a BytesIO object: - if isinstance(file_data, str): - file_data = io.BytesIO(files.read_file(file_data, mode = "rb")) - file_data.seek(0) - - # Hash the file's data: - hasher = Hasher() - hasher.update(file_data.getvalue()) - file_info.hash = hasher.hexdigest() - - # Save the file to the database: - file_data.seek(0) - response.success = await mongo_conn.easy_upload( - source = file_data, - file_name = file_info.filename, - file_metadata = file_info.model_dump(), - file_id = file_info.fileId, - chunk_size = chunk_size, - raise_exception = True - ) - - # Note down the results: - response.message = ( - "file saved successfully" if response.success - else "file saving failed" - ) - response.data = True if response.success else False - - # If something goes wrong: - except Exception as exception: - self._printer(exception) - response.exception = exception - response.message = str(exception) - response.success = False - response.data = None - - # Done here: - return response - - async def upload_from_stream(self): pass - - # ┳┓ ┓ • - # ┃┃┏┓┃┏┓╋┓┏┓┏┓ - # ┻┛┗ ┗┗ ┗┗┛┗┗┫ - # ┛ - - async def delete_file(self): pass - - # ┏┓ • • - # ┃┃┏┓┏┓┏┳┓┓┏┏┓┏┓┏┓┏ - # ┣┛┗ ┛ ┛┗┗┗┛┛┗┗┛┛┗┛ - - async def make_public(self): pass - - async def make_private(self): pass - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - import asyncio - import time - - async def main(): - - files_mongo = AsyncMongoStorage( - connection_string = r"mongodb://del.ditscentre.in:27017,wtt.ditscentre.in:27017,mum.arh.001.ditscentre.in:27017/admin?tls=true&tlsCAFile=%2Fetc%2Fssl%2Fcerts%2Fmongo_data_ca.pem&tlsCertificateKeyFile=%2Fetc%2Fssl%2Fcerts%2Fmongo_data_cert.pem&replicaSet=dits_mongod_rep&readPreference=primary&authMechanism=MONGODB-X509&authSource=%24external", - database_name = "converseStore", - max_connections = 10, - debug = True - ) - await files_mongo.connect() - my_files = FileManagementModel() - - user_bhopli = CoreUserInfoModel( - fullName = "Bhopli Narangi", - userId = 1, - entityId = 2, - billingAccountId = 3, - departmentId = 4, - branchId = 5, - industry = "technology" - ) - - file_info = await my_files.info( - mongo_conn = files_mongo, - file_id = "67598d48c1bf89b25695f20b" - ) - print("FILE INFO:", json.to_string(file_info.model_dump(), default = str)) - - - asyncio.run(main()) diff --git a/models/behaviour/file/file_object.py b/models/behaviour/file/file_object.py deleted file mode 100644 index 808e2be..0000000 --- a/models/behaviour/file/file_object.py +++ /dev/null @@ -1,807 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - Tuesday, 10th Dec., 2024 - - OBJECTIVE: - - To define all file-management activities in one place. - - REFERENCES: - - N/A - - DOWNLOADS: - - N/A - -""" -import io -# ***************************************************************************************************************** -# ***** **** -# *** IMPORT *** -# ***** **** -# ***************************************************************************************************************** - - -# To make sibling directories accessible for imports: -import sys -sys.path.append(".") -sys.path.append("..") - -# My async utils: -from utils_v2.string import json -from utils_v2.date_time import date_time -from utils_v2.system import files -from utils_v2.security.hash import Hasher -from utils_v2.database.async_mysql_v2 import AsyncMySQL -from utils_v2.database.async_mongo_v2 import AsyncMongo, AsyncMongoStorage - -# Base model: -from models.behaviour.base import BaseModel - -# Data models: -from models.data.core.user import CoreUserInfoModel -from models.data.core.file_object import ( - CoreFileObjectInfoModel, - CoreFileObjectSharingModel, - CoreFileObjectPermissionsModel, - CoreFileObjectAccessResponseModel -) - -# To work with MongoDB: -from bson import ObjectId - -# To work with datatypes: -from typing import Any, Literal, List - -# To make deep copies: -import copy - -# For debugging: -from icecream import IceCreamDebugger - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** CLASSES *** -# ***** **** -# ***************************************************************************************************************** - - -class FileObjectManagementModel: - - # Define class-level variables: - FILE_OBJECTS_COLLECTION = "_fileObjects" - - def __init__( - self, - debug = True, - debug_prefix = "File Objs. | ", - ): - - # Debugging: - self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) - if not debug: self._printer.disable() - - def enable_debug(self): - self._printer.enable() - - def disable_debug(self): - self._printer.disable() - - # ┓┏ ┓ - # ┣┫┏┓┃┏┓┏┓┏┓┏ - # ┛┗┗ ┗┣┛┗ ┛ ┛ - # ┛ - - @staticmethod - def users_match( - user_p: CoreUserInfoModel, - user_r: CoreUserInfoModel, - ignore_null: bool = True - ) -> bool: - - """ - To match if a user that is requesting a resource is the same as the user known to have access to the resource. - :param user_p: One of the dicts to check. - :param user_r: The other dict to check. - :param ignore_null: Whether to consider only non-null values, or all values. - :return: True if they match, else False. - """ - - # Start by assuming success: - are_matching = True - - # Iterate through the required items: - for rk, rv in user_r.model_dump().items(): - - # Do not consider fields that are nulls if asked to ignore them: - if ignore_null and rv is None: continue - - # Extract the corresponding value from the other user, - # and test it for being equal: - pv = getattr(user_p, rk, None) - if ( - (not isinstance(rv, type(pv))) or - (rv != pv) - ): - are_matching = False - break - - # Done here: - return are_matching - - # ┏┓ • ┓ ┏┓ • - # ┃┃┓┏┓┏┃┏ ┃┃┓┏┏┓┏┓┓┏┓┏ - # ┗┻┗┻┗┗┛┗ ┗┻┗┻┗ ┛ ┗┗ ┛ - - async def exists( - self, - mongo_conn: AsyncMongoStorage, - file_object_id: ObjectId | str - ) -> CoreFileObjectAccessResponseModel: - - """ - To check whether, or not, a particular file's record exists in the database. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_object_id: The id of the file to check. - :return: A structured response where the existence of the file is noted in the 'result' field. - """ - - # Create a response: - response = CoreFileObjectAccessResponseModel() - - try: - - # Run the query: - record = await mongo_conn.find_one( - collection = self.FILE_OBJECTS_COLLECTION, - filter = {"_id": ObjectId(file_object_id)}, - projection = {"_id": True, "user": True, "isDir": True}, - raise_exception = True - ) - - # Note down the result: - if record: - response.result = True - response.success = True - response.message = "ok" - - except Exception as exception: - response.exception = exception - response.message = str(exception) - response.success = False - response.result = None - - # Done here: - return response - - async def info( - self, - mongo_conn: AsyncMongoStorage, - file_object_id: ObjectId | str - ) -> CoreFileObjectAccessResponseModel: - - """ - To get the information about this file. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_object_id: The id of the file to check. - :return: A structured response where the info of the file is noted in the 'result' field. - """ - - # Create a response: - response = CoreFileObjectAccessResponseModel() - - try: - - # Run the query: - record = await mongo_conn.find_one( - collection = self.FILE_OBJECTS_COLLECTION, - filter = {"_id": ObjectId(file_object_id)}, - raise_exception = True - ) - - # Note down the result: - response.success = True - if record: - response.result = CoreFileObjectInfoModel(**record) - response.message = "ok" - else: - response.message = "no such file object" - - # If something goes wrong: - except Exception as exception: - response.exception = exception - response.message = str(exception) - response.success = False - response.result = None - - # Done here: - return response - - async def is_private( - self, - mongo_conn: AsyncMongoStorage, - file_object_id: ObjectId | str - ) -> bool | None: - - """ - To check whether, or not, a particular file is publicly readable. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_object_id: The id of the file to check. - :return: True if private, else False. None if it doesn't exist at all. - """ - - # Create a response: - response = CoreFileObjectAccessResponseModel() - - try: - - # Run the query: - record = await mongo_conn.find_one( - collection = self.FILE_OBJECTS_COLLECTION, - filter = {"_id": ObjectId(file_object_id)}, - projection = {"_id": False, "isPrivate": True}, - raise_exception = True - ) - - # Note down the result: - if record: - response.result = record["isPrivate"] - response.success = True - response.message = "ok" - - # If something goes wrong: - except Exception as exception: - response.exception = exception - response.message = str(exception) - response.success = False - response.result = None - - # Done here: - return response - - def is_owner( - self, - mongo_conn: AsyncMongoStorage, - user_info: CoreUserInfoModel, - file_object_info: CoreFileObjectInfoModel, - ignore_null: bool = True - ) -> CoreFileObjectAccessResponseModel: - - """ - To check if a specific user is the owner of a specific file. - :param mongo_conn: The instance of the database connection to perform this action. - :param user_info: The details of the user who needs to have permissions to this file. - :param file_object_info: The information about the file/dir. Fetch it from the 'info' method. - :param ignore_null: Whether to consider only non-null values, or all values. - :return: True if owner, else False. None if something goes wrong. - """ - - # Create a response: - response = CoreFileObjectAccessResponseModel() - - try: - - # Test for a match: - if self.users_match( - user_p = file_object_info.user, - user_r = user_info, - ignore_null = ignore_null - ): - response.result = True - response.message = "user is the owner of this resource" - else: - response.result = False - response.message = "user is not the owner of this resource" - response.success = True - - # If something goes wrong: - except Exception as exception: - response.exception = exception - response.message = str(exception) - response.success = False - response.result = None - - # Done here: - return response - - def has_permission( - self, - mongo_conn: AsyncMongoStorage, - user_info: CoreUserInfoModel, - file_object_info: CoreFileObjectInfoModel, - permission: Literal["read", "write", "delete", "changePermissions"], - ignore_null: bool = True - ) -> CoreFileObjectAccessResponseModel: - - """ - To check if a particular user has permissions to a given file obj. You may pass either an instance of the file's - info, or you may send the file's id to check. If you pass just the file's id, a database call will be needed. - :param mongo_conn: The instance of the database connection to perform this action. - :param user_info: The details of the user who needs to have permissions to this file. - :param permission: The name of the permission that the said user must have on this file. - :param file_object_info: The information about the file/dir. Fetch it from the 'info' method. - :param ignore_null: Whether to consider only non-null values, or all values. - :return: True if the user has said permission, else False. None if something goes wrong. - """ - - # If this is a public file/dir, - # and the permission requested is 'read': - if (not file_object_info.isPrivate) and permission == "read": - return CoreFileObjectAccessResponseModel( - success = True, - message = "this resource is publicly available", - result = True, - exception = None - ) - - # The owner always has all permissions: - response = self.is_owner( - mongo_conn = mongo_conn, - user_info = user_info, - file_object_info = file_object_info, - ignore_null = ignore_null - ) - if not response.success: return response - if response.result is True: return response - - # Note down the failure of the ownership test: - response.message = "this user does not have the requested permission over this resource" - - # Since The person requesting this is not the owner, - # we check with the sharing details: - for sharing_data in file_object_info.sharedWith: - - # We match the users. - # If they don't match, we move to the next user: - if not self.users_match( - user_p = sharing_data.user, - user_r = user_info, - ignore_null = ignore_null - ): continue - - # If we found a matching user, - # we check for the permission: - if getattr(sharing_data.permissions, permission, False): - response.result = True - response.success = True - response.message = "this user has the requested permission over this resource" - break - - # Done here: - return response - - # ┓ • • - # ┃ ┓┏╋┓┏┓┏┓ - # ┗┛┗┛┗┗┛┗┗┫ - # ┛ - - async def list_owned_dirs( - self, - mongo_conn: AsyncMongoStorage, - user_info: CoreUserInfoModel, - limit: int = 50, - skip: int = 0, - ) -> List[CoreFileObjectInfoModel] | None: - - """ - To list all the dirs that are owned by the described user. - :param mongo_conn: The connection instance to use to make the check. - :param user_info: The information about the user that we must match. - :param limit: How many max. records to fetch. - :param skip: How many initial records to skip. Useful for pagination. - :return: The list of files owned by the user (can be empty), or None if something goes wrong. - """ - - # Run the query: - records = await mongo_conn.find_many( - collection = self.FILE_OBJECTS_COLLECTION, - filter = mongo_conn.dict_to_dot_notation({ - "user": {k: v for k, v in user_info.model_dump().items() if v is not None}, - "isDir": True - }), - limit = limit, - skip = skip - ) - - # If something went wrong, we receive null for the records. - # We pass that null on: - if records is None: return None - - # Otherwise, we format and return the records: - return [CoreFileObjectInfoModel(**record) for record in records] - - async def list_shared_dirs( - self, - mongo_conn: AsyncMongoStorage, - user_info: CoreUserInfoModel, - limit: int = 50, - skip: int = 0, - ) -> List[CoreFileObjectInfoModel] | None: - - """ - To list all the dirs that have been shared with the described user. - :param mongo_conn: The connection instance to use to make the check. - :param user_info: The information about the user that we must match. - :param limit: How many max. records to fetch. - :param skip: How many initial records to skip. Useful for pagination. - :return: The list of files owned by the user (can be empty), or None if something goes wrong. - """ - - # Run the query: - records = await mongo_conn.find_many( - collection = self.FILE_OBJECTS_COLLECTION, - filter = mongo_conn.dict_to_dot_notation({ - "sharedWith.user": {k: v for k, v in user_info.model_dump().items() if v is not None}, - "isDir": True - }), - limit = limit, - skip = skip - ) - - # If something went wrong, we receive null for the records. - # We pass that null on: - if records is None: return None - - # Otherwise, we format and return the records: - return [CoreFileObjectInfoModel(**record) for record in records] - - async def list_owned_files( - self, - mongo_conn: AsyncMongoStorage, - user_info: CoreUserInfoModel, - limit: int = 50, - skip: int = 0, - ) -> List[CoreFileObjectInfoModel] | None: - - """ - To list all the files that are owned by the described user. - :param mongo_conn: The connection instance to use to make the check. - :param user_info: The information about the user that we must match. - :param limit: How many max. records to fetch. - :param skip: How many initial records to skip. Useful for pagination. - :return: The list of files owned by the user (can be empty), or None if something goes wrong. - """ - - # Run the query: - records = await mongo_conn.find_many( - collection = self.FILE_OBJECTS_COLLECTION, - filter = mongo_conn.dict_to_dot_notation({ - "user": {k: v for k, v in user_info.model_dump().items() if v is not None}, - "isDir": False - }), - limit = limit, - skip = skip - ) - - # If something went wrong, we receive null for the records. - # We pass that null on: - if records is None: return None - - # Otherwise, we format and return the records: - return [CoreFileObjectInfoModel(**record) for record in records] - - async def list_shared_files( - self, - mongo_conn: AsyncMongoStorage, - user_info: CoreUserInfoModel, - limit: int = 50, - skip: int = 0, - ) -> List[CoreFileObjectInfoModel] | None: - - """ - To list all the files that have been shared with the described user. - :param mongo_conn: The connection instance to use to make the check. - :param user_info: The information about the user that we must match. - :param limit: How many max. records to fetch. - :param skip: How many initial records to skip. Useful for pagination. - :return: The list of files owned by the user (can be empty), or None if something goes wrong. - """ - - # Run the query: - records = await mongo_conn.find_many( - collection = self.FILE_OBJECTS_COLLECTION, - filter = mongo_conn.dict_to_dot_notation({ - "sharedWith.user": {k: v for k, v in user_info.model_dump().items() if v is not None}, - "isDir": False - }), - limit = limit, - skip = skip - ) - - # If something went wrong, we receive null for the records. - # We pass that null on: - if records is None: return None - - # Otherwise, we format and return the records: - return [CoreFileObjectInfoModel(**record) for record in records] - - # ┳┓ ┓• - # ┣┫┏┓┏┓┏┫┓┏┓┏┓ - # ┛┗┗ ┗┻┗┻┗┛┗┗┫ - # ┛ - - @staticmethod - async def download_file( - mongo_conn: AsyncMongoStorage, - file_id: ObjectId - ) -> io.BytesIO | None: - - """ - To quickly download small files. Do not use this for larger files because the file will be held in RAM first - and any large file will end up filling RAM fast. It's okay for smaller files that won't block up the memory. - WARNING: Check for permissions before using this method. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_id: The id of the file to fetch. - :return: The file in a BytesIO buffer, or None if the file doesn't exist. - """ - - # Get the file from the database: - buffer = io.BytesIO() - success = await mongo_conn.easy_download( - destination = buffer, - file_id = ObjectId(file_id), - raise_exception = True - ) - buffer.seek(0) - - # Return the result: - if not success: return None - else: return buffer - - @staticmethod - async def get_file_download_stream( - mongo_conn: AsyncMongoStorage, - file_id: ObjectId - ) -> Any: - - """ - To download any file as a stream. Better than the simple 'download' method because it doesn't block RAM. Once - the stream is created you can read from it, and terminate it like this: - READ: await stream.read(chunk_size) - CLOSE (without awaiting): stream.close() - WARNING: Check for permissions before using this method. - :param mongo_conn: The instance of the database connection to perform this action. - :param file_id: The id of the file whose stream you would like to fetch. - :return: Returns the stream object that will allow more efficient downloads of files on the user's end. - """ - - return await mongo_conn.get_download_stream(file_id = file_id) - - # ┓ ┏ • • - # ┃┃┃┏┓┓╋┓┏┓┏┓ - # ┗┻┛┛ ┗┗┗┛┗┗┫ - # ┛ - - async def make_dir( - self, - mongo_conn: AsyncMongoStorage, - user_info: CoreUserInfoModel, - dir_name: str, - dir_metadata: dict = None, - dir_tags: list = None, - parent_id: ObjectId | str = None - ) -> bool: - - # Create an instance of the directory's model: - dir_model = CoreFileObjectInfoModel( - _id = mongo_conn.generate_id(), - user = user_info, - isDir = True, - name = dir_name, - createTs = date_time.get_current_utc_date_time(as_string = False), - metadata = dir_metadata, - tags = dir_tags, - parentId = parent_id, - isPrivate = True - ) - - print("DIRECTORY:", json.to_string(dir_model.model_dump(), default = str)) - - # Insert this document into the database: - inserted_id = await mongo_conn.insert_one( - collection = self.FILE_OBJECTS_COLLECTION, - document = dir_model.model_dump() - ) - - # Done here: - return True if inserted_id else False - - async def upload_file( - self, - mongo_conn: AsyncMongoStorage, - file_info: CoreFileObjectInfoModel, - file_data: io.BytesIO | str - ): - - # Start by assuming failure: - file_uploaded = False - - # Start a session: - async with await (await mongo_conn.client).start_session() as session: - - # Define the transaction options: - options = { - # "read_concern": {"level": "snapshot"}, # ... Optional: ensures consistent reads. - # "write_concern": {"w": "majority"}, # ...... Ensures writes are acknowledged. - # "read_preference": "primary", # ............ Specify where to read from (e.g., primary). - } - - # Start the transaction: - async with session.start_transaction(**options): - - try: - - # Check that we indeed have a file that we are uploading: - if file_info.isDir: raise ValueError("cannot data to upload directory object") - - # Generate an ObjectId: - file_info.fileObjectId = mongo_conn.generate_id(as_str = False) - - # Read the file's data into a BytesIO object: - if isinstance(file_data, str): - file_data = io.BytesIO(files.read_file(file_data, mode = "rb")) - file_data.seek(0) - - # Hash the file's data: - hasher = Hasher() - hasher.update(file_data.getvalue()) - file_info.hash = hasher.hexdigest() - - # Now we upload the actual content of the file with the same id: - success = await mongo_conn.easy_upload( - source = file_data, - file_name = file_info.name, - file_metadata = None, - file_id = file_info.fileObjectId, - session = session, - raise_exception = True - ) - - # Safety check to ensure that the data was written: - if not success: - raise ValueError("file object's bytes weren't uploaded") - - # Write the file object's info model now with the same id: - inserted_id = await mongo_conn.insert_one( - collection = self.FILE_OBJECTS_COLLECTION, - document = file_info.model_dump(), - session = session, - raise_exception = True - ) - - # Safety check to ensure that the info was written: - if inserted_id is None: - raise ValueError("file object's info wasn't inserted") - - # If we've reached this far: - file_uploaded = True - - # If something goes wrong: - except Exception as exception: - session.abort_transaction() - file_uploaded = False - self._printer(exception) - - # Done here: - return file_uploaded - - async def upload_from_stream(self): pass - - # ┳┓ ┓ • - # ┃┃┏┓┃┏┓╋┓┏┓┏┓ - # ┻┛┗ ┗┗ ┗┗┛┗┗┫ - # ┛ - - async def delete_dir(self): pass - - async def delete_file(self): pass - - # ┏┓ • • - # ┃┃┏┓┏┓┏┳┓┓┏┏┓┏┓┏┓┏ - # ┣┛┗ ┛ ┛┗┗┗┛┛┗┗┛┛┗┛ - - async def update_permissions(self): pass - - async def make_public(self): pass - - async def make_private(self): pass - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - import asyncio - import time - - async def main(): - - files_mongo = AsyncMongoStorage( - connection_string = r"mongodb://del.ditscentre.in:27017,wtt.ditscentre.in:27017,mum.arh.001.ditscentre.in:27017/admin?tls=true&tlsCAFile=%2Fetc%2Fssl%2Fcerts%2Fmongo_data_ca.pem&tlsCertificateKeyFile=%2Fetc%2Fssl%2Fcerts%2Fmongo_data_cert.pem&replicaSet=dits_mongod_rep&readPreference=primary&authMechanism=MONGODB-X509&authSource=%24external", - database_name = "converseStore", - max_connections = 10, - debug = True - ) - await files_mongo.connect() - my_fs = FileObjectManagementModel() - - user_bhopli = CoreUserInfoModel( - fullName = "Bhopli Narangi", - userId = 1, - entityId = 2, - billingAccountId = 3, - departmentId = 4, - branchId = 5, - industry = "technology" - ) - - user_polki = CoreUserInfoModel( - fullName = "Polki Muchhwaali", - userId = 6, - entityId = 7, - billingAccountId = 8, - departmentId = 9, - branchId = 10, - # industry = "finance" - ) - - file_info = await my_fs.info( - mongo_conn = files_mongo, - file_object_id = "67598d48c1bf89b25695f20b" - ) - print("SUCCESS:", file_info.success) - print("MESSAGE:", file_info.message) - print("FILE INFO:", json.to_string(file_info.result.model_dump(), default = str)) - is_owner = my_fs.has_permission( - mongo_conn = files_mongo, - user_info = user_polki, - file_object_info = file_info.result, - permission = "write" - ) - print("HAS PERMISSION:", is_owner.model_dump_json(indent = 4)) - - - asyncio.run(main()) diff --git a/models/behaviour/mail/retrieve.py b/models/behaviour/mail/retrieve.py deleted file mode 100644 index c90a76f..0000000 --- a/models/behaviour/mail/retrieve.py +++ /dev/null @@ -1,215 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - Tuesday, 3rd Dec., 2024 - - OBJECTIVE: - - To enlist and retrieve mails for various filtering conditions. - - REFERENCES: - - N/A - - DOWNLOADS: - - N/A - -""" - -# ***************************************************************************************************************** -# ***** **** -# *** IMPORT *** -# ***** **** -# ***************************************************************************************************************** - - -# To make sibling directories accessible for imports: -import sys -sys.path.append(".") -sys.path.append("..") - -# My async utils: -from utils_v2.string import json -from utils_v2.date_time import date_time -from utils_v2.database.async_mongo_v2 import AsyncMongo - -# Base model: -from models.behaviour.base import BaseModel - -# To work with MongoDB: -from bson import ObjectId - -# To work with datatypes: -from typing import Literal, List - -# For asynchronous activities: -import asyncio - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** CLASSES *** -# ***** **** -# ***************************************************************************************************************** - - -class MailRetrieveModel(BaseModel): - - # For MongoDB: - AUTH_COLLECTION = "_authTokens" - MAIL_COLLECTION = "_messages" - - async def get_mail( - self, - mongo_conn: AsyncMongo, - mail_id: str | ObjectId - ): - - """ - Retrieves one full mail from the database. - :param mongo_conn: The instance of the database connector to use to get the mail's data. - :param mail_id: The '_id' of the document that holds the mail. - :return: Either the JSON that describes the mail or None if such a mail does not exist. - """ - - # Get the data from the database: - mail_data = await mongo_conn.find_one( - collection = self.MAIL_COLLECTION, - filter = {"_id": ObjectId(mail_id)}, - projection = { - "_id": True, - "serviceType": True, - "client": True, - "ts": True, - "readTs": True, - "payload.ts": True, - "payload.readTs": True, - "payload.from": True, - "payload.to": True, - "payload.cc": True, - "payload.bcc": True, - "payload.parts": True, - "payload.attachments": True, - "payload.labels": True, - "payload.snippet": True, - "payload.aiSnippet": True, - } - ) - - # Format the data: - if mail_data: - mail_data["mailId"] = str(mail_data.pop("_id")) - # mail_data["payload"]["ts"] = mail_data["payload"]["ts"].isoformat() - # mail_data["payload"]["readTs"] = mail_data["payload"]["readTs"].isoformat() - mail_data["ts"] = mail_data["ts"].isoformat() - mail_data["readTs"] = mail_data["readTs"].isoformat() - - # Done here: - return mail_data - - async def list_for_token_id( - self, - mongo_conn: AsyncMongo, - token_id: str | ObjectId | List[str | ObjectId], - limit: int = 25, - skip: int = 0 - ): - - """ - To enlist mails for one account. - :param mongo_conn: The instance of the database connector to use to get the mail's data. - :param token_id: The id(s) of the document in the database that holds the tokens to access the account. - :param limit: How many records to fetch. - :param skip: How many initial records to skip. useful for pagination. - :return: Either the JSON that describes the mails or None if something failed. - """ - - # Ensure that the token ids are in expected format: - if not isinstance(token_id, list): token_id = [token_id] - token_id = [ObjectId(t) for t in token_id] - - # Get the data from the database: - mails_list = await mongo_conn.find_many( - collection = self.MAIL_COLLECTION, - filter = { - "tokenId": {"$in": token_id}, - "serviceType": "email" - }, - projection = { - "_id": True, - "serviceType": True, - "client": True, - "ts": True, - "readTs": True, - "payload.ts": True, - "payload.readTs": True, - "payload.from": True, - "payload.subject": True, - "payload.labels": True, - "payload.snippet": True, - "payload.aiSnippet": True, - }, - limit = limit, - skip = skip, - sort = {"payload.ts": -1} - ) - - # Format the data: - if mails_list: - for mail_data in mails_list: - mail_data["mailId"] = str(mail_data.pop("_id")) - # mail_data["payload"]["ts"] = mail_data["payload"]["ts"].isoformat() - # mail_data["payload"]["readTs"] = mail_data["payload"]["readTs"].isoformat() - mail_data["ts"] = mail_data["ts"].isoformat() - mail_data["readTs"] = mail_data["readTs"].isoformat() - - # Done here: - return mails_list - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - pass diff --git a/models/behaviour/mail/sync_v2.py b/models/behaviour/mail/sync_v2.py deleted file mode 100644 index a6da37c..0000000 --- a/models/behaviour/mail/sync_v2.py +++ /dev/null @@ -1,593 +0,0 @@ -""" - - 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.api.mail.sync import MailSyncOneResult, MailSyncManyResults - -# To work with MongoDB: -from bson import ObjectId -from pymongo import InsertOne, UpdateOne, ReplaceOne - -# To work with LLMs: -from models.behaviour.ai.llm.open_ai import LLMOpenAI -from models.data.api.ai.llm import LLMInput - -# 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 = [ - { - "role": "system", - "content": ( - "You're a mail summary expert that summarizes mails in 150 chars or less. " - "If available, show login info like username and OTPs in your summary." - "If no login info is provided, please don't worry; just summarize what you see." - ) - } - ] - - # ┏┓ ┓ - # ┣┫╋╋┏┓┏┣┓┏┳┓┏┓┏┓╋┏ - # ┛┗┗┗┗┻┗┛┗┛┗┗┗ ┛┗┗┛ - - @staticmethod - async def __save_one_attachment( - 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, - user_info: dict, - mongo_conn: AsyncMongo, - mail_client: AsyncGMailClient, - tokens: GoogleAuthTokens, - message_id: str, - llm: LLMOpenAI = 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 user_info: The information of the user (derived from his session token). - :param mongo_conn: The instance of the connection to the database to use. - :param mail_client: The instance of the mail client to use to perform the action. - :param tokens: The tokens to use to fetch the mails. - :param 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 - }, - "user_info": { - "entityId": user_info["entityId"], - "billingAccountId": user_info["billingAccountId"] - } - }), - 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 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 - - # 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: - - # Invoke the LLM: - llm_response = response = await llm.invoke( - mongo_conn = mongo_conn, - user_info = user_info, - llm_input = LLMInput( - messages = self.PROMPT_TEMPLATE + [ - { - "role": "human", - "content": f"Please summarize this mail: \"\"\"{client_response.data['unformattedText']}\"\"\"" - } - ] - ) - ) - - # Format the response: - llm_json = { - "ts": llm_response.ts, - "snippet": llm_response.output, - "tokens": llm_response.tokens.model_dump() - } - - # Add the LLM's response to the main data: - 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, - user_info: dict, - mongo_conn: AsyncMongo, - token_id: ObjectId, - mail_client: AsyncGMailClient, - tokens: GoogleAuthTokens, - llm: LLMOpenAI = 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 user_info: The information of the user (derived from his session token). - :param mongo_conn: The instance of the connection to the database to use. - :param token_id: The id of the document in the database that holds the tokens to access the account. - Needed only for refreshing the tokens and saving them. - :param mail_client: The instance of the mail client to use to perform the action. - :param tokens: The tokens to use to fetch the mails. - :param llm: The instance of the LLM to use to summarize the mail's content. - :param force_sync: Whether you would like to forcefully re-sync the mail even if it is already present in the - database. - :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, - token_id = token_id, - client_user_id = tokens.client_user_id, - 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, - user_info = user_info, - 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", - "tokenId": ObjectId(token_id), - "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 - ) - - # Apply the labels to the read messages: - try: - client_response = await mail_client.modify_messages( - tokens = tokens, - message_ids = [v["id"] for v in messages_list.values()], - add_label_ids = [tokens.labels.get("TCAOFF", {}).get("id")] - ) - except Exception as exception: - self._printer(exception) - - # 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, - user_info: dict, - mongo_conn: AsyncMongo, - token_id: ObjectId, - llm: LLMOpenAI = 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 user_info: The information of the user (derived from his session token). - :param mongo_conn: The instance of the connection to the database to use. - :param token_id: The id of the document in the database that holds the tokens to access the account. - Needed only for refreshing the tokens and saving them. - :param 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, - token_id = token_id, - ) - - # If we failed to load the authorization tokens: - if not auth_json: - sync_results.message = f"no such token id '{token_id}'" - return sync_results - - # ┏┓ ┏┓┳┳┓ •┓ - # ┣ ┏┓┏┓ ┃┓┃┃┃┏┓┓┃ - # ┻ ┗┛┛ ┗┛┛ ┗┗┻┗┗ - - if auth_json["client"] == "gmail": - return await self.__sync_many_gmail( - session_token = session_token, - user_info = user_info, - mongo_conn = mongo_conn, - token_id = token_id, - 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/behaviour/sms/auth.py b/models/behaviour/sms/auth.py deleted file mode 100644 index 6d5e667..0000000 --- a/models/behaviour/sms/auth.py +++ /dev/null @@ -1,240 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - Thursday, 5th Dec., 2024 - - OBJECTIVE: - - To work with auth details of SMS clients like Nimbus SMS (India) and Savvy Bulk SMS (Kenya). - - REFERENCES: - - N/A - - DOWNLOADS: - - N/A - -""" - -# ***************************************************************************************************************** -# ***** **** -# *** IMPORT *** -# ***** **** -# ***************************************************************************************************************** - - -# To make sibling directories accessible for imports: -import sys -sys.path.append(".") -sys.path.append("..") - -# 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 - -# Base model: -from models.behaviour.base import BaseModel - -# To work with MongoDB: -from bson import ObjectId - -# To work with datatypes: -from typing import Literal - -# To make deep-copies: -import copy - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** CLASSES *** -# ***** **** -# ***************************************************************************************************************** - - -class SMSAuthModel(BaseModel): - - AUTH_COLLECTION = "_authTokens" - - async def set( - self, - db_conn: AsyncMySQL, - mongo_conn: AsyncMongo, - user_info: dict, - client_user_id: dict, - auth: dict, - token: dict, - service_client: Literal["nimbusSmsIndia", "savvyBulkSmsKenya"], - auth_type: Literal["auth"], - sync_freq: Literal[60, 300, 900] = 300, - session_token: str = None - ) -> ObjectId | None: - - """ - To store auth/tokens for a particular service to the database. - :param db_conn: The database connection (MariaDB) to use to perform the action. - :param mongo_conn: The database connection (MongoDB) to use to perform the action. - :param user_info: The dictionary that has the user's session information. - :param client_user_id: The way the third-party client recognizes your user. - :param auth: The authentication details of the account. - :param token: The token granted by the third-party service. - :param service_client: The name of the company or brand that is providing this service that is being integrated. - :param auth_type: To identify the type of authentication being done here. This could indicate simple password - authentication, more advance OAuth2.0 authentication, etc. - :param sync_freq: The time interval in which mails need to be sync'd. Specify this in seconds. - :param session_token: The session token of the user who requested this service. - :return: An ObjectId to later store the granted tokens. - """ - - # Note down the timestamp at which this event occurred: - request_ts = date_time.get_current_utc_date_time(as_string = False) - - # Get the identifier from the database: - mongo_json = await mongo_conn.find_one_and_update( - collection = self.AUTH_COLLECTION, - filter = mongo_conn.dict_to_dot_notation({ - "serviceType": "email", - "user": { - "entityId": user_info["entityId"], - "billingAccountId": user_info["billingAccountId"] - }, - "clientUserId": client_user_id - }), - update = { - "$set": { - "lastRequestTs": request_ts, - "status": "active", - "syncFreq": max(sync_freq, 60) - }, - "$setOnInsert": { - "version": "1.0.0", - "serviceType": "sms", - "client": service_client, - "authType": auth_type, - "user": user_info, - "clientUserId": client_user_id, - "auth": auth, - "token": token, - "firstRefreshTs": None, - "lastRefreshTs": None, - "firstRequestTs": request_ts - } - }, - projection = { - "_id": True - }, - upsert = True, - return_updated = True - ) - - # Tell MariaDB that an authorization request was initiated: - db_json = {} - if mongo_json is not None: - db_json = await self.call_procedure( - db_conn = db_conn, - proc_name = "entity_integration_save", - proc_args = ( - user_info["entityId"], # ........................................... 'p_entity_id' - service_client, # .................................................. 'p_provider' - "Active", # ........................................................ 'p_current_status' - "Auth Details Accepted", # ......................................... 'p_last_action' - None, # ............................................................ 'p_display_name' - None, # ............................................................ 'p_display_picture' - str(mongo_json["_id"]), # .......................................... 'p_token_id' - json.to_string(python_data = client_user_id, no_space = True), # ... 'p_notes' - user_info["userId"] # .............................................. 'p_created_by' - ), - session_token = session_token - ) - - # Done here: - return mongo_json["_id"] if mongo_json and db_json.get("status") == 1 else None - - async def get( - self, - mongo_conn: AsyncMongo, - token_id: ObjectId | str = None, - **kwargs - ) -> dict | None: - - """ - To retrieve stored auth/tokens from the database. - :param mongo_conn: The database connection (MongoDB) to use to perform the action. - :param token_id: The identifier granted providing auth details for the first time in 'set_token'. - :param kwargs: Any set of key-value pairs to build custom search criteria. This could be things like the user - info, the client, the type of authentication used, or even the kind of service. - :return: The retrieved record that has the token, and information about the service and client if found, else - None when there is no matching record. - """ - - # Build the filter: - filter_json = {k: v for k, v in kwargs.items()} - if token_id: filter_json["_id"] = ObjectId(token_id) - - # If there is no search criteria, we exit with failure: - if not filter_json: return None - - # If there is some filtering possible, - # we fetch and return the token: - return await mongo_conn.find_one( - collection = self.AUTH_COLLECTION, - filter = filter_json, - projection = { - "_id": True, - "serviceType": True, - "authType": True, - "client": True, - "clientUserId": True, - "token": True - } - ) - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - pass diff --git a/models/behaviour/sms/auth_v2.py b/models/behaviour/sms/auth_v2.py deleted file mode 100644 index 2e2320f..0000000 --- a/models/behaviour/sms/auth_v2.py +++ /dev/null @@ -1,227 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - ORIGINAL: Thursday, 5th Dec., 2024 - UPGRADED: Monday, 9th Dec., 2024 - - OBJECTIVE: - - To work with auth details of SMS clients like Nimbus SMS (India) and Savvy Bulk SMS (Kenya). - - REFERENCES: - - N/A - - DOWNLOADS: - - N/A - -""" - -# ***************************************************************************************************************** -# ***** **** -# *** IMPORT *** -# ***** **** -# ***************************************************************************************************************** - - -# To make sibling directories accessible for imports: -import sys -sys.path.append(".") -sys.path.append("..") - -# 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 - -# Base model: -from models.behaviour.base import BaseModel - -# Data models: -from models.data.core.auth_token import CoreAuthTokenModel - -# To work with MongoDB: -from bson import ObjectId - -# To work with datatypes: -from typing import Literal - -# To make deep-copies: -import copy - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** CLASSES *** -# ***** **** -# ***************************************************************************************************************** - - -class SMSAuthModel(BaseModel): - - AUTH_COLLECTION = "_authTokens" - - async def set( - self, - db_conn: AsyncMySQL, - mongo_conn: AsyncMongo, - auth_token: CoreAuthTokenModel, - session_token: str = None - ) -> ObjectId | None: - - """ - To store auth/tokens for a particular service to the database. - :param db_conn: The database connection (MariaDB) to use to perform the action. - :param mongo_conn: The database connection (MongoDB) to use to perform the action. - :param auth_token: An instance of the core auth-token model that holds data in the database. - :param session_token: The session token of the user who requested this service. - :return: An ObjectId to later store the granted tokens. - """ - - # Note down the timestamp at which this event occurred: - request_ts = date_time.get_current_utc_date_time(as_string = False) - - # Get the identifier from the database: - # BE CAREFUL WITH THE KEYS HERE, THEY SHOULD MATCH THE FIELDS OF THE CORE AUTH-TOKEN MODEL: - mongo_json = await mongo_conn.find_one_and_update( - collection = self.AUTH_COLLECTION, - filter = mongo_conn.dict_to_dot_notation({ - "serviceType": auth_token.serviceType, - "user": { - "entityId": auth_token.user.entityId, - "billingAccountId": auth_token.user.billingAccountId - }, - "clientUserId": auth_token.clientUserId - }), - update = { - "$set": { - "lastRequestTs": auth_token.lastRequestTs, - "status": auth_token.status, - "syncFreq": auth_token.syncFreq - }, - "$setOnInsert": { - "version": auth_token.version, - "serviceType": auth_token.serviceType, - "client": auth_token.client, - "authType": auth_token.authType, - "user": auth_token.user.model_dump(), - "clientUserId": auth_token.clientUserId, - "auth": auth_token.auth, - "token": auth_token.token, - "firstRefreshTs": auth_token.firstRefreshTs, - "lastRefreshTs": auth_token.lastRefreshTs, - "firstRequestTs": auth_token.firstRequestTs or request_ts - } - }, - projection = { - "_id": True - }, - upsert = True, - return_updated = True - ) - - # Tell MariaDB that an authorization request was initiated: - db_json = {} - if mongo_json is not None: - token_notes = auth_token.clientUserId - db_json = await self.call_procedure( - db_conn = db_conn, - proc_name = "entity_integration_save", - proc_args = ( - auth_token.user.entityId, # ..................................... 'p_entity_id' - auth_token.client, # ............................................ 'p_provider' - auth_token.status, # ............................................ 'p_current_status' - "Auth Details Accepted", # ...................................... 'p_last_action' - None, # ......................................................... 'p_display_name' - None, # ......................................................... 'p_display_picture' - str(mongo_json["_id"]), # ....................................... 'p_token_id' - json.to_string(python_data = token_notes, no_space = True), # ... 'p_notes' - auth_token.user.userId # ........................................ 'p_created_by' - ), - session_token = session_token - ) - - # Done here: - return mongo_json["_id"] if mongo_json and db_json.get("status") == 1 else None - - async def get( - self, - mongo_conn: AsyncMongo, - token_id: ObjectId | str = None, - **kwargs - ) -> dict | None: - - """ - To retrieve stored auth/tokens from the database. - :param mongo_conn: The database connection (MongoDB) to use to perform the action. - :param token_id: The identifier granted providing auth details for the first time in 'set_token'. - :param kwargs: Any set of key-value pairs to build custom search criteria. This could be things like the user - info, the client, the type of authentication used, or even the kind of service. - :return: The retrieved record that has the token, and information about the service and client if found, else - None when there is no matching record. - """ - - # Build the filter: - filter_json = {k: v for k, v in kwargs.items()} - if token_id: filter_json["_id"] = ObjectId(token_id) - - # If there is no search criteria, we exit with failure: - if not filter_json: return None - - # If there is some filtering possible, we fetch the token: - token = await mongo_conn.find_one( - collection = self.AUTH_COLLECTION, - filter = filter_json, - ) - - # Done here: - return CoreAuthTokenModel(**token) if token else None - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - pass diff --git a/models/behaviour/sms/send.py b/models/behaviour/sms/send.py deleted file mode 100644 index 922fb4a..0000000 --- a/models/behaviour/sms/send.py +++ /dev/null @@ -1,214 +0,0 @@ -""" - - AUTHOR: - - Khushal P Soonderji - - DATE: - - Monday, 9th Dec., 2024 - - OBJECTIVE: - - To send SMS from clients like Nimbus SMS (India) and Savvy Bulk SMS (Kenya). - - REFERENCES: - - N/A - - DOWNLOADS: - - N/A - -""" - -# ***************************************************************************************************************** -# ***** **** -# *** IMPORT *** -# ***** **** -# ***************************************************************************************************************** - - -# To make sibling directories accessible for imports: -import sys -sys.path.append(".") -sys.path.append("..") - -# 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 - -# SMS-related utils: -from utils_v2.sms.models.behaviour.nimbus.async_nimbus import AsyncNimbusSMS -from utils_v2.sms.models.behaviour.savvy_bulk_sms.async_savvy_bulk_sms import AsyncSavvyBulkSMS - -# Base model: -from models.behaviour.base import BaseModel - -# Data models: -from models.data.core.auth_token import CoreAuthTokenModel -from models.data.core.message import CoreMessageModel -from models.data.api.sms.send import ( - SMSSendRequestHeaders, - SMSSendRequestData, - NimbusSMSIndiaMessage, - SavvyBulkSMSKenyaMessage -) -from utils_v2.sms.models.data.sms_message import SentSMSMessageModel - -# To work with MongoDB: -from bson import ObjectId - -# To work with datatypes: -from typing import Literal - -# To make deep-copies: -import copy - - -# ***************************************************************************************************************** -# ***** **** -# *** MACROS / ONE-TIME INIT *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** VARIABLES *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** FUNCTIONS *** -# ***** **** -# ***************************************************************************************************************** - - -# --- Nothing Yet - - -# ***************************************************************************************************************** -# ***** **** -# *** CLASSES *** -# ***** **** -# ***************************************************************************************************************** - - -class SMSSendModel(BaseModel): - - MESSAGES_COLLECTION = "_messages" - - async def send_sms( - self, - mongo_conn: AsyncMongo, - token_id: ObjectId | str, - auth_token: CoreAuthTokenModel, - inbound_data: SMSSendRequestData, - session_token: str = None - ) -> SentSMSMessageModel: - - """ - To store auth/tokens for a particular service to the database. - :param mongo_conn: The database connection (MongoDB) to use to perform the action. - :param auth_token: An instance of the core auth-token model that holds data in the database. - :param session_token: The session token of the user who requested this service. - :return: An ObjectId to later store the granted tokens. - """ - - # Basic prep: - event_ts = date_time.get_current_utc_date_time(as_string = False) - client_response = None - message_id = None - sms_sent = None - - # ┏┓ ┳┓• ┓ ┏┓┳┳┓┏┓ ┳ ┓• - # ┣ ┏┓┏┓ ┃┃┓┏┳┓┣┓┓┏┏ ┗┓┃┃┃┗┓ ┃┏┓┏┫┓┏┓ - # ┻ ┗┛┛ ┛┗┗┛┗┗┗┛┗┻┛ ┗┛┛ ┗┗┛ ┻┛┗┗┻┗┗┻ - - if isinstance(inbound_data.message, NimbusSMSIndiaMessage): - - # Prepare the client: - sms_client = AsyncNimbusSMS( - entity_id = auth_token.auth.get("entityId"), - sender_id = auth_token.auth.get("senderId"), - user_id = auth_token.auth.get("userId"), - api_key = auth_token.auth.get("apiKey"), - http_client = self._http_client - ) - - # Send the SMS: - client_response = await sms_client.send_sms( - recipient_number = inbound_data.message.recipientNo, - message = inbound_data.message.text, - template_id = inbound_data.message.templateId - ) - - # ┏┓ ┏┓ ┳┓ ┓┓ ┏┓┳┳┓┏┓ ┓┏┓ - # ┣ ┏┓┏┓ ┗┓┏┓┓┏┓┏┓┏ ┣┫┓┏┃┃┏ ┗┓┃┃┃┗┓ ┃┫ ┏┓┏┓┓┏┏┓ - # ┻ ┗┛┛ ┗┛┗┻┗┛┗┛┗┫ ┻┛┗┻┗┛┗ ┗┛┛ ┗┗┛ ┛┗┛┗ ┛┗┗┫┗┻ - # ┛ ┛ - - elif isinstance(inbound_data.message, SavvyBulkSMSKenyaMessage): - - # Prepare the client: - sms_client = AsyncSavvyBulkSMS( - api_key = auth_token.auth.get("apiKey"), - partner_id = auth_token.auth.get("partnerId"), - short_code = auth_token.auth.get("shortCode"), - http_client = self._http_client - ) - - # Send the SMS: - client_response = await sms_client.send_sms( - recipient_number = inbound_data.message.recipientNo, - message = inbound_data.message.text - ) - - # ┏┓ ┏┳┓┓ ┳┳┓ - # ┗┓┏┓┓┏┏┓ ┃ ┣┓┏┓ ┃┃┃┏┓┏┏┏┓┏┓┏┓ - # ┗┛┗┻┗┛┗ ┻ ┛┗┗ ┛ ┗┗ ┛┛┗┻┗┫┗ - # ┛ - - # Save the message: - if client_response: - message_id = await mongo_conn.insert_one( - collection = self.MESSAGES_COLLECTION, - document = CoreMessageModel( - ts = event_ts, - readTs = event_ts, - tokenId = ObjectId(token_id), - serviceType = auth_token.serviceType, - client = auth_token.client, - clientMessageId = client_response.messageId, - clientThreadId = None, - isInward = False, - sentSuccessfully = client_response.success, - payload = client_response.model_dump() - ).model_dump() - ) - - # Done here: - return client_response - - -# ***************************************************************************************************************** -# ***** **** -# *** MAIN PROGRAM *** -# ***** **** -# ***************************************************************************************************************** - - -if __name__ == "__main__": - - pass diff --git a/models/data/api/mail/__init__.py b/models/core/__init__.py similarity index 100% rename from models/data/api/mail/__init__.py rename to models/core/__init__.py diff --git a/models/data/api/sms/__init__.py b/models/core/ai/__init__.py similarity index 100% rename from models/data/api/sms/__init__.py rename to models/core/ai/__init__.py diff --git a/models/core/ai/llm.py b/models/core/ai/llm.py new file mode 100644 index 0000000..0989246 --- /dev/null +++ b/models/core/ai/llm.py @@ -0,0 +1,255 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Thursday, 5th Dec., 2024. + + OBJECTIVE: + + To provide a structure to normalize input to and output from a standardized LLM wrapper. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For making data behaviour_models: +from pydantic import BaseModel, Field, field_validator, PastDatetime, AwareDatetime +from typing import Optional, Literal, Union, List, Any + +# My utils: +from utils_v2.string import regex +from utils_v2.date_time import date_time + +# To work with date and time: +import datetime + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# RegEx Patterns: +REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$" + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +class LLMInputMessage(BaseModel): + + role: Literal["system", "ai", "human"] = Field( + description = "the role of this message", + frozen = True + ) + + content: str = Field( + description = "the message sent by the 'role'", + frozen = True + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + + +# --------------------------------------------------------------------------------------------------------------------- + + +class LLMInput(BaseModel): + + messages: List[LLMInputMessage] + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + + # ┓┏ ┓• ┓ • + # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ + # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ + + @field_validator("messages") + def validate_messages(cls, value): + + # Maintain counter(s): + system_message_index = -1 + system_message_count = 0 + + # Loop through the messages and check them: + for index, message in enumerate(value): + + # For 'system' messages: + if message.role == "system": + system_message_index = index + system_message_count += 1 + + # Verify that there is AT MOST ONE 'system' message, + # and verify that the 'system' message is the first message: + if system_message_count > 1: raise ValueError(f"there can be at most 1 'system' message, found {system_message_count}") + if system_message_index > 0: raise ValueError(f"'system' message must always be at index 0, found it at index {system_message_index}") + + # Done here: + return value + + +# --------------------------------------------------------------------------------------------------------------------- + + +class LLMUsageTokens(BaseModel): + + input: int = Field( + description = "how many tokens were given in the input", + frozen = True + ) + + output: int = Field( + description = "how many tokens were generated as the output", + frozen = True + ) + + total: int = Field( + description = "the sum of the input and output tokens", + frozen = True + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + + +# --------------------------------------------------------------------------------------------------------------------- + + +class LLMOutput(BaseModel): + + ts: AwareDatetime = Field( + description = "the time at which the llm was invoked", + default_factory = date_time.get_current_utc_date_time, + frozen = True + ) + + messages: List[LLMInputMessage] = Field( + description = "the messages that came in that invoked the llm", + frozen = True + ) + + output: str | None = Field( + description = "what the llm generated", + default = None, + frozen = True + ) + + client: Literal["openai"] = Field( + description = "the co./brand that was used to use an llm", + frozen = True + ) + + model: str = Field( + description = "to know which model used in the process", + frozen = True + ) + + tokens: LLMUsageTokens = Field( + description = "to know how many tokens were used in the process", + default = LLMUsageTokens(input = 0, output = 0, total = 0), + frozen = True + ) + + invocationId: Any | None = Field( + description = "the id of the document that notes this invocation; useful for reconciliation", + frozen = False, + default = None + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + # llm_messages = [ + # { + # "role": "system", + # "content": "You are an office assistant." + # }, + # { + # "role": "ai", + # "content": "Hello, sir. How may I help you today?" + # }, + # { + # "role": "human", + # "content": "Please summarize this mail for me..." + # } + # ] + # + # llm_input = LLMInput(messages = llm_messages) + # print(llm_input.model_dump_json(indent = 4)) + + llm_output = LLMOutput( + messages=[LLMInputMessage(role='system', content="You are an office assistant. It's Christmas, so definitley respond like Santa Claus."), LLMInputMessage(role='ai', content='Hello, sir. How may I help you today?'), LLMInputMessage(role='human', content='Please summarize this mail for me...')], + client = "openai", + model = "o1" + ) diff --git a/models/data/core/auth_token.py b/models/core/auth_token.py similarity index 99% rename from models/data/core/auth_token.py rename to models/core/auth_token.py index 5f74088..a97e0db 100644 --- a/models/data/core/auth_token.py +++ b/models/core/auth_token.py @@ -44,7 +44,7 @@ from utils_v2.string import regex from utils_v2.date_time import date_time # Other core models: -from models.data.core.user import CoreUserInfoModel +from models.core.user import CoreUserInfoModel # To work with MongoDB: from bson.objectid import ObjectId diff --git a/models/data/core/file.py b/models/core/file.py similarity index 100% rename from models/data/core/file.py rename to models/core/file.py diff --git a/models/data/core/file_object.py b/models/core/file_object.py similarity index 100% rename from models/data/core/file_object.py rename to models/core/file_object.py diff --git a/models/data/core/message.py b/models/core/message.py similarity index 90% rename from models/data/core/message.py rename to models/core/message.py index 0eaacee..93904ea 100644 --- a/models/data/core/message.py +++ b/models/core/message.py @@ -37,7 +37,7 @@ sys.path.append("..") # For making data behaviour_models: from pydantic import BaseModel, Field, field_validator, PastDatetime, AwareDatetime -from typing import Optional, Literal, Union +from typing import Optional, Literal, Union, List, Any # My utils: from utils_v2.string import regex @@ -46,6 +46,9 @@ from utils_v2.date_time import date_time # To work with MongoDB: from bson.objectid import ObjectId +# Data models: +from models.core.ai.llm import LLMOutput + # To work with date and time: import datetime @@ -141,7 +144,8 @@ class CoreMessageModel(BaseModel): isSent: bool = Field( description = "to understand whether this message was an incoming message or outgoing message", - frozen = True + frozen = False, + default = False ) isBroadcast: bool = Field( @@ -156,11 +160,28 @@ class CoreMessageModel(BaseModel): default = False ) + aiSnippet: LLMOutput | None = Field( + description = "holds a short summary generated by ", + frozen = False + ) + + preview: str = Field( + description = "a truncated version of the actual textual content of the message", + frozen = False + ) + message: dict = Field( description = "the actual contents of the message; will differ for each client", frozen = True ) + tags: List[Any] = Field( + description = "a list of keywords to apply to this file/dir to filter it later", + frozen = False, + default = [], + examples = ["urgent", "otp", "GST"] + ) + usedAi: bool | None = Field( description = "to mark when a sent message was generated by ai; null means the status is not known", frozen = True, @@ -195,6 +216,11 @@ class CoreMessageModel(BaseModel): except: pass return value + @field_validator("tags", mode = "before") + def validate_tags(cls, value): + if value is None: value = [] + return value + # ***************************************************************************************************************** # ***** **** diff --git a/models/data/core/payment.py b/models/core/payment.py similarity index 95% rename from models/data/core/payment.py rename to models/core/payment.py index e651fbc..a3b18ed 100644 --- a/models/data/core/payment.py +++ b/models/core/payment.py @@ -37,7 +37,7 @@ sys.path.append("..") # For making data behaviour_models: from pydantic import BaseModel, Field, field_validator, PastDatetime, AwareDatetime -from typing import Optional, Literal, Union, List +from typing import Optional, Literal, Union, List, Any # My utils: from utils_v2.string import regex @@ -171,6 +171,13 @@ class CorePaymentModel(BaseModel): frozen = True ) + tags: List[Any] = Field( + description = "a list of keywords to apply to this file/dir to filter it later", + frozen = False, + default = [], + examples = ["renewal", "subscription"] + ) + client: Literal["razorpay", "safaricomMPesaExpress"] = Field( description = "the third-part client that was used", frozen = True @@ -221,6 +228,11 @@ class CorePaymentModel(BaseModel): if currency is None: raise ValueError("invalid currency code, please use iso 4217 standard") return value + @field_validator("tags", mode = "before") + def validate_tags(cls, value): + if value is None: value = [] + return value + # ***************************************************************************************************************** # ***** **** diff --git a/models/data/core/user.py b/models/core/user.py similarity index 100% rename from models/data/core/user.py rename to models/core/user.py diff --git a/models/data/api/tech/__init__.py b/models/data/api/tech/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/models/data/core/__init__.py b/models/data/core/__init__.py deleted file mode 100644 index e69de29..0000000