""" AUTHOR: Khushal P Soonderji DATE: Friday, 17th Jan., 2025. OBJECTIVE: To automatically synchronize the mails by fetching data from the user's third-party mail client's server to your database. This is done in batches of auth-tokens. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import io import os import socket # My utils: from utils_v2.string import json from utils_v2.string import regex from utils_v2.system import files from utils_v2.date_time import date_time from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache from utils_v2.serialization.json_serializer import JSONSerializer # Controllers: from controllers_v2.message.mail.all_mail import AllMailController from controllers_v2.message.mail.gmail import GmailController from controllers_v2.core.ai.llm import CoreLLMController # Mail clients: from utils_v2.goog.controllers.gmail.gmail_client import AsyncGmailClient # To make HTTP calls: import httpx # To work with date and time: import datetime import time # Models: 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 # To work with various datatypes: from typing import List # For random values: import random # Debugging: from icecream import IceCreamDebugger # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Debugging: printer = IceCreamDebugger(prefix = "Mail Sync. | ", includeContext = True) no_context_printer = IceCreamDebugger(prefix = "Mail Sync. | ", includeContext = False) # To make API calls: http_client = httpx.AsyncClient( limits = httpx.Limits( max_connections = 100, # ............ Maximum number of connections allowed in the pool. max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. ), timeout = httpx.Timeout( pool = 120.0, # .... Time to wait for a free connection from the pool. connect = 2.5, # ... Time to wait for establishing a connection to the server. write = 10.0, # .... Time to wait for sending data. read = 9.9 # ....... Time to wait for receiving data. ) ) # General: SERVER_HOSTNAME = str(socket.gethostname()) # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # Session-awareness and maintenance of this script's state: SCRIPT_DATA = {} exclusive_lock = asyncio.Semaphore(1) # For databases: data_mongo: AsyncMongo | None = None sql_writer: AsyncMySQL | None = None # For caching: redis_cache: AsyncRedisCache | None = None # Controllers: mail_controller: AllMailController | None = None gmail_controller: GmailController | None = None llm_controller: CoreLLMController | None = None # Mail clients: gmail_client: AsyncGmailClient | None = None # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** async def init( script_id: str, debug: bool ): """ To initialize all credentials, instances, and connectivity for this whole script. :param script_id: The id to use to load cred and data from the internal service. :param debug: Whether, or not, you would like to print the debug messages. :return: True if initialized successfully, else False. """ # Declare the required global variables: global SCRIPT_DATA global data_mongo global sql_writer global redis_cache global mail_controller global gmail_controller global llm_controller global gmail_client # Basic stuff: if debug: printer.enable() no_context_printer("Initializing.") # ┏┓ ┓ ┓ ┳┓ # ┃ ┏┓┏┓┏┫ ┏┓┏┓┏┫ ┃┃┏┓╋┏┓ # ┗┛┛ ┗ ┗┻ ┗┻┛┗┗┻ ┻┛┗┻┗┗┻ # Get the script credentials: response = await http_client.get( url = r"https://nexcom.ditscentre.in/internal/cred/get", headers = {"X-Script-Id": script_id} ) if response.status_code not in [200]: print("FATAL: SCRIPT CREDENTIALS LOADING FAILED!") return False script_cred = response.json().get("data") # Get the script data: response = await http_client.get( url = r"https://nexcom.ditscentre.in/internal/data/get", headers = {"X-Script-Id": script_id} ) if response.status_code not in [200]: print("FATAL: SCRIPT DATA LOADING FAILED!") return False SCRIPT_DATA = response.json().get("data") # Done with this step: no_context_printer("Cred and Data loaded.") # ┳┳┓ • ┳┓┳┓ # ┃┃┃┏┓┏┓┓┏┓┃┃┣┫ # ┛ ┗┗┻┛ ┗┗┻┻┛┻┛ sql_writer = AsyncMySQL( pool_size = script_cred["mariaDb"]["write"]["poolSize"], host = script_cred["mariaDb"]["write"]["host"], user = script_cred["mariaDb"]["write"]["user"], password = script_cred["mariaDb"]["write"]["password"], database = script_cred["mariaDb"]["write"]["database"] ) if not await sql_writer.connect(): print("FATAL: MARIA-DB NOT CONNECTED!") return False no_context_printer("MariaDB ready.") # ┳┳┓ # ┃┃┃┏┓┏┓┏┓┏┓ # ┛ ┗┗┛┛┗┗┫┗┛ # ┛ data_mongo = AsyncMongo( connection_string = script_cred["mongoDb"]["data"]["connectionString"], database_name = script_cred["mongoDb"]["data"]["dbName"], max_connections = script_cred["mongoDb"]["data"]["poolSize"], debug = debug ) if not await data_mongo.connect(): print("FATAL: MONGO-DB NOT CONNECTED!") return False no_context_printer("MongoDB ready.") # ┳┓ ┓• ┏┓ ┓ # ┣┫┏┓┏┫┓┏ ━━ ┃ ┏┓┏┣┓┏┓ # ┛┗┗ ┗┻┗┛ ┗┛┗┻┗┛┗┗ redis_cache = AsyncRedisCache( connection_string = script_cred["redisCache"]["funcReturn"]["connectionString"], serializer = JSONSerializer(), debug = debug, debug_prefix = "General Cache | " ) if not await redis_cache.connect(): print("FATAL: REDIS CACHE NOT CONNECTED!") return False no_context_printer("Redis cache ready.") # ┏┓ ┓┓ ┓┏┏┓ # ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ ┃┃┏┛ # ┗┛┗┛┛┗┗┛ ┗┛┗┗┗ ┛ ┛ ┗┛┗━ # Messages / Mail Controllers: mail_controller = AllMailController( cache = redis_cache, http_client = http_client, alert_url = SCRIPT_DATA["alerts"]["url"], debug = False ) gmail_controller = GmailController( cache = redis_cache, http_client = http_client, alert_url = SCRIPT_DATA["alerts"]["url"], debug = False ) # ┏┓ ┓ ┏┓┓• # ┃ ┏┓┏┓┏┓┏┓┏╋┏┓┏┓┏ ┏┓┏┓┏┫ ┃ ┃┓┏┓┏┓╋┏ # ┗┛┗┛┛┗┛┗┗ ┗┗┗┛┛ ┛ ┗┻┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛ # Create an instance to handle GMail-related activities: gmail_client = AsyncGmailClient( service_name = "gmail", oauth_json = script_cred["google"]["oauth"]["tcaoff"], http_client = http_client, redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail", debug = False, debug_prefix = "Gmail (M) | ", debug_only_errors = True ) # ┏┓┳ ┳┳┓ • # ┣┫┃ ┃┃┃┏┓┏┓┓┏ # ┛┗┻ ┛ ┗┗┻┗┫┗┗ # ┛ # For LLMs: llm_controller = CoreLLMController( llm_creds = { "model": script_cred["openAi"]["model"], "openai_api_key": script_cred["openAi"]["openai_api_key"] }, cache = redis_cache, alert_url = SCRIPT_DATA["alerts"]["url"], http_client = http_client, debug = debug, debug_prefix = "AI (LLM) | ", debug_only_errors = True ) # ┳┓ # ┃┃┏┓┏┓┏┓ # ┻┛┗┛┛┗┗ # If everything went well, we return with success: no_context_printer("Initialization done.") return True # --------------------------------------------------------------------------------------------------------------------- async def sync_one_account(auth_token: CoreAuthTokenModel): """ To sync one third-party mail account. :param auth_token: The auth-token that gives access to the mail account. :return: ?? """ # 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) # Start by assuming failure: client_controller = None client_connector = None sync_results = MailSyncManyResults() # Figure out which mail client has to be used: match auth_token.client: case "gmail": client_controller, client_connector = gmail_controller, gmail_client case _: client_controller, client_connector = None, None # Try to sync mails: if client_controller is not None and client_connector is not None: try: sync_results = await client_controller.sync_mails( sql_conn = sql_writer, mongo_data_conn = data_mongo, mail_client = client_connector, auth_token = auth_token, user_info = None, llm = llm_controller, force_sync = False, start_date = now - datetime.timedelta(days = 1), end_date = now, max_count = 100, session_token = None ) except Exception as exception: pass # printer(exception) # else: printer("Invalid/unimplemented client.", auth_token.client) # Release the auth-token from the batch: 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), last_sync_ts = now ) # Done here: if sync_results.totalCount: printer( auth_token.clientUserId["email"], sync_results.totalCount, sync_results.newCount, sync_results.attemptedCount, sync_results.successCount, sync_results.failureCount ) # --------------------------------------------------------------------------------------------------------------------- async def sync_accounts(batch_size: int) -> None: """ To go into an indefinite loop and keep sync'ing mails for many accounts. :param batch_size: The no. of accounts to pick in every batch. :return: None. """ while True: # Get batches of auth-tokens to work with: 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)) # Process each batch: now = date_time.get_current_utc_date_time(as_string = False) tasks = [sync_one_account(auth_token = auth_token) for auth_token in auth_tokens_batch] results = await asyncio.gather(*tasks) # Small delay to not overload the database: 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)) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": # To get args from the terminal: import argparse # Get the config from the command-line: parser = argparse.ArgumentParser(description = f"To automatically sync. the mails for all users.") parser.add_argument( "-b", "--batch-size", type = int, help = "How many auth-tokens to load at once to sync.", default = 25 ) parser.add_argument( "-s", "--script-id", type = str, help = "The id of this script (will affect the loaded config)." ) parser.add_argument( "-d", "--debug", action = "store_true", help = "Whether, or not, you want to see debugging messages in the terminal.", default = False ) args = parser.parse_args() # Startup message: printer.enable() debugging_enabled = args.debug printer(debugging_enabled) printer.disable() async def main(): if await init( script_id = args.script_id, debug = args.debug ): await sync_accounts( batch_size = args.batch_size ) asyncio.run(main())