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

(20260110) Started writing code for working hours counting.

Browse Source
This commit is contained in:
Khushal P Soonderji
2026-01-10 15:26:06 +05:30
parent 5bec648313
commit b1ff42d126
4 changed files with 211 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cron/reports.py
+201 -106
View File
@@ -46,9 +46,6 @@ import datetime
# To work with tabulate data:
import pandas as pd
# To make API calls:
import requests
# Cosec-related:
from cosec_web.cosec_web import CosecWeb
@@ -112,19 +109,7 @@ downloads_dir = os.path.join(proj_dir, r"downloads")
# *****************************************************************************************************************
# def remove_special_chars(s: str) -> str:
#
# return regex.replace(
# text = s,
# pattern = r"[^\w\d\- _]",
# substitute_text = "_"
# )
# ---------------------------------------------------------------------------------------------------------------------
def add_branches_to_tcaoff(
def sync_branches_to_tcaoff(
tcaoff_creds: dict,
cosec_muster_roll: dict,
) -> Dict[str, int]:
@@ -160,7 +145,7 @@ def add_branches_to_tcaoff(
# ---------------------------------------------------------------------------------------------------------------------
def add_departments_to_tcaoff(
def sync_departments_to_tcaoff(
tcaoff_creds: dict,
cosec_muster_roll: dict,
) -> Dict[str, int]:
@@ -198,7 +183,7 @@ def add_departments_to_tcaoff(
# ---------------------------------------------------------------------------------------------------------------------
def add_teams_to_tcaoff(
def sync_teams_to_tcaoff(
tcaoff_creds: dict,
cosec_muster_roll: dict,
) -> Dict[str, int]:
@@ -217,8 +202,19 @@ def add_teams_to_tcaoff(
# Get the list of existing departments 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("Sync'd 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"]]
@@ -227,7 +223,7 @@ def add_teams_to_tcaoff(
# 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 tcaoff_teams:
if cosec_team["User ID"] not in synced_team_ids:
# Check the branch id and department id:
branch_id = tcaoff_branches_lookup.get(tcaoff.remove_special_chars(cosec_team["Branch Name"]))
@@ -242,43 +238,46 @@ def add_teams_to_tcaoff(
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": regex.replace(
text = cosec_team['User Name'],
pattern = r"[^\w\d\._]",
substitute_text = ""
).strip().lower() + "@velankanigroup.com",
"email": tcaoff_username + "@velankanigroup.com",
"phoneNo": "9876543210",
"role": cosec_team['Category Name'].strip(),
"username": cosec_team['User ID'].strip(),
"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["Category Name"].strip(),
username = cosec_team["User ID"].strip(),
password = "Vispl@123"
)
# 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
break
# if success: response["success"] += 1
# else: response["fail"] += 1
if response["total"] >= 100: break
# Done here:
print("TCAOFF-Cosec Depts. Sync.:", json.to_string(response))
@@ -288,6 +287,72 @@ def add_teams_to_tcaoff(
# ---------------------------------------------------------------------------------------------------------------------
def sync_attendance_to_tcaoff(
tcaoff_creds: dict,
cosec_in_out_summary: dict,
) -> Dict[str, int]:
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 = defaultdict(float)
# 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]:
user_allowed_events = in_out_df[
(in_out_df["User ID"] == user_id) &
(in_out_df["Event Status"] == "Allowed")
]
print("\n\n---\n\n")
print("USER ID:", user_id)
print("\n")
print("Allowed Events:")
print("--------------")
print(user_allowed_events[:10].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
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))
# ---------------------------------------------------------------------------------------------------------------------
def kill_chrome() -> None:
# First kill the previous processes,
@@ -340,7 +405,7 @@ def get_muster_roll(cosec_creds: dict) -> bool:
# # Close the browser window:
# cosec.quit()
report_path = r"D:\kps\PycharmProjects\cosec\downloads\Monthly_Details.xls"
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:
@@ -402,48 +467,50 @@ def get_in_out_summary(cosec_creds: dict) -> bool:
:return: True if the automated fetch was successful, else False.
"""
# Start by assuming failure:
success = 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()
# 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()
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:
@@ -503,8 +570,8 @@ def get_reports(
# # # 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 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:
@@ -516,41 +583,68 @@ def get_reports(
# branch_id = 124,
# dept_id = 851,
# reporting_to = 2857,
# team_name = "Test - 20260108",
# team_name = "Test - 20260110",
# email = "user@domain.com",
# phone_no = "9876543219",
# role = "Python Dev",
# username = "test.user.20260108",
# username = "test.user.20260110",
# password = "12345678",
# application_notes = {
# "cosec": {
# "User ID": "VI-1234",
# "User Name": "John Doe"
# }
# }
# )
# MUSTER-ROLL REPORT:
# # 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:
try:
# Get the Muster Roll and then wait
# for the driver's resources to get freed:
success = get_muster_roll(cosec_creds = cosec_creds)
success = get_in_out_summary(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)
# add_branches_to_tcaoff(
# tcaoff_creds = tcaoff_creds,
# cosec_muster_roll = cosec_muster_roll,
# )
# add_departments_to_tcaoff(
# tcaoff_creds = tcaoff_creds,
# cosec_muster_roll = cosec_muster_roll,
# )
add_teams_to_tcaoff(
print("IN/OUT SUMMARY: Sync'ing with TCAOFF")
cosec_in_out_summary = json.from_file(in_out_summary_cache_file)
sync_attendance_to_tcaoff(
tcaoff_creds = tcaoff_creds,
cosec_muster_roll = cosec_muster_roll,
cosec_in_out_summary = cosec_in_out_summary,
)
# If something goes wrong:
except Exception as e:
print("MUSTER ROLL FETCH FAILED!")
print("IN-OUT SUMMARY FETCH FAILED!")
raise
# # Get the In-Out Summary and then wait
@@ -570,6 +664,7 @@ def get_reports(
except Exception as e:
print("REPORT CRON FAILED LOOP!")
print("EXCEPTION:", e)
tcaoff.logout(tcaoff_creds)
raise
helpers/tcaoff.py
+8 -1
View File
@@ -374,6 +374,7 @@ def team_add(
role: str,
username: str,
password: str,
application_notes: dict | list | str = None
) -> bool:
"""
@@ -388,9 +389,14 @@ def team_add(
:param role: The role of the team member in the organization.
:param username: The unique username of the team member. Cannot be the same as anyone else.
:param password: The password for this team member's login.
:return:
:param application_notes: Optional notes about the team member.
:return: True if successful, else False.
"""
# Ensure that JSON notes are converted to string:
if isinstance(application_notes, (list, dict)):
application_notes = json.to_string(application_notes, no_space = True)
# Make the API call:
response = requests.post(
url = tcaoff_creds["urls"]["teamAdd"],
@@ -406,6 +412,7 @@ def team_add(
"username": username,
"password": password,
"hierarchy": 1,
"applicantNotes": application_notes
}
)
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