khushalps/cosec_automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20260109) Almost done with Cosec-TCAOFF sync.

Browse Source
This commit is contained in:
Khushal P Soonderji
2026-01-09 16:35:49 +05:30
parent d41bf34101
commit 5bec648313
11 changed files with 1809 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cron/reports.py
+377 -49
View File
@@ -33,9 +33,6 @@
# To make sibling directories accessible for imports:
import sys
import pandas as pd
sys.path.append(".")
sys.path.append("..")
@@ -46,14 +43,28 @@ import os
import time
import datetime
# To work with tabulate data:
import pandas as pd
# To make API calls:
import requests
# Cosec-related:
from cosec_web.cosec_web import CosecWeb
# TCAOFF-related:
from helpers import tcaoff
# 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
# *****************************************************************************************************************
# ***** ****
@@ -66,6 +77,7 @@ from utils_v2.date_time import date_time
file_dir = files.get_file_directory(include_filename = False)
proj_dir = files.get_parent_directory(file_dir, depth = 1)
cosec_creds_file = os.path.join(proj_dir, "creds", "cosec.json")
tcaoff_creds_file = os.path.join(proj_dir, "creds", "tcaoff.json")
muster_roll_cache_file = os.path.join(proj_dir, "local", "cache", "muster_roll_cache.json")
in_out_summary_cache_file = os.path.join(proj_dir, "local", "cache", "in_out_summary_cache.json")
chrome_driver_dir = os.path.join(proj_dir, r"drivers","chrome")
@@ -100,6 +112,182 @@ downloads_dir = os.path.join(proj_dir, r"downloads")
# *****************************************************************************************************************
# def remove_special_chars(s: str) -> str:
#
# return regex.replace(
# text = s,
# pattern = r"[^\w\d\- _]",
# substitute_text = "_"
# )
# ---------------------------------------------------------------------------------------------------------------------
def add_branches_to_tcaoff(
tcaoff_creds: dict,
cosec_muster_roll: dict,
) -> Dict[str, int]:
# Start with a basic response structure:
response = defaultdict(int)
# Get the list of existing branches from TCAOFF:
tcaoff_branches = tcaoff.branch_list(tcaoff_creds)
tcaoff_branches = [_["branch_name"] for _ in tcaoff_branches]
tcaoff_branches = list(set(tcaoff_branches))
# Get only the branch names from Cosec:
cosec_branches = [tcaoff.remove_special_chars(_["Branch Name"]) for _ in cosec_muster_roll["report"]]
cosec_branches = list(set(cosec_branches))
# Loop through the data from Cosec and add missing branches to TCAOFF:
for cosec_branch in cosec_branches:
if cosec_branch not in tcaoff_branches:
success = tcaoff.branch_add(
tcaoff_creds,
branch_name = cosec_branch
)
response["total"] += 1
if success: response["success"] += 1
else: response["fail"] += 1
# Done here:
print("TCAOFF-Cosec Branches Sync.:", json.to_string(response))
return response
# ---------------------------------------------------------------------------------------------------------------------
def add_departments_to_tcaoff(
tcaoff_creds: dict,
cosec_muster_roll: dict,
) -> Dict[str, int]:
# Start with a basic response structure:
response = defaultdict(int)
# Get the list of existing departments from TCAOFF:
tcaoff_depts = tcaoff.department_list(tcaoff_creds)
tcaoff_depts = [_["department_name"] for _ in tcaoff_depts]
tcaoff_depts = list(set(tcaoff_depts))
# print("TCAOFF DEPTS:", tcaoff_depts)
# Get only the department names from Cosec:
cosec_depts = [tcaoff.remove_special_chars(_["Department Name"]) for _ in cosec_muster_roll["report"]]
cosec_depts = list(set(cosec_depts))
# print("COSEC DEPTS:", cosec_depts)
# Loop through the data from Cosec and add missing departments to TCAOFF:
for cosec_dept in cosec_depts:
if cosec_dept not in tcaoff_depts:
success = tcaoff.department_add(
tcaoff_creds,
department_name = cosec_dept
)
response["total"] += 1
if success: response["success"] += 1
else: response["fail"] += 1
# Done here:
print("TCAOFF-Cosec Depts. Sync.:", json.to_string(response))
return response
# ---------------------------------------------------------------------------------------------------------------------
def add_teams_to_tcaoff(
tcaoff_creds: dict,
cosec_muster_roll: dict,
) -> Dict[str, int]:
# Start with a basic response structure:
response = defaultdict(int)
# Map out branch ids:
tcaoff_branches = tcaoff.branch_list(tcaoff_creds)
tcaoff_branches_lookup = {d["branch_name"]: d["branch_id"] for d in tcaoff_branches}
# Map out dept. ids:
tcaoff_depts = tcaoff.department_list(tcaoff_creds)
tcaoff_depts_lookup = {d["department_name"]:d["department_id"] for d in tcaoff_depts}
# Get the list of existing departments from TCAOFF:
# NOTE: `pseudonym` is the unique username of the user.
tcaoff_teams = tcaoff.team_list(tcaoff_creds)
tcaoff_teams = [_["pseudonym"] for _ in tcaoff_teams]
tcaoff_teams = list(set(tcaoff_teams))
# # Get only the department names from Cosec:
# cosec_depts = [remove_special_chars(_["Department Name"]) for _ in cosec_muster_roll["report"]]
# cosec_depts = list(set(cosec_depts))
# # print("COSEC DEPTS:", cosec_depts)
# Loop through the data from Cosec and add missing departments to TCAOFF:
for cosec_team in cosec_muster_roll["report"]:
if cosec_team["User ID"] not in tcaoff_teams:
# Check the branch id and department id:
branch_id = tcaoff_branches_lookup.get(tcaoff.remove_special_chars(cosec_team["Branch Name"]))
dept_id = tcaoff_depts_lookup[tcaoff.remove_special_chars(cosec_team["Department Name"])]
if branch_id is None:
print(f"TEAM SYNC ERR: Branch '{tcaoff.remove_special_chars(cosec_team['Branch Name'])}' not found in TCAOFF")
response["fail"] += 1
continue
if dept_id is None:
print(f"TEAM SYNC ERR: Dept. '{tcaoff.remove_special_chars(cosec_team['Department Name'])}' not found in TCAOFF")
response["fail"] += 1
continue
# Add the team:
team_json = {
"branchId": branch_id,
"idDepartment": dept_id,
"reportingTo": None,
"name": cosec_team['User Name'].strip(),
"email": regex.replace(
text = cosec_team['User Name'],
pattern = r"[^\w\d\._]",
substitute_text = ""
).strip().lower() + "@velankanigroup.com",
"phoneNo": "9876543210",
"role": cosec_team['Category Name'].strip(),
"username": cosec_team['User ID'].strip(),
"password": "Vispl@123",
"hierarchy": 1,
}
print("Need to Add:", json.to_string(team_json))
success = tcaoff.team_add(
tcaoff_creds,
branch_id = branch_id,
dept_id = dept_id,
reporting_to = None,
team_name = cosec_team["User Name"].strip(),
email = regex.replace(
text = cosec_team['User Name'],
pattern = r"[^\w\d\._]",
substitute_text = ""
).strip().lower() + "@velankanigroup.com",
phone_no = "9876543210",
role = cosec_team["Category Name"].strip(),
username = cosec_team["User ID"].strip(),
password = "Vispl@123"
)
response["total"] += 1
if success: response["success"] += 1
else: response["fail"] += 1
break
# Done here:
print("TCAOFF-Cosec Depts. Sync.:", json.to_string(response))
return response
# ---------------------------------------------------------------------------------------------------------------------
def kill_chrome() -> None:
# First kill the previous processes,
@@ -111,37 +299,48 @@ def kill_chrome() -> None:
# ---------------------------------------------------------------------------------------------------------------------
def get_muster_roll(cosec_creds: dict):
def get_muster_roll(cosec_creds: dict) -> bool:
# Force close other running Chrome processes:
kill_chrome()
"""
Get the latest Muster-Roll from Matrix Cosec. It saves the data into a local cache file.
:param cosec_creds: The credentials (and config) to operate Cosec Matrix.
:return: True if the automated fetch was successful, else False.
"""
# 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,
)
# Start by assuming failure:
success = False
# Perform the login:
cosec.login(initial_sleep = 2.5)
# # 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:
# report_path = cosec.get_muster_roll(
# initial_sleep = 1.0,
# on_date = date_time.get_current_utc_date_time(),
# group_ids = cosec_creds["musterRollConfig"]["groupIds"],
# download_timeout = 60.0
# )
#
# # Log out to end the cycle:
# cosec.logout()
#
# # Close the browser window:
# cosec.quit()
# Get the in/out report:
report_path = cosec.get_muster_roll(
initial_sleep = 1.0,
on_date = date_time.get_current_utc_date_time(),
group_ids = cosec_creds["musterRollConfig"]["groupIds"],
download_timeout = 60.0
)
# Log out to end the cycle:
cosec.logout()
# Close the browser window:
cosec.quit()
report_path = r"D:\kps\PycharmProjects\cosec\downloads\Monthly_Details.xls"
# Now process the report,
# and save it to the JSON file:
@@ -159,6 +358,22 @@ def get_muster_roll(cosec_creds: dict):
report_data = report_data.where(report_data.notna(), None)
report_data = report_data.to_dict(orient = "records")
# Loop through the report for inferred data.
# Add the ids of the "Direct Reporting" and "Level-1" values:
for person in report_data:
person["Direct Reporting ID"] = None
person["Level-1 ID"] = None
for check_against in report_data:
if person["Direct Reporting"] == check_against["User Name"]:
person["Direct Reporting ID"] = check_against["User ID"]
if person["Level-1"] == check_against["User Name"]:
person["Level-1 ID"] = check_against["User ID"]
# Sort by hierarchy (BFS):
pass
print(f"MUSTER ROLL ({len(report_data)}):", json.to_string(report_data))
# Save the data to a JSON file:
json.to_file(
file = muster_roll_cache_file,
@@ -169,11 +384,26 @@ def get_muster_roll(cosec_creds: dict):
no_space = True
)
# Note down success:
success = True
# Done here:
return success
# ---------------------------------------------------------------------------------------------------------------------
def get_in_out_summary(cosec_creds: dict):
def get_in_out_summary(cosec_creds: dict) -> 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.
:return: True if the automated fetch was successful, else False.
"""
# Start by assuming failure:
success = False
# Force close other running Chrome processes:
kill_chrome()
@@ -250,28 +480,97 @@ def get_in_out_summary(cosec_creds: dict):
no_space = True
)
# Note down success:
success = True
# Done here:
return success
# ---------------------------------------------------------------------------------------------------------------------
def get_reports(cosec_creds: dict) -> None:
def get_reports(
cosec_creds: dict,
tcaoff_creds: dict,
) -> None:
# Get the Muster Roll and then wait
# for the driver's resources to get freed:
# try:
# get_muster_roll(cosec_creds = cosec_creds)
# time.sleep(2.5)
# except Exception as e:
# print("MUSTER ROLL FETCH FAILED!")
# Get the In-Out Summary and then wait
# for the driver's resources to get freed:
# Try the whole process once:
try:
get_in_out_summary(cosec_creds = cosec_creds)
time.sleep(2.5)
# Log in to TCAOFF:
tcaoff.login(tcaoff_creds)
# # # TEST SECTION:
# # # Test out listing API calls:
print("TCAOFF Branches:", json.to_string(tcaoff.branch_list(tcaoff_creds)))
print("TCAOFF Depts.:", json.to_string(tcaoff.department_list(tcaoff_creds)))
# print("TCAOFF Team:", json.to_string(tcaoff.team_list(tcaoff_creds)))
# # TEST SECTION:
# # Test out listing API calls:
# tcaoff.branch_add(tcaoff_creds, branch_name = "Test - 20260108")
# tcaoff.department_add(tcaoff_creds, department_name = "Test - 20260108")
# tcaoff.team_add(
# tcaoff_creds,
# branch_id = 124,
# dept_id = 851,
# reporting_to = 2857,
# team_name = "Test - 20260108",
# email = "user@domain.com",
# phone_no = "9876543219",
# role = "Python Dev",
# username = "test.user.20260108",
# password = "12345678",
# )
# MUSTER-ROLL REPORT:
try:
# Get the Muster Roll and then wait
# for the driver's resources to get freed:
success = get_muster_roll(cosec_creds = cosec_creds)
# Sync data between Cosec and TCAOFF:
if success:
print("MUSTER ROLL: Sync'ing with TCAOFF")
cosec_muster_roll = json.from_file(muster_roll_cache_file)
# add_branches_to_tcaoff(
# tcaoff_creds = tcaoff_creds,
# cosec_muster_roll = cosec_muster_roll,
# )
# add_departments_to_tcaoff(
# tcaoff_creds = tcaoff_creds,
# cosec_muster_roll = cosec_muster_roll,
# )
add_teams_to_tcaoff(
tcaoff_creds = tcaoff_creds,
cosec_muster_roll = cosec_muster_roll,
)
# If something goes wrong:
except Exception as e:
print("MUSTER ROLL FETCH FAILED!")
raise
# # Get the In-Out Summary and then wait
# # for the driver's resources to get freed:
# try:
# success = get_in_out_summary(cosec_creds = cosec_creds)
# time.sleep(2.5)
# if success: pass
# except Exception as e:
# print("IN-OUT SUMMARY FETCH FAILED!")
# raise
# Log out from TCAOFF:
tcaoff.logout(tcaoff_creds)
# If TCAOFF's login or logout fails:
except Exception as e:
print("IN-OUT SUMMARY FETCH FAILED!")
raise e
print("REPORT CRON FAILED LOOP!")
print("EXCEPTION:", e)
raise
# ---------------------------------------------------------------------------------------------------------------------
@@ -279,12 +578,16 @@ def get_reports(cosec_creds: dict) -> None:
def loop(
cosec_creds: dict,
tcaoff_creds: dict,
interval_seconds: int | float = 900
) -> None:
# Just keep fetching the reports in an infinite loop:
while True:
get_reports(cosec_creds = cosec_creds)
get_reports(
cosec_creds = cosec_creds,
tcaoff_creds = tcaoff_creds
)
time.sleep(interval_seconds)
@@ -303,7 +606,7 @@ if __name__ == "__main__":
print("M-ROLL CACHE:", muster_roll_cache_file)
print("IN-OUT CACHE:", in_out_summary_cache_file)
# Read the credentials.
# Read Cosec's credentials.
# It should be in the format:
"""
{
@@ -330,9 +633,34 @@ if __name__ == "__main__":
}
}
"""
cosec_creds = json.from_file(cosec_creds_file)
# Read TCAOFF's credentials.
# It should be in the format:
"""
{
"creds": {
"username": "<usr>",
"password": "<pwd>",
"sessionToken": null,
"user": null
},
"urls": {
"login": "https://api.thecaoffice.com/ca/login",
"logout": "https://api.thecaoffice.com/user/logout",
"branchList": "https://api.thecaoffice.com/commons/branch/list",
"branchAdd": "https://api.thecaoffice.com/commons/branch/add",
"deptList": "https://api.thecaoffice.com/commons/departments/list",
"deptAdd": "https://api.thecaoffice.com/commons/departments/add",
"teamList": "https://api.thecaoffice.com/team/list",
"teamAdd": "https://api.thecaoffice.com/team/add"
}
}
"""
tcaoff_creds = json.from_file(tcaoff_creds_file)
loop(
cosec_creds = cosec_creds,
tcaoff_creds = tcaoff_creds,
interval_seconds = cosec_creds["generalConfig"]["pollInterval"],
)