749 lines
28 KiB
Python
749 lines
28 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
CREATED: Thu, 5th Feb, 2026
|
|
UPDATED: Thu, 5th Feb, 2026
|
|
|
|
OBJECTIVE:
|
|
|
|
There will be some common actions across various scripts. This script holds those common actions.
|
|
|
|
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
|
|
|
|
# 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 helpers.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.date_time import date_time
|
|
|
|
# To work with datatypes:
|
|
from typing import List, Dict, Any, Union
|
|
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
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# File paths:
|
|
FILE_DIR = files.get_file_directory(include_filename = False)
|
|
PROJ_DIR = files.get_parent_directory(FILE_DIR, depth = 1)
|
|
CACHE_DIR = os.path.join(PROJ_DIR, "local", "cache")
|
|
CREDS_DIR = os.path.join(PROJ_DIR, "creds")
|
|
# ---
|
|
COSEC_CREDS_FILE = os.path.join(CREDS_DIR, "cosec.json")
|
|
TCAOFF_CREDS_FILE = os.path.join(CREDS_DIR, "tcaoff.json")
|
|
MUSTER_ROLL_CACHE_FILE = os.path.join(CACHE_DIR, "muster_roll_cache.json")
|
|
IN_OUT_SUMMARY_CACHE_FILE = os.path.join(CACHE_DIR, "in_out_summary_cache.json")
|
|
PREV_DAY_IN_OUT_SUMMARY_CACHE_FILE = os.path.join(CACHE_DIR, "prev_day_in_out_summary_cache.json")
|
|
CHROME_DRIVER_DIR = os.path.join(PROJ_DIR, "drivers", "chrome")
|
|
USER_DATA_DIR = os.path.join(PROJ_DIR, "browser", "user_data")
|
|
DOWNLOADS_DIR = os.path.join(PROJ_DIR, "downloads")
|
|
# ---
|
|
TEST_MODE_MUSTER_ROLL_FILE_PATH = os.path.join(PROJ_DIR, "cosec_web", "sample_files", "muster_roll.xls")
|
|
TEST_MODE_IN_OUT_SUMMARY_FILE_PATH = os.path.join(PROJ_DIR, "cosec_web", "sample_files", "in_out_summary.xls")
|
|
|
|
# Debugging:
|
|
printer = IceCreamDebugger(prefix = "Common | ", includeContext = True)
|
|
err_printer = IceCreamDebugger(prefix = "[ERR] Common | ", includeContext = True)
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
def kill_chrome() -> None:
|
|
|
|
"""
|
|
To kill running Chrome processes so that they don't interfere with the one that will be spun up by automation code.
|
|
:return: None
|
|
"""
|
|
|
|
# First kill the previous processes,
|
|
# then wait if old processes were killed:
|
|
kill_count = CosecWeb.kill_chrome_processes()
|
|
if kill_count > 0: time.sleep(2.5)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def get_muster_roll(
|
|
cosec_creds: dict,
|
|
on_date: datetime.datetime,
|
|
cache_file: str = MUSTER_ROLL_CACHE_FILE,
|
|
test_mode: bool = False
|
|
) -> bool:
|
|
|
|
"""
|
|
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.
|
|
:param on_date: The date on which you want .
|
|
:param cache_file: The cache file to store the results in.
|
|
:param test_mode: If set to True, a past file will be used instead of getting new reports from Cosec.
|
|
:return: True if the automated fetch was successful, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
report_path = None
|
|
|
|
# If test mode:
|
|
if test_mode:
|
|
report_path = TEST_MODE_MUSTER_ROLL_FILE_PATH
|
|
|
|
# When not working in test mode:
|
|
else:
|
|
|
|
# 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 = on_date,
|
|
group_ids = cosec_creds["musterRollConfig"]["groupIds"],
|
|
download_timeout = 60.0
|
|
)
|
|
|
|
# Log out to end the cycle:
|
|
cosec.logout()
|
|
|
|
# Close the browser window:
|
|
cosec.quit()
|
|
|
|
# Now process the report,
|
|
# and save it to the JSON file:
|
|
if report_path is not None:
|
|
|
|
# Read the data:
|
|
report_data = CosecWeb.read_muster_roll_xls(report_path)
|
|
|
|
# Do the remaining cleanup and formatting:
|
|
report_data = report_data[[
|
|
"User ID", "User Name", "Category Name",
|
|
"Grade Name", "Branch Name", "Department Name",
|
|
"Direct Reporting", "Level-1"
|
|
]]
|
|
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 = cache_file,
|
|
python_data = {
|
|
"ts": date_time.get_current_utc_date_time(as_string = False).timestamp(),
|
|
"report": report_data
|
|
},
|
|
no_space = True
|
|
)
|
|
|
|
# Note down success:
|
|
success = True
|
|
|
|
# Done here:
|
|
return success
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def sync_branches_to_tcaoff(
|
|
tcaoff_client: AsyncTheCAOffice,
|
|
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 = 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:
|
|
if cosec_branch.lower() not in tcaoff_branches:
|
|
success = await tcaoff_client.branch_add(branch_name = cosec_branch)
|
|
response["total"] += 1
|
|
if success: response["success"] += 1
|
|
else: response["fail"] += 1
|
|
|
|
# Done here:
|
|
return response
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def sync_departments_to_tcaoff(
|
|
tcaoff_client: AsyncTheCAOffice,
|
|
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 = await tcaoff_client.department_list()
|
|
tcaoff_depts = [_["department_name"].lower() for _ in tcaoff_depts]
|
|
tcaoff_depts = list(set(tcaoff_depts))
|
|
|
|
# Get only the department names from Cosec:
|
|
cosec_depts = [tcaoff_client.remove_special_chars(_["Department Name"]) for _ in cosec_muster_roll["report"]]
|
|
cosec_depts = list(set(cosec_depts))
|
|
|
|
# Loop through the data from Cosec and add missing departments to TCAOFF:
|
|
for cosec_dept in cosec_depts:
|
|
if cosec_dept.lower() not in tcaoff_depts:
|
|
success = tcaoff_client.department_add(department_name = cosec_dept)
|
|
response["total"] += 1
|
|
if success: response["success"] += 1
|
|
else: response["fail"] += 1
|
|
|
|
# Done here:
|
|
return response
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def sync_teams_to_tcaoff(
|
|
tcaoff_client: AsyncTheCAOffice,
|
|
cosec_muster_roll: dict,
|
|
) -> Dict[str, int]:
|
|
|
|
# Start with a basic response structure:
|
|
response = defaultdict(int)
|
|
|
|
# Map out branch ids:
|
|
tcaoff_branches = await tcaoff_client.branch_list()
|
|
tcaoff_branches_lookup = {d["branch_name"].lower(): d["branch_id"] for d in tcaoff_branches}
|
|
|
|
# Map out dept. ids:
|
|
tcaoff_depts = await tcaoff_client.department_list()
|
|
tcaoff_depts_lookup = {d["department_name"].lower(): d["department_id"] for d in tcaoff_depts}
|
|
|
|
# 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()
|
|
|
|
# # 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"]:
|
|
|
|
# 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
|
|
|
|
# Check the branch id and department id:
|
|
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", cosec_branch_name)
|
|
response["fail"] += 1
|
|
continue
|
|
if dept_id is None:
|
|
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_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_user_name,
|
|
email = tcaoff_email_id,
|
|
phone_no = "9876543210",
|
|
role = cosec_grade_name,
|
|
username = cosec_user_id,
|
|
password = "Vispl@123",
|
|
applicant_notes = {"cosec": cosec_team}
|
|
)
|
|
if success: response["success"] += 1
|
|
else:
|
|
err_printer("COULDN'T ADD", cosec_team)
|
|
response["fail"] += 1
|
|
|
|
# Done here:
|
|
printer("TCAOFF-Cosec Team Sync.:", json.to_string(response))
|
|
return response
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def get_in_out_summary(
|
|
cosec_creds: dict,
|
|
from_dt: date_time.datetime = None,
|
|
to_dt: date_time.datetime = None,
|
|
cache_file: str = IN_OUT_SUMMARY_CACHE_FILE,
|
|
test_mode: bool = False
|
|
) -> 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.
|
|
:param from_dt: The date from which to fetch In-Out-Summary.
|
|
:param to_dt: The date till which to fetch In-Out-Summary.
|
|
:param cache_file: The cache file to use to store the results.
|
|
:param test_mode: If set to True, a past file will be used instead of getting new reports from Cosec.
|
|
:return: True if the automated fetch was successful, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
report_path = None
|
|
|
|
# If test mode:
|
|
if test_mode:
|
|
report_path = TEST_MODE_IN_OUT_SUMMARY_FILE_PATH
|
|
|
|
# When not working in test mode:
|
|
else:
|
|
|
|
# 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:
|
|
if to_dt is None: to_dt = date_time.get_current_ist_date_time()
|
|
if from_dt is None: from_dt = to_dt - datetime.timedelta(days = 1)
|
|
report_path = cosec.get_in_out_summary(
|
|
initial_sleep = 1.0,
|
|
from_date = from_dt,
|
|
to_date = to_dt,
|
|
group_ids = cosec_creds["inOutConfig"]["groupIds"],
|
|
download_timeout = 60.0,
|
|
timezone = cosec_creds["generalConfig"]["timezone"],
|
|
)
|
|
|
|
# Log out to end the cycle:
|
|
cosec.logout()
|
|
|
|
# Close the browser window:
|
|
cosec.quit()
|
|
|
|
# Now process the report,
|
|
# and save it to the JSON file:
|
|
if report_path is not None:
|
|
|
|
# Read the data:
|
|
report_data = CosecWeb.read_in_out_summary_xls(report_path)
|
|
|
|
# Assume that the punch time in the data is IST data,
|
|
# then normalize it to UTC:
|
|
def parse_dt(x):
|
|
if pd.isnull(x): return None
|
|
else: return date_time.to_timezone(
|
|
datetime_object = date_time.as_if_timezone(
|
|
datetime_object = date_time.parse_date_time(x),
|
|
timezone = cosec_creds["generalConfig"]["timezone"]
|
|
),
|
|
timezone = date_time.TIMEZONE_UTC
|
|
).timestamp()
|
|
report_data["Punch Time"] = report_data["Punch Time"].apply(lambda x: parse_dt(x))
|
|
|
|
# Do the remaining cleanup and formatting:
|
|
report_data = report_data.where(report_data.notna(), None)
|
|
report_data = report_data.to_dict(orient = "records")
|
|
report_data = {
|
|
"ts": date_time.get_current_utc_date_time(as_string = False).timestamp(),
|
|
"report": report_data
|
|
}
|
|
|
|
# Save the data to a JSON file:
|
|
json.to_file(
|
|
file = cache_file,
|
|
python_data = report_data,
|
|
no_space = True
|
|
)
|
|
|
|
# Note down success:
|
|
success = True
|
|
|
|
# Done here:
|
|
return success
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def compute_work_done(
|
|
in_out_df: pd.DataFrame
|
|
) -> List[Dict[str, Union[str, int, float, None]]]:
|
|
|
|
"""
|
|
To calculate the full work done by all the employees on all the provided dates.
|
|
:param in_out_df: The table that contains all the work done by all the employees on the date-range that was
|
|
selected.
|
|
:return: A list of dicts that contain the information of all the work done.
|
|
"""
|
|
|
|
# Create the structure that will be given as the output:
|
|
flattened_work_reports = []
|
|
|
|
# Create an internal dict with the structure:
|
|
# work_reports["user"]["date"] = {...}
|
|
work_reports = defaultdict(lambda: defaultdict(dict))
|
|
|
|
# Loop through the full DataFrame once,
|
|
# and figure out the first in and last out times:
|
|
for index, row in in_out_df.iterrows():
|
|
|
|
# Ignore if the event wasn't a success:
|
|
if row["Event Status"] != "Allowed":
|
|
continue
|
|
|
|
# Extract user and date:
|
|
user_id = row["User ID"]
|
|
punch_dt = date_time.parse_date_time(row["Punch Time"], timezone = date_time.TIMEZONE_IST)
|
|
punch_date = punch_dt.strftime("%Y-%m-%d")
|
|
punch_ts = punch_dt.timestamp()
|
|
punch_loc = row[" Device/Source Detail"]
|
|
|
|
# Handle the first in time:
|
|
if row["I/O Type"] == "In":
|
|
if work_reports[user_id][punch_date].get("first_in") is None:
|
|
work_reports[user_id][punch_date]["first_in"] = punch_ts
|
|
work_reports[user_id][punch_date]["first_in_loc"] = punch_loc
|
|
|
|
# Handle the last out time:
|
|
if row["I/O Type"] == "Out":
|
|
if work_reports[user_id][punch_date].get("first_in") is not None:
|
|
work_reports[user_id][punch_date]["last_out"] = punch_ts
|
|
work_reports[user_id][punch_date]["last_out_loc"] = punch_loc
|
|
|
|
# Now use the first_in and last out times of each record to figure out the amount of work done:
|
|
for user_id, user_reports in work_reports.items():
|
|
for punch_date, punch_info in user_reports.items():
|
|
|
|
# Some defaults:
|
|
min_work_seconds = 10.0 * 60.0 * 60.0
|
|
work_seconds = 0.0
|
|
work_ot_seconds = 0.0
|
|
work_status = "A"
|
|
|
|
# Extract, clean and compute punch timing:
|
|
first_in = punch_info.get("first_in")
|
|
first_in_loc = punch_info.get("first_in_loc")
|
|
last_out = punch_info.get("last_out")
|
|
last_out_loc = punch_info.get("last_out_loc")
|
|
|
|
# When the user has a valid in-time, but no known out time,
|
|
# we assume that he worked a full day:
|
|
if first_in is not None and not last_out:
|
|
last_out = first_in + min_work_seconds
|
|
work_seconds = last_out - first_in
|
|
work_ot_seconds = 0.0
|
|
|
|
# When the user has neither an in-time, nor an out-time,
|
|
# we assume that he was absent the whole day:
|
|
elif not first_in and not last_out:
|
|
work_seconds = 0.0
|
|
work_ot_seconds = 0.0
|
|
|
|
# When we have both - an in-time and an out-time - we compute the work done.
|
|
# Any work over 10 hours will be counted as overtime:
|
|
elif first_in and last_out:
|
|
work_seconds = last_out - first_in
|
|
work_ot_seconds = max(0.0, work_seconds - min_work_seconds)
|
|
|
|
# Finally, compute the work status.
|
|
# 'A' --> Absent
|
|
# 'H' --> Half Day
|
|
# 'P' --> Present (Full Day)
|
|
# 'OT' -> Overtime
|
|
work_hours = work_seconds / (60 * 60)
|
|
if work_hours > 10.0: work_status = "OT"
|
|
elif 7.5 < work_hours <= 10.0: work_status = "P"
|
|
elif 2.5 < work_hours <= 5.0: work_status = "H"
|
|
else: work_status = "A"
|
|
|
|
# Save the data:
|
|
flattened_work_reports.append({
|
|
"user_id": user_id,
|
|
"work_date": punch_date,
|
|
"first_in": first_in,
|
|
"first_in_loc": first_in_loc,
|
|
"last_out": last_out,
|
|
"last_out_loc": last_out_loc,
|
|
"work_seconds": work_seconds,
|
|
"work_hours": work_hours,
|
|
"work_ot_seconds": work_ot_seconds,
|
|
"work_ot_hours": work_ot_seconds / (60.0 * 60.0),
|
|
"work_status": work_status,
|
|
})
|
|
|
|
# Done here:
|
|
return flattened_work_reports
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def sync_attendance_to_tcaoff(
|
|
tcaoff_client: AsyncTheCAOffice,
|
|
cosec_in_out_summary: dict,
|
|
verbose: bool = False
|
|
) -> Dict[str, int]:
|
|
|
|
"""
|
|
|
|
:param tcaoff_client: The asynchronous client object that interfaces with TCAOFF.
|
|
:param cosec_in_out_summary: he table that contains all the work done by all the employees on the date-range that
|
|
was selected.
|
|
:param verbose: If True, internal debugging print will be more verbose.
|
|
:return: The dict that gives you the count of the successful and failed attendance marking API calls.
|
|
"""
|
|
|
|
# Results:
|
|
results = defaultdict(int)
|
|
|
|
# Convert the In/Out data to a DataFrame:
|
|
in_out_df = pd.DataFrame(cosec_in_out_summary["report"])
|
|
|
|
# Compute the work done:
|
|
work_reports = compute_work_done(in_out_df)
|
|
|
|
# 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()
|
|
|
|
# Get a mapping from Cosec id to TCAOFF record:
|
|
cosec_id_to_tcaoff_team = {}
|
|
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 not app_notes: continue
|
|
cosec_notes = app_notes.get("cosec", {})
|
|
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 = []
|
|
for wr in work_reports:
|
|
|
|
# Extract basic details:
|
|
cosec_user_id = wr["user_id"]
|
|
|
|
# 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 not found in TCAOFF", cosec_user_id)
|
|
continue
|
|
|
|
# Create the task for this attendance:
|
|
tasks.append(
|
|
tcaoff_client.attendance_mark(
|
|
user_id = tcaoff_team["user_id"],
|
|
status = wr["work_status"],
|
|
over_time = wr["work_ot_hours"],
|
|
attendance_date = wr["work_date"],
|
|
json_notes = {
|
|
"totHours": wr["work_hours"],
|
|
"firstIn": wr["first_in"],
|
|
"lastOut": wr["last_out"],
|
|
}
|
|
)
|
|
)
|
|
|
|
printer("TASKS COUNT", len(tasks))
|
|
|
|
def chunks(lst, size = 10):
|
|
for i in range(0, len(lst), size):
|
|
yield lst[i:i + size]
|
|
|
|
count = 0
|
|
for chunk in chunks(tasks[:], size = 25):
|
|
count += 1
|
|
printer("Task Chunk:", count)
|
|
_res = await asyncio.gather(*chunk)
|
|
success = sum(_res)
|
|
failure = len(_res) - success
|
|
results["success"] += success
|
|
results["failure"] += failure
|
|
results["total"] += len(_res)
|
|
|
|
# Done here:
|
|
printer(results)
|
|
return results
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
pass
|