""" 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_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 # 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]]]: # 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(): # 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() # 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 # 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 # 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") last_out = punch_info.get("last_out") # 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, "last_out": last_out, "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]: # 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: print(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"], } ) ) print("TASKS COUNT:", len(tasks)) def chunks(lst, size = 10): for i in range(0, len(lst), size): yield lst[i:i + size] results = [] count = 1 for chunk in chunks(tasks[-35:]): print("Task Chunk:", count) _res = await asyncio.gather(*chunk) results += _res count += 1 print(f"RESULTS ({len(results)}):", results) print("SUCCESS:", sum(results)) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass