diff --git a/cron/reports.py b/cron/reports.py index a6fb3b3..2c24210 100644 --- a/cron/reports.py +++ b/cron/reports.py @@ -33,9 +33,6 @@ # To make sibling directories accessible for imports: import sys - -import pandas as pd - sys.path.append(".") sys.path.append("..") @@ -46,14 +43,28 @@ import os import time import datetime +# To work with tabulate data: +import pandas as pd + +# To make API calls: +import requests + # Cosec-related: from cosec_web.cosec_web import CosecWeb +# TCAOFF-related: +from helpers import tcaoff + # My utils: from utils_v2.system import files from utils_v2.string import json +from utils_v2.string import regex from utils_v2.date_time import date_time +# To work with datatypes: +from typing import List, Dict, Any +from collections import defaultdict + # ***************************************************************************************************************** # ***** **** @@ -66,6 +77,7 @@ from utils_v2.date_time import date_time file_dir = files.get_file_directory(include_filename = False) proj_dir = files.get_parent_directory(file_dir, depth = 1) cosec_creds_file = os.path.join(proj_dir, "creds", "cosec.json") +tcaoff_creds_file = os.path.join(proj_dir, "creds", "tcaoff.json") muster_roll_cache_file = os.path.join(proj_dir, "local", "cache", "muster_roll_cache.json") in_out_summary_cache_file = os.path.join(proj_dir, "local", "cache", "in_out_summary_cache.json") chrome_driver_dir = os.path.join(proj_dir, r"drivers","chrome") @@ -100,6 +112,182 @@ downloads_dir = os.path.join(proj_dir, r"downloads") # ***************************************************************************************************************** +# def remove_special_chars(s: str) -> str: +# +# return regex.replace( +# text = s, +# pattern = r"[^\w\d\- _]", +# substitute_text = "_" +# ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +def add_branches_to_tcaoff( + tcaoff_creds: dict, + cosec_muster_roll: dict, +) -> Dict[str, int]: + + # Start with a basic response structure: + response = defaultdict(int) + + # Get the list of existing branches from TCAOFF: + tcaoff_branches = tcaoff.branch_list(tcaoff_creds) + tcaoff_branches = [_["branch_name"] for _ in tcaoff_branches] + tcaoff_branches = list(set(tcaoff_branches)) + + # Get only the branch names from Cosec: + cosec_branches = [tcaoff.remove_special_chars(_["Branch Name"]) for _ in cosec_muster_roll["report"]] + cosec_branches = list(set(cosec_branches)) + + # Loop through the data from Cosec and add missing branches to TCAOFF: + for cosec_branch in cosec_branches: + if cosec_branch not in tcaoff_branches: + success = tcaoff.branch_add( + tcaoff_creds, + branch_name = cosec_branch + ) + response["total"] += 1 + if success: response["success"] += 1 + else: response["fail"] += 1 + + # Done here: + print("TCAOFF-Cosec Branches Sync.:", json.to_string(response)) + return response + + +# --------------------------------------------------------------------------------------------------------------------- + + +def add_departments_to_tcaoff( + tcaoff_creds: dict, + cosec_muster_roll: dict, +) -> Dict[str, int]: + + # Start with a basic response structure: + response = defaultdict(int) + + # Get the list of existing departments from TCAOFF: + tcaoff_depts = tcaoff.department_list(tcaoff_creds) + tcaoff_depts = [_["department_name"] for _ in tcaoff_depts] + tcaoff_depts = list(set(tcaoff_depts)) + # print("TCAOFF DEPTS:", tcaoff_depts) + + # Get only the department names from Cosec: + cosec_depts = [tcaoff.remove_special_chars(_["Department Name"]) for _ in cosec_muster_roll["report"]] + cosec_depts = list(set(cosec_depts)) + # print("COSEC DEPTS:", cosec_depts) + + # Loop through the data from Cosec and add missing departments to TCAOFF: + for cosec_dept in cosec_depts: + if cosec_dept not in tcaoff_depts: + success = tcaoff.department_add( + tcaoff_creds, + department_name = cosec_dept + ) + response["total"] += 1 + if success: response["success"] += 1 + else: response["fail"] += 1 + + # Done here: + print("TCAOFF-Cosec Depts. Sync.:", json.to_string(response)) + return response + + +# --------------------------------------------------------------------------------------------------------------------- + + +def add_teams_to_tcaoff( + tcaoff_creds: dict, + cosec_muster_roll: dict, +) -> Dict[str, int]: + + # Start with a basic response structure: + response = defaultdict(int) + + # Map out branch ids: + tcaoff_branches = tcaoff.branch_list(tcaoff_creds) + tcaoff_branches_lookup = {d["branch_name"]: d["branch_id"] for d in tcaoff_branches} + + # Map out dept. ids: + tcaoff_depts = tcaoff.department_list(tcaoff_creds) + tcaoff_depts_lookup = {d["department_name"]:d["department_id"] for d in tcaoff_depts} + + # Get the list of existing departments from TCAOFF: + # NOTE: `pseudonym` is the unique username of the user. + tcaoff_teams = tcaoff.team_list(tcaoff_creds) + tcaoff_teams = [_["pseudonym"] for _ in tcaoff_teams] + tcaoff_teams = list(set(tcaoff_teams)) + + # # Get only the department names from Cosec: + # cosec_depts = [remove_special_chars(_["Department Name"]) for _ in cosec_muster_roll["report"]] + # cosec_depts = list(set(cosec_depts)) + # # print("COSEC DEPTS:", cosec_depts) + + # Loop through the data from Cosec and add missing departments to TCAOFF: + for cosec_team in cosec_muster_roll["report"]: + if cosec_team["User ID"] not in tcaoff_teams: + + # Check the branch id and department id: + branch_id = tcaoff_branches_lookup.get(tcaoff.remove_special_chars(cosec_team["Branch Name"])) + dept_id = tcaoff_depts_lookup[tcaoff.remove_special_chars(cosec_team["Department Name"])] + if branch_id is None: + print(f"TEAM SYNC ERR: Branch '{tcaoff.remove_special_chars(cosec_team['Branch Name'])}' not found in TCAOFF") + response["fail"] += 1 + continue + if dept_id is None: + print(f"TEAM SYNC ERR: Dept. '{tcaoff.remove_special_chars(cosec_team['Department Name'])}' not found in TCAOFF") + response["fail"] += 1 + continue + + # Add the team: + team_json = { + "branchId": branch_id, + "idDepartment": dept_id, + "reportingTo": None, + "name": cosec_team['User Name'].strip(), + "email": regex.replace( + text = cosec_team['User Name'], + pattern = r"[^\w\d\._]", + substitute_text = "" + ).strip().lower() + "@velankanigroup.com", + "phoneNo": "9876543210", + "role": cosec_team['Category Name'].strip(), + "username": cosec_team['User ID'].strip(), + "password": "Vispl@123", + "hierarchy": 1, + } + print("Need to Add:", json.to_string(team_json)) + success = tcaoff.team_add( + tcaoff_creds, + branch_id = branch_id, + dept_id = dept_id, + reporting_to = None, + team_name = cosec_team["User Name"].strip(), + email = regex.replace( + text = cosec_team['User Name'], + pattern = r"[^\w\d\._]", + substitute_text = "" + ).strip().lower() + "@velankanigroup.com", + phone_no = "9876543210", + role = cosec_team["Category Name"].strip(), + username = cosec_team["User ID"].strip(), + password = "Vispl@123" + ) + response["total"] += 1 + if success: response["success"] += 1 + else: response["fail"] += 1 + break + + # Done here: + print("TCAOFF-Cosec Depts. Sync.:", json.to_string(response)) + return response + + +# --------------------------------------------------------------------------------------------------------------------- + + def kill_chrome() -> None: # First kill the previous processes, @@ -111,37 +299,48 @@ def kill_chrome() -> None: # --------------------------------------------------------------------------------------------------------------------- -def get_muster_roll(cosec_creds: dict): +def get_muster_roll(cosec_creds: dict) -> bool: - # Force close other running Chrome processes: - kill_chrome() + """ + Get the latest Muster-Roll from Matrix Cosec. It saves the data into a local cache file. + :param cosec_creds: The credentials (and config) to operate Cosec Matrix. + :return: True if the automated fetch was successful, else False. + """ - # Create an instance of the automation object: - cosec = CosecWeb( - cosec_url = cosec_creds["creds"]["url"], - username = cosec_creds["creds"]["username"], - password = cosec_creds["creds"]["password"], - driver_dir = chrome_driver_dir, - user_data_dir = user_data_dir, - downloads_dir = downloads_dir, - ) + # Start by assuming failure: + success = False - # Perform the login: - cosec.login(initial_sleep = 2.5) + # # Force close other running Chrome processes: + # kill_chrome() + # + # # Create an instance of the automation object: + # cosec = CosecWeb( + # cosec_url = cosec_creds["creds"]["url"], + # username = cosec_creds["creds"]["username"], + # password = cosec_creds["creds"]["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 = date_time.get_current_utc_date_time(), + # group_ids = cosec_creds["musterRollConfig"]["groupIds"], + # download_timeout = 60.0 + # ) + # + # # Log out to end the cycle: + # cosec.logout() + # + # # Close the browser window: + # cosec.quit() - # Get the in/out report: - report_path = cosec.get_muster_roll( - initial_sleep = 1.0, - on_date = date_time.get_current_utc_date_time(), - group_ids = cosec_creds["musterRollConfig"]["groupIds"], - download_timeout = 60.0 - ) - - # Log out to end the cycle: - cosec.logout() - - # Close the browser window: - cosec.quit() + report_path = r"D:\kps\PycharmProjects\cosec\downloads\Monthly_Details.xls" # Now process the report, # and save it to the JSON file: @@ -159,6 +358,22 @@ def get_muster_roll(cosec_creds: dict): report_data = report_data.where(report_data.notna(), None) report_data = report_data.to_dict(orient = "records") + # Loop through the report for inferred data. + # Add the ids of the "Direct Reporting" and "Level-1" values: + for person in report_data: + person["Direct Reporting ID"] = None + person["Level-1 ID"] = None + for check_against in report_data: + if person["Direct Reporting"] == check_against["User Name"]: + person["Direct Reporting ID"] = check_against["User ID"] + if person["Level-1"] == check_against["User Name"]: + person["Level-1 ID"] = check_against["User ID"] + + # Sort by hierarchy (BFS): + pass + + print(f"MUSTER ROLL ({len(report_data)}):", json.to_string(report_data)) + # Save the data to a JSON file: json.to_file( file = muster_roll_cache_file, @@ -169,11 +384,26 @@ def get_muster_roll(cosec_creds: dict): no_space = True ) + # Note down success: + success = True + + # Done here: + return success + # --------------------------------------------------------------------------------------------------------------------- -def get_in_out_summary(cosec_creds: dict): +def get_in_out_summary(cosec_creds: dict) -> bool: + + """ + Get the latest In-Out-Summary from Matrix Cosec. It saves the data into a local cache file. + :param cosec_creds: The credentials (and config) to operate Cosec Matrix. + :return: True if the automated fetch was successful, else False. + """ + + # Start by assuming failure: + success = False # Force close other running Chrome processes: kill_chrome() @@ -250,28 +480,97 @@ def get_in_out_summary(cosec_creds: dict): no_space = True ) + # Note down success: + success = True + + # Done here: + return success + # --------------------------------------------------------------------------------------------------------------------- -def get_reports(cosec_creds: dict) -> None: +def get_reports( + cosec_creds: dict, + tcaoff_creds: dict, +) -> None: - # Get the Muster Roll and then wait - # for the driver's resources to get freed: - # try: - # get_muster_roll(cosec_creds = cosec_creds) - # time.sleep(2.5) - # except Exception as e: - # print("MUSTER ROLL FETCH FAILED!") - - # Get the In-Out Summary and then wait - # for the driver's resources to get freed: + # Try the whole process once: try: - get_in_out_summary(cosec_creds = cosec_creds) - time.sleep(2.5) + + # Log in to TCAOFF: + tcaoff.login(tcaoff_creds) + + # # # TEST SECTION: + # # # Test out listing API calls: + print("TCAOFF Branches:", json.to_string(tcaoff.branch_list(tcaoff_creds))) + print("TCAOFF Depts.:", json.to_string(tcaoff.department_list(tcaoff_creds))) + # print("TCAOFF Team:", json.to_string(tcaoff.team_list(tcaoff_creds))) + + # # TEST SECTION: + # # Test out listing API calls: + # tcaoff.branch_add(tcaoff_creds, branch_name = "Test - 20260108") + # tcaoff.department_add(tcaoff_creds, department_name = "Test - 20260108") + # tcaoff.team_add( + # tcaoff_creds, + # branch_id = 124, + # dept_id = 851, + # reporting_to = 2857, + # team_name = "Test - 20260108", + # email = "user@domain.com", + # phone_no = "9876543219", + # role = "Python Dev", + # username = "test.user.20260108", + # password = "12345678", + # ) + + # MUSTER-ROLL REPORT: + try: + + # Get the Muster Roll and then wait + # for the driver's resources to get freed: + success = get_muster_roll(cosec_creds = cosec_creds) + + # Sync data between Cosec and TCAOFF: + if success: + print("MUSTER ROLL: Sync'ing with TCAOFF") + cosec_muster_roll = json.from_file(muster_roll_cache_file) + # add_branches_to_tcaoff( + # tcaoff_creds = tcaoff_creds, + # cosec_muster_roll = cosec_muster_roll, + # ) + # add_departments_to_tcaoff( + # tcaoff_creds = tcaoff_creds, + # cosec_muster_roll = cosec_muster_roll, + # ) + add_teams_to_tcaoff( + tcaoff_creds = tcaoff_creds, + cosec_muster_roll = cosec_muster_roll, + ) + + # If something goes wrong: + except Exception as e: + print("MUSTER ROLL FETCH FAILED!") + raise + + # # Get the In-Out Summary and then wait + # # for the driver's resources to get freed: + # try: + # success = get_in_out_summary(cosec_creds = cosec_creds) + # time.sleep(2.5) + # if success: pass + # except Exception as e: + # print("IN-OUT SUMMARY FETCH FAILED!") + # raise + + # Log out from TCAOFF: + tcaoff.logout(tcaoff_creds) + + # If TCAOFF's login or logout fails: except Exception as e: - print("IN-OUT SUMMARY FETCH FAILED!") - raise e + print("REPORT CRON FAILED LOOP!") + print("EXCEPTION:", e) + raise # --------------------------------------------------------------------------------------------------------------------- @@ -279,12 +578,16 @@ def get_reports(cosec_creds: dict) -> None: def loop( cosec_creds: dict, + tcaoff_creds: dict, interval_seconds: int | float = 900 ) -> None: # Just keep fetching the reports in an infinite loop: while True: - get_reports(cosec_creds = cosec_creds) + get_reports( + cosec_creds = cosec_creds, + tcaoff_creds = tcaoff_creds + ) time.sleep(interval_seconds) @@ -303,7 +606,7 @@ if __name__ == "__main__": print("M-ROLL CACHE:", muster_roll_cache_file) print("IN-OUT CACHE:", in_out_summary_cache_file) - # Read the credentials. + # Read Cosec's credentials. # It should be in the format: """ { @@ -330,9 +633,34 @@ if __name__ == "__main__": } } """ - cosec_creds = json.from_file(cosec_creds_file) + + # Read TCAOFF's credentials. + # It should be in the format: + """ + { + "creds": { + "username": "", + "password": "", + "sessionToken": null, + "user": null + }, + "urls": { + "login": "https://api.thecaoffice.com/ca/login", + "logout": "https://api.thecaoffice.com/user/logout", + "branchList": "https://api.thecaoffice.com/commons/branch/list", + "branchAdd": "https://api.thecaoffice.com/commons/branch/add", + "deptList": "https://api.thecaoffice.com/commons/departments/list", + "deptAdd": "https://api.thecaoffice.com/commons/departments/add", + "teamList": "https://api.thecaoffice.com/team/list", + "teamAdd": "https://api.thecaoffice.com/team/add" + } + } + """ + tcaoff_creds = json.from_file(tcaoff_creds_file) + loop( cosec_creds = cosec_creds, + tcaoff_creds = tcaoff_creds, interval_seconds = cosec_creds["generalConfig"]["pollInterval"], ) diff --git a/helpers/__init__.py b/helpers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/helpers/tcaoff.py b/helpers/tcaoff.py new file mode 100644 index 0000000..a1b1c19 --- /dev/null +++ b/helpers/tcaoff.py @@ -0,0 +1,436 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + CREATED: Wed, 26th Nov, 2025 + UPDATED: Wed, 26th Nov, 2025 + + OBJECTIVE: + + To achieve so-and-so-objective... + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For system-level activities: +import os + +# To work with date and time: +import time +import datetime + +# To work with tabulate data: +import pandas as pd + +# To make API calls: +import requests + +# Cosec-related: +from cosec_web.cosec_web import CosecWeb + +# My utils: +from utils_v2.system import files +from utils_v2.string import json +from utils_v2.string import regex +from utils_v2.date_time import date_time + +# To work with datatypes: +from typing import List, Dict, Any + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +def remove_special_chars(s: str) -> str: + + return regex.replace( + text = s, + pattern = r"[^\w\d\- _]", + substitute_text = "_" + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +def login( + tcaoff_creds: dict +) -> bool: + + """ + Log in to TCAOFF. The retrieved session token is then updated in the input dict itself. + :param tcaoff_creds: The set of credentials to use to log in. + :return: True if logged in, else False. + """ + + # Make the API call: + response = requests.post( + url = tcaoff_creds["urls"]["login"], + json = { + "username": tcaoff_creds["creds"]["username"], + "password": tcaoff_creds["creds"]["password"], + "mode": "cosec" + } + ) + + # If login succeeded: + if response.status_code in [200]: + response_json = response.json() + tcaoff_creds["creds"]["sessionToken"] = response_json["sessionToken"] + tcaoff_creds["creds"]["user"] = { + "userId": 1234 + } + print("TCAOFF: logged in") + return True + + # If login failed: + else: + print("TCAOFF: log-in failed") + return False + + +# --------------------------------------------------------------------------------------------------------------------- + + +def logout( + tcaoff_creds: dict +) -> bool: + + """ + Log out from TCAOFF. The session token is cleared from the in-mem creds. + :param tcaoff_creds: The set of credentials to use to log out. + :return: True if logged out, else False. + """ + + # Make the API call: + response = requests.post( + url = tcaoff_creds["urls"]["logout"], + json = {"username": tcaoff_creds["creds"]["sessionToken"]} + ) + + # Clear the session details from the in-mem creds: + tcaoff_creds["creds"]["sessionToken"] = None + tcaoff_creds["creds"]["user"] = None + + # If logout succeeded: + if response.status_code in [200]: + print("TCAOFF: logged out") + return True + + # If login failed: + else: + print("TCAOFF: log-out failed") + return False + + +# --------------------------------------------------------------------------------------------------------------------- + + +def branch_list( + tcaoff_creds: dict +) -> List[Dict[str, Any]] | None: + + """ + List the existing branches. + :param tcaoff_creds: The set of credentials to use with the API call. + :return: True if logged out, else False. + """ + + # Make the API call: + response = requests.post( + url = tcaoff_creds["urls"]["branchList"], + headers = {"X-Session-Token": tcaoff_creds["creds"]["sessionToken"]}, + json = {"idUser": tcaoff_creds["creds"]["user"]["userId"]} + ) + + # If the call succeeded: + if response.status_code in [200]: + response_json = response.json() + branches = response_json["data"]["rs0"] + print("TCAOFF: branches listed") + return branches + + # If the call failed: + else: + print("TCAOFF: branch-list failed") + return None + + +# --------------------------------------------------------------------------------------------------------------------- + + +def branch_add( + tcaoff_creds: dict, + branch_name: str +) -> bool: + + """ + Add a new branch. + :param tcaoff_creds: The set of credentials to use with the API call. + :param branch_name: The name of the branch to add. + :return: True if logged out, else False. + """ + + # Make the API call: + response = requests.post( + url = tcaoff_creds["urls"]["branchAdd"], + headers = {"X-Session-Token": tcaoff_creds["creds"]["sessionToken"]}, + json = { + "idUser": tcaoff_creds["creds"]["user"]["userId"], + "branchName": branch_name + } + ) + + # # Debugging: + # response_json = response.json() + # print("TCAOFF Branch-Add:", json.to_string(response_json)) + + # If the call succeeded: + if response.status_code in [200]: + print(f"TCAOFF: branch '{branch_name}' added") + return True + + # If the call failed: + else: + print(f"TCAOFF: branch '{branch_name}' NOT added") + return False + + +# --------------------------------------------------------------------------------------------------------------------- + + +def department_list( + tcaoff_creds: dict +) -> List[Dict[str, Any]] | None: + + """ + List the existing departments. + :param tcaoff_creds: The set of credentials to use with the API call. + :return: True if logged out, else False. + """ + + # Make the API call: + response = requests.post( + url = tcaoff_creds["urls"]["deptList"], + headers = {"X-Session-Token": tcaoff_creds["creds"]["sessionToken"]}, + json = {"idUser": tcaoff_creds["creds"]["user"]["userId"]} + ) + + # If the call succeeded: + if response.status_code in [200]: + response_json = response.json() + branches = response_json["data"]["rs0"] + print("TCAOFF: depts. listed") + return branches + + # If the call failed: + else: + print("TCAOFF: dept-list failed") + return None + + +# --------------------------------------------------------------------------------------------------------------------- + + +def department_add( + tcaoff_creds: dict, + department_name: str +) -> bool: + + """ + Add a new department. + :param tcaoff_creds: The set of credentials to use with the API call. + :param department_name: The name of the department to add. + :return: True if logged out, else False. + """ + + # Make the API call: + response = requests.post( + url = tcaoff_creds["urls"]["deptAdd"], + headers = {"X-Session-Token": tcaoff_creds["creds"]["sessionToken"]}, + json = { + "idUser": tcaoff_creds["creds"]["user"]["userId"], + "departmentName": department_name + } + ) + + # # Debugging: + # response_json = response.json() + # print("TCAOFF Dept-Add:", json.to_string(response_json)) + + # If the call succeeded: + if response.status_code in [200]: + print(f"TCAOFF: dept. '{department_name}' added") + return True + + # If the call failed: + else: + print(f"TCAOFF: dept. '{department_name}' NOT added") + return False + + +# --------------------------------------------------------------------------------------------------------------------- + + +def team_list( + tcaoff_creds: dict +) -> List[Dict[str, Any]] | None: + + """ + List the existing team members. + :param tcaoff_creds: The set of credentials to use with the API call. + :return: True if logged out, else False. + """ + + # Make the API call: + response = requests.post( + url = tcaoff_creds["urls"]["teamList"], + headers = {"X-Session-Token": tcaoff_creds["creds"]["sessionToken"]}, + json = {"idUser": tcaoff_creds["creds"]["user"]["userId"]} + ) + + # If the call succeeded: + if response.status_code in [200]: + response_json = response.json() + branches = response_json["data"]["rs0"] + print("TCAOFF: team listed") + return branches + + # If the call failed: + else: + print("TCAOFF: team-list failed") + return None + + +# --------------------------------------------------------------------------------------------------------------------- + + +def team_add( + tcaoff_creds: dict, + branch_id: int, + dept_id: int, + reporting_to: int | None, + team_name: str, + email: str, + phone_no: str, + role: str, + username: str, + password: str, +) -> bool: + + """ + Add a new team member. + :param tcaoff_creds: The set of credentials to use with the API call. + :param dept_id: The id of the department that this team member is working in. + :param branch_id: The id of the branch that this team member is working in. + :param reporting_to: The id of the senior to whom this team member will report. + :param team_name: The name of the team member. This is the full display name. Can be the same as others. + :param email: The email id of the team member. + :param phone_no: The phone no. of the team member. + :param role: The role of the team member in the organization. + :param username: The unique username of the team member. Cannot be the same as anyone else. + :param password: The password for this team member's login. + :return: + """ + + # Make the API call: + response = requests.post( + url = tcaoff_creds["urls"]["teamAdd"], + headers = {"X-Session-Token": tcaoff_creds["creds"]["sessionToken"]}, + json = { + "branchId": branch_id, + "idDepartment": dept_id, + "reportingTo": reporting_to, + "name": team_name, + "email": email, + "phoneNo": phone_no, + "role": role, + "username": username, + "password": password, + "hierarchy": 1, + } + ) + + # Debugging: + response_json = response.json() + print("TCAOFF Team-Add:", json.to_string(response_json)) + + # If the call succeeded: + if response.status_code in [200]: + print(f"TCAOFF: team '{team_name} ({username})' added") + return True + + # If the call failed: + else: + print(f"TCAOFF: team '{team_name} ({username})' NOT added") + return False + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/playground/__init__.py b/playground/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/playground/cache_v2_test.py b/playground/cache_v2_test.py new file mode 100644 index 0000000..7786fd2 --- /dev/null +++ b/playground/cache_v2_test.py @@ -0,0 +1,55 @@ +import time +from utils_v2.system import files +from utils_v2.cache_v2.async_mem_cache import AsyncMemCache +from utils_v2.cache_v2.async_disk_cache import AsyncDiskCache +import asyncio + +async def main(): + + proj_dir = files.get_file_directory(include_filename = False) + proj_dir = files.get_parent_directory(proj_dir, depth = 1) + print("PROJ. DIR.:", proj_dir) + + memcache = AsyncMemCache() + # diskcache = AsyncDiskCache( + # caching_dir = os.path.join(proj_dir, "local"), + # lock_on_read = False, + # lock_wait_timeout = 5.0, + # expiry_check_interval = 60.0 + # ) + + cache = memcache + + # print("\n\n---\n\n") + # print("EXPIRY CHECK:") + # await cache.set("k1", "v1", expiry = 3.0) + # print(await cache.get("k1")) + # time.sleep(3.1) + # print(await cache.get("k1")) + + print("\n\n---\n\n") + print("DELETE CHECK:") + await cache.set("k1", "v1", expiry = 30.0) + await cache.set("k2", "v2", expiry = 30.0) + await cache.set("k3", "v3", expiry = 30.0) + print("LIST:", await cache.list_keys()) + print("LIST:", await cache.list_keys(match = "k[1-2]")) + print(await cache.get("k1")) + await cache.delete("k1") + print(await cache.get("k1")) + print("LIST:", await cache.list_keys()) + + # print("\n\n---\n\n") + # print("COUNTING:") + # count = await cache.count("c1", expiry = 3.1) + # for i in range(10): + # time.sleep(1.0) + # count = await cache.count("c1", expiry = 3.0, raise_exception = True) + # ttl = await cache.ttl("c1") + # print(count, await cache.get("c1"), ttl) + # print("OUTSIDE LOOP!") + # print(await cache.get("c1")) + # time.sleep(3.1) + # print(await cache.get("c1")) + +asyncio.run(main()) diff --git a/playground/grade_and_category.py b/playground/grade_and_category.py new file mode 100644 index 0000000..5182891 --- /dev/null +++ b/playground/grade_and_category.py @@ -0,0 +1,9 @@ +from utils_v2.string import json + +data = json.from_file(r"D:\kps\PycharmProjects\cosec\local\cache\muster_roll_cache.json") + +for d in data["report"]: + if d["Branch Name"] == "Client": continue + if str(d["Category Name"]).lower().strip() != str(d["Grade Name"]).lower().strip(): + print(json.to_string(d)) + print("\n\n---\n\n") \ No newline at end of file diff --git a/playground/ipdr.md b/playground/ipdr.md new file mode 100644 index 0000000..1132580 --- /dev/null +++ b/playground/ipdr.md @@ -0,0 +1,141 @@ +# IPDR at VKNTPL + +Implementation by **Bhushan C Thakkar** and **Khushal P Soonderji** in November of 2025. + +_Documentation made on 20251128_ + +--- + +## 1. Login: + +The username is `vknipdr` and the server address is `ipdr.prysmnet.com`. You may SSH into the device to perform your activities. + +>ssh vknipdr@ipdr.prysmnet.com + +--- + +## 2. Capturing Logs With Syslog-ng: + +The system uses **syslog-ng** to capture IPDR-related logs from **Mikrotik** hardware and saves them to a file on the server in fast and efficient **CSV format**. + +The file that hold the config can be accessed by the following command (`sudo` required): + +> nano /etc/syslog-ng/conf.d/mikrotik.conf + +Here is the code snippet that makes the magic work: + +```text +# ------------------------------------------------------------ +# SOURCE — receive logs from MikroTik +# ------------------------------------------------------------ +source s_mikrotik_nat { + udp(ip(0.0.0.0) port(514) keep-hostname(yes)); + +}; + +# ------------------------------------------------------------ +# Filter - accept on Forward Chains +# ------------------------------------------------------------ + +filter f_forward { + match("forward:") and not match("src-mac"); +}; + + +# ------------------------------------------------------------ +# PARSER — extract NAT fields using regex (flat declaration) +# ------------------------------------------------------------ + +parser p_nat_forward_regex { + regexp-parser( + template("${MESSAGE}") + patterns( + "forward: in:(?.*) out:(?.*), proto (?.*), (?[\\d\\.]*):(?[\\d]+)->(?[\\d\\.]*):(?[\\d]+)(?:, NAT \\(.*?(?[\\d\\.]+):(?[\\d]+)\\)[^,]*)?, len (?\\d+)" + ) + ); +}; + +# ------------------------------------------------------------ +# TEMPLATE — +# ------------------------------------------------------------ +template t_nat_csv { + template("${DATE},$src_ip,$src_port,$nat_trans_ip,$nat_trans_port,$dst_ip,$dst_port\n"); + template-escape(no); +}; + +# ------------------------------------------------------------ +# DESTINATION — MongoDB +# ------------------------------------------------------------ + + +# ------------------------------------------------------------ +# DESTINATION — Raw log file (for testing) +# ------------------------------------------------------------ +destination d_rawfile { + file( + "/mnt/storage/syslog-ng/prysmnet/mikrotik-raw.log" + template(t_nat_csv) + flush_lines(1) + ); +}; + +# ------------------------------------------------------------ +# LOG PATH — Write all incoming MikroTik messages directly to file +# ------------------------------------------------------------ +log { + source(s_mikrotik_nat); + filter(f_forward); + parser(p_nat_forward_regex); + destination(d_rawfile); +}; +``` + +--- + +## 3. Restarting Syslog-ng: + +Once you have saved your config file, you will need to restart the process with the following command (`sudo` required): + +> systemctl restart syslog-ng + +--- + +## 4. Checking The Logs: + +You can see a *live trail* of the logs by running the following command: + +> tail -f -n 10 /mnt/storage/syslog-ng/prysmnet/mikrotik-raw.log + +You will see the logs in the following format: + +```text +Nov 11 14:39:03,10.252.247.94,54084,103.171.2.187,54084,142.250.194.234,443 +Nov 11 14:39:03,10.252.255.135,56424,103.171.2.219,56424,123.63.54.23,443 +Nov 11 14:39:03,10.252.247.94,54084,,,142.250.194.234,443 +``` + +--- + +## 5. Log Rotation: + +Logs can add up very fast and the file can become unreasonable large and tough to manage. For this, we need to perform +log rotation. A service named `logrotate` has been used for the same. At the time of first setup, the file was set up such +that the configuration could be seen by running the following command: + +> cat /etc/logrotate.d/syslog-forward + +And the contents were: + +```text +/var/log/syslog-ng/mikrotik-raw.log{ + size 20G + rotate 365 + compress + delaycompress + missingok + notifempty + copytruncate + dateext + dateformat -%Y-%m-%d_%H-%M-%S +} +``` \ No newline at end of file diff --git a/utils_v2/cache_v2/base.py b/utils_v2/cache_v2/async_base.py similarity index 95% rename from utils_v2/cache_v2/base.py rename to utils_v2/cache_v2/async_base.py index 8221296..ebdb517 100644 --- a/utils_v2/cache_v2/base.py +++ b/utils_v2/cache_v2/async_base.py @@ -49,6 +49,7 @@ import base64 # Other utils: from utils_v2.string import json +from utils_v2.date_time import date_time # To work with datatypes: from typing import List, Any @@ -86,10 +87,13 @@ from functools import wraps class AsyncCachingBase(ABC): + # Defaults: + LONG_EXPIRY = 3_15_36_000 + def __init__( self, debug: bool = True, - debug_prefix: str = "A-Cache | " + debug_prefix: str = "ABaseCache | " ): """ @@ -99,8 +103,8 @@ class AsyncCachingBase(ABC): """ # For debugging: - self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) - if not debug: self.__printer.disable() + self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) + if not debug: self._printer.disable() # ┳┓ ┓ • # ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓ @@ -108,16 +112,27 @@ class AsyncCachingBase(ABC): # ┛ ┛ ┛ def enable_debug(self): - self.__printer.enable() + self._printer.enable() def disable_debug(self): - self.__printer.disable() + self._printer.disable() # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ + @property + def now_utc(self) -> int | float: + + """ + Returns the current UTC time as a timestamp. + :return: The current UTC time as a timestamp. + """ + + # Return the current time in UTC as a timestamp: + return date_time.get_current_utc_date_time(as_string = False).timestamp() + @staticmethod def make_key(*args, **kwargs) -> str: diff --git a/utils_v2/cache_v2/async_disk_cache.py b/utils_v2/cache_v2/async_disk_cache.py new file mode 100644 index 0000000..3372dcb --- /dev/null +++ b/utils_v2/cache_v2/async_disk_cache.py @@ -0,0 +1,642 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + CREATED: Thu, 27th Nov, 2025 + UPDATED: Thu, 27th Nov, 2025 + + OBJECTIVE: + + To define a caching class that uses the persistent disk to hold cached data. + This is a good way to cache data across multiple processes on the same machine (as long as they have access to + the directory where the cached data is being stored). + + WARNING: + -------- + THIS IS A MODERATELY STATEFUL WAY OF IMPLEMENTING CACHING. THE USER'S REQUEST NEEDS TO HIT THE EXACT SAME + MACHINE AGAIN FOR THE CACHING TO BE MEANINGFUL. IT CAN HIT ANY PROCESS ON THE SAME MACHINE, BUT THE PHYSICAL + MACHINE WILL HAVE TO BE THE SAME. + + The idea is simple: + ------------------- + Everything that needs to be cached should have a key (identifier) and a value (the actual data). The key becomes + the name of the file that holds the data on disk. The data itself is pickled using Python's pickle library. This + way it retains its native datatypes. + + TROUBLESHOOTING: + ---------------- + There's always a possibility that another user or process may accidentally mess with the caching directory and + throw the whole system off. The simplest workaround is to delete the caching directory and rerun the program. + On a fresh restart, the program will re-create the directory and things start from zero. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For system-level activities: +import os + +# For data-processing: +import pickle + +# For working with tabulated data: +import pandas as pd + +# Other utils: +from utils_v2.date_time import date_time +from utils_v2.string import regex +from utils_v2.cache_v2.async_base import AsyncCachingBase +from utils_v2.system import files + +# To work with datatypes: +from typing import List, Any + +# For asynchronous activities: +import asyncio + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class AsyncDiskCache(AsyncCachingBase): + + def __init__( + self, + caching_dir: str, + lock_on_read: bool = False, + lock_wait_timeout: int | float = 5.0, + expiry_check_interval: int | float = 60.0, + debug = True, + debug_prefix = "ADiskCache | " + ): + + """ + Implements a simple cache in disk that holds and returns all native datatypes. All timestamps are normalized to + UTC timezone. + NOTE: It is NOT async, actually. It has been built on top of an async class so it has been declared as if it is. + :param caching_dir: Directory where the cached data is stored. + :param lock_on_read: Whether, or not, read operations lock the cached data. All modification operations (write, + delete, update) will trigger a lock. Reading can be done without such a constraint. + :param lock_wait_timeout: When cached data is being accessed, it may be locked. This is the default timeout + for the file to get unlocked. + :param expiry_check_interval: How frequently to check for expiry of all the cached data. When you request very + specific data, its expiry will be checked before serving; but this interval defines a general cleanup to + release memory. + :param debug: Whether, or not, you want to show debugging messages from the start. + :param debug_prefix: The prefix text to show with the debugging messages. + """ + + # Invoke the parent class's constructor: + super().__init__( + debug = debug, + debug_prefix = debug_prefix + ) + + # Note down the variables: + self.__caching_dir = caching_dir + self.__lock_on_read = lock_on_read + self.__lock_wait_timeout = lock_wait_timeout + self.__expiry_check_interval = expiry_check_interval + self.__last_expiry_check_ts = date_time.get_current_utc_date_time(as_string = False).timestamp() + + # Create a local dir that will hold the cached data: + if not os.path.exists(self.__caching_dir): + files.make_directory(self.__caching_dir) + + # Create a lookup table and store that as a separate file: + if not os.path.exists(os.path.join(self.__caching_dir, "lookup.pkl")): + lookup_df = pd.DataFrame(columns = ["key", "exp"]) + self.to_disk(key = "lookup", value = lookup_df) + + # If we yet don't have a lookup file: + if not os.path.exists(os.path.join(self.__caching_dir, "lookup.pkl")): + exception = RuntimeError("CRITICAL: Lookup file not found!") + self._printer("LOOKUP FAIL!", exception) + + # ┓┏ ┓ + # ┣┫┏┓┃┏┓┏┓┏┓┏ + # ┛┗┗ ┗┣┛┗ ┛ ┛ + # ┛ + + @property + def now_utc(self) -> int | float: + + """ + Returns the current UTC time as a timestamp. + :return: The current UTC time as a timestamp. + """ + + # Return the current time in UTC as a timestamp: + return date_time.get_current_utc_date_time(as_string = False).timestamp() + + async def exists( + self, + key: str + ) -> bool: + + """ + Checks if a particular key exists in the cached data. + :param key: + :return: + """ + + async def __lock( + self, + key: str + ) -> bool: + + """ + To mark a key as locked so that no other process tries to access it at the same time. Read access may yet be + granted depending on the value of 'lock_on_read'. + :param key: The identifier of the cached data. + :return: True if successful, False otherwise. + """ + + try: + + # Try to write a lock file with the + # UTC timestamp in it for reference of when it was created: + files.write_file( + file_path = os.path.join(self.__caching_dir, f"{key}.lock"), + file_data = str(self.now_utc), + mode = "w", + raise_exception = True + ) + + # No exception means success: + return True + + except Exception as exception: + self._printer("LOCK FAIL!", key, exception) + self._printer(exception) + return False + + async def __unlock( + self, + key: str + ) -> bool: + + """ + To mark a key as unlocked so that other processes may start accessing it. Reading access may always be unlocked + depending on the value of 'lock_on_read'. + :param key: The identifier of the cached data. + :return: True if successful, False otherwise. + """ + + try: + + # Delete the file that indicates that a key is locked: + files.delete_file( + file_path = os.path.join(self.__caching_dir, f"{key}.lock"), + raise_exception = True + ) + + # No exception means success: + return True + + except Exception as exception: + self._printer("UNLOCK FAIL!", key, exception) + return False + + async def __wait_for_unlock( + self, + key: str, + lock_wait_timeout: int | float = None + ) -> None: + + """ + THis waits for a particular key to be unlocked. + :param key: The identifier of the cached data. + :param lock_wait_timeout: The max amount to wait for a particular key to be unlocked. If null, the default value + set in 'lock_wait_timeout' (from the constructor) will be used. You may override that value by passing a + custom value here. + :return: None. + """ + + # Figure out when the time will run out: + timeout = lock_wait_timeout or self.__lock_wait_timeout + exp_utc = self.now_utc + timeout + + # Wait for either the key to get unlocked, + # or the time to run out: + while os.path.exists(os.path.join(self.__caching_dir, f"{key}.lock")): + if self.now_utc >= exp_utc: + exception = TimeoutError(f"Key '{key}' was locked for more than {timeout:.2} seconds.") + self._printer("LOCK WAIT TIMED OUT!", key, timeout, exception) + raise exception + await asyncio.sleep(0.05) + + async def to_disk( + self, + key: str, + value: Any, + lock_wait_timeout: int | float = None, + ) -> bool: + + """ + Stores the 'value' to the disk and keeps the 'key' as the filename. + :param key: The identifier of the cached data. Becomes the name of the file when stored on disk. + :param value: The actual cached data. + :param lock_wait_timeout: How long (in seconds) to wait for the key to get unlocked. + :return: True if cached, else False. + """ + + # We first wait for the key to get unlocked: + await self.__wait_for_unlock( + key = key, + lock_wait_timeout = lock_wait_timeout + ) + + try: + + # First lock the key so that no other process can modify it: + if not await self.__lock(key = key): + raise RuntimeError("Failed to lock key '{key}'.") + + # Try to store the data: + files.write_file( + file_path = os.path.join(self.__caching_dir, f"{key}.pkl"), + file_data = pickle.dumps(value), + mode = "wb", + raise_exception = True + ) + + # Unlock the key: + if not await self.__unlock(key = key): + raise RuntimeError("Failed to unlock key '{key}'.") + + # Done here: + return True + + # If something goes wrong: + except Exception as exception: + self._printer("CACHE SAVING FAIL!", exception) + await self.__unlock(key = key) + return False + + async def from_disk( + self, + key: str + ) -> Any: + + """ + Reads cached data from the disk. + :param key: The identifier of the cached data. It is the name of the file when stored on disk. + :return: The read data. + """ + + # If configured that way, + # we must lock the key before reading: + if self.__lock_on_read: + if not await self.__lock(key = key): + raise RuntimeError("Failed to lock key '{key}'.") + + try: + + # Try to read the data: + data = files.read_file( + file_path = os.path.join(self.__caching_dir, f"{key}.pkl"), + mode = "rb", + raise_exception = True + ) + data = pickle.loads(data) + + # Unlock the key: + if self.__lock_on_read: + if not await self.__unlock(key = key): + raise RuntimeError("Failed to unlock key '{key}'.") + + # Done here: + return data + + # If something goes wrong: + except Exception as exception: + self._printer("CACHE MISS!", exception) + await self.__unlock(key = key) + return False + + async def clear_expired_keys(self) -> None: + + """ + Clears all the expired cached data. + :return: None. + """ + + # If the expiry check interval has not been crossed, + # we need not go through the process: + ref_utc = self.now_utc + if ref_utc - self.__last_expiry_check_ts < self.__expiry_check_interval: return + + # Otherwise, we note down the current timestamp and proceed: + self.__last_expiry_check_ts = ref_utc + + # Load the lookup and check what all needs to be deleted: + ref_utc = self.now_utc + lookup_df = await self.from_disk(key = "lookup") + to_del = lookup_df[lookup_df["exp"] <= ref_utc]["key"].to_list() + + + + # # Enlist all the keys that need to be deleted: + # to_del = [] + # for k, v in self.__cached_data.items(): + # if ref_utc >= v["exp"]: to_del.append(k) + # + # # Delete the expired keys: + # for k in to_del: + # self.__cached_data.pop(k) + + # ┏┓ + # ┃ ┏┓┏┓┏┓ + # ┗┛┗┛┛ ┗ + + async def list_keys( + self, + match: str | None = None, + raise_exception: bool = False + ) -> List[str] | None: + + """ + To get a list of all the keys that match the pattern. + :param match: To match a glob-style pattern. THIS IS NOT FULL-FLEDGED REGEX. + :param raise_exception: Whether to raise or suppress exceptions. + :return: The list of keys if successful, else None. + """ + + # Clear the expired keys: + await self.clear_expired_keys() + + # Start with an empty list: + keys_list = [] + + try: + + # Enlist all the keys, + # test the pattern against all the keys and keep only those that match: + ref_utc = self.now_utc + for k, v in self.__cached_data.items(): + if v["exp"] > ref_utc: + if match is None: keys_list.append(k) + elif regex.match(text = str(k), pattern = match): + keys_list.append(k) + + # Done here: + return list(set(keys_list)) if keys_list else None + + # If an exception occurs in the process: + except Exception as exception: + self._printer(exception) + if raise_exception: raise + else: return None + + async def ttl( + self, + key: str | bytes, + raise_exception: bool = False + ) -> int | float | None: + + """ + To get the no. of seconds till the expiry of some key. + :param key: The key to check the expiry of. + :param raise_exception: Whether to raise or suppress exceptions. + :return: -1 if the key is persistent (i.e., no expiry time set), -2 if the key does not exist, or the time left + in seconds if the key exists and is not persistent. If something goes wrong, you will get a null value. + """ + + # Clear the expired keys: + await self.clear_expired_keys() + + # Start by assuming failure: + ttl = None + + try: + + # Get the expiry: + exp_utc = self.__cached_data.get(key, {}).get("exp", None) + if exp_utc is None: ttl = -2 + else: ttl = exp_utc - self.now_utc() + return ttl + + # If an exception occurs in the process: + except Exception as exception: + self._printer(exception) + if raise_exception: raise + else: return None + + async def set( + self, + key: str | bytes, + value: Any, + expiry: float = None, + raise_exception: bool = False + ) -> bool: + + """ + Saves some value to the cache. If an expiry is specified, the data will be deleted after that many seconds. + :param key: The key with which the data will be stored and retrieved. + :param value: The value to store. + :param expiry: The time in seconds after which the data will expire. Must be a positive number. + :param raise_exception: Whether to raise or suppress exceptions. + :return: True if cached, else False. + """ + + # Clear the expired keys: + await self.clear_expired_keys() + + try: + + # If the expiry is not specified, + # make it an unreasonably far off future date: + if expiry is None: expiry = self.now_utc + 3_15_36_000 + + # Here we actually try to set the data: + self.__cached_data[key] = { + "val": value, + "exp": self.now_utc + expiry + } + + # Done here: + return True + + # If an exception occurs in the process: + except Exception as exception: + self._printer(exception) + if raise_exception: raise + else: return False + + async def get( + self, + key: str | bytes, + raise_exception: bool = False, + on_fail: Any = None + ) -> Any: + + """ + Retrieve the cached value. + :param key: The key with which the data was saved. + :param raise_exception: Whether to raise or suppress exceptions. + :param on_fail: What to return if the process fails due to an exception. + :return: The retrieved data or null if not found. + """ + + # Clear the expired keys: + await self.clear_expired_keys() + + try: + + # Here we try to fetch the data: + data = None + raw = self.__cached_data.get(key) + + # Check for expiry if it was a cache hit: + if raw: data = raw["val"] if raw["exp"] > self.now_utc else None + + # Done here: + return data + + # If an exception occurs in the process: + except Exception as exception: + self._printer(exception) + if raise_exception: raise + else: return on_fail + + async def delete( + self, + key: str | bytes, + raise_exception: bool = False + ) -> bool: + + """ + Prematurely delete the value from the cache before it expires. + :param key: The key with which the data was saved. + :param raise_exception: Whether to raise or suppress exceptions. + :return: True if deleted, else False. + """ + + # Clear the expired keys: + await self.clear_expired_keys() + + try: + + # Try to manually delete the key before expiry: + response = self.__cached_data.pop(key, None) + return True if response else False + + # If an exception occurs in the process: + except Exception as exception: + self._printer(exception) + if raise_exception: raise + else: return False + + async def count( + self, + key: str | bytes, + value: int = 1, + expiry: float = None, + raise_exception: bool = False + ) -> int | None: + + """ + To use simple counters. If the counter (identified by the 'key') exists, it will be incremented, else the + counter will be created and the value will be incremented from 0. + :param key: The name of the counter. + :param value: The amount to increment the value by. Send negative values to count backwards. + :param expiry: The time (in seconds) in which the counter expires. Starts from the time the counter is created. + This value has to be an integer. If a float is passed, the value will be rounded off. + :param raise_exception: Whether to raise or suppress exceptions. + :return: The latest value of the counter. Will be null if something went wrong and the exception was suppressed. + """ + + # Clear the expired keys: + await self.clear_expired_keys() + + try: + + # Check if + already_existed = False if self.__cached_data.get(key, None) is None else True + + # If it already exists, we just increment the value; + # else we create the value and increment the value starting from zero: + if already_existed: self.__cached_data[key]["val"] = self.__cached_data[key]["val"] + value + else: self.__cached_data[key] = { + "val": value, + "exp": self.now_utc + expiry + } + + # Return the new value: + return self.__cached_data[key]["val"] + + # If an exception occurs in the process: + except Exception as exception: + self._printer(exception) + if raise_exception: raise + else: return None + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/utils_v2/cache_v2/async_local_cache.py b/utils_v2/cache_v2/async_mem_cache.py similarity index 66% rename from utils_v2/cache_v2/async_local_cache.py rename to utils_v2/cache_v2/async_mem_cache.py index 2ea446d..0942872 100644 --- a/utils_v2/cache_v2/async_local_cache.py +++ b/utils_v2/cache_v2/async_mem_cache.py @@ -7,11 +7,24 @@ DATE: CREATED: Tue, 25th Nov, 2025 - UPDATED: Tue, 25th Nov, 2025 + UPDATED: Thu, 27th Nov, 2025 OBJECTIVE: To define a caching class that uses in-RAM dicts to hold cached data. + This is a good way to cache data for a single-machine, single-process service. The cached data is rapidly + available (since it's held in RAM), but it cannot be shared across various instances. + + WARNING: + -------- + THIS IS A STRONGLY STATEFUL WAY OF IMPLEMENTING CACHING. THE USER'S REQUEST WILL HAVE TO HIT EXACTLY THE SAME + MACHINE AND EXACTLY THE SAME PROCESS ON THAT MACHINE FOR THE CACHE TO BE ANY GOOD. + + The idea is simple: + ------------------- + Everything that needs to be cached should have a key (identifier) and a value (the actual data). We hold the + cache in a dictionary in Python where the identifier becomes the key of the dict, and the value is the cached + data. REFERENCES: @@ -33,24 +46,17 @@ # To make sibling directories accessible for imports: import sys -from unittest import case - sys.path.append(".") sys.path.append("..") # Other utils: -from utils_v2.string import json -from utils_v2.serialization.pickle_serializer import PickleSerializer from utils_v2.date_time import date_time from utils_v2.string import regex -from utils_v2.cache_v2.base import AsyncCachingBase +from utils_v2.cache_v2.async_base import AsyncCachingBase # To work with datatypes: from typing import List, Any -# For async activities: -import asyncio - # ***************************************************************************************************************** # ***** **** @@ -79,17 +85,23 @@ import asyncio # ***************************************************************************************************************** -class AsyncLocalCache(AsyncCachingBase): +class AsyncMemCache(AsyncCachingBase): def __init__( self, - debug = False, - debug_prefix = "R-Cache | " + expiry_check_interval: int | float = 60.0, + debug = True, + debug_prefix = "AMemCache | " ): """ - Implements a simple cache in RAM that holds and returns all native datatypes like ints, floats, bools, - strings, dicts, lists, sets, and tuples :) + Implements a simple cache in RAM that holds and returns all native datatypes. This is good for single-process + implementations that run on a single machine because the cached data will not be accessible to other processes + even when running on the same machine. Also, all timestamps are normalized to UTC. + NOTE: It is NOT async, actually. It has been built on top of an async class so it has been declared as if it is. + :param expiry_check_interval: How frequently to check for expiry of all the cached data. When you request very + specific data, its expiry will be checked before serving; but this interval defines a general cleanup to + release memory. :param debug: Whether, or not, you want to show debugging messages from the start. :param debug_prefix: The prefix text to show with the debugging messages. """ @@ -102,51 +114,57 @@ class AsyncLocalCache(AsyncCachingBase): # Create a local dict that will hold the cached data: self.__cached_data = {} - - # ┳┓ ┓ • - # ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓ - # ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫ - # ┛ ┛ ┛ - - def enable_debug(self): - self.__printer.enable() - - def disable_debug(self): - self.__printer.disable() + self.__expiry_check_interval = expiry_check_interval + self.__last_expiry_check_ts = date_time.get_current_utc_date_time(as_string = False).timestamp() # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ - @staticmethod - def now_utc() -> int | float: + async def __expired( + self, + key: str + ) -> bool: """ - Returns the current UTC time as a timestamp. - :return: The current UTC time as a timestamp. + Tells you if a key has expired. + :param key: The key that you want to identify. + :return: True if expired, False if valid. """ - # Return the current time in UTC as a timestamp: - return date_time.get_current_utc_date_time(as_string = False).timestamp() + # If the key doesn't exist, it has expired: + if key not in self.__cached_data: return True - async def clear_expired_keys(self) -> None: + # If the key exists, but the current timestamp has crossed the expiry timestamp; + # we also try to delete the key in that case: + elif self.now_utc >= self.__cached_data[key]["exp"]: + await self.delete(key, raise_exception = False) + return True + + # Otherwise it is yet valid (not expired): + else: return False + + async def __clear_expired_keys(self) -> None: """ Clears all the expired cached data. :return: None. """ - # Keys to delete: - to_del = [] + # If the expiry check interval has not been crossed, + # we need not go through the process: + ref_utc = self.now_utc + if ref_utc - self.__last_expiry_check_ts < self.__expiry_check_interval: return + + # Otherwise, we note down the current timestamp and proceed: + self.__last_expiry_check_ts = ref_utc # Enlist all the keys that need to be deleted: - for k, v in self.__cached_data.items(): - if self.now_utc() >= v["exp"]: to_del.append(k) + to_del = [k for k in self.__cached_data.keys() if self.__expired(k)] # Delete the expired keys: - for k in to_del: - self.__cached_data.pop(k) + for k in to_del: await self.delete(k, raise_exception = False) # ┏┓ # ┃ ┏┓┏┓┏┓ @@ -166,7 +184,7 @@ class AsyncLocalCache(AsyncCachingBase): """ # Clear the expired keys: - await self.clear_expired_keys() + await self.__clear_expired_keys() # Start with an empty list: keys_list = [] @@ -175,17 +193,18 @@ class AsyncLocalCache(AsyncCachingBase): # Enlist all the keys, # test the pattern against all the keys and keep only those that match: - for k in self.__cached_data.keys(): - if match is None: keys_list.append(k) - elif regex.match(text = str(k), pattern = match): - keys_list.append(k) + for k, v in self.__cached_data.items(): + if not await self.__expired(k): + if match is None: keys_list.append(k) + elif regex.match(text = str(k), pattern = match): + keys_list.append(k) # Done here: return list(set(keys_list)) if keys_list else None - # If an exception occurs in th eprocess: + # If an exception occurs in the process: except Exception as exception: - self.__printer(exception) + self._printer("LIST ERR!", exception) if raise_exception: raise else: return None @@ -204,7 +223,7 @@ class AsyncLocalCache(AsyncCachingBase): """ # Clear the expired keys: - await self.clear_expired_keys() + await self.__clear_expired_keys() # Start by assuming failure: ttl = None @@ -214,12 +233,12 @@ class AsyncLocalCache(AsyncCachingBase): # Get the expiry: exp_utc = self.__cached_data.get(key, {}).get("exp", None) if exp_utc is None: ttl = -2 - else: ttl = exp_utc - self.now_utc() + else: ttl = exp_utc - self.now_utc return ttl - # If an exception occurs in th eprocess: + # If an exception occurs in the process: except Exception as exception: - self.__printer(exception) + self._printer("TTL ERR!", key, exception) if raise_exception: raise else: return None @@ -241,28 +260,26 @@ class AsyncLocalCache(AsyncCachingBase): """ # Clear the expired keys: - await self.clear_expired_keys() + await self.__clear_expired_keys() try: # If the expiry is not specified, # make it an unreasonably far off future date: - if expiry is None: - expiry = date_time.get_current_utc_date_time(as_string = False) + date_time.timedelta(days = 365) - expiry = expiry.timestamp() + if expiry is None: expiry = self.now_utc + self.LONG_EXPIRY # Here we actually try to set the data: self.__cached_data[key] = { "val": value, - "exp": self.now_utc() + expiry + "exp": self.now_utc + expiry } # Done here: return True - # If an exception occurs in th eprocess: + # If an exception occurs in the process: except Exception as exception: - self.__printer(exception) + self._printer("SET ERR!", key, exception) if raise_exception: raise else: return False @@ -282,17 +299,27 @@ class AsyncLocalCache(AsyncCachingBase): """ # Clear the expired keys: - await self.clear_expired_keys() + await self.__clear_expired_keys() try: # Here we try to fetch the data: - data = self.__cached_data.get(key, {}).get("val", None) + data = on_fail + raw = self.__cached_data[key] + + # Check for expiry if it was a cache hit: + if await self.__expired(key) and raise_exception: raise RuntimeError(f"Key '{key}' not found.") + else: data = raw["val"] + + # if self.now_utc >= raw["exp"]: await self.delete(key, raise_exception = True) + # else: data = raw["val"] + + # Done here: return data - # If an exception occurs in th eprocess: + # If an exception occurs in the process: except Exception as exception: - self.__printer(exception) + self._printer("CACHE MISS!", key, exception) if raise_exception: raise else: return on_fail @@ -310,7 +337,7 @@ class AsyncLocalCache(AsyncCachingBase): """ # Clear the expired keys: - await self.clear_expired_keys() + await self.__clear_expired_keys() try: @@ -318,9 +345,9 @@ class AsyncLocalCache(AsyncCachingBase): response = self.__cached_data.pop(key, None) return True if response else False - # If an exception occurs in th eprocess: + # If an exception occurs in the process: except Exception as exception: - self.__printer(exception) + self._printer("DEL ERR!", key, exception) if raise_exception: raise else: return False @@ -344,27 +371,29 @@ class AsyncLocalCache(AsyncCachingBase): """ # Clear the expired keys: - await self.clear_expired_keys() + await self.__clear_expired_keys() try: - # Check if - already_existed = False if self.__cached_data.get(key, None) is None else True + # If the key already exists and hasn't expired yet: + if not await self.__expired(key): + self.__cached_data[key]["val"] = self.__cached_data[key]["val"] + value - # If it already exists, we just increment the value; - # else we create the value and increment the value starting from zero: - if already_existed: self.__cached_data[key]["val"] = self.__cached_data[key]["val"] + value - else: self.__cached_data[key] = { - "val": value, - "exp": self.now_utc() + expiry - } + # If the key has expired or if it is new: + else: + await self.set( + key = key, + value = value, + expiry = expiry, + raise_exception = True + ) # Return the new value: - return self.__cached_data[key]["val"] + return await self.get(key, raise_exception = True) - # If an exception occurs in th eprocess: + # If an exception occurs in the process: except Exception as exception: - self.__printer(exception) + self._printer("COUNT ERR!", key, value, exception) if raise_exception: raise else: return None diff --git a/utils_v2/cache_v2/async_redis_cache.py b/utils_v2/cache_v2/async_redis_cache.py index 51f822b..db10959 100644 --- a/utils_v2/cache_v2/async_redis_cache.py +++ b/utils_v2/cache_v2/async_redis_cache.py @@ -11,7 +11,9 @@ OBJECTIVE: - To define a caching class that uses Redis to asynchronously cache information. + To define a caching class that uses Redis to asynchronously cache information. This is truly stateless caching + since yu can run your backend script from any number of servers and yet have the same cache data sync'd through + Redis. REFERENCES: @@ -43,7 +45,7 @@ from redis.asyncio.sentinel import Sentinel # Other utils: from utils_v2.string import json from utils_v2.serialization.pickle_serializer import PickleSerializer -from utils_v2.cache_v2.base import AsyncCachingBase +from utils_v2.cache_v2.async_base import AsyncCachingBase # To work with datatypes: from typing import List, Any @@ -80,7 +82,7 @@ class AsyncRedisCache(AsyncCachingBase): def __init__( self, - connection_string: str | dict = None, + connection_params: str | dict = None, serializer = None, ping_counter = 1_000, debug = False, @@ -90,7 +92,7 @@ class AsyncRedisCache(AsyncCachingBase): """ Implements a simple cache in Redis which holds and returns all native datatypes like ints, floats, bools, strings, dicts, lists, sets, and tuples :) - :param connection_string: The connection URL or Sentinel JSON for connecting to Redis. + :param connection_params: The connection URL or Sentinel JSON for connecting to Redis. :param serializer: The serializer to use. :param ping_counter: The number of requests to Redis after which you want to ping to ensure connection. :param debug: Whether, or not, you want to show debugging messages from the start. @@ -110,30 +112,19 @@ class AsyncRedisCache(AsyncCachingBase): self.__requests_since_last_ping = 0 # Figure out the connection mechanism: - if isinstance(connection_string, dict): - self.__sentinel_json = connection_string + if isinstance(connection_params, dict): + self.__sentinel_json = connection_params self.__connection_string = None - self.__printer("Received Sentinel JSON.") - elif isinstance(connection_string, str): + self._printer("Received Sentinel JSON.") + elif isinstance(connection_params, str): try: - self.__sentinel_json = json.from_string(connection_string) + self.__sentinel_json = json.from_string(connection_params) self.__connection_string = None - self.__printer("Seems like a Sentinel JSON String.") + self._printer("Seems like a Sentinel JSON String.") except: self.__sentinel_json = None - self.__connection_string = connection_string - self.__printer("Seems like a regular Connection String.") - - # ┳┓ ┓ • - # ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓ - # ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫ - # ┛ ┛ ┛ - - def enable_debug(self): - self.__printer.enable() - - def disable_debug(self): - self.__printer.disable() + self.__connection_string = connection_params + self._printer("Seems like a regular Connection String.") # ┏┓ • # ┃ ┏┓┏┓┏┓┏┓┏╋┓┏┓┏┓ @@ -174,7 +165,7 @@ class AsyncRedisCache(AsyncCachingBase): else: raise ValueError("Either a Sentinel JSON or a Connection String is needed.") except Exception as exception: - self.__printer(exception) + self._printer("CONN. ERR!", exception) return False async def disconnect(self) -> bool: @@ -187,7 +178,7 @@ class AsyncRedisCache(AsyncCachingBase): if self.__client is not None: try: await self.__client.close() except Exception as exception: - self.__printer(exception) + self._printer(exception) return False return True return True @@ -210,7 +201,7 @@ class AsyncRedisCache(AsyncCachingBase): self.__requests_since_last_ping = 0 return True except Exception as exception: - self.__printer(exception) + self._printer(exception) return await self.connect() else: self.__requests_since_last_ping += 1 @@ -254,7 +245,7 @@ class AsyncRedisCache(AsyncCachingBase): # If an exception occurs in th eprocess: except Exception as exception: - self.__printer(exception) + self._printer("LIST ERR!", exception) if raise_exception: raise else: return None @@ -288,7 +279,7 @@ class AsyncRedisCache(AsyncCachingBase): # If an exception occurs in th eprocess: except Exception as exception: - self.__printer(exception) + self._printer("TTL ERR!", key, exception) if raise_exception: raise else: return None @@ -322,7 +313,7 @@ class AsyncRedisCache(AsyncCachingBase): # If an exception occurs in th eprocess: except Exception as exception: - self.__printer(exception) + self._printer("SET ERR!", key, exception) if raise_exception: raise else: return False @@ -353,7 +344,7 @@ class AsyncRedisCache(AsyncCachingBase): # If an exception occurs in th eprocess: except Exception as exception: - self.__printer(exception) + self._printer("CACHE MISS!", key, exception) if raise_exception: raise else: return on_fail @@ -381,7 +372,7 @@ class AsyncRedisCache(AsyncCachingBase): # If an exception occurs in th eprocess: except Exception as exception: - self.__printer(exception) + self._printer("DEL ERR!", key, exception) if raise_exception: raise else: return False @@ -409,7 +400,7 @@ class AsyncRedisCache(AsyncCachingBase): try: - # check if the key already exists, + # Check if the key already exists, # regardless of that, increment the counter: already_existed = await self.__client.exists(key) new_value = await self.__client.incrby(key, value) @@ -422,7 +413,7 @@ class AsyncRedisCache(AsyncCachingBase): # If an exception occurs in th eprocess: except Exception as exception: - self.__printer(exception) + self._printer("COUNT ERR!", key, value, exception) if raise_exception: raise else: return None