""" AUTHOR: Khushal P Soonderji DATE: CREATED: Thu, 5th Feb, 2026 UPDATED: Thu, 5th Feb, 2026 OBJECTIVE: To run report fetching and sync'ing in periodic intervals. 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 tcaoff.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 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 # Common and Shared: from scripts import common # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Debugging: printer = IceCreamDebugger(prefix = "Cron | ", includeContext = True) err_printer = IceCreamDebugger(prefix = "[ERR] Cron | ", includeContext = True) # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** async def yesterday_cron( cosec_creds: dict, tcaoff_client: AsyncTheCAOffice, test_mode: bool = False, ) -> None: printer("YESTERDAY CRON") # Try the whole process once: try: # Log in to TCAOFF: success = await tcaoff_client.login() if not success: raise RuntimeError("TCAOFF Login Failed!") # Try the part that needs COSEC: try: # Get the previous day's In/Out Summary and then wait # for the driver's resources to get freed: success = common.get_in_out_summary( cosec_creds = cosec_creds, from_dt = date_time.get_current_ist_date_time() - datetime.timedelta(days = 3), to_dt = date_time.get_current_ist_date_time(), cache_file = common.PREV_DAY_IN_OUT_SUMMARY_CACHE_FILE, test_mode = test_mode ) # Sync data between Cosec and TCAOFF: if success: printer("IN/OUT SUMMARY: Sync'ing with TCAOFF") await common.sync_attendance_to_tcaoff( tcaoff_client = tcaoff_client, cosec_in_out_summary = json.from_file(common.PREV_DAY_IN_OUT_SUMMARY_CACHE_FILE), chunk_size = 10 ) # If something goes wrong in the COSEC step: except Exception as exception: err_printer(exception) if test_mode: raise # Log out of TCAOFF: success = await tcaoff_client.logout() if not success: raise RuntimeError("TCAOFF Login Failed!") # If something goes wrong: except Exception as exception: err_printer(exception) await tcaoff_client.logout() if test_mode: raise # --------------------------------------------------------------------------------------------------------------------- async def today_cron( cosec_creds: dict, tcaoff_client: AsyncTheCAOffice, test_mode: bool = False, ) -> None: """ To get today's stats about the team. :param cosec_creds: The credentials to use to log into Matrix COSEC. :param tcaoff_client: The client to interact with TCAOFF. :param test_mode: If True, the execution will be done in a controlled (but live) test setup. :return: None. """ printer("TODAY CRON") # Try the whole process once: try: # Log in to TCAOFF: success = await tcaoff_client.login() if not success: raise RuntimeError("TCAOFF Login Failed!") # Try the part that needs COSEC: try: # MUSTER-ROLL: # Get the Muster Roll and then wait # for the driver's resources to get freed: success = common.get_muster_roll( cosec_creds = cosec_creds, on_date = date_time.get_current_ist_date_time().replace( hour = 0, minute = 0, second = 0 ), cache_file = common.MUSTER_ROLL_CACHE_FILE, test_mode = test_mode ) # Sync data between Cosec and TCAOFF: if success: print("MUSTER ROLL: Sync'ing with TCAOFF") cosec_muster_roll = json.from_file(common.MUSTER_ROLL_CACHE_FILE) await common.sync_branches_to_tcaoff( tcaoff_client = tcaoff_client, cosec_muster_roll = cosec_muster_roll, ) await common.sync_departments_to_tcaoff( tcaoff_client = tcaoff_client, cosec_muster_roll = cosec_muster_roll, ) await common.sync_teams_to_tcaoff( tcaoff_client = tcaoff_client, cosec_muster_roll = cosec_muster_roll, ) # IN-OUT SUMMARY: # Get the previous day's In/Out Summary and then wait # for the driver's resources to get freed: success = common.get_in_out_summary( cosec_creds = cosec_creds, from_dt = date_time.get_current_ist_date_time().replace(hour = 0, minute = 0, second = 0, microsecond = 0), to_dt = date_time.get_current_ist_date_time(), cache_file = common.IN_OUT_SUMMARY_CACHE_FILE, test_mode = test_mode ) # Sync data between Cosec and TCAOFF: if success: printer("IN/OUT SUMMARY: Sync'ing with TCAOFF") await common.sync_attendance_to_tcaoff( tcaoff_client = tcaoff_client, cosec_in_out_summary = json.from_file(common.IN_OUT_SUMMARY_CACHE_FILE), chunk_size = 10 ) # If something goes wrong in the COSEC step: except Exception as exception: err_printer(exception) if test_mode: raise # If something goes wrong: except Exception as exception: err_printer(exception) await tcaoff_client.logout() if test_mode: raise # --------------------------------------------------------------------------------------------------------------------- async def set_scheduler( cosec_creds: dict, tcaoff_client: AsyncTheCAOffice, test_mode: bool = False ) -> None: """ Set up the scheduler (cron) that will time the activities. The background task happens at fixed times, the foreground task happens in a loop. :param cosec_creds: The credentials to use to log into Matrix COSEC. :param tcaoff_client: The client to interact with TCAOFF. :param test_mode: If True, the scheduler will be ignored and the process will be run once immediately. :return: None. """ # If in test mode: if test_mode: printer("Starting Test") await yesterday_cron( cosec_creds = copy.deepcopy(cosec_creds), tcaoff_client = tcaoff_client, # test_mode = test_mode ) await today_cron( cosec_creds = copy.deepcopy(cosec_creds), tcaoff_client = tcaoff_client, # test_mode = test_mode ) printer("Test Done") return # If not in test mode, we continue with the scheduler. # Create the scheduler: scheduler = AsyncIOScheduler() # Populate the tasks in the scheduler: for h, m in zip( [8, 23], [30, 30] ): scheduler.add_job( yesterday_cron, "cron", hour = h, minute = m, args = [ copy.deepcopy(cosec_creds), tcaoff_client ] ) # Populate the tasks in the scheduler: for h in range(0, 23): for m in [0]: scheduler.add_job( today_cron, "cron", hour = h, minute = m, args = [ copy.deepcopy(cosec_creds), tcaoff_client ] ) # Run the scheduler: scheduler.start() while True: now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") printer("Heartbeat", now) await asyncio.sleep(60.0) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": # Parse args: import argparse ap = argparse.ArgumentParser() ap.add_argument( "--test", action = "store_true", default = False, ) ap.add_argument( "--verbose", action = "store_true", default = False, ) args = ap.parse_args() # Explicitly mention the expected file paths for other devs to maintain: print("PROJ. DIR. :", common.PROJ_DIR) print("COSEC CREDS :", common.COSEC_CREDS_FILE) print("TCAOFF CREDS:", common.TCAOFF_CREDS_FILE) # Read required credentials: cosec_creds = json.from_file(common.COSEC_CREDS_FILE) tcaoff_creds = json.from_file(common.TCAOFF_CREDS_FILE) # Create the clients: tcaoff_client = AsyncTheCAOffice( username = tcaoff_creds["creds"]["username"], password = tcaoff_creds["creds"]["password"], debug = True, debug_only_errors = False if args.verbose else True, ) # Schedule the activities: asyncio.run(set_scheduler( cosec_creds = cosec_creds, tcaoff_client = tcaoff_client, test_mode = args.test ))