339 lines
12 KiB
Python
339 lines
12 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
CREATED: Wed, 26th Nov, 2025
|
|
UPDATED: Wed, 26th Nov, 2025
|
|
|
|
OBJECTIVE:
|
|
|
|
To achieve so-and-so-objective...
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
|
|
import pandas as pd
|
|
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# For system-level activities:
|
|
import os
|
|
|
|
# To work with date and time:
|
|
import time
|
|
import datetime
|
|
|
|
# Cosec-related:
|
|
from cosec_web.cosec_web import CosecWeb
|
|
|
|
# My utils:
|
|
from utils_v2.system import files
|
|
from utils_v2.string import json
|
|
from utils_v2.date_time import date_time
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** 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")
|
|
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")
|
|
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 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):
|
|
|
|
# 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()
|
|
|
|
# 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")
|
|
|
|
# Save the data to a JSON file:
|
|
json.to_file(
|
|
file = muster_roll_cache_file,
|
|
python_data = {
|
|
"ts": date_time.get_current_utc_date_time(as_string = False).timestamp(),
|
|
"report": report_data
|
|
},
|
|
no_space = True
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def get_in_out_summary(cosec_creds: dict):
|
|
|
|
# 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()
|
|
|
|
# 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 = in_out_summary_cache_file,
|
|
python_data = report_data,
|
|
no_space = True
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def get_reports(cosec_creds: dict) -> None:
|
|
|
|
# Get the Muster Roll and then wait
|
|
# for the driver's resources to get freed:
|
|
# try:
|
|
# get_muster_roll(cosec_creds = cosec_creds)
|
|
# time.sleep(2.5)
|
|
# except Exception as e:
|
|
# print("MUSTER ROLL FETCH FAILED!")
|
|
|
|
# Get the In-Out Summary and then wait
|
|
# for the driver's resources to get freed:
|
|
try:
|
|
get_in_out_summary(cosec_creds = cosec_creds)
|
|
time.sleep(2.5)
|
|
except Exception as e:
|
|
print("IN-OUT SUMMARY FETCH FAILED!")
|
|
raise e
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def loop(
|
|
cosec_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)
|
|
time.sleep(interval_seconds)
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** 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 the credentials.
|
|
# It should be in the format:
|
|
"""
|
|
{
|
|
"creds": {
|
|
"url": "http://x.x.x.x/COSEC/",
|
|
"username": "<usr>",
|
|
"password": "<pwd>"
|
|
},
|
|
"generalConfig": {
|
|
"pollInterval": 3600,
|
|
"timezone": "Asia/Kolkata",
|
|
"timedelta": {
|
|
"days": 0,
|
|
"hours": 24,
|
|
"minutes": 0,
|
|
"seconds": 0
|
|
}
|
|
},
|
|
"musterRollConfig": {
|
|
"groupIds": ["2", "3", "4"]
|
|
},
|
|
"inOutConfig": {
|
|
"groupIds": ["2", "3", "4"]
|
|
}
|
|
}
|
|
"""
|
|
|
|
cosec_creds = json.from_file(cosec_creds_file)
|
|
loop(
|
|
cosec_creds = cosec_creds,
|
|
interval_seconds = cosec_creds["generalConfig"]["pollInterval"],
|
|
)
|