(20250122) Started making a cron script for work reminders. Modifying the mail module to work with it.

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

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())

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"
     )