318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""
|
|
|
|
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,
|
|
)
|
|
)
|