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

Compare commits

base: khushalps/cosec_automation:6952bf73f560e8ce16fd781934ab63b58a98e559
Branches Tags
khushalps/cosec_automation:master
...
compare: khushalps/cosec_automation:093e1de3fb365d18e00232f6d7a6308842e8f35a
Branches Tags
khushalps/cosec_automation:master

10 Commits
6952bf73f5 ... 093e1de3fb

Author SHA1 Message Date
khushalps 093e1de3fb (20260623) Migrate to new server. 2026-06-23 17:16:03 +05:30
khushalps c3ebad3b35 (20260404) Team fixing on TCAOFF completed. 2026-04-04 13:14:44 +05:30
khushalps 6838648a92 (20260404) Team fixing on TCAOFF completed. 2026-04-04 12:57:28 +05:30
khushalps 1102b826da (20260403) Started working on Manual Team Sync. and Fix 2026-04-03 10:05:43 +05:30
yatmesh e2167e3f5d Added the mandatory fields for the team update api 2026-04-02 13:10:22 +00:00
khushalps 3a68412b1b (20260317) Tried headless and parallel mode. 2026-03-17 18:09:04 +05:30
khushalps e25daf5fd5 (20260311) Cosec scripts can now accept custom file names for the credentials, but they must be in the same 'creds' directory. 2026-03-11 11:23:52 +05:30
khushalps 0adb27bae3 (20260310) Modified the work computation to label the event as "In" or "Out" by even considering the label of the device. 2026-03-11 11:22:56 +05:30
khushalps fe752a0028 (20260310) Modified the work computation to label the event as "In" or "Out" by even considering the label of the device. 2026-03-10 13:18:39 +05:30
khushalps 172cfdc92d (20260309) Added caching to avoid unchanged attendance calls and added logic to check for custom overtime thresholds. 2026-03-09 20:24:24 +05:30
19 changed files with 16351 additions and 14491 deletions
Expand all files Collapse all files
README.md
+5 -1
View File
@@ -116,4 +116,8 @@ Run the file `<proj-dir>/cron/reports.py`, or `<proj-dir>/run_cron.sh`
### 👉 Running The Manual Attendance Script
Run the file `<proj-dir>/cron/attendance.py` with the args `--date <yyyy-mm-dd>`, or run `<proj-dir>/run_cron.sh` and it will prompt you to type the date.
Run the file `<proj-dir>/cron/attendance.py` with the args `--date <yyyy-mm-dd>`, or run `<proj-dir>/run_cron.sh` and it will prompt you to type the date.
---
End of notes.
cosec_web/cosec_web.py
+36 -7
View File
@@ -105,6 +105,9 @@ class CosecWeb:
driver_dir: str,
user_data_dir: str,
downloads_dir: str,
window_width: int = 1920,
window_height: int = 1080,
headless: bool = True,
debug: bool = True,
debug_prefix: str = "Cosec-Web | "
):
@@ -117,6 +120,9 @@ class CosecWeb:
:param driver_dir: The directory where the driver is placed.
:param user_data_dir: The directory where you want the web-driver to store user data.
:param downloads_dir: The directory where you want to hold the downloaded files.
:param window_width: The width of the browser window.
:param window_height: The height of the browser window.
:param headless: Whether, or not, to use Chrome browser in headless (No UI) mode.
:param debug: Whether, or not, to show debug messages.
:param debug_prefix: The prefix to show before the debug messages.
"""
@@ -135,6 +141,9 @@ class CosecWeb:
self.driver_dir = driver_dir
self.user_data_dir = user_data_dir
self.downloads_dir = downloads_dir
self.headless = headless
self.window_width = window_width
self.window_height = window_height
# Figure out which driver to use:
self.driver_path = None
@@ -148,7 +157,7 @@ class CosecWeb:
# If we don't have a matching driver:
if self.driver_path is None: raise NotImplementedError("OS and/or CPU Architecture Not Supported.")
# Else, we initialize the driver:
# Else, we start constructing the preferred configuration::
self._nc_printer("Using driver:", self.driver_path)
self._service = Service(self.driver_path)
self._options = webdriver.ChromeOptions()
@@ -158,26 +167,42 @@ class CosecWeb:
self._options.add_experimental_option(
"prefs",
{
"downloads.default_directory": downloads_dir, # ... The custom downloads path.
"downloads.prompt_for_download": False, # ......... Do not show save dialog.
"downloads.directory_upgrade": True, # ............ If path doesn't exist, try to create it.
"safebrowsing.enabled": True, # .................. Allow safe downloads.
"downloads.default_directory": downloads_dir, # ....... The custom 'downloads' path.
"downloads.prompt_for_download": False, # ............. Do not show save dialog.
"downloads.directory_upgrade": True, # ................ If path doesn't exist, try to create it.
"safebrowsing.enabled": True, # ....................... Allow safe downloads.
"safebrowsing.disable_download_protection": True, # ... Don't show "keep" button before downloading.
"profile.default_content_settings.popups": 0,
"profile.default_content_setting_values.automatic_downloads": 1,
"profile.content_settings.exceptions.automatic_downloads.*.setting": 1,
}
)
# For headless mode:
if headless:
self._printer("HEADLESS MODE!")
self._options.add_argument("--headless=new") # .................................. Modern headless mode.
self._options.add_argument("--disable-gpu") # ................................... Optional (mostly for Windows).
self._options.add_argument(f"--window-size={window_width},{window_height}") # ... The internally simulated screen size.
# Initialize the driver:
self._driver = webdriver.Chrome(service = self._service, options = self._options)
self._driver.set_page_load_timeout(300)
self._driver.execute_cdp_cmd("Page.setDownloadBehavior", {
"behavior": "allow",
"downloadPath": downloads_dir
})
self._printer("Driver ready.")
# For later use, we will need variables to hold window information:
self._original_window = None
self._current_window = None
def __del__(self):
self._printer("Deinitializing!")
try: self.quit()
except Exception as exception: self._printer(exception)
# ┳┓ ┓ •
# ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓
# ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫
@@ -748,8 +773,8 @@ class CosecWeb:
self._driver.switch_to.window(self._current_window)
self._printer("Context switched.", self._current_window)
# Make the window full-screen for standardized behaviour:
self._driver.maximize_window()
# Set the size of the window so that you are certain that all the elements will be visible:
self._driver.set_window_size(self.window_width, self.window_height)
# The username text field:
element = wait.until(EC.presence_of_element_located((By.ID, "loginid")))
@@ -838,6 +863,8 @@ class CosecWeb:
:return: The path to the downloaded report file.
"""
self._printer("MUSTER ROLL")
# Empty out the past downloads:
for file_name in files.list_files(
self.downloads_dir,
@@ -1027,6 +1054,8 @@ class CosecWeb:
:return: The path to the downloaded report file.
"""
self._printer("IN-OUT SUMMARY")
# Apply the timezone if given:
if timezone:
from_date = date_time.to_timezone(from_date, timezone)
cron/__init__.py
View File
cron/attendance.py
-220
View File
@@ -1,220 +0,0 @@
"""
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 tcaoff 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
# Needed functions:
from cron.reports import *
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
def manual_attendance(
cosec_creds: dict,
tcaoff_creds: dict,
target_date: date_time.datetime
) -> 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.
:param target_date: The date to sync attendance.
:return: None.
"""
print("ATTENDANCE SYNC")
# 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 = target_date,
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)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description = "Manual attendance sync. script.")
parser.add_argument(
"--from-date",
help = "The date from which you want to sync attendance.",
default = date_time.get_current_ist_date_time()
)
parser.add_argument(
"--to-date",
help = "The date till when you want to sync attendance.",
default = date_time.get_current_ist_date_time()
)
args = parser.parse_args()
args.from_date = date_time.parse_date_time(args.from_date, timezone = date_time.TIMEZONE_IST)
args.to_date = date_time.parse_date_time(args.to_date, timezone = date_time.TIMEZONE_IST)
if args.from_date > args.to_date: args.from_date, args.to_date = args.to_date, args.from_date
print("From Date:", args.from_date)
print(" To Date:", args.to_date)
dates_list = [args.from_date]
while args.from_date < args.to_date:
args.from_date = args.from_date + datetime.timedelta(days = 1)
dates_list.append(args.from_date)
print(f"DATES ({len(dates_list)}):", dates_list)
# Load the credentials:
cosec_creds = json.from_file(cosec_creds_file)
tcaoff_creds = json.from_file(tcaoff_creds_file)
# Run the attendance for each specified date:
for target_date in dates_list:
print("TARGET DATE:", target_date)
manual_attendance(
cosec_creds = cosec_creds,
tcaoff_creds = tcaoff_creds,
target_date = date_time.parse_date_time(
target_date,
date_formats = [
"%Y%m%d",
"%Y-%m-%d"
],
) + date_time.timedelta(
hours = 23,
minutes = 59
)
)
cron/reports.py
-926
View File
@@ -1,926 +0,0 @@
"""
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 tcaoff 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)
print("Cosec Notes:", cosec_notes)
# cosec_id_to_tcaoff_team[cosec_notes["User ID"]] = t
cosec_id_to_tcaoff_team[cosec_notes.get("User ID") or cosec_notes.get("UserID")] = 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.0 * 60.0)
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!")
print("EXCEPTION:", e)
# 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
)
# 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!")
print("EXCEPTION:", e)
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)
raise
# ---------------------------------------------------------------------------------------------------------------------
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!")
print("EXCEPTION:", e)
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.
"""
# Create the scheduler:
scheduler = BackgroundScheduler()
# Populate the tasks in the scheduler:
for h, m in zip(
[8, 23],
[30, 30]
):
scheduler.add_job(
yesterday_cron,
"cron",
hour = h,
minute = m,
args = [
copy.deepcopy(cosec_creds),
copy.deepcopy(tcaoff_creds)
]
)
# Populate the tasks in the scheduler:
for h in range(0, 23):
for m in [0]:
scheduler.add_job(
today_cron,
"cron",
hour = h,
minute = m,
args = [
copy.deepcopy(cosec_creds),
copy.deepcopy(tcaoff_creds)
]
)
# Run the scheduler:
scheduler.start()
while True: time.sleep(10.0)
# *****************************************************************************************************************
# ***** ****
# *** 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 required credentials:
cosec_creds = json.from_file(cosec_creds_file)
tcaoff_creds = json.from_file(tcaoff_creds_file)
# Schedule the activities:
set_scheduler(
cosec_creds = cosec_creds,
tcaoff_creds = tcaoff_creds,
)
drivers/chrome/linux64/THIRD_PARTY_NOTICES.chromedriver
+14550 -13188
View File
File diff suppressed because it is too large Load Diff
drivers/chrome/linux64/chromedriver
BIN
View File
Binary file not shown.
drivers/chrome/win64/THIRD_PARTY_NOTICES.chromedriver
+1050 -59
View File
File diff suppressed because it is too large Load Diff
drivers/chrome/win64/chromedriver.exe
BIN
View File
Binary file not shown.
local/cache/in_out_summary_cache.json
Vendored
+1 -1
View File
File diff suppressed because one or more lines are too long
local/cache/muster_roll_cache.json
Vendored
+1 -1
View File
File diff suppressed because one or more lines are too long
local/cache/prev_day_in_out_summary_cache.json
Vendored
+1 -1
View File
File diff suppressed because one or more lines are too long
playground/20260309_attendance_register_test.py
+2 -1
View File
@@ -32,6 +32,7 @@ async def main():
target_month = datetime.datetime.now(),
raise_exception = True
)
print(f"RAW RECORDS ({len(raw_records)}):", json.to_string(raw_records))
# Format the data for preview:
formatted_records = []
@@ -45,7 +46,7 @@ async def main():
formatted_records.append(cosec_data)
# Show the formatted data for debugging:
print(f"RECORDS ({len(formatted_records)}):", json.to_string(formatted_records))
print(f"FORMATTED RECORDS ({len(formatted_records)}):", json.to_string(formatted_records[:10]))
print(f"Found {len(formatted_records)} record(s).")
# Log out:
scripts/common.py
+165 -40
View File
@@ -22,8 +22,7 @@
N/A
"""
import pathlib
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
@@ -60,12 +59,9 @@ 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 typing import List, Dict, Any, Union, Callable
from collections import defaultdict
# To run a cron-like scheduler:
from apscheduler.schedulers.asyncio import AsyncIOScheduler
# For async activities:
import asyncio
@@ -85,20 +81,27 @@ 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")
MANUAL_SUMMARY_CACHE_FILE = os.path.join(CACHE_DIR, "manual_in_out_summary_cache.json")
# # ---
# 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")
# MANUAL_SUMMARY_CACHE_FILE = os.path.join(CACHE_DIR, "manual_in_out_summary_cache.json")
# WORK_REPORTS_CACHE_FILE = os.path.join(CACHE_DIR, "work_reports_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")
# USER_DATA_DIR = os.path.join(PROJ_DIR, "browser", "user_data", os.environ.get("ORG", "default"))
# DOWNLOADS_DIR = os.path.join(PROJ_DIR, "downloads", os.environ.get("ORG", "default"))
# ---
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")
# Defaults:
DEFAULT_WEEKLY_WORKING_DAYS = 5
DEFAULT_WEEKLY_WORKING_HOURS = 50.0
DEFAULT_DAILY_WORKING_HOURS = DEFAULT_WEEKLY_WORKING_HOURS / DEFAULT_WEEKLY_WORKING_DAYS
# Debugging:
printer = IceCreamDebugger(prefix = "Common | ", includeContext = True)
err_printer = IceCreamDebugger(prefix = "[ERR] Common | ", includeContext = True)
@@ -131,6 +134,42 @@ err_printer = IceCreamDebugger(prefix = "[ERR] Common | ", includeContext = True
# *****************************************************************************************************************
def get_muster_roll_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "muster_roll_cache.json")
def get_in_out_summary_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "in_out_summary_cache.json")
def get_prev_day_in_out_summary_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "prev_day_in_out_summary_cache.json")
def get_manual_in_out_summary_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "manual_in_out_summary_cache.json")
def get_manual_muster_roll_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "manual_muster_roll_cache.json")
def get_work_reports_cache_file_path(): return os.path.join(CACHE_DIR, os.environ.get("ORG", "default"), "work_reports_cache.json")
# ---
def get_browser_user_data_directory(): return os.path.join(PROJ_DIR, "browser", "user_data", os.environ.get("ORG", "default"))
def get_browser_downloads_directory(): return os.path.join(PROJ_DIR, "downloads", os.environ.get("ORG", "default"))
# ---------------------------------------------------------------------------------------------------------------------
def init_paths():
"""
We will be segregating the data of the various organizations by their sub-dirs. This quick function ensures that
those segregated paths exist. Call it at the start of your script when you have set the organization name to the
temporary environment variable.
"""
for path in [
os.path.join(CACHE_DIR, os.environ.get("ORG", "default")),
os.path.join(PROJ_DIR, "browser", "user_data", os.environ.get("ORG", "default")),
os.path.join(PROJ_DIR, "downloads", os.environ.get("ORG", "default"))
]:
if not os.path.exists(path):
printer("Making", path)
files.make_directory(path)
# ---------------------------------------------------------------------------------------------------------------------
def kill_chrome() -> None:
"""
@@ -142,6 +181,7 @@ def kill_chrome() -> None:
# then wait if old processes were killed:
kill_count = CosecWeb.kill_chrome_processes()
if kill_count > 0: time.sleep(2.5)
printer(kill_count)
# ---------------------------------------------------------------------------------------------------------------------
@@ -150,7 +190,7 @@ def kill_chrome() -> None:
def get_muster_roll(
cosec_creds: dict,
on_date: datetime.datetime,
cache_file: str = MUSTER_ROLL_CACHE_FILE,
cache_file: str | Callable | None = None,
test_mode: bool = False
) -> bool:
@@ -163,6 +203,12 @@ def get_muster_roll(
:return: True if the automated fetch was successful, else False.
"""
# Figure out path(s):
if not isinstance(cache_file, str):
if isinstance(cache_file, Callable): cache_file = cache_file()
else: cache_file = get_muster_roll_cache_file_path()
printer(cache_file)
# Start by assuming failure:
success = False
report_path = None
@@ -183,8 +229,9 @@ def get_muster_roll(
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,
user_data_dir = get_browser_user_data_directory(),
downloads_dir = get_browser_downloads_directory(),
headless = False
)
# Perform the login:
@@ -308,7 +355,7 @@ async def sync_departments_to_tcaoff(
# 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)
success = await tcaoff_client.department_add(department_name = cosec_dept)
response["total"] += 1
if success: response["success"] += 1
else: response["fail"] += 1
@@ -428,7 +475,7 @@ 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,
cache_file: str | Callable | None = None,
test_mode: bool = False
) -> bool:
@@ -442,6 +489,12 @@ def get_in_out_summary(
:return: True if the automated fetch was successful, else False.
"""
# Figure out path(s):
if not isinstance(cache_file, str):
if isinstance(cache_file, Callable): cache_file = cache_file()
else: cache_file = get_in_out_summary_cache_file_path()
printer(cache_file)
# Start by assuming failure:
success = False
report_path = None
@@ -462,8 +515,9 @@ def get_in_out_summary(
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,
user_data_dir = get_browser_user_data_directory(),
downloads_dir = get_browser_downloads_directory(),
headless = False
)
# Perform the login:
@@ -516,6 +570,7 @@ def get_in_out_summary(
}
# Save the data to a JSON file:
printer("Saving", cache_file)
json.to_file(
file = cache_file,
python_data = report_data,
@@ -533,16 +588,30 @@ def get_in_out_summary(
def compute_work_done(
in_out_df: pd.DataFrame
in_out_df: pd.DataFrame,
working_hours_lookup: dict,
today: datetime.datetime = None
) -> 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.
:param working_hours_lookup: The lookup table that tells you how many hours a day is the employee expected to work.
:param today: Today's date to use as reference when calculating work done. Useful when the team member has not
logged out, and we need to assume calculations.
:return: A list of dicts that contain the information of all the work done.
"""
# Input cleaning:
now = date_time.get_current_ist_date_time()
now_ts = now.timestamp()
if today is None: today = date_time.get_current_ist_date_time()
today = today.replace(hour = 0, minute = 0, second = 0, microsecond = 0)
today_ts = today.timestamp()
if not isinstance(working_hours_lookup, dict):
working_hours_lookup = {}
# Create the structure that will be given as the output:
flattened_work_reports = []
@@ -565,8 +634,13 @@ def compute_work_done(
punch_ts = punch_dt.timestamp()
punch_loc = row[" Device/Source Detail"]
# Figure out the event type:
loc_event_type = None
if punch_loc.lower().find("out") >= 0: loc_event_type = "Out"
io_event_type = row["I/O Type"] if loc_event_type is None else loc_event_type
# Handle the first in time:
if row["I/O Type"] == "In":
if io_event_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
@@ -577,7 +651,7 @@ def compute_work_done(
work_reports[user_id][punch_date]["last_in_loc"] = punch_loc
# Handle the last out time:
if row["I/O Type"] == "Out":
if io_event_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
@@ -587,7 +661,7 @@ def compute_work_done(
for punch_date, punch_info in user_reports.items():
# Some defaults:
min_work_seconds = 10.0 * 60.0 * 60.0
min_work_seconds = working_hours_lookup.get("user_id", {}).get("daily_working_hours", DEFAULT_DAILY_WORKING_HOURS) * 60.0 * 60.0
work_seconds = 0.0
work_ot_seconds = 0.0
work_status = "A"
@@ -600,11 +674,17 @@ def compute_work_done(
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:
# When the user has a valid in-time, but no known out time.
# In such a case we must assume the work done.
# If the punch date is today's date, we assume the user is yet working till 'now'.
# If the punch date is one from the past, we assume the user worked till his minimum daily hours:
if first_in is not None and last_out is None:
last_out = first_in + min_work_seconds
work_seconds = min_work_seconds
if first_in >= today_ts:
last_out = now_ts
work_seconds = now_ts - first_in
else:
last_out = first_in + min_work_seconds
work_seconds = min_work_seconds
work_ot_seconds = 0.0
# When the user has neither an in-time, nor an out-time,
@@ -626,10 +706,15 @@ def compute_work_done(
# 'HD2' --> Half Day (2nd Half)
# '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 4.5 < work_hours <= 7.5: work_status = "HD1"
# --- [Multi-Layered Logic]:
# if work_hours > 10.0: work_status = "OT"
# elif 7.5 < work_hours <= 10.0: work_status = "P"
# elif 4.5 < work_hours <= 7.5: work_status = "HD1"
# else: work_status = "A"
# --- [Simple Present/Absent Logic]:
if first_in: work_status = "P"
else: work_status = "A"
# Save the data:
@@ -664,7 +749,7 @@ async def sync_attendance_to_tcaoff(
) -> Dict[str, int]:
"""
To mark attendance on TCAOFF from Cosec records.
: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.
@@ -673,15 +758,12 @@ async def sync_attendance_to_tcaoff(
:return: The dict that gives you the count of the successful and failed attendance marking API calls.
"""
# Figure out path(s):
work_reports_cache_file = get_work_reports_cache_file_path()
# 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()
@@ -698,6 +780,47 @@ async def sync_attendance_to_tcaoff(
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 a mapping of TCAOFF team working hours details
# where the key will be the unique employee id and the value will be a dict of their working hours expectations:
# print("TCAOFF TEAM:", json.to_string(tcaoff_teams[:10]))
working_hours_lookup = {}
for t in tcaoff_teams:
json_notes = json.from_string(t["json_notes"])
weekly_working_hours = json_notes.get("weeklyWorkingHours") or DEFAULT_WEEKLY_WORKING_HOURS
weekly_off_days = json_notes.get("weeklyOff") or [str(n) for n in range(7 - DEFAULT_WEEKLY_WORKING_DAYS)]
weekly_working_days = 7 - len(weekly_off_days)
working_hours_lookup[t["pseudonym"]] = {
"weekly_working_days": weekly_working_days,
"weekly_working_hours": weekly_working_hours,
"daily_working_hours": weekly_working_hours / weekly_working_days
}
# Convert the In/Out data to a DataFrame:
in_out_df = pd.DataFrame(cosec_in_out_summary["report"])
# Compute the work done:
new_work_reports = compute_work_done(in_out_df, working_hours_lookup = working_hours_lookup)
# TO AVOID DUPLICATE HITS:
# Now we save the work reports for next time,
# and then we check for the ones that have changed:
work_reports = []
if os.path.exists(work_reports_cache_file):
cached_work_reports = json.from_file(work_reports_cache_file)
hashed_new_work_reports = {
wr["user_id"] + "." + wr["work_date"] : wr
for wr in new_work_reports
}
hashed_old_work_reports = {
wr["user_id"] + "." + wr["work_date"]: wr
for wr in cached_work_reports
}
for k, v in hashed_new_work_reports.items():
if v == hashed_old_work_reports.get(k, {}): continue
work_reports.append(v)
else: work_reports = new_work_reports
json.to_file(work_reports_cache_file, new_work_reports)
# Create the tasks to fire:
tasks = []
for wr in work_reports:
@@ -722,6 +845,8 @@ async def sync_attendance_to_tcaoff(
"cosec": {
"workSeconds": wr["work_seconds"],
"workHours": wr["work_hours"],
"workOtSeconds": wr["work_ot_seconds"],
"workOtHours": wr["work_ot_hours"],
"firstIn": wr["first_in"],
"firstInLoc": wr["first_in_loc"],
"lastIn": wr["last_in"],
scripts/cron.py
+63 -27
View File
@@ -112,7 +112,7 @@ err_printer = IceCreamDebugger(prefix = "[ERR] Cron | ", includeContext = True)
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# *****************************************************************************************************************-
async def yesterday_cron(
@@ -139,7 +139,7 @@ async def yesterday_cron(
cosec_creds = cosec_creds,
from_dt = date_time.get_current_ist_date_time() - datetime.timedelta(days = 3),
to_dt = date_time.get_current_ist_date_time(),
cache_file = common.PREV_DAY_IN_OUT_SUMMARY_CACHE_FILE,
cache_file = common.get_prev_day_in_out_summary_cache_file_path,
test_mode = test_mode
)
@@ -148,7 +148,7 @@ async def yesterday_cron(
printer("IN/OUT SUMMARY: Sync'ing with TCAOFF")
await common.sync_attendance_to_tcaoff(
tcaoff_client = tcaoff_client,
cosec_in_out_summary = json.from_file(common.PREV_DAY_IN_OUT_SUMMARY_CACHE_FILE),
cosec_in_out_summary = json.from_file(common.get_prev_day_in_out_summary_cache_file_path()),
chunk_size = 10
)
@@ -198,7 +198,7 @@ async def today_cron(
try:
# MUSTER-ROLL:
# -----------
# Get the Muster Roll and then wait
# for the driver's resources to get freed:
success = common.get_muster_roll(
@@ -208,14 +208,14 @@ async def today_cron(
minute = 0,
second = 0
),
cache_file = common.MUSTER_ROLL_CACHE_FILE,
cache_file = common.get_muster_roll_cache_file_path,
test_mode = test_mode
)
# Sync data between Cosec and TCAOFF:
if success:
print("MUSTER ROLL: Sync'ing with TCAOFF")
cosec_muster_roll = json.from_file(common.MUSTER_ROLL_CACHE_FILE)
cosec_muster_roll = json.from_file(common.get_muster_roll_cache_file_path())
await common.sync_branches_to_tcaoff(
tcaoff_client = tcaoff_client,
cosec_muster_roll = cosec_muster_roll,
@@ -230,14 +230,14 @@ async def today_cron(
)
# IN-OUT SUMMARY:
# --------------
# Get the previous day's In/Out Summary and then wait
# for the driver's resources to get freed:
success = common.get_in_out_summary(
cosec_creds = cosec_creds,
from_dt = date_time.get_current_ist_date_time().replace(hour = 0, minute = 0, second = 0, microsecond = 0),
to_dt = date_time.get_current_ist_date_time(),
cache_file = common.IN_OUT_SUMMARY_CACHE_FILE,
cache_file = common.get_in_out_summary_cache_file_path,
test_mode = test_mode
)
@@ -246,7 +246,7 @@ async def today_cron(
printer("IN/OUT SUMMARY: Sync'ing with TCAOFF")
await common.sync_attendance_to_tcaoff(
tcaoff_client = tcaoff_client,
cosec_in_out_summary = json.from_file(common.IN_OUT_SUMMARY_CACHE_FILE),
cosec_in_out_summary = json.from_file(common.get_in_out_summary_cache_file_path()),
chunk_size = 10
)
@@ -268,6 +268,7 @@ async def today_cron(
async def set_scheduler(
cosec_creds: dict,
tcaoff_client: AsyncTheCAOffice,
immediate: bool = False,
test_mode: bool = False
) -> None:
@@ -276,25 +277,26 @@ async def set_scheduler(
foreground task happens in a loop.
:param cosec_creds: The credentials to use to log into Matrix COSEC.
:param tcaoff_client: The client to interact with TCAOFF.
:param test_mode: If True, the scheduler will be ignored and the process will be run once immediately.
:param immediate: If True, the process will run immediately first and then the scheduler will be set. If False, only
the scheduler will be set.
:param test_mode: If True, records will be fetched from cache instead of Cosec.
:return: None.
"""
# If in test mode:
if test_mode:
if immediate:
printer("Starting Test")
await yesterday_cron(
cosec_creds = copy.deepcopy(cosec_creds),
tcaoff_client = tcaoff_client,
# test_mode = test_mode
test_mode = test_mode
)
await today_cron(
cosec_creds = copy.deepcopy(cosec_creds),
tcaoff_client = tcaoff_client,
# test_mode = test_mode
test_mode = test_mode
)
printer("Test Done")
return
printer("Immediate run done.")
# If not in test mode, we continue with the scheduler.
# Create the scheduler:
@@ -352,24 +354,55 @@ if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument(
"--test", "--test-mode",
help = "Picks the last cache file instead of fetching records from COSEC.",
action = "store_true",
default = False,
)
ap.add_argument(
"--immediate",
help = "To do one run immediately first, then set the scheduler.",
action = "store_true",
default = False
)
ap.add_argument(
"--verbose",
action = "store_true",
default = False,
default = False
)
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()
# Explicitly mention the expected file paths for other devs to maintain:
print("PROJ. DIR. :", common.PROJ_DIR)
print("COSEC CREDS :", common.COSEC_CREDS_FILE)
print("TCAOFF CREDS:", common.TCAOFF_CREDS_FILE)
# Hold temporary environment variables:
os.environ["ORG"] = args.org
# Read required credentials:
cosec_creds = json.from_file(common.COSEC_CREDS_FILE)
tcaoff_creds = json.from_file(common.TCAOFF_CREDS_FILE)
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()
# Create the clients:
tcaoff_client = AsyncTheCAOffice(
@@ -380,8 +413,11 @@ if __name__ == "__main__":
)
# Schedule the activities:
asyncio.run(set_scheduler(
cosec_creds = cosec_creds,
tcaoff_client = tcaoff_client,
test_mode = args.test
))
asyncio.run(
set_scheduler(
cosec_creds = cosec_creds,
tcaoff_client = tcaoff_client,
immediate = args.immediate,
test_mode = args.test
)
)
scripts/manual_attendance.py
+39 -17
View File
@@ -84,8 +84,8 @@ from scripts import common
# Debugging:
printer = IceCreamDebugger(prefix = "Manual | ", includeContext = True)
err_printer = IceCreamDebugger(prefix = "[ERR] Manual | ", includeContext = True)
printer = IceCreamDebugger(prefix = "Manual Att | ", includeContext = True)
err_printer = IceCreamDebugger(prefix = "[ERR] Manual Att | ", includeContext = True)
# *****************************************************************************************************************
@@ -149,16 +149,17 @@ async def manual_attendance(
cosec_creds = cosec_creds,
from_dt = from_date,
to_dt = to_date,
cache_file = common.MANUAL_SUMMARY_CACHE_FILE,
cache_file = common.get_manual_in_out_summary_cache_file_path,
test_mode = test_mode
)
# Sync data between Cosec and TCAOFF:
if success:
printer("IN/OUT SUMMARY: Sync'ing with TCAOFF")
print("SYNC FROM:", common.get_manual_in_out_summary_cache_file_path())
await common.sync_attendance_to_tcaoff(
tcaoff_client = tcaoff_client,
cosec_in_out_summary = json.from_file(common.MANUAL_SUMMARY_CACHE_FILE),
cosec_in_out_summary = json.from_file(common.get_manual_in_out_summary_cache_file_path()),
chunk_size = 10
)
@@ -191,14 +192,14 @@ if __name__ == "__main__":
import argparse
ap = argparse.ArgumentParser()
ap.add_argument(
"--test",
"--test", "--test-mode",
action = "store_true",
default = False,
default = False
)
ap.add_argument(
"--verbose",
action = "store_true",
default = False,
default = False
)
ap.add_argument(
"--from-date",
@@ -210,28 +211,49 @@ if __name__ == "__main__":
help = "The date till when you want to sync attendance.",
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 :", common.COSEC_CREDS_FILE)
print("TCAOFF CREDS:", common.TCAOFF_CREDS_FILE)
print("COSEC CREDS :", cosec_creds_path)
print("TCAOFF CREDS:", tcaoff_creds_path)
# Read required credentials:
cosec_creds = json.from_file(common.COSEC_CREDS_FILE)
tcaoff_creds = json.from_file(common.TCAOFF_CREDS_FILE)
# Ensure that certain required paths exist:
common.init_paths()
# Date-handling:
args.from_date = date_time.parse_date_time(args.from_date, timezone = date_time.TIMEZONE_IST)
args.to_date = date_time.parse_date_time(args.to_date, timezone = date_time.TIMEZONE_IST)
if args.from_date > args.to_date: args.from_date, args.to_date = args.to_date, args.from_date
args.from_date = args.from_date.replace(hour = 0, minute = 0, second = 0, microsecond = 0)
args.to_date = args.to_date.replace(hour = 23, minute = 59, second = 59, microsecond = 999999)
print("From Date:", args.from_date)
print(" To Date:", args.to_date)
# dates_list = [args.from_date]
# while args.from_date < args.to_date:
# args.from_date = args.from_date + datetime.timedelta(days = 1)
# dates_list.append(args.from_date)
# print(f"DATES ({len(dates_list)}):", dates_list)
# Create the clients:
tcaoff_client = AsyncTheCAOffice(
scripts/manual_team_sync.py
+344
View File
@@ -0,0 +1,344 @@
"""
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,
)
)
tcaoff/async_tcaoff.py
+84 -2
View File
@@ -186,8 +186,15 @@ class AsyncTheCAOffice:
# ┛
@staticmethod
def remove_special_chars(s: str) -> str:
def remove_special_chars(s: str | Any) -> str:
# If the input is not a string,
# we convert that to a string:
if not isinstance(s, str):
s = str(s)
# If the input is a string,
# we ensure we remove unsupported chars:
return regex.replace(
text = s,
pattern = r"[^\w\d\- _]",
@@ -524,7 +531,7 @@ class AsyncTheCAOffice:
role: str,
username: str,
password: str,
applicant_notes: dict | list | str = None,
applicant_notes: dict | list = None,
raise_exception: bool = False,
) -> bool:
@@ -597,6 +604,81 @@ class AsyncTheCAOffice:
# Done here:
return success
async def team_update(
self,
user_id: int,
dept_id: int,
team_name: str,
email: str,
phone_no: str,
role: str,
applicant_notes: dict | list | str = None,
raise_exception: bool = False,
) -> bool:
"""
Update an existing team member.
:param user_id: The id of the team member.
:param dept_id: The id of the department that this team member is working in.
:param team_name: The name of the team member. This is the full display name. Can be the same as others.
:param email: The email id of the team member.
:param phone_no: The phone no. of the team member.
:param role: The role of the team member in the organization.
:param applicant_notes: Optional notes about the team member.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: True if successful, else False.
"""
# Start by assuming failure:
success = False
try:
# Make the API call:
response = await self._http_client.post(
url = self.TEAM_UPDATE_URL,
headers = {"X-Session-Token": self.__session_token},
json = {
"idUser": user_id,
"idDepartment": dept_id,
"name": team_name,
"email": email,
"phoneNo": phone_no,
"role": role,
"hierarchy": 1,
"applicantNotes": applicant_notes
}
)
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Team updated.",
user_id,
team_name,
)
success = True
# If the call failed:
else:
response_json = response.json()
self._err_printer(
"Team NOT updated.",
user_id,
team_name,
response_json
)
success = False
# If something goes wrong:
except Exception as exception:
self._err_printer("Team NOT updated.", exception)
if raise_exception: raise
success = False
# Done here:
return success
# ┏┓ ┓
# ┣┫╋╋┏┓┏┓┏┫┏┓┏┓┏┏┓
# ┛┗┗┗┗ ┛┗┗┻┗┻┛┗┗┗
thecaoffice-hr.py
+10
View File
@@ -366,6 +366,16 @@ Paylod = {
// Team update api
url : https://api.thecaoffice.com/team/update
//mandatory fields.
//1.name
//2.phoneNo
//3.role
//4.idDepartment
//5.email
//6.idUser
//Input json
final teamUpdateParam = {
'idUser': 12345, //Pass the id of the user whose details are being updated