""" AUTHOR: Khushal P Soonderji DATE: CREATED: Thu, 5th Feb, 2026 UPDATED: Thu, 5th Feb, 2026 OBJECTIVE: There will be some common actions across various scripts. This script holds those common actions. 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 import copy # To work with date and time: import time import datetime # To work with tabulate data: import pandas as pd # Cosec-related: from cosec_web.cosec_web import CosecWeb # TCAOFF-related: from helpers.async_tcaoff import AsyncTheCAOffice # 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, Union from collections import defaultdict # To run a cron-like scheduler: from apscheduler.schedulers.asyncio import AsyncIOScheduler # For async activities: import asyncio # For debugging: from icecream import IceCreamDebugger # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # File paths: FILE_DIR = files.get_file_directory(include_filename = False) PROJ_DIR = files.get_parent_directory(FILE_DIR, depth = 1) CACHE_DIR = os.path.join(PROJ_DIR, "local", "cache") CREDS_DIR = os.path.join(PROJ_DIR, "creds") # --- COSEC_CREDS_FILE = os.path.join(CREDS_DIR, "cosec.json") TCAOFF_CREDS_FILE = os.path.join(CREDS_DIR, "tcaoff.json") MUSTER_ROLL_CACHE_FILE = os.path.join(CACHE_DIR, "muster_roll_cache.json") IN_OUT_SUMMARY_CACHE_FILE = os.path.join(CACHE_DIR, "in_out_summary_cache.json") PREV_DAY_IN_OUT_SUMMARY_CACHE_FILE = os.path.join(CACHE_DIR, "prev_day_in_out_summary_cache.json") CHROME_DRIVER_DIR = os.path.join(PROJ_DIR, "drivers", "chrome") USER_DATA_DIR = os.path.join(PROJ_DIR, "browser", "user_data") DOWNLOADS_DIR = os.path.join(PROJ_DIR, "downloads") # --- TEST_MODE_MUSTER_ROLL_FILE_PATH = os.path.join(PROJ_DIR, "cosec_web", "sample_files", "muster_roll.xls") TEST_MODE_IN_OUT_SUMMARY_FILE_PATH = os.path.join(PROJ_DIR, "cosec_web", "sample_files", "in_out_summary.xls") # Debugging: printer = IceCreamDebugger(prefix = "Common | ", includeContext = True) err_printer = IceCreamDebugger(prefix = "[ERR] Common | ", includeContext = True) # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** def kill_chrome() -> None: """ To kill running Chrome processes so that they don't interfere with the one that will be spun up by automation code. :return: None """ # First kill the previous processes, # then wait if old processes were killed: kill_count = CosecWeb.kill_chrome_processes() if kill_count > 0: time.sleep(2.5) # --------------------------------------------------------------------------------------------------------------------- def get_muster_roll( cosec_creds: dict, on_date: datetime.datetime, cache_file: str = MUSTER_ROLL_CACHE_FILE, test_mode: bool = False ) -> bool: """ 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. :param on_date: The date on which you want . :param cache_file: The cache file to store the results in. :param test_mode: If set to True, a past file will be used instead of getting new reports from Cosec. :return: True if the automated fetch was successful, else False. """ # Start by assuming failure: success = False report_path = None # If test mode: if test_mode: report_path = TEST_MODE_MUSTER_ROLL_FILE_PATH # When not working in test mode: else: # 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 = on_date, group_ids = cosec_creds["musterRollConfig"]["groupIds"], download_timeout = 60.0 ) # Log out to end the cycle: cosec.logout() # Close the browser window: cosec.quit() # Now process the report, # and save it to the JSON file: if report_path is not None: # Read the data: report_data = CosecWeb.read_muster_roll_xls(report_path) # Do the remaining cleanup and formatting: report_data = report_data[[ "User ID", "User Name", "Category Name", "Grade Name", "Branch Name", "Department Name", "Direct Reporting", "Level-1" ]] 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 = cache_file, python_data = { "ts": date_time.get_current_utc_date_time(as_string = False).timestamp(), "report": report_data }, no_space = True ) # Note down success: success = True # Done here: return success # --------------------------------------------------------------------------------------------------------------------- async def sync_branches_to_tcaoff( tcaoff_client: AsyncTheCAOffice, 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 = await tcaoff_client.branch_list() tcaoff_branches = [_["branch_name"].lower() for _ in tcaoff_branches] tcaoff_branches = list(set(tcaoff_branches)) # Get only the branch names from Cosec: cosec_branches = [tcaoff_client.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.lower() not in tcaoff_branches: success = await tcaoff_client.branch_add(branch_name = cosec_branch) response["total"] += 1 if success: response["success"] += 1 else: response["fail"] += 1 # Done here: return response # --------------------------------------------------------------------------------------------------------------------- async def sync_departments_to_tcaoff( tcaoff_client: AsyncTheCAOffice, 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 = await tcaoff_client.department_list() tcaoff_depts = [_["department_name"].lower() for _ in tcaoff_depts] tcaoff_depts = list(set(tcaoff_depts)) # Get only the department names from Cosec: cosec_depts = [tcaoff_client.remove_special_chars(_["Department Name"]) for _ in cosec_muster_roll["report"]] cosec_depts = list(set(cosec_depts)) # Loop through the data from Cosec and add missing departments to TCAOFF: for cosec_dept in cosec_depts: if cosec_dept.lower() not in tcaoff_depts: success = tcaoff_client.department_add(department_name = cosec_dept) response["total"] += 1 if success: response["success"] += 1 else: response["fail"] += 1 # Done here: return response # --------------------------------------------------------------------------------------------------------------------- async def sync_teams_to_tcaoff( tcaoff_client: AsyncTheCAOffice, cosec_muster_roll: dict, ) -> Dict[str, int]: # Start with a basic response structure: response = defaultdict(int) # Map out branch ids: tcaoff_branches = await tcaoff_client.branch_list() tcaoff_branches_lookup = {d["branch_name"].lower(): d["branch_id"] for d in tcaoff_branches} # Map out dept. ids: tcaoff_depts = await tcaoff_client.department_list() tcaoff_depts_lookup = {d["department_name"].lower(): d["department_id"] for d in tcaoff_depts} # Get the list of existing team members from TCAOFF: # NOTE: `pseudonym` is the unique username of the user. tcaoff_teams = await tcaoff_client.team_list() synced_team_ids = [] for t in tcaoff_teams: app_notes = json.from_string(t["json_notes"]).get("applicantNotes", {}) if isinstance(app_notes, str): app_notes = json.from_string(app_notes) if app_notes is None: app_notes = {} cosec_notes = app_notes.get("cosec", {}) print("COSEC NOTES:", cosec_notes) if cosec_notes: user_id = cosec_notes.get("User ID") if not user_id: user_id = cosec_notes["UserID"] synced_team_ids.append(user_id) tcaoff_teams = [_["pseudonym"] for _ in tcaoff_teams] tcaoff_teams = list(set(tcaoff_teams)) # Loop through the data from Cosec and add missing team-members to TCAOFF: for cosec_team in cosec_muster_roll["report"]: if cosec_team["User ID"] not in synced_team_ids: # Count the user: response["total"] += 1 # Remove special chars: branch_name = tcaoff_client.remove_special_chars(cosec_team["Branch Name"]).lower() dept_name = tcaoff_client.remove_special_chars(cosec_team["Branch Name"]).lower() # Check the branch id and department id: branch_id = tcaoff_branches_lookup.get(branch_name) dept_id = tcaoff_depts_lookup.get(dept_name) if branch_id is None: err_printer("TEAM SYNC ERR", "Branch Not Found in TCAOFF", branch_name) response["fail"] += 1 continue if dept_id is None: err_printer("TEAM SYNC ERR", "Dept. Not Found in TCAOFF", dept_name) response["fail"] += 1 continue # Add the team: tcaoff_username = regex.replace( text = cosec_team["User Name"], pattern = r"[^\w\d\._]", substitute_text = "" ).strip().lower() # print("Need to Add:", json.to_string(team_json)) success = await tcaoff_client.team_add( branch_id = branch_id, dept_id = dept_id, reporting_to = None, team_name = cosec_team["User Name"].strip(), email = tcaoff_username + "@velankanigroup.com", phone_no = "9876543210", role = (cosec_team.get("Grade Name") or "Unknown").strip(), username = tcaoff_username, password = "Vispl@123", applicant_notes = {"cosec": cosec_team} ) if success: response["success"] += 1 else: err_printer("COULDN'T ADD", cosec_team) response["fail"] += 1 # Done here: printer("TCAOFF-Cosec Depts. Sync.:", json.to_string(response)) return response # --------------------------------------------------------------------------------------------------------------------- def get_in_out_summary( cosec_creds: dict, from_dt: date_time.datetime = None, to_dt: date_time.datetime = None, cache_file: str = IN_OUT_SUMMARY_CACHE_FILE, test_mode: bool = False ) -> 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. :param from_dt: The date from which to fetch In-Out-Summary. :param to_dt: The date till which to fetch In-Out-Summary. :param cache_file: The cache file to use to store the results. :param test_mode: If set to True, a past file will be used instead of getting new reports from Cosec. :return: True if the automated fetch was successful, else False. """ # Start by assuming failure: success = False report_path = None # If test mode: if test_mode: report_path = TEST_MODE_IN_OUT_SUMMARY_FILE_PATH # When not working in test mode: else: # 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: if to_dt is None: to_dt = date_time.get_current_ist_date_time() if from_dt is None: from_dt = to_dt - datetime.timedelta(days = 1) report_path = cosec.get_in_out_summary( initial_sleep = 1.0, from_date = from_dt, to_date = to_dt, group_ids = cosec_creds["inOutConfig"]["groupIds"], download_timeout = 60.0, timezone = cosec_creds["generalConfig"]["timezone"], ) # Log out to end the cycle: cosec.logout() # Close the browser window: cosec.quit() # Now process the report, # and save it to the JSON file: if report_path is not None: # Read the data: report_data = CosecWeb.read_in_out_summary_xls(report_path) # Assume that the punch time in the data is IST data, # then normalize it to UTC: def parse_dt(x): if pd.isnull(x): return None else: return date_time.to_timezone( datetime_object = date_time.as_if_timezone( datetime_object = date_time.parse_date_time(x), timezone = cosec_creds["generalConfig"]["timezone"] ), timezone = date_time.TIMEZONE_UTC ).timestamp() report_data["Punch Time"] = report_data["Punch Time"].apply(lambda x: parse_dt(x)) # Do the remaining cleanup and formatting: report_data = report_data.where(report_data.notna(), None) report_data = report_data.to_dict(orient = "records") report_data = { "ts": date_time.get_current_utc_date_time(as_string = False).timestamp(), "report": report_data } # Save the data to a JSON file: json.to_file( file = cache_file, python_data = report_data, no_space = True ) # Note down success: success = True # Done here: return success # --------------------------------------------------------------------------------------------------------------------- def compute_work_done( in_out_df: pd.DataFrame ) -> List[Dict[str, Union[str, int, float, None]]]: """ To calculate the full work done by all the employees on all the provided dates. :param in_out_df: The table that contains all the work done by all the employees on the date-range that was selected. :return: A list of dicts that contain the information of all the work done. """ # Create the structure that will be given as the output: flattened_work_reports = [] # Create an internal dict with the structure: # work_reports["user"]["date"] = {...} work_reports = defaultdict(lambda: defaultdict(dict)) # Loop through the full DataFrame once, # and figure out the first in and last out times: for index, row in in_out_df.iterrows(): # Ignore if the event wasn't a success: if row["Event Status"] != "Allowed": continue # Extract user and date: user_id = row["User ID"] punch_dt = date_time.parse_date_time(row["Punch Time"], timezone = date_time.TIMEZONE_IST) punch_date = punch_dt.strftime("%Y-%m-%d") punch_ts = punch_dt.timestamp() punch_loc = row[" Device/Source Detail"] # Handle the first in time: if row["I/O Type"] == "In": if work_reports[user_id][punch_date].get("first_in") is None: work_reports[user_id][punch_date]["first_in"] = punch_ts work_reports[user_id][punch_date]["first_in_loc"] = punch_loc # Handle the last out time: if row["I/O Type"] == "Out": if work_reports[user_id][punch_date].get("first_in") is not None: work_reports[user_id][punch_date]["last_out"] = punch_ts work_reports[user_id][punch_date]["last_out_loc"] = punch_loc # Now use the first_in and last out times of each record to figure out the amount of work done: for user_id, user_reports in work_reports.items(): for punch_date, punch_info in user_reports.items(): # Some defaults: min_work_seconds = 10.0 * 60.0 * 60.0 work_seconds = 0.0 work_ot_seconds = 0.0 work_status = "A" # Extract, clean and compute punch timing: first_in = punch_info.get("first_in") first_in_loc = punch_info.get("first_in_loc") last_out = punch_info.get("last_out") last_out_loc = punch_info.get("last_out_loc") # When the user has a valid in-time, but no known out time, # we assume that he worked a full day: if first_in is not None and not last_out: last_out = first_in + min_work_seconds work_seconds = last_out - first_in work_ot_seconds = 0.0 # When the user has neither an in-time, nor an out-time, # we assume that he was absent the whole day: elif not first_in and not last_out: work_seconds = 0.0 work_ot_seconds = 0.0 # When we have both - an in-time and an out-time - we compute the work done. # Any work over 10 hours will be counted as overtime: elif first_in and last_out: work_seconds = last_out - first_in work_ot_seconds = max(0.0, work_seconds - min_work_seconds) # Finally, compute the work status. # 'A' --> Absent # 'H' --> Half Day # 'P' --> Present (Full Day) # 'OT' -> Overtime work_hours = work_seconds / (60 * 60) if work_hours > 10.0: work_status = "OT" elif 7.5 < work_hours <= 10.0: work_status = "P" elif 2.5 < work_hours <= 5.0: work_status = "H" else: work_status = "A" # Save the data: flattened_work_reports.append({ "user_id": user_id, "work_date": punch_date, "first_in": first_in, "first_in_loc": first_in_loc, "last_out": last_out, "last_out_loc": last_out_loc, "work_seconds": work_seconds, "work_hours": work_hours, "work_ot_seconds": work_ot_seconds, "work_ot_hours": work_ot_seconds / (60.0 * 60.0), "work_status": work_status, }) # Done here: return flattened_work_reports # --------------------------------------------------------------------------------------------------------------------- async def sync_attendance_to_tcaoff( tcaoff_client: AsyncTheCAOffice, cosec_in_out_summary: dict, verbose: bool = False ) -> Dict[str, int]: """ :param tcaoff_client: The asynchronous client object that interfaces with TCAOFF. :param cosec_in_out_summary: he table that contains all the work done by all the employees on the date-range that was selected. :param verbose: If True, internal debugging print will be more verbose. :return: The dict that gives you the count of the successful and failed attendance marking API calls. """ # Results: results = defaultdict(int) # Convert the In/Out data to a DataFrame: in_out_df = pd.DataFrame(cosec_in_out_summary["report"]) # Compute the work done: work_reports = compute_work_done(in_out_df) # Get the list of existing team members from TCAOFF: # NOTE: `pseudonym` is the unique username of the user. tcaoff_teams = await tcaoff_client.team_list() # Get a mapping from Cosec id to TCAOFF record: cosec_id_to_tcaoff_team = {} for t in tcaoff_teams: app_notes = json.from_string(t["json_notes"]).get("applicantNotes", {}) if isinstance(app_notes, str): app_notes = json.from_string(app_notes) if not app_notes: continue cosec_notes = app_notes.get("cosec", {}) if not cosec_notes: continue # print("Cosec Notes:", cosec_notes) cosec_id_to_tcaoff_team[cosec_notes.get("User ID") or cosec_notes.get("UserID")] = t # Create the tasks to fire: tasks = [] for wr in work_reports: # Extract basic details: cosec_user_id = wr["user_id"] # Match it to the TCAOFF team-member: tcaoff_team = cosec_id_to_tcaoff_team.get(cosec_user_id) if tcaoff_team is None: printer(f"TCAOFF SYNC ERR: Cosesc User Id '{cosec_user_id}' not found in TCAOFF") continue # Create the task for this attendance: tasks.append( tcaoff_client.attendance_mark( user_id = tcaoff_team["user_id"], status = wr["work_status"], over_time = wr["work_ot_hours"], attendance_date = wr["work_date"], json_notes = { "totHours": wr["work_hours"], "firstIn": wr["first_in"], "lastOut": wr["last_out"], } ) ) printer("TASKS COUNT", len(tasks)) def chunks(lst, size = 10): for i in range(0, len(lst), size): yield lst[i:i + size] count = 0 for chunk in chunks(tasks[:30]): start_time = time.time() count += 1 printer("Task Chunk:", count) _res = await asyncio.gather(*chunk) success = sum(_res) failure = len(_res) - success results["success"] += success results["failure"] += failure results["total"] += len(_res) print("BATCH TIME TAKEN:", time.time() - start_time) print("\n\n---\n\n") # Done here: printer(results) return results # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass