diff --git a/api/blueprints/message/mail/send/send.py b/api/blueprints/message/mail/send/send.py index 14d9e93..4d90afd 100644 --- a/api/blueprints/message/mail/send/send.py +++ b/api/blueprints/message/mail/send/send.py @@ -219,7 +219,10 @@ async def send_one_mail( # ┛┗┗┻┗┛┗ ┗┛┛┗┗ ┗┛┗ # If the session token is invalid/expired: - if kwargs.get("session_info") is None: + if ( + kwargs.get("session_info") is None and + inbound_headers["Remote-IP"] not in current_app.whitelisted_ips + ): return ResponseModel( status_code = StatusCodes.FAILED, http_code = HttpCodes.UNAUTHORIZED diff --git a/cron/misc/__init__.py b/cron/misc/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/cron/misc/reminders_for_work.py b/cron/misc/reminders_for_work.py new file mode 100644 index 0000000..a571863 --- /dev/null +++ b/cron/misc/reminders_for_work.py @@ -0,0 +1,514 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Wednesday, 22nd Jan., 2025. + + OBJECTIVE: + + To keep checking for and sending reminders for work. + + 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 working with tabulated data: +import pandas as pd + +# For asynchronous activities: +import asyncio + +# To work with various datatypes: +from typing import List, Literal + +# For random values: +import random + +# Debugging: +from icecream import IceCreamDebugger + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# Debugging: +printer = IceCreamDebugger(prefix = "Work Rmndr. | ", includeContext = True) +no_context_printer = IceCreamDebugger(prefix = "Work Rmndr. | ", 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_reader: AsyncMySQL | 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_reader + + # 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_reader = AsyncMySQL( + pool_size = script_cred["mariaDb"]["read"]["poolSize"], + host = script_cred["mariaDb"]["read"]["host"], + user = script_cred["mariaDb"]["read"]["user"], + password = script_cred["mariaDb"]["read"]["password"], + database = script_cred["mariaDb"]["read"]["database"] + ) + if not await sql_reader.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.") + + # ┳┓ + # ┃┃┏┓┏┓┏┓ + # ┻┛┗┛┛┗┗ + + # If everything went well, we return with success: + no_context_printer("Initialization done.") + return True + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def send_telegram( + message: str, + chat_id: str = None, + message_type: Literal["info", "warning", "error"] = "info" +) -> None: + + """ + To send out alerts and heartbeats to inform about th script being alive. + :param message: The text to send. + :param chat_id: The destination chat identifier. + :param message_type: The kind of message to send. Decides the presentation of the header. + :return: None + """ + + try: + + # Create the JSON for sending to the API endpoint: + json_input = { + "chatClient": "telegram", + "message": message, + "type": message_type + } + if chat_id: json_input["chatId"] = json_input + + # Make the API call to send the ticks: + response = await http_client.post( + url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend", + json = json_input + ) + + # Raise an exception if the call was not successful: + response.raise_for_status() + + # If something goes wrong: + except Exception as exception: + printer(exception) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def get_reminders_to_send() -> pd.DataFrame | None: + + """ + To get a tabulated list of reminders to send out. + :return: A DataFrame of the reminders to send, or None if the request fails. + """ + + # Start with blank variables: + reminders_df = None + db_status = None + db_message = None + + try: + + # Ask the database: + db_json, exception = await sql_reader.call_procedure_and_get_json( + procedure_name = "reminder_for_work", + procedure_args = (), + retry_count = 3, + backoff_seconds = 1.0, + backoff_multiplier = 1.1, + return_exception = True + ) + + # Capture needed parts of the response: + db_status = db_json["status"] + db_message = db_json["message"] + + # Create the response DataFrame: + reminders_df = pd.DataFrame(db_json["data"]["rs0"]) + reminders_df.rename( + columns = { + "token_id": "tokenKey", + "recepient": "to", + "integration_type": "serviceType", + "content": "content" + }, + inplace = True + ) + + # If something goes wrong: + except Exception as exception: + printer(exception) + + # # Send out any needed alert: + # if exception or not db_status: + # await send_telegram( + # message = ( + # "*Reminders For Work*\n\n" + # f"Status: `{db_status}`\n\n" + # f"Message: `{db_message}`\n\n" + # f"Exception: `{exception}`" + # ), + # message_type = "error" + # ) + + # Done here: + return reminders_df + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def send_reminders_by_chat(chat_df: pd.DataFrame) -> None: + + """ + To send out reminders by chat apps like WhatsApp or Telegram. + :param chat_df: The part of the tabulated data that has reminders that need to be sent over chat apps. + :return: None + """ + + # Get a list of unique token-keys: + unique_token_keys = chat_df["tokenKey"].unique().tolist() + + # For every unique token key, we make an API call: + for token_key in unique_token_keys: + + # Construct the input(s) needed for the API call: + token_key_df = chat_df[chat_df["tokenKey"] == token_key] + messages = [ + { + "recipientNo": row["to"], + "message": row["content"] + } for index, row in token_key_df.iterrows() + ] + + # We are now ready to make the API call: + exception = None + try: + + # Make the API call: + response = await http_client.post( + url = r"https://api.thecaoffice.com/converse/chat", + json = { + "tokenKey": token_key, + "message": messages, + "tags": ["Work Reminder"] + } + ) + + # If the API call failed: + response.raise_for_status() + + # If the API call goes wrong: + except Exception as e: + exception = e + printer(exception) + + # Send out an alert if needed: + if exception is not None: + await send_telegram( + message = f"*Reminders For Work (Chat)*\n\nException: `{exception}`", + message_type = "error" + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def send_reminders_by_mail(mail_df: pd.DataFrame) -> None: + + """ + To send out reminders by e-mail. + :param mail_df: The part of the tabulated data that has reminders that need to be sent over e-mail. + :return: None + """ + + # Iterate over all the rows and send out the mails: + for index, row in mail_df.iterrows(): + + # Create the input(s) needed for making the API call: + input_json = { + "tokenKey": row["tokenKey"], + "to": [row["to"]], + "subject": f"Reminder for Work - {datetime.datetime.now().strftime('%d/%m/%Y')}", + "body": [ + { + "type": "html", + "part": { + "content": row["content"] + } + } + ] + } + + # We are now ready to make the API call: + exception = None + try: + + # Make the API call: + response = await http_client.post( + url = r"https://api.thecaoffice.com/converse/mail", + json = input_json + ) + + # If the API call failed: + response.raise_for_status() + + # If the API call goes wrong: + except Exception as e: + exception = e + printer(exception) + + # Send out an alert if needed: + if exception is not None: + await send_telegram( + message = f"*Reminders For Work (Mail)*\n\nException: `{exception}`", + message_type = "error" + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def send_reminders_by_sms(chat_df: pd.DataFrame): + + print("SMS DF:", chat_df.to_string()) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def send_reminders_once() -> None: + + """ + To ask the database if there are any reminders to send, and then send any needed work reminders. + :return: None. + """ + + reminders_df = await get_reminders_to_send() + if reminders_df is not None: + tasks = [ + # send_reminders_by_chat(reminders_df[reminders_df["serviceType"] == "chat"]), + send_reminders_by_mail(reminders_df[reminders_df["serviceType"] == "mail"]), + # send_reminders_by_sms(reminders_df[reminders_df["serviceType"] == "sms"]) + ] + await asyncio.gather(*tasks) + + +# ***************************************************************************************************************** +# ***** **** +# *** 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( + "-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 runner(): + if await init( + script_id = args.script_id, + debug = args.debug + ): await send_reminders_once() + + asyncio.run(runner()) diff --git a/models/api/message/mail/send.py b/models/api/message/mail/send.py index 35ff663..9fe2a66 100644 --- a/models/api/message/mail/send.py +++ b/models/api/message/mail/send.py @@ -87,6 +87,7 @@ class MailSendRequestHeaders(BaseModel): description = "the session token of the user who is requesting the service", pattern = REGEX_SESSION_TOKEN, frozen = True, + default = None, alias = "X-Session-Token" )