959 lines
34 KiB
Python
959 lines
34 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
CREATED: Wed, 26th Nov, 2025
|
|
UPDATED: Wed, 26th Nov, 2025
|
|
|
|
OBJECTIVE:
|
|
|
|
To achieve so-and-so-objective...
|
|
|
|
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 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
|
|
|
|
# To run a cron-like scheduler:
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** 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)
|
|
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")
|
|
prev_day_in_out_summary_cache_file = os.path.join(proj_dir, "local", "cache", "prev_day_in_out_summary_cache.json")
|
|
chrome_driver_dir = os.path.join(proj_dir, r"drivers","chrome")
|
|
user_data_dir = os.path.join(proj_dir, r"browser", "user_data")
|
|
downloads_dir = os.path.join(proj_dir, r"downloads")
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
def sync_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"].lower() 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.lower() 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 sync_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"].lower() 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.lower() 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 sync_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"].lower(): 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"].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 = tcaoff.team_list(tcaoff_creds)
|
|
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:
|
|
# print("NULL APP NOTES??:", t["json_notes"])
|
|
app_notes = {}
|
|
cosec_notes = app_notes.get("cosec", {})
|
|
# print("COSEC NOTES:", cosec_notes)
|
|
if cosec_notes: synced_team_ids.append(cosec_notes["User ID"])
|
|
tcaoff_teams = [_["pseudonym"] for _ in tcaoff_teams]
|
|
tcaoff_teams = list(set(tcaoff_teams))
|
|
print("Team Ids:", synced_team_ids)
|
|
|
|
# # 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 synced_team_ids:
|
|
|
|
# Count the user:
|
|
response["total"] += 1
|
|
|
|
# Check the branch id and department id:
|
|
branch_id = tcaoff_branches_lookup.get(tcaoff.remove_special_chars(cosec_team["Branch Name"]).lower())
|
|
dept_id = tcaoff_depts_lookup.get(tcaoff.remove_special_chars(cosec_team["Department Name"]).lower())
|
|
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:
|
|
tcaoff_username = regex.replace(
|
|
text = cosec_team["User Name"],
|
|
pattern = r"[^\w\d\._]",
|
|
substitute_text = ""
|
|
).strip().lower()
|
|
# team_json = {
|
|
# "branchId": branch_id,
|
|
# "idDepartment": dept_id,
|
|
# "reportingTo": None,
|
|
# "name": cosec_team['User Name'].strip(),
|
|
# "email": tcaoff_username + "@velankanigroup.com",
|
|
# "phoneNo": "9876543210",
|
|
# "role": cosec_team['Grade Name'].strip(),
|
|
# "username": tcaoff_username,
|
|
# "password": "Vispl@123",
|
|
# "hierarchy": 1,
|
|
# "application_notes": {"cosec": cosec_team}
|
|
# }
|
|
# 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 = tcaoff_username + "@velankanigroup.com",
|
|
phone_no = "9876543210",
|
|
role = (cosec_team.get("Grade Name") or "Unknown").strip(),
|
|
username = tcaoff_username,
|
|
password = "Vispl@123",
|
|
applicant_notes = {"cosec": cosec_team}
|
|
)
|
|
# response["total"] += 1
|
|
if success: response["success"] += 1
|
|
else:
|
|
print("COULDN'T ADD:", json.to_string(cosec_team))
|
|
response["fail"] += 1
|
|
# if response["total"] >= 100: break
|
|
|
|
# Done here:
|
|
print("TCAOFF-Cosec Depts. Sync.:", json.to_string(response))
|
|
return response
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def sync_attendance_to_tcaoff(
|
|
tcaoff_creds: dict,
|
|
cosec_in_out_summary: dict,
|
|
) -> Dict[str, int]:
|
|
|
|
# Get the list of existing team members from TCAOFF:
|
|
# NOTE: `pseudonym` is the unique username of the user.
|
|
tcaoff_teams = tcaoff.team_list(tcaoff_creds)
|
|
print("TCAOFF TEAMS:", json.to_string(tcaoff_teams))
|
|
|
|
# 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["User ID"]] = t
|
|
print("COSEC to TCAOFF TEAMS:", json.to_string(cosec_id_to_tcaoff_team))
|
|
|
|
# print("Hi, attendance!")
|
|
# print(json.to_string(cosec_in_out_summary["report"][0]))
|
|
|
|
# Convert the data to a DataFrame:
|
|
in_out_df = pd.DataFrame(cosec_in_out_summary["report"])
|
|
# print(in_out_df[:35].to_string())
|
|
|
|
# Get the unique user ids:
|
|
user_ids = in_out_df["User ID"].unique().tolist()
|
|
print(f"USER IDS ({len(user_ids)}):", user_ids)
|
|
|
|
# Start with a blank user to time mapping:
|
|
user_id_to_time = {}
|
|
status_count = defaultdict(int)
|
|
|
|
# For each unique user id,
|
|
# Check the no of times when he successfully triggered the attendance-punching device:
|
|
for user_id in user_ids:
|
|
|
|
# Fetch only the successful events:
|
|
user_allowed_events = in_out_df[
|
|
(in_out_df["User ID"] == user_id) &
|
|
(in_out_df["Event Status"] == "Allowed")
|
|
]
|
|
|
|
# Now get the last date (since records can be spread across days),
|
|
# and keep the records of only the last day:
|
|
last_dt = user_allowed_events["Punch Time"].iloc[-1]
|
|
last_dt = date_time.parse_date_time(last_dt)
|
|
last_dt = date_time.datetime(
|
|
year = last_dt.year,
|
|
month = last_dt.month,
|
|
day = last_dt.day,
|
|
hour = 0,
|
|
minute = 0,
|
|
second = 0
|
|
).timestamp()
|
|
print("LAST DATE-TIME:", last_dt)
|
|
user_allowed_events = user_allowed_events[user_allowed_events["Punch Time"] >= last_dt]
|
|
|
|
print("\n\n---\n\n")
|
|
print("USER ID:", user_id)
|
|
print("\n")
|
|
print("Allowed Events:")
|
|
print("--------------")
|
|
print(user_allowed_events.to_string())
|
|
print("\n")
|
|
print("Interpretation:")
|
|
print("--------------")
|
|
|
|
# We note down the first "In" time of the user
|
|
# and the last "Out" time of the user:
|
|
first_in = None
|
|
last_out = None
|
|
for idx, row in user_allowed_events.iterrows():
|
|
print(f"{row['I/O Type']: >4} at row: {idx: <5} | ts: {row['Punch Time']: <15} | dt: {date_time.parse_date_time(row['Punch Time'])}")
|
|
if row["I/O Type"] == "In" and first_in is None: first_in = row["Punch Time"]
|
|
if row["I/O Type"] == "Out" and first_in is not None: last_out = row["Punch Time"]
|
|
|
|
print("\n")
|
|
print("Timestamps:")
|
|
print("----------")
|
|
print("FIRST-IN:", date_time.parse_date_time(first_in))
|
|
print("LAST-OUT:", date_time.parse_date_time(last_out))
|
|
|
|
# Figure out the worked time:
|
|
if first_in is None and last_out is None:
|
|
user_id_to_time[user_id] = {
|
|
"work_seconds": 0.0,
|
|
"work_date": last_dt
|
|
}
|
|
elif first_in is None or last_out is None:
|
|
user_id_to_time[user_id] = {
|
|
"work_seconds": 60 * 60 * 10.0, # ... 10 hours represented in seconds.
|
|
"work_date": last_dt
|
|
}
|
|
else:
|
|
user_id_to_time[user_id] = {
|
|
"work_seconds": last_out - first_in,
|
|
"work_date": last_dt
|
|
}
|
|
|
|
# Add fixed known details:
|
|
user_id_to_time[user_id]["first_in"] = first_in
|
|
user_id_to_time[user_id]["last_out"] = last_out
|
|
|
|
print("\n")
|
|
print("Result:")
|
|
print("------")
|
|
print("Total Seconds :", user_id_to_time[user_id]["work_seconds"])
|
|
print("Total Hours :", user_id_to_time[user_id]["work_seconds"] / (60 * 60))
|
|
|
|
# break
|
|
|
|
# # Wait till you get an "IN" event, ten wait till you get an "OUT" event.
|
|
# # Keep totalling between them. If you end with an "IN" event, consider the
|
|
# # current time as the "OUT" event:
|
|
# in_at = None
|
|
# for idx, row in user_allowed_events.iterrows():
|
|
# if row["I/O Type"] == "In" and in_at is None:
|
|
# print(f" In at row: {idx: <5} | ts: {row['Punch Time']: <15} | dt: {date_time.parse_date_time(row['Punch Time'])}")
|
|
# in_at = row["Punch Time"]
|
|
# if row["I/O Type"] == "Out" and in_at is not None:
|
|
# print(f"Out at row: {idx: <5} | ts: {row['Punch Time']: <15} | dt: {date_time.parse_date_time(row['Punch Time'])}")
|
|
# user_id_to_time[user_id] += row["Punch Time"] - in_at
|
|
# in_at = None
|
|
# # if in_at is not None:
|
|
# # user_id_to_time[user_id] += date_time.get_current_date_time().timestamp() - in_at
|
|
# print("\n")
|
|
# print("Result:")
|
|
# print("------")
|
|
# print("Total Seconds :", user_id_to_time[user_id])
|
|
# print("Total Hours :", user_id_to_time[user_id] / (60*60))
|
|
# # break
|
|
|
|
# # Show the results:
|
|
# user_id_to_hours = {k:v/(60*60) for k, v in user_id_to_time.items()}
|
|
# print("\n\n---\n\n")
|
|
# print("WORK-HOURS DONE:", json.to_string(user_id_to_hours))
|
|
|
|
# Start with a basic response structure:
|
|
response = defaultdict(int)
|
|
|
|
# At this point, you have the amount of work done be each Cosec team member.
|
|
# Now we loop through the results and sync them on TCAOFF:
|
|
for user_id, work in user_id_to_time.items():
|
|
|
|
response["total"] += 1
|
|
|
|
# Find the equivalent TCAOFF team member record:
|
|
tcaoff_team = cosec_id_to_tcaoff_team.get(user_id)
|
|
if tcaoff_team is None:
|
|
print(f"TCAOFF SYNC ERR: Cosesc User Id '{user_id}' not found in TCAOFF")
|
|
response["fail"] += 1
|
|
continue
|
|
|
|
# Update the attendance on TCAOFF:
|
|
hours_worked = work["work_seconds"] // (60 * 60)
|
|
if hours_worked > 10.0: status = "OT"
|
|
elif 7.5 < hours_worked <= 10.0: status = "P"
|
|
elif 2.5 < hours_worked <= 5.0: status = "H"
|
|
else: status = "A"
|
|
status_count[status] += 1
|
|
success = tcaoff.attendance_mark(
|
|
tcaoff_creds,
|
|
user_id = tcaoff_team["user_id"],
|
|
status = status,
|
|
over_time = max(0.0, hours_worked - 10.0),
|
|
attendance_date = work["work_date"],
|
|
json_notes = {
|
|
"totHours": hours_worked,
|
|
"firstIn": work["first_in"],
|
|
"lastOut": work["last_out"],
|
|
}
|
|
)
|
|
if success: response["success"] += 1
|
|
else: response["fail"] += 1
|
|
|
|
# Show the stats:
|
|
print("WORKING HOUR STATUS STATS:", json.to_string(status_count))
|
|
|
|
# Done here:
|
|
print("TCAOFF-Cosec Attendance Sync.:", json.to_string(response))
|
|
return response
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def kill_chrome() -> 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,
|
|
cache_file: str = muster_roll_cache_file
|
|
) -> 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 cache_file: The cache file to store the results in.
|
|
:return: True if the automated fetch was successful, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
|
|
# 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()
|
|
|
|
# report_path = r"D:\kps\PycharmProjects\cosec\cosec_web\sample_files\muster_roll.xls"
|
|
|
|
# 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def get_in_out_summary(
|
|
cosec_creds: dict,
|
|
target_date: date_time.datetime = None,
|
|
cache_file: str = in_out_summary_cache_file,
|
|
) -> 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 target_date: The date to check for in Out-Of-Out-Summary.
|
|
:param cache_file: The cache file to use to store the results.
|
|
:return: True if the automated fetch was successful, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
|
|
# 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 target_date is None: target_date = date_time.get_current_ist_date_time()
|
|
from_dt = target_date - datetime.timedelta(
|
|
days = cosec_creds["generalConfig"]["timedelta"]["days"],
|
|
hours = cosec_creds["generalConfig"]["timedelta"]["hours"],
|
|
minutes = cosec_creds["generalConfig"]["timedelta"]["minutes"],
|
|
seconds = cosec_creds["generalConfig"]["timedelta"]["seconds"],
|
|
)
|
|
to_dt = target_date
|
|
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()
|
|
|
|
# report_path = r"D:\kps\PycharmProjects\cosec\cosec_web\sample_files\in_out_summary.xls"
|
|
|
|
# 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 today_cron(
|
|
cosec_creds: dict,
|
|
tcaoff_creds: dict,
|
|
) -> None:
|
|
|
|
"""
|
|
To get today's stats about the team.
|
|
:param cosec_creds: The credentials to use to log into Matrix COSEC.
|
|
:param tcaoff_creds: The credentials to use to log into TCAOFF.
|
|
:return: None.
|
|
"""
|
|
|
|
print("TODAY CRON")
|
|
|
|
# Try the whole process once:
|
|
try:
|
|
|
|
# Log in to TCAOFF:
|
|
tcaoff.login(tcaoff_creds)
|
|
|
|
# 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,
|
|
cache_file = muster_roll_cache_file
|
|
)
|
|
|
|
# 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)
|
|
sync_branches_to_tcaoff(
|
|
tcaoff_creds = tcaoff_creds,
|
|
cosec_muster_roll = cosec_muster_roll,
|
|
)
|
|
sync_departments_to_tcaoff(
|
|
tcaoff_creds = tcaoff_creds,
|
|
cosec_muster_roll = cosec_muster_roll,
|
|
)
|
|
sync_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
|
|
|
|
# IN-OUT REPORT:
|
|
try:
|
|
|
|
# Get the previous day's In/Out Summary and then wait
|
|
# for the driver's resources to get freed:
|
|
success = get_in_out_summary(
|
|
cosec_creds = cosec_creds,
|
|
target_date = date_time.get_current_ist_date_time(),
|
|
cache_file = in_out_summary_cache_file
|
|
)
|
|
|
|
# If something goes wrong:
|
|
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("REPORT CRON FAILED LOOP!")
|
|
print("EXCEPTION:", e)
|
|
tcaoff.logout(tcaoff_creds)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def yesterday_cron(
|
|
cosec_creds: dict,
|
|
tcaoff_creds: dict,
|
|
) -> None:
|
|
|
|
"""
|
|
To mark yesterday's attendance.
|
|
:param cosec_creds: The credentials to use to log into Matrix COSEC.
|
|
:param tcaoff_creds: The credentials to use to log into TCAOFF.
|
|
:return: None.
|
|
"""
|
|
|
|
print("YESTERDAY CRON")
|
|
|
|
# Try the whole process once:
|
|
try:
|
|
|
|
# Log in to TCAOFF:
|
|
tcaoff.login(tcaoff_creds)
|
|
|
|
# IN-OUT SUMMARY REPORT:
|
|
try:
|
|
|
|
# Get the previous day's In/Out Summary and then wait
|
|
# for the driver's resources to get freed:
|
|
success = get_in_out_summary(
|
|
cosec_creds = cosec_creds,
|
|
target_date = date_time.get_current_ist_date_time() - datetime.timedelta(days = 1),
|
|
cache_file = prev_day_in_out_summary_cache_file
|
|
)
|
|
|
|
# Sync data between Cosec and TCAOFF:
|
|
if success:
|
|
print("IN/OUT SUMMARY: Sync'ing with TCAOFF")
|
|
sync_attendance_to_tcaoff(
|
|
tcaoff_creds = tcaoff_creds,
|
|
cosec_in_out_summary = json.from_file(prev_day_in_out_summary_cache_file),
|
|
)
|
|
|
|
# If something goes wrong:
|
|
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("REPORT CRON FAILED LOOP!")
|
|
print("EXCEPTION:", e)
|
|
tcaoff.logout(tcaoff_creds)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def set_scheduler(
|
|
cosec_creds: dict,
|
|
tcaoff_creds: dict,
|
|
) -> None:
|
|
|
|
"""
|
|
Set up the scheduler (cron) that will time the activities. The background task happens at fixed times, the
|
|
foreground task happens in a loop.
|
|
:param cosec_creds: The credentials to use to log into Matrix COSEC.
|
|
:param tcaoff_creds: The credentials to use to log into TCAOFF.
|
|
:return: None.
|
|
"""
|
|
|
|
scheduler = BackgroundScheduler()
|
|
for h, m in zip(
|
|
[8, 23, 15],
|
|
[30, 30, 40]
|
|
):
|
|
scheduler.add_job(
|
|
yesterday_cron,
|
|
"cron",
|
|
hour = h,
|
|
minute = m,
|
|
args = [
|
|
copy.deepcopy(cosec_creds),
|
|
copy.deepcopy(tcaoff_creds)
|
|
]
|
|
)
|
|
scheduler.start()
|
|
|
|
try:
|
|
|
|
while True:
|
|
start = time.time()
|
|
today_cron(
|
|
cosec_creds = copy.deepcopy(cosec_creds),
|
|
tcaoff_creds = copy.deepcopy(tcaoff_creds),
|
|
)
|
|
elapsed = time.time() - start
|
|
time.sleep(max(0, int(3600 - elapsed)))
|
|
|
|
except (KeyboardInterrupt, SystemExit):
|
|
scheduler.shutdown()
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
# Explicitly mention the expected file paths for other devs to maintain:
|
|
print("PROJ. DIR. :", proj_dir)
|
|
print("CREDS FILE :", cosec_creds_file)
|
|
print("M-ROLL CACHE:", muster_roll_cache_file)
|
|
print("IN-OUT CACHE:", in_out_summary_cache_file)
|
|
|
|
# Read Cosec's credentials.
|
|
# It should be in the format:
|
|
"""
|
|
{
|
|
"creds": {
|
|
"url": "http://x.x.x.x/COSEC/",
|
|
"username": "<usr>",
|
|
"password": "<pwd>"
|
|
},
|
|
"generalConfig": {
|
|
"pollInterval": 3600,
|
|
"timezone": "Asia/Kolkata",
|
|
"timedelta": {
|
|
"days": 0,
|
|
"hours": 24,
|
|
"minutes": 0,
|
|
"seconds": 0
|
|
}
|
|
},
|
|
"musterRollConfig": {
|
|
"groupIds": ["2", "3", "4"]
|
|
},
|
|
"inOutConfig": {
|
|
"groupIds": ["2", "3", "4"]
|
|
}
|
|
}
|
|
"""
|
|
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",
|
|
"teamUpdate": "https://api.thecaoffice.com/team/update",
|
|
"attendanceList": "https://api.thecaoffice.com/user/attendance/register",
|
|
"attendanceMark": "https://api.thecaoffice.com/user/attendance/mark"
|
|
}
|
|
}
|
|
"""
|
|
tcaoff_creds = json.from_file(tcaoff_creds_file)
|
|
|
|
set_scheduler(
|
|
cosec_creds = cosec_creds,
|
|
tcaoff_creds = tcaoff_creds,
|
|
)
|