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

(20260309) Added caching to avoid unchanged attendance calls and added logic to check for custom overtime thresholds.

Browse Source
This commit is contained in:
Khushal P Soonderji
2026-03-09 20:24:24 +05:30
parent 6952bf73f5
commit 172cfdc92d
7 changed files with 92 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cosec_web/cosec_web.py
+4
View File
@@ -838,6 +838,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 +1029,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)
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
+69 -16
View File
@@ -92,6 +92,7 @@ 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")
@@ -99,6 +100,11 @@ DOWNLOADS_DIR = os.path.join(PROJ_DIR, "downloads")
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)
@@ -533,16 +539,22 @@ def get_in_out_summary(
def compute_work_done(
in_out_df: pd.DataFrame
in_out_df: pd.DataFrame,
working_hours_lookup: dict
) -> 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.
:return: A list of dicts that contain the information of all the work done.
"""
# Input cleaning:
if not isinstance(working_hours_lookup, dict):
working_hours_lookup = {}
# Create the structure that will be given as the output:
flattened_work_reports = []
@@ -587,7 +599,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 +612,10 @@ 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:
if first_in is not None and last_out is None:
last_out = first_in + min_work_seconds
work_seconds = min_work_seconds
# 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 +637,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"
# ---
# 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"
# ---
if first_in: work_status = "P"
else: work_status = "A"
# Save the data:
@@ -664,7 +680,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.
@@ -676,12 +692,6 @@ async def sync_attendance_to_tcaoff(
# 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 +708,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 +773,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
+14 -6
View File
@@ -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:
@@ -355,6 +357,11 @@ if __name__ == "__main__":
action = "store_true",
default = False,
)
ap.add_argument(
"--immediate",
action = "store_true",
default = False,
)
ap.add_argument(
"--verbose",
action = "store_true",
@@ -383,5 +390,6 @@ if __name__ == "__main__":
asyncio.run(set_scheduler(
cosec_creds = cosec_creds,
tcaoff_client = tcaoff_client,
immediate = args.immediate,
test_mode = args.test
))