(20251119) Added caching to the muster roll data.

backend/api/helpers/muster_roll.py

@@ -0,0 +1,211 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        CREATED: Wed, 19th Nov, 2025
+        UPDATED: Wed, 19th Nov, 2025
+
+    OBJECTIVE:
+
+        To load and manage offline master muster roll data.
+
+    REFERENCES:
+
+        N/A
+
+    DOWNLOADS:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                     IMPORT                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+sys.path.append(".")
+sys.path.append("..")
+
+# System-level:
+import os
+import datetime
+
+# For using Quart:
+from quart import current_app
+
+# To work with tabulated data:
+import pandas as pd
+
+# Cosec-related:
+from cosec_web.cosec_web import CosecWeb
+
+# My utils:
+from utils_v2.date_time import date_time
+from utils_v2.system import files
+
+# To work with datatypes:
+from typing import List
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                             MACROS / ONE-TIME INIT                                            ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   VARIABLES                                                   ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                    CLASSES                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   FUNCTIONS                                                   ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+def init() -> None:
+
+    current_app.muster_roll = None
+    current_app.last_muster_roll_load_ts = date_time.get_current_utc_date_time(as_string = False)
+
+
+# ---------------------------------------------------------------------------------------------------------------------
+
+
+def load_from_cosec(
+        cosec_url: str,
+        username: str,
+        password: str,
+        group_ids: List[str],
+        on_date: datetime.datetime = None,
+) -> pd.DataFrame | None:
+
+    # Figure out the paths:
+    base_dir = files.get_parent_directory(
+        files.get_file_directory(include_filename = False),
+        depth = 3
+    )
+
+    # Put together the directory for the drivers, the downloads, etc.:
+    chrome_driver_dir = os.path.join(base_dir, r"drivers/chrome")
+    user_data_dir = os.path.join(base_dir, r"browser/user_data")
+    downloads_dir = os.path.join(base_dir, r"downloads")
+    masters_dir = os.path.join(base_dir, r"masters")
+
+    # Create an instance of the automation object:
+    cosec = CosecWeb(
+        cosec_url = cosec_url,
+        username = username,
+        password = password,
+        driver_dir = chrome_driver_dir,
+        user_data_dir = user_data_dir,
+        downloads_dir = downloads_dir,
+    )
+
+    # Perform the login:
+    cosec.login(initial_sleep = 2.5)
+
+    # Get the in/out report:
+    report_path = cosec.get_muster_roll(
+        initial_sleep = 1.0,
+        on_date = on_date or date_time.get_current_utc_date_time(),
+        group_ids = group_ids,
+        download_timeout = 60.0
+    )
+
+    # Log out to end the cycle:
+    cosec.logout()
+
+    # Close the browser window:
+    cosec.quit()
+
+    # If we didn't get any path, the download failed:
+    if report_path is None:
+        current_app.printer("FAILED to fetch Muster Roll.")
+        current_app.muster_roll = None
+
+    # Else, we parse the data and hold it:
+    else:
+        current_app.printer("Fetched Muster Roll.")
+        report_df = cosec.read_muster_roll_xls(report_path)
+        report_df = report_df[[
+            "User ID", "User Name", "Category Name",
+            "Grade Name", "Branch Name", "Department Name",
+            "Direct Reporting", "Level-1"
+        ]]
+        current_app.muster_roll = report_df
+        current_app.last_muster_roll_load_ts = date_time.get_current_utc_date_time(as_string = False)
+
+    # Done here:
+    return current_app.muster_roll
+
+
+# ---------------------------------------------------------------------------------------------------------------------
+
+
+def load(
+        cosec_url: str,
+        username: str,
+        password: str,
+        group_ids: List[str],
+        on_date: datetime.datetime = None,
+) -> pd.DataFrame:
+
+    # If the data either doesn't exist at all,
+    # or the data has gone stale, reload:
+    if (
+            current_app.muster_roll is None
+            # or current_app.last_muster_roll_load_ts.timestamp() - date_time.get_current_utc_date_time(as_string = False).timestamp() >= 300
+    ): load_from_cosec(
+        cosec_url = cosec_url,
+        username = username,
+        password = password,
+        group_ids = group_ids,
+        on_date = on_date,
+    )
+
+    # Done here:
+    return current_app.muster_roll
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  MAIN PROGRAM                                                 ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass