""" 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 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 CoreMessageController(BaseModel): # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ # For MongoDB: MESSAGES_COLLECTION = "_messages" # ┏┓┳┓┳┳┳┓ ┏┓ # ┃ ┣┫┃┃┃┃ ━━ ┃ ┏┓┏┓┏┓╋┏┓ # ┗┛┛┗┗┛┻┛ ┗┛┛ ┗ ┗┻┗┗ async def insert( self, mongo_conn: AsyncMongo, message: CoreMessageModel ) -> ObjectId: """ Simply insert one message document into the database. :param mongo_conn: The instance of the database connector to use for the operation. :param message: The message to save into the database. :return: The object id of the inserted document. """ # 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: list ) -> int: """ Needed in cases like forcing re-sync of mails where you need to perform actions like bulk replacements of existing documents. Not recommended to use. Please use very carefully to ensure document integrity. :param mongo_conn: The instance of the database connector to use for the operation. :param mongo_operations: The list operations that are supported by MongoDB's Bulk Write system. :return: The no. of documents affected. """ return await mongo_conn.bulk_write( collection = self.MESSAGES_COLLECTION, requests = mongo_operations, raise_exception = True ) # ┏┓┳┓┳┳┳┓ ┳┓ • # ┃ ┣┫┃┃┃┃ ━━ ┣┫┏┓╋┏┓┓┏┓┓┏┏┓ # ┗┛┛┗┗┛┻┛ ┛┗┗ ┗┛ ┗┗ ┗┛┗ 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. :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, "syncTs": True, "tokenId": True, "serviceType": True, "client": True, "clientMessageId": True, "clientThreadId": True, "isSent": True, "isBroadcast": True, "sentSuccessfully": True, "sender": True, "chat": True, "snippet": True, "aiSnippet": True, "tags": True }, raise_exception = True ) # Convert the fetched records to instances of the data model and return: for record in records: record["message"] = {} 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. """ # 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 ) # 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. :param mongo_conn: The instance of the database connector to use for the operation. :param message_id: The id of the message that needs to be read. :return: The contents of that one message in a structured format. """ # We fetch the whole payload of that one message: record = await mongo_conn.find_one( collection = self.MESSAGES_COLLECTION, filter = {"_id": ObjectId(message_id)}, 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. async def update_tags( self, mongo_conn: AsyncMongo, message_id: ObjectId | str, unset_tags: List[str] = None, set_tags: List[str] = None ) -> bool: """ Updates the tags on one message. The tags to remove are processed first, the ones to add are processed later. :param mongo_conn: The instance of the database connector to use for the operation. :param message_id: The id of the message that needs to be read. :param unset_tags: The tags to remove from the message. :param set_tags: The tags to add to the message. :return: True if the update was successful, else False. """ # Update the tags: return await mongo_conn.update_one( collection = self.MESSAGES_COLLECTION, filter = {"_id": ObjectId(message_id)}, update = [{ "$set": { "tags": { "$let": { "vars": { "removed_tags": { "$setDifference": [ "$tags", unset_tags ] } }, "in": { "$setUnion": [ "$$removed_tags", set_tags ] } } } } }], raise_exception = True ) # ┏┓┳┓┳┳┳┓ ┳┓ ┓ # ┃ ┣┫┃┃┃┃ ━━ ┃┃┏┓┃┏┓╋┏┓ # ┗┛┛┗┗┛┻┛ ┻┛┗ ┗┗ ┗┗ # No support whatsoever for deleting messages. # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass