diff --git a/scripts/common.py b/scripts/common.py index f25cc0f..bb34075 100644 --- a/scripts/common.py +++ b/scripts/common.py @@ -138,6 +138,7 @@ def get_muster_roll_cache_file_path(): return os.path.join(CACHE_DIR, os.environ def get_in_out_summary_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "in_out_summary_cache.json") def get_prev_day_in_out_summary_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "prev_day_in_out_summary_cache.json") def get_manual_in_out_summary_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "manual_in_out_summary_cache.json") +def get_manual_muster_roll_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "manual_muster_roll_cache.json") def get_work_reports_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "work_reports_cache.json") # --- def get_browser_user_data_directory(): return os.path.join(PROJ_DIR, "browser", "user_data", os.environ.get("ORG", "default")) diff --git a/scripts/manual_attendance.py b/scripts/manual_attendance.py index 0b4aab4..80670f4 100644 --- a/scripts/manual_attendance.py +++ b/scripts/manual_attendance.py @@ -84,8 +84,8 @@ from scripts import common # Debugging: -printer = IceCreamDebugger(prefix = "Manual | ", includeContext = True) -err_printer = IceCreamDebugger(prefix = "[ERR] Manual | ", includeContext = True) +printer = IceCreamDebugger(prefix = "Manual Att | ", includeContext = True) +err_printer = IceCreamDebugger(prefix = "[ERR] Manual Att | ", includeContext = True) # ***************************************************************************************************************** diff --git a/scripts/manual_team_sync.py b/scripts/manual_team_sync.py new file mode 100644 index 0000000..abdc73d --- /dev/null +++ b/scripts/manual_team_sync.py @@ -0,0 +1,317 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + CREATED: Thu, 2nd Apr, 2026 + + OBJECTIVE: + + To fix team information issues. + + Cosec has some identification points for each team member. These are noted in TCAOFF in 'json_notes.cosec'. This + script is meant to go through the team members on TCAOFF, see whose cosec information is either missing or + malformed, and recreate it. + + 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 +from pathlib import Path + +# 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.string import fuzzy +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 = "Manual Team | ", includeContext = True) +err_printer = IceCreamDebugger(prefix = "[ERR] Manual Team | ", includeContext = True) + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +def enlist_teams_to_fix( + tcaoff_team_list: list, + cosec_team_list: list, +) -> list: + + # Prepare the variables: + teams_to_fix = [] + cosec_id_to_cosec_team_map = {ct["User ID"]: ct for ct in cosec_team_list} + print("COSEC_MAP:", json.to_string(cosec_id_to_cosec_team_map)) + + for tt in tcaoff_team_list: + + # Start by assuming everything is alright, + # then extract the notes: + need_to_sync = False + json_notes = json.from_string(tt.get("json_notes", "{}")) + applicant_notes = json_notes.get("applicantNotes") or {} + cosec_notes = applicant_notes.get("cosec") or {} + + # Now check if any fix is required: + user_id = cosec_notes.get("User ID", cosec_notes.get("UserID")) + if not user_id: + print("NEED TO FIX:", f"{tt['full_name']} ({tt['pseudonym']})") + + # if tt['pseudonym'] == "VIS07070": + # print("RAKESH:", json.to_string(tt)) + # print("RAKESH:", user_id) + # print("RAKESH:", json.to_string(cosec_notes)) + + + +# --------------------------------------------------------------------------------------------------------------------- + + +def enlist_missing_teams( + tcaoff_team_list: list, + cosec_team_list: list +) -> list: + + pass + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def manual_team_sync( + cosec_creds: dict, + tcaoff_client: AsyncTheCAOffice, + on_date: date_time.datetime = None, + test_mode: bool = False, +) -> None: + + printer("MANUAL TEAM SYNC") + + # 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: + + # Figure out the date: + now = date_time.get_current_ist_date_time() + if not on_date: on_date = now + + # # Get the previous day's In/Out Summary and then wait + # # for the driver's resources to get freed: + # success = common.get_muster_roll( + # cosec_creds = cosec_creds, + # on_date = on_date, + # cache_file = common.get_manual_muster_roll_cache_file_path, + # test_mode = test_mode + # ) + success = True + + # Sync data between Cosec and TCAOFF: + if success: + printer("MUSTER ROLL: Sync'ing with TCAOFF") + print("SYNC FROM:", common.get_manual_muster_roll_cache_file_path()) + cosec_teams = json.from_file(common.get_manual_muster_roll_cache_file_path())["report"] + # print("COSEC:", json.to_string(cosec_teams)) + tcaoff_teams = await tcaoff_client.team_list() + # print("TCAOFF:", json.to_string(tcaoff_teams)) + # --- + teams_to_fix = enlist_teams_to_fix(tcaoff_teams, cosec_teams) + + + # 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 Logout Failed!") + + # If something goes wrong: + except Exception as exception: + err_printer(exception) + await tcaoff_client.logout() + if test_mode: raise + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + # Parse args: + import argparse + ap = argparse.ArgumentParser() + ap.add_argument( + "--test", "--test-mode", + action = "store_true", + default = False + ) + ap.add_argument( + "--verbose", + action = "store_true", + default = False + ) + ap.add_argument( + "--from-date", + help = "The date from which you want to sync attendance.", + default = date_time.get_current_ist_date_time() + ) + ap.add_argument( + "--on-date", + help = "The date for which you want to sync. teams.", + default = date_time.get_current_ist_date_time() + ) + ap.add_argument( + "--cosec-creds", + help = "The credentials JSON from which you would like to connect to COSEC.", + default = "cosec.json" + ) + ap.add_argument( + "--tcaoff-creds", + help = "The credentials JSON from which you would like to connect to TheCAOffice.", + default = "tcaoff.json" + ) + ap.add_argument( + "--org", "--organization", + help = "When handling multiple organizations, this word will be used to keep their files separate.", + default = "default" + ) + args = ap.parse_args() + + # Hold temporary environment variables: + os.environ["ORG"] = args.org + + # Read required credentials: + cosec_creds_path = os.path.join(common.CREDS_DIR, args.cosec_creds) + tcaoff_creds_path = os.path.join(common.CREDS_DIR, args.tcaoff_creds) + # --- + cosec_creds = json.from_file(cosec_creds_path) + tcaoff_creds = json.from_file(tcaoff_creds_path) + + # Explicitly mention the expected file paths for other devs to maintain: + print("PROJ. DIR. :", common.PROJ_DIR) + print("COSEC CREDS :", cosec_creds_path) + print("TCAOFF CREDS:", tcaoff_creds_path) + + # Ensure that certain required paths exist: + common.init_paths() + + # Date-handling: + args.on_date = date_time.parse_date_time(args.on_date, timezone = date_time.TIMEZONE_IST) + args.on_date = args.on_date.replace(hour = 0, minute = 0, second = 0, microsecond = 0) + print("On Date:", args.on_date) + + # 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( + manual_team_sync( + cosec_creds = cosec_creds, + tcaoff_client = tcaoff_client, + on_date = args.on_date, + test_mode = args.test, + ) + )