345 lines
12 KiB
Python
345 lines
12 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:
|
|
|
|
"""
|
|
Gives you a list of TCAOFF team members whose Cosec details are missing or malformed.
|
|
This does NOT enlist missing names. The focus of this function is only fixing.
|
|
:param tcaoff_team_list: The list of team members from TCAOFF's '/team/list' API.
|
|
:param cosec_team_list: The list of team members from Cosec's Muster Roll report.
|
|
:return: The list of TCAOFF team members with read-to-use correct data.
|
|
"""
|
|
|
|
# Prepare the variables:
|
|
teams_to_fix = []
|
|
cosec_id_to_cosec_team_map = {ct["User ID"]: ct for ct in cosec_team_list}
|
|
|
|
for tt in tcaoff_team_list:
|
|
|
|
# Extract Cosec notes:
|
|
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.
|
|
# Fixes will be required when:
|
|
# 1. The team member is present in TCAOFF,
|
|
# 2. the team member either has no cosec notes, or the notes are malformed.
|
|
user_id = cosec_notes.get("User ID", cosec_notes.get("UserID"))
|
|
if not user_id:
|
|
if (valid_cose_notes := cosec_id_to_cosec_team_map.get(tt["pseudonym"])) is not None:
|
|
applicant_notes["cosec"] = valid_cose_notes
|
|
tt.update({"applicantNotes": applicant_notes})
|
|
teams_to_fix.append(tt)
|
|
|
|
# Done here:
|
|
return teams_to_fix
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
# 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:
|
|
|
|
"""
|
|
To manually sync. teams between Cosec (source) and TCAOFF (dest).
|
|
:param cosec_creds: The credentials to log into Cosec.
|
|
:param tcaoff_client: The credentials to log into TCAOFF.
|
|
:param on_date: The date for which the teams need to be synchronized. That day's Cosec Muster Roll will be fetched
|
|
and those records will be sync'd.
|
|
:param test_mode: Enable this during development or local testing.
|
|
:return: 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:
|
|
# --- Data Fetch:
|
|
printer("MUSTER ROLL: Sync'ing with TCAOFF")
|
|
cosec_teams = json.from_file(common.get_manual_muster_roll_cache_file_path())["report"]
|
|
tcaoff_teams = await tcaoff_client.team_list()
|
|
# --- Team Fix:
|
|
teams_to_fix = enlist_teams_to_fix(tcaoff_teams, cosec_teams)
|
|
for t in teams_to_fix:
|
|
printer("Fixing", t["user_id"], t["full_name"])
|
|
await tcaoff_client.team_update(
|
|
user_id = t["user_id"],
|
|
dept_id = t["department_id"],
|
|
team_name = t["full_name"],
|
|
email = t["email"],
|
|
phone_no = t["phone_number"],
|
|
role = t["role"],
|
|
applicant_notes = t["applicantNotes"],
|
|
)
|
|
|
|
|
|
# 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,
|
|
)
|
|
)
|