diff --git a/api/blueprints/ai/llm/invoke.py b/api/blueprints/ai/llm/invoke.py index 523bed0..ca06063 100644 --- a/api/blueprints/ai/llm/invoke.py +++ b/api/blueprints/ai/llm/invoke.py @@ -164,7 +164,7 @@ async def invoke_llm( # Call the LLM and see if its service worked or not: llm_response = await current_app.llm.invoke( - mongo_conn = current_app.data_mongo, + mongo_data_conn = current_app.data_mongo, user_info = CoreUserInfoModel(**kwargs["session_info"]), llm_input = inbound_data ) diff --git a/api/blueprints/message/mail/sync/sync_v3.py b/api/blueprints/message/mail/sync/sync_v3.py index af974c6..a401b89 100644 --- a/api/blueprints/message/mail/sync/sync_v3.py +++ b/api/blueprints/message/mail/sync/sync_v3.py @@ -268,8 +268,11 @@ async def sync_mail( message = sync_results.message, data = { "totalCount": sync_results.totalCount, + "attemptedCount": sync_results.attemptedCount, "successCount": sync_results.successCount, - "failureCount": sync_results.failureCount + "failureCount": sync_results.failureCount, + "newCount": sync_results.newCount, + "existingCount": sync_results.totalCount - sync_results.newCount } ) diff --git a/controllers/api/mail.py b/controllers/api/mail.py index 2a05fca..45b4829 100644 --- a/controllers/api/mail.py +++ b/controllers/api/mail.py @@ -55,7 +55,7 @@ from models.message.mail.send import MailSendOneResult from models.api.message.mail.send import MailSendRequestData # Mail Clients: -from utils_v2.goog.controllers.gmail.gmail_client import AsyncGMailClient +from utils_v2.goog.controllers.gmail.gmail_client import AsyncGmailClient from utils_v2.goog.models.auth_tokens import GoogleAuthTokens # To work with MongoDB: @@ -342,7 +342,7 @@ class MailController: mongo_conn: AsyncMongo, user_info: CoreUserInfoModel, auth_token: CoreAuthTokenModel, - mail_client: AsyncGMailClient, + mail_client: AsyncGmailClient, google_tokens: GoogleAuthTokens, message_id: str, llm: CoreLLMController = None, @@ -443,7 +443,7 @@ class MailController: mongo_conn: AsyncMongo, user_info: CoreUserInfoModel, auth_token: CoreAuthTokenModel, - mail_client: AsyncGMailClient, + mail_client: AsyncGmailClient, llm: CoreLLMController = None, force_sync: bool = False, start_date: datetime.datetime = None, diff --git a/controllers_v2/core/ai/llm.py b/controllers_v2/core/ai/llm.py index 92c543b..3b6baf7 100644 --- a/controllers_v2/core/ai/llm.py +++ b/controllers_v2/core/ai/llm.py @@ -128,7 +128,7 @@ class CoreLLMController(BaseModel): async def invoke( self, - mongo_conn: AsyncMongo, + mongo_data_conn: AsyncMongo, user_info: CoreUserInfoModel, llm_input: LLMInput ) -> LLMOutput: @@ -158,7 +158,7 @@ class CoreLLMController(BaseModel): # Store this into MongoDB: mongo_document = {"user": user_info.model_dump()} for k, v in llm_response.model_dump().items(): mongo_document[k] = v - inserted_id = await mongo_conn.insert_one( + inserted_id = await mongo_data_conn.insert_one( collection = self.AI_USAGE_COLLECTION, document = mongo_document ) diff --git a/controllers_v2/message/mail/gmail.py b/controllers_v2/message/mail/gmail.py index 0a153aa..fc72764 100644 --- a/controllers_v2/message/mail/gmail.py +++ b/controllers_v2/message/mail/gmail.py @@ -474,31 +474,32 @@ class GmailController(MailController): # Start by assuming failure: sync_result = MailSyncOneResult() - # If we've not been forced to re-sync the mail message, - # we first check if the mail already exists in our database: - if not force_sync: - mail_records = await self.get_message_previews( - mongo_data_conn = mongo_data_conn, - token_ids = [ObjectId(auth_token.authTokenId)], - limit = 1, - skip = 0, - additional_filter = { - "tokenId": ObjectId(auth_token.authTokenId), - "serviceType": auth_token.serviceType, - "client": auth_token.client, - "clientMessageId": message_id - } + # Check if the mail already exists in your database: + mail_records = await self.get_message_previews( + mongo_data_conn = mongo_data_conn, + token_ids = [ObjectId(auth_token.authTokenId)], + limit = 1, + skip = 0, + additional_filter = { + "tokenId": ObjectId(auth_token.authTokenId), + "serviceType": auth_token.serviceType, + "client": auth_token.client, + "clientMessageId": message_id + } + ) + if mail_records: + sync_result.success = True + sync_result.isNew = False + sync_result.message = ( + f"Gmail message '{message_id}' already " + f"sync'd on {mail_records[0].syncTs} (UTC)." ) - if mail_records: - sync_result.success = True - sync_result.message = ( - f"Gmail message '{message_id}' already " - f"sync'd on {mail_records[0].syncTs} (UTC)." - ) - print(sync_result) - return sync_result + + # If the mail already exists and we have not been asked to force-sync: + if not sync_result.isNew and not force_sync: return sync_result # Now that we know that we have to fetch the mail from GMail: + sync_result.attempted = True client_response = await mail_client.get_message( tokens = google_tokens, message_id = message_id, @@ -660,8 +661,14 @@ class GmailController(MailController): 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 + + # Maintain the result counters: + if result.attempted: + if result.success: sync_results.successCount += 1 + else: sync_results.failureCount += 1 + if result.isNew: sync_results.newCount += 1 + + # Create the database requests: if result.mailMessage: replacement_json = result.mailMessage.model_dump() replacement_json.pop("_id", None) @@ -678,7 +685,7 @@ class GmailController(MailController): ) ) - # Make the bulk insert operation: + # Make the bulk operation: if mongo_operations: sync_count = await self.bulk_operate_messages( mongo_data_conn = mongo_data_conn, diff --git a/cron/message/mail/auto_sync.py b/cron/message/mail/auto_sync.py index a97b164..52bb087 100644 --- a/cron/message/mail/auto_sync.py +++ b/cron/message/mail/auto_sync.py @@ -71,6 +71,7 @@ import time from models.core.auth_token import CoreAuthTokenModel from models.core.user import CoreUserInfoModel from models.message.mail.sync import MailSyncOneResult, MailSyncManyResults +from models.core.ai.llm import LLMInput, LLMInputMessage # For asynchronous activities: import asyncio @@ -78,6 +79,9 @@ import asyncio # To work with various datatypes: from typing import List +# For random values: +import random + # Debugging: from icecream import IceCreamDebugger @@ -317,8 +321,6 @@ async def sync_one_account(auth_token: CoreAuthTokenModel): :return: ?? """ - # printer("Sync'ing account.", auth_token.clientUserId["email"], auth_token.authTokenId) - # Note down the time at which the attempt to sync the account is being made: now = date_time.get_current_utc_date_time(as_string = False) @@ -356,13 +358,20 @@ async def sync_one_account(auth_token: CoreAuthTokenModel): await mail_controller.release_token_from_batch_by_id( mongo_data_conn = data_mongo, token_id = auth_token.authTokenId, - # sync_after_ts = now + datetime.timedelta(seconds = auth_token.syncFreq or 300), - sync_after_ts = now + datetime.timedelta(seconds = 1), + sync_after_ts = now + datetime.timedelta(seconds = auth_token.syncFreq or 300), last_sync_ts = now ) # Done here: - if sync_results.totalCount: printer(auth_token.clientUserId["email"], sync_results.successCount, sync_results.totalCount) + if sync_results.totalCount: + printer( + auth_token.clientUserId["email"], + sync_results.totalCount, + sync_results.newCount, + sync_results.attemptedCount, + sync_results.successCount, + sync_results.failureCount + ) # --------------------------------------------------------------------------------------------------------------------- @@ -379,14 +388,12 @@ async def sync_accounts(batch_size: int) -> None: while True: # Get batches of auth-tokens to work with: - # no_context_printer("Getting batch.") - # no_context_printer(batch_size) auth_tokens_batch = await mail_controller.get_batches_to_sync( mongo_data_conn = data_mongo, limit = batch_size, batch_timeout_seconds = 10 ) - # no_context_printer(len(auth_tokens_batch)) + no_context_printer(len(auth_tokens_batch)) # Process each batch: now = date_time.get_current_utc_date_time(as_string = False) @@ -394,9 +401,8 @@ async def sync_accounts(batch_size: int) -> None: results = await asyncio.gather(*tasks) # Small delay to not overload the database: - # if len(auth_tokens_batch) == 0: await asyncio.sleep(5.0) - await asyncio.sleep(1.0) - # no_context_printer() + if len(auth_tokens_batch) == 0: await asyncio.sleep(4.0 + (random.random() * 2.0)) + else: await asyncio.sleep(1.0 + (random.random() * 2.0)) # ***************************************************************************************************************** diff --git a/kill_mail_sync.sh b/kill_mail_sync.sh index d6d6db8..74b1a9c 100644 --- a/kill_mail_sync.sh +++ b/kill_mail_sync.sh @@ -2,7 +2,7 @@ # Kill all the scripts: echo "Killing the script." -pkill -9 -f "$(pwd)/background/finstitutions/trading/tick_in_stateful.py" +pkill -9 -f "$(pwd)/cron/message/mail/auto_sync.py" # All done: echo "Done!" diff --git a/models/message/mail/sync.py b/models/message/mail/sync.py index a8212da..f29e419 100644 --- a/models/message/mail/sync.py +++ b/models/message/mail/sync.py @@ -85,6 +85,16 @@ class MailSyncOneResult(BaseModel): default = False ) + attempted: bool = Field( + description = "whether, or not, sync'ing was tried for this mail", + default = False + ) + + isNew: bool = Field( + description = "whether, or not, this mail is new", + default = True + ) + message: str | None = Field( description = "a brief message to summarize the result of the process", default = None @@ -114,6 +124,16 @@ class MailSyncManyResults(BaseModel): default = 0 ) + newCount: int = Field( + description = "how many of the total mails were new", + default = 0 + ) + + attemptedCount: int = Field( + description = "how many mails were attempted to sync.", + default = 0 + ) + successCount: int = Field( description = "the no. of mails that were successfully sync'd", default = 0