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

(20260217) Ready to test on prodcuction now

Browse Source
This commit is contained in:
Khushal P Soonderji
2026-02-17 14:53:09 +05:30
parent df92fb08e9
commit 64a6eac20f
5 changed files with 102 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/common.py
+46 -32
View File
@@ -267,10 +267,12 @@ async def sync_branches_to_tcaoff(
tcaoff_branches = await tcaoff_client.branch_list()
tcaoff_branches = [_["branch_name"].lower() for _ in tcaoff_branches]
tcaoff_branches = list(set(tcaoff_branches))
print("TCAOFF BRANCHES: ", tcaoff_branches)
# Get only the branch names from Cosec:
cosec_branches = [tcaoff_client.remove_special_chars(_["Branch Name"]) for _ in cosec_muster_roll["report"]]
cosec_branches = list(set(cosec_branches))
print("COSEC BRANCHES: ", cosec_branches)
# Loop through the data from Cosec and add missing branches to TCAOFF:
for cosec_branch in cosec_branches:
@@ -338,61 +340,75 @@ async def sync_teams_to_tcaoff(
# Get the list of existing team members from TCAOFF:
# NOTE: `pseudonym` is the unique username of the user.
tcaoff_teams = await tcaoff_client.team_list()
synced_team_ids = []
for t in tcaoff_teams:
app_notes = json.from_string(t["json_notes"]).get("applicantNotes", {})
if isinstance(app_notes, str): app_notes = json.from_string(app_notes)
if app_notes is None:
app_notes = {}
cosec_notes = app_notes.get("cosec", {})
print("COSEC NOTES:", cosec_notes)
if cosec_notes:
user_id = cosec_notes.get("User ID")
if not user_id: user_id = cosec_notes["UserID"]
synced_team_ids.append(user_id)
# # OLD WAY - Based on human-readable user names:
# # Identify which users have not already been sync'd:
# synced_team_ids = []
# for t in tcaoff_teams:
# app_notes = json.from_string(t["json_notes"]).get("applicantNotes", {})
# if isinstance(app_notes, str): app_notes = json.from_string(app_notes)
# if app_notes is None:
# app_notes = {}
# cosec_notes = app_notes.get("cosec", {})
# # print("COSEC NOTES:", cosec_notes)
# if cosec_notes:
# user_id = cosec_notes.get("User ID")
# if not user_id: user_id = cosec_notes["UserID"]
# synced_team_ids.append(user_id)
# tcaoff_teams = [_["pseudonym"] for _ in tcaoff_teams]
# tcaoff_teams = list(set(tcaoff_teams))
# NEW WAY - Based on unique Employee Ids:
# Identify which users have not already been sync'd:
synced_team_ids = [_["pseudonym"] for _ in tcaoff_teams]
tcaoff_teams = [_["pseudonym"] for _ in tcaoff_teams]
tcaoff_teams = list(set(tcaoff_teams))
# Loop through the data from Cosec and add missing team-members to TCAOFF:
for cosec_team in cosec_muster_roll["report"]:
if cosec_team["User ID"] not in synced_team_ids:
# Extract Cosec Details:
cosec_user_id = cosec_team["User ID"].strip()
cosec_user_name = cosec_team["User Name"].strip()
cosec_branch_name = tcaoff_client.remove_special_chars(cosec_team["Branch Name"]).lower()
cosec_dept_name = tcaoff_client.remove_special_chars(cosec_team["Department Name"]).lower()
cosec_grade_name = (cosec_team.get("Grade Name") or "Unknown").strip()
if cosec_user_id not in synced_team_ids:
# Count the user:
response["total"] += 1
# Remove special chars:
branch_name = tcaoff_client.remove_special_chars(cosec_team["Branch Name"]).lower()
dept_name = tcaoff_client.remove_special_chars(cosec_team["Branch Name"]).lower()
# Check the branch id and department id:
branch_id = tcaoff_branches_lookup.get(branch_name)
dept_id = tcaoff_depts_lookup.get(dept_name)
branch_id = tcaoff_branches_lookup.get(cosec_branch_name)
dept_id = tcaoff_depts_lookup.get(cosec_dept_name)
if branch_id is None:
err_printer("TEAM SYNC ERR", "Branch Not Found in TCAOFF", branch_name)
err_printer("TEAM SYNC ERR", "Branch Not Found in TCAOFF", cosec_branch_name)
response["fail"] += 1
continue
if dept_id is None:
err_printer("TEAM SYNC ERR", "Dept. Not Found in TCAOFF", dept_name)
err_printer("TEAM SYNC ERR", "Dept. Not Found in TCAOFF", cosec_dept_name)
response["fail"] += 1
continue
# Add the team:
tcaoff_username = regex.replace(
text = cosec_team["User Name"],
text = cosec_user_name,
pattern = r"[^\w\d\._]",
substitute_text = ""
).strip().lower()
tcaoff_email_id = tcaoff_username + "@velankanigroup.com"
# print("Need to Add:", json.to_string(team_json))
success = await tcaoff_client.team_add(
branch_id = branch_id,
dept_id = dept_id,
reporting_to = None,
team_name = cosec_team["User Name"].strip(),
email = tcaoff_username + "@velankanigroup.com",
team_name = cosec_user_name,
email = tcaoff_email_id,
phone_no = "9876543210",
role = (cosec_team.get("Grade Name") or "Unknown").strip(),
username = tcaoff_username,
role = cosec_grade_name,
username = cosec_user_id,
password = "Vispl@123",
applicant_notes = {"cosec": cosec_team}
)
@@ -402,7 +418,7 @@ async def sync_teams_to_tcaoff(
response["fail"] += 1
# Done here:
printer("TCAOFF-Cosec Depts. Sync.:", json.to_string(response))
printer("TCAOFF-Cosec Team Sync.:", json.to_string(response))
return response
@@ -668,6 +684,7 @@ async def sync_attendance_to_tcaoff(
if not cosec_notes: continue
# print("Cosec Notes:", cosec_notes)
cosec_id_to_tcaoff_team[cosec_notes.get("User ID") or cosec_notes.get("UserID")] = t
cosec_id_to_tcaoff_team = {k:v for k, v in cosec_id_to_tcaoff_team.items() if k == v["pseudonym"]}
# Create the tasks to fire:
tasks = []
@@ -679,7 +696,7 @@ async def sync_attendance_to_tcaoff(
# Match it to the TCAOFF team-member:
tcaoff_team = cosec_id_to_tcaoff_team.get(cosec_user_id)
if tcaoff_team is None:
printer(f"TCAOFF SYNC ERR: Cosesc User Id '{cosec_user_id}' not found in TCAOFF")
printer(f"TCAOFF SYNC ERR: Cosesc User Id not found in TCAOFF", cosec_user_id)
continue
# Create the task for this attendance:
@@ -704,8 +721,7 @@ async def sync_attendance_to_tcaoff(
yield lst[i:i + size]
count = 0
for chunk in chunks(tasks[:30]):
start_time = time.time()
for chunk in chunks(tasks[:], size = 100):
count += 1
printer("Task Chunk:", count)
_res = await asyncio.gather(*chunk)
@@ -714,8 +730,6 @@ async def sync_attendance_to_tcaoff(
results["success"] += success
results["failure"] += failure
results["total"] += len(_res)
print("BATCH TIME TAKEN:", time.time() - start_time)
print("\n\n---\n\n")
# Done here:
printer(results)