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

(20260110) Attendance-sync'ing ready with a manual attendance script also.

Browse Source
This commit is contained in:
Khushal P Soonderji
2026-01-13 18:03:21 +05:30
parent b1ff42d126
commit 74619bdec8
5 changed files with 695 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cron/attendance.py
+200
View File
@@ -0,0 +1,200 @@
"""
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
# 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(
"-date", "--d",
help = "The date on which you want to sync attendance.",
default = date_time.get_current_ist_date_time()
)
args = parser.parse_args()
print("Target Date:", args.d)
cosec_creds = json.from_file(cosec_creds_file)
tcaoff_creds = json.from_file(tcaoff_creds_file)
manual_attendance(
cosec_creds = cosec_creds,
tcaoff_creds = tcaoff_creds,
target_date = date_time.parse_date_time(
args.d,
date_formats = [
"%Y%m%d",
"%Y-%m-%d"
],
) + date_time.timedelta(
hours = 23,
minutes = 59
)
)
cron/reports.py
+417 -235
View File
@@ -38,6 +38,7 @@ sys.path.append("..")
# For system-level activities:
import os
import copy
# To work with date and time:
import time
@@ -62,6 +63,9 @@ from utils_v2.date_time import date_time
from typing import List, Dict, Any
from collections import defaultdict
# To run a cron-like scheduler:
from apscheduler.schedulers.background import BackgroundScheduler
# *****************************************************************************************************************
# ***** ****
@@ -77,6 +81,7 @@ 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")
@@ -119,7 +124,7 @@ def sync_branches_to_tcaoff(
# Get the list of existing branches from TCAOFF:
tcaoff_branches = tcaoff.branch_list(tcaoff_creds)
tcaoff_branches = [_["branch_name"] for _ in tcaoff_branches]
tcaoff_branches = [_["branch_name"].lower() for _ in tcaoff_branches]
tcaoff_branches = list(set(tcaoff_branches))
# Get only the branch names from Cosec:
@@ -128,7 +133,7 @@ def sync_branches_to_tcaoff(
# Loop through the data from Cosec and add missing branches to TCAOFF:
for cosec_branch in cosec_branches:
if cosec_branch not in tcaoff_branches:
if cosec_branch.lower() not in tcaoff_branches:
success = tcaoff.branch_add(
tcaoff_creds,
branch_name = cosec_branch
@@ -155,7 +160,7 @@ def sync_departments_to_tcaoff(
# Get the list of existing departments from TCAOFF:
tcaoff_depts = tcaoff.department_list(tcaoff_creds)
tcaoff_depts = [_["department_name"] for _ in tcaoff_depts]
tcaoff_depts = [_["department_name"].lower() for _ in tcaoff_depts]
tcaoff_depts = list(set(tcaoff_depts))
# print("TCAOFF DEPTS:", tcaoff_depts)
@@ -166,7 +171,7 @@ 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 not in tcaoff_depts:
if cosec_dept.lower() not in tcaoff_depts:
success = tcaoff.department_add(
tcaoff_creds,
department_name = cosec_dept
@@ -193,13 +198,13 @@ def sync_teams_to_tcaoff(
# Map out branch ids:
tcaoff_branches = tcaoff.branch_list(tcaoff_creds)
tcaoff_branches_lookup = {d["branch_name"]: d["branch_id"] for d in tcaoff_branches}
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"]:d["department_id"] for d in tcaoff_depts}
tcaoff_depts_lookup = {d["department_name"].lower():d["department_id"] for d in tcaoff_depts}
# Get the list of existing departments from TCAOFF:
# 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 = []
@@ -207,14 +212,14 @@ def sync_teams_to_tcaoff(
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"])
# print("NULL APP NOTES??:", t["json_notes"])
app_notes = {}
cosec_notes = app_notes.get("cosec", {})
print("COSEC NOTES:", cosec_notes)
# 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("Sync'd Team Ids:", synced_team_ids)
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"]]
@@ -225,9 +230,12 @@ def sync_teams_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"]))
dept_id = tcaoff_depts_lookup[tcaoff.remove_special_chars(cosec_team["Department Name"])]
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
@@ -243,41 +251,39 @@ def sync_teams_to_tcaoff(
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 = regex.replace(
# text = cosec_team['User Name'],
# pattern = r"[^\w\d\._]",
# substitute_text = ""
# ).strip().lower() + "@velankanigroup.com",
# phone_no = "9876543210",
# role = cosec_team["Grade Name"].strip(),
# username = cosec_team["User ID"].strip(),
# password = "Vispl@123",
# application_notes = {"cosec": cosec_team}
# )
response["total"] += 1
# if success: response["success"] += 1
# else: response["fail"] += 1
if response["total"] >= 100: break
# 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",
application_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))
@@ -292,8 +298,25 @@ def sync_attendance_to_tcaoff(
cosec_in_out_summary: dict,
) -> Dict[str, int]:
print("Hi, attendance!")
print(json.to_string(cosec_in_out_summary["report"][0]))
# 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"])
@@ -304,50 +327,144 @@ def sync_attendance_to_tcaoff(
print(f"USER IDS ({len(user_ids)}):", user_ids)
# Start with a blank user to time mapping:
user_id_to_time = defaultdict(float)
user_id_to_time = {}
# For each unique user id,
# Check the no of times when he successfully triggered the attendance-punching device:
for user_id in user_ids[:10]:
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[:10].to_string())
print(user_allowed_events.to_string())
print("\n")
print("Interpretation:")
print("--------------")
# 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
# 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():
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(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
}
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))
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
# 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))
# # 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"
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"],
)
if success: response["success"] += 1
else: response["fail"] += 1
# Done here:
print("TCAOFF-Cosec Attendance Sync.:", json.to_string(response))
return response
# ---------------------------------------------------------------------------------------------------------------------
@@ -364,48 +481,52 @@ def kill_chrome() -> None:
# ---------------------------------------------------------------------------------------------------------------------
def get_muster_roll(cosec_creds: dict) -> bool:
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()
# Force close other running Chrome processes:
kill_chrome()
report_path = r"D:\kps\PycharmProjects\cosec\cosec_web\sample_files\muster_roll.xls"
# 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:
@@ -441,7 +562,7 @@ def get_muster_roll(cosec_creds: dict) -> bool:
# Save the data to a JSON file:
json.to_file(
file = muster_roll_cache_file,
file = cache_file,
python_data = {
"ts": date_time.get_current_utc_date_time(as_string = False).timestamp(),
"report": report_data
@@ -459,58 +580,64 @@ def get_muster_roll(cosec_creds: dict) -> bool:
# ---------------------------------------------------------------------------------------------------------------------
def get_in_out_summary(cosec_creds: dict) -> bool:
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:
# now_utc = date_time.get_current_utc_date_time()
# from_dt = now_utc - 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 = now_utc
# 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()
# Start by assuming failure:
success = False
report_path = r"D:\kps\PycharmProjects\cosec\cosec_web\sample_files\in_out_summary.xls"
# 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:
@@ -542,7 +669,7 @@ def get_in_out_summary(cosec_creds: dict) -> bool:
# Save the data to a JSON file:
json.to_file(
file = in_out_summary_cache_file,
file = cache_file,
python_data = report_data,
no_space = True
)
@@ -557,106 +684,74 @@ def get_in_out_summary(cosec_creds: dict) -> bool:
# ---------------------------------------------------------------------------------------------------------------------
def get_reports(
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)
# # # TEST SECTION:
# # # Test out listing API calls:
# print("TCAOFF Branches:", json.to_string(tcaoff.branch_list(tcaoff_creds)))
# print("TCAOFF Depts.:", json.to_string(tcaoff.department_list(tcaoff_creds)))
# print("TCAOFF Team:", json.to_string(tcaoff.team_list(tcaoff_creds)))
# # TEST SECTION:
# # Test out listing API calls:
# tcaoff.branch_add(tcaoff_creds, branch_name = "Test - 20260108")
# tcaoff.department_add(tcaoff_creds, department_name = "Test - 20260108")
# tcaoff.team_add(
# tcaoff_creds,
# branch_id = 124,
# dept_id = 851,
# reporting_to = 2857,
# team_name = "Test - 20260110",
# email = "user@domain.com",
# phone_no = "9876543219",
# role = "Python Dev",
# username = "test.user.20260110",
# password = "12345678",
# application_notes = {
# "cosec": {
# "User ID": "VI-1234",
# "User Name": "John Doe"
# }
# }
# )
# # MUSTER-ROLL REPORT:
# try:
#
# # Get the Muster Roll and then wait
# # for the driver's resources to get freed:
# success = get_muster_roll(cosec_creds = cosec_creds)
#
# # Sync data between Cosec and TCAOFF:
# if success:
# print("MUSTER ROLL: Sync'ing with TCAOFF")
# cosec_muster_roll = json.from_file(muster_roll_cache_file)
# # 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:
# MUSTER-ROLL REPORT:
try:
# Get the Muster Roll and then wait
# for the driver's resources to get freed:
success = get_in_out_summary(cosec_creds = cosec_creds)
success = get_muster_roll(
cosec_creds = cosec_creds,
cache_file = muster_roll_cache_file
)
# Sync data between Cosec and TCAOFF:
if success:
print("IN/OUT SUMMARY: Sync'ing with TCAOFF")
cosec_in_out_summary = json.from_file(in_out_summary_cache_file)
sync_attendance_to_tcaoff(
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_in_out_summary = cosec_in_out_summary,
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
# # Get the In-Out Summary and then wait
# # for the driver's resources to get freed:
# try:
# success = get_in_out_summary(cosec_creds = cosec_creds)
# time.sleep(2.5)
# if success: pass
# except Exception as e:
# print("IN-OUT SUMMARY FETCH FAILED!")
# raise
# Log out from TCAOFF:
tcaoff.logout(tcaoff_creds)
@@ -665,25 +760,111 @@ def get_reports(
print("REPORT CRON FAILED LOOP!")
print("EXCEPTION:", e)
tcaoff.logout(tcaoff_creds)
raise
# ---------------------------------------------------------------------------------------------------------------------
def loop(
def yesterday_cron(
cosec_creds: dict,
tcaoff_creds: dict,
interval_seconds: int | float = 900
) -> None:
# Just keep fetching the reports in an infinite loop:
while True:
get_reports(
cosec_creds = cosec_creds,
tcaoff_creds = tcaoff_creds
"""
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)
]
)
time.sleep(interval_seconds)
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()
# *****************************************************************************************************************
@@ -748,14 +929,15 @@ if __name__ == "__main__":
"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"
"teamAdd": "https://api.thecaoffice.com/team/add",
"attendanceList": "https://api.thecaoffice.com/user/attendance/register",
"attendanceMark": "https://api.thecaoffice.com/user/attendance/mark"
}
}
"""
tcaoff_creds = json.from_file(tcaoff_creds_file)
loop(
set_scheduler(
cosec_creds = cosec_creds,
tcaoff_creds = tcaoff_creds,
interval_seconds = cosec_creds["generalConfig"]["pollInterval"],
)
helpers/tcaoff.py
+76 -9
View File
@@ -59,7 +59,7 @@ 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 typing import List, Dict, Any, Literal
# *****************************************************************************************************************
@@ -239,9 +239,10 @@ def branch_add(
}
)
# # Debugging:
# response_json = response.json()
# print("TCAOFF Branch-Add:", json.to_string(response_json))
# Debugging:
if response.status_code not in [200]:
response_json = response.json()
print("TCAOFF Branch-Add:", json.to_string(response_json))
# If the call succeeded:
if response.status_code in [200]:
@@ -312,9 +313,10 @@ def department_add(
}
)
# # Debugging:
# response_json = response.json()
# print("TCAOFF Dept-Add:", json.to_string(response_json))
# Debugging:
if response.status_code not in [200]:
response_json = response.json()
print("TCAOFF Dept-Add:", json.to_string(response_json))
# If the call succeeded:
if response.status_code in [200]:
@@ -417,8 +419,9 @@ def team_add(
)
# Debugging:
response_json = response.json()
print("TCAOFF Team-Add:", json.to_string(response_json))
if response.status_code not in [200]:
response_json = response.json()
print("TCAOFF Team-Add:", json.to_string(response_json))
# If the call succeeded:
if response.status_code in [200]:
@@ -431,6 +434,70 @@ def team_add(
return False
# ---------------------------------------------------------------------------------------------------------------------
def attendance_mark(
tcaoff_creds: dict,
user_id: int,
status: Literal["P", "H", "A", "OT"],
over_time: int | float,
attendance_date: datetime.datetime | None = None,
) -> bool:
"""
Add a new department.
:param tcaoff_creds: The set of credentials to use with the API call.
:param user_id: The id of the user whose attendance is being marked.
:param status: The status of the attendance.
1. "P" for present,
2. "H" for half-day,
3. "A" for absent,
4. "OT" for over-time.
:param over_time: The amount of over-time work in hours.
:param attendance_date: The date of the attendance. If not given, today's date will be used.
:return: True if logged out, else False.
"""
# Prepare the payload:
json_payload = {
"date": (
date_time.parse_date_time(attendance_date) or
date_time.get_current_date_time()
).strftime("%Y-%m-%d"),
"idUser": user_id,
"status": status if over_time <= 0.0 else "OT",
"ot": over_time,
}
# Make the API call:
response = requests.post(
url = tcaoff_creds["urls"]["attendanceMark"],
headers = {"X-Session-Token": tcaoff_creds["creds"]["sessionToken"]},
json = json_payload
)
# Debugging:
if response.status_code not in [200]:
try:
response_json = response.json()
print("TCAOFF Attendance-Mark:", json.to_string(response_json))
except Exception as e:
print("TCAOFF Attendance-Mark:", e)
print("TCAOFF Attendance-Mark:", response.content)
print("Payload:", json.to_string(json_payload))
# If the call succeeded:
if response.status_code in [200]:
print(f"TCAOFF: attendance for '{user_id}' marked")
return True
# If the call failed:
else:
print(f"TCAOFF: attendance for '{user_id}' NOT marked")
return False
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
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