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

(20251126) Now a separate Cron Script does the Selenium automation and caches the results in a JSON file.

Browse Source
This commit is contained in:
Khushal P Soonderji
2025-11-26 13:56:48 +05:30
parent f81693306f
commit d9090be865
7 changed files with 337 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -9,3 +9,4 @@ __pycache__/
/downloads/
/browser/user_data/
/browser/
/creds/
backend/api/blueprints/reports/in_out_report.py
+7 -57
View File
@@ -162,68 +162,18 @@ async def in_out_report_generate(
report_data = None
# Figure out the paths:
base_dir = files.get_parent_directory(
proj_dir = files.get_parent_directory(
files.get_file_directory(include_filename = False),
depth = 4
)
cache_file = os.path.join(proj_dir, "local", "cache", "in_out_summary_cache.json")
# Put together the directory for the drivers, the downloads, etc.:
chrome_driver_dir = os.path.join(base_dir, r"drivers/chrome")
user_data_dir = os.path.join(base_dir, r"browser/user_data")
downloads_dir = os.path.join(base_dir, r"downloads")
# Read the report:
try: report_data = json.from_file(cache_file)
except Exception as e: pass
# Show all the paths for debugging:
current_app.printer("PATHS:", chrome_driver_dir, user_data_dir, downloads_dir)
# Wait for the semaphore so that only one instance is automating Chrome at a time:
async with current_app.chrome_semaphore:
# First kill the previous processes:
kill_count = CosecWeb.kill_chrome_processes()
current_app.printer("Killed Chrome Procs.", kill_count)
if kill_count > 0: time.sleep(2.5)
# Create an instance of the automation object:
cosec = CosecWeb(
cosec_url = inbound_headers.cosecUrl,
username = inbound_headers.cosecUsername,
password = inbound_headers.cosecPassword,
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_in_out_summary(
initial_sleep = 1.0,
from_date = inbound_data.fromDate,
to_date = inbound_data.toDate,
group_ids = inbound_data.groupIds,
download_timeout = 60.0
)
# Log out to end the cycle:
cosec.logout()
# Close the browser window:
cosec.quit()
# If we didn't get any path, the download failed:
if report_path is None:
success = False
# If the data was loaded successfully:
else:
success = True
report_df = cosec.read_in_out_summary_xls(report_path)
##############
#Added by Yatmesh on 22 nov 17:40
report_df = report_df.where(report_df.notna(),None)
###################################
report_data = report_df.to_dict(orient = "records")
# Note down the status of success or failure:
success = True if report_data else False
# ┳┓
# ┣┫┏┓┏┏┓┏┓┏┓┏┏┓
backend/api/blueprints/reports/muster_roll.py
+11 -31
View File
@@ -175,39 +175,19 @@ async def muster_roll_report_generate(
success = False
report_data = None
# Get the muster roll data:
async with current_app.chrome_semaphore:
report_data = muster_roll.load(
cosec_url = inbound_headers.cosecUrl,
username = inbound_headers.cosecUsername,
password = inbound_headers.cosecPassword,
group_ids = inbound_data.groupIds,
on_date = inbound_data.onDate,
# Figure out the paths:
proj_dir = files.get_parent_directory(
files.get_file_directory(include_filename = False),
depth = 4
)
cache_file = os.path.join(proj_dir, "local", "cache", "muster_roll_cache.json")
# If data was loaded:
if report_data is not None:
# report_data = report_data.copy()
# report_data = report_data[[
# "User ID", "User Name", "Category Name",
# "Grade Name", "Branch Name", "Department Name",
# "Direct Reporting", "Level-1"
# ]]
# unique_branches = report_data["Branch Name"].unique().tolist()
# unique_depts = report_data[["Branch Name", "Department Name"]].drop_duplicates().to_dict(orient = "records")
# unique_reportees = report_data[["Branch Name", "Department Name", "Direct Reporting"]].drop_duplicates().to_dict(orient = "records")
# unique_combos = {
# "Branch Name": unique_branches,
# "Department Name": unique_depts,
# "Direct Reporting": unique_reportees
# }
# report_data = unique_combos
##############
# Added by Yatmesh on 25 nov 15:08
report_data = report_data.where(report_data.notna(), None)
###################################
report_data = report_data.to_dict(orient = "records")
success = True
# Read the report:
try: report_data = json.from_file(cache_file)
except Exception as e: pass
# Note down the status of success or failure:
success = True if report_data else False
# ┳┓
# ┣┫┏┓┏┏┓┏┓┏┓┏┏┓
cron/__init__.py
View File
cron/reports.py
+315
View File
@@ -0,0 +1,315 @@
"""
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
# To work with date and time:
import time
# 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.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()
report_path = cosec.get_in_out_summary(
initial_sleep = 1.0,
from_date = now_utc,
to_date = now_utc,
group_ids = cosec_creds["inOutConfig"]["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_in_out_summary_xls(report_path)
# Assume that the punch time in the data is IST data,
# then normalize it to UTC:
report_data["Punch Time"] = report_data["Punch Time"].apply(
lambda x: date_time.to_timezone(
datetime_object = date_time.as_if_timezone(
datetime_object = x,
timezone = date_time.TIMEZONE_IST
),
timezone = date_time.TIMEZONE_UTC
).timestamp()
)
# 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:
print("MUSTER ROLL FETCH FAILED!")
pass
# 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>"
},
"musterRollConfig": {
"groupIds": "2,3,4"
},
"inOutConfig": {
"groupIds": "2,3,4"
}
}
"""
loop(
cosec_creds = json.from_file(cosec_creds_file),
interval_seconds = 900 # ... Run every 15 mins.
)
local/cache/in_out_summary_cache.json
Vendored
+1
View File
File diff suppressed because one or more lines are too long
local/cache/muster_roll_cache.json
Vendored
+1
View File
File diff suppressed because one or more lines are too long