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

(20260212) Moved to async version of cron reports.

Browse Source
This commit is contained in:
Khushal P Soonderji
2026-02-12 11:00:19 +05:30
parent 94293cf2fc
commit beca5e3127
3 changed files with 356 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/common.py
+299 -11
View File
@@ -146,6 +146,268 @@ def kill_chrome() -> None:
# ---------------------------------------------------------------------------------------------------------------------
def get_muster_roll(
cosec_creds: dict,
on_date: datetime.datetime,
cache_file: str = MUSTER_ROLL_CACHE_FILE,
test_mode: bool = False
) -> 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 on_date: The date on which you want .
:param cache_file: The cache file to store the results in.
:param test_mode: If set to True, a past file will be used instead of getting new reports from Cosec.
:return: True if the automated fetch was successful, else False.
"""
# Start by assuming failure:
success = False
# If test mode:
if test_mode:
report_path = TEST_MODE_MUSTER_ROLL_FILE_PATH
# When not working in test mode:
else:
# 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 = on_date,
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")
# Loop through the report for inferred data.
# Add the ids of the "Direct Reporting" and "Level-1" values:
for person in report_data:
person["Direct Reporting ID"] = None
person["Level-1 ID"] = None
for check_against in report_data:
if person["Direct Reporting"] == check_against["User Name"]:
person["Direct Reporting ID"] = check_against["User ID"]
if person["Level-1"] == check_against["User Name"]:
person["Level-1 ID"] = check_against["User ID"]
# Sort by hierarchy (BFS):
pass
print(f"MUSTER ROLL ({len(report_data)}):", json.to_string(report_data))
# Save the data to a JSON file:
json.to_file(
file = cache_file,
python_data = {
"ts": date_time.get_current_utc_date_time(as_string = False).timestamp(),
"report": report_data
},
no_space = True
)
# Note down success:
success = True
# Done here:
return success
# ---------------------------------------------------------------------------------------------------------------------
async def sync_branches_to_tcaoff(
tcaoff_client: AsyncTheCAOffice,
cosec_muster_roll: dict,
) -> Dict[str, int]:
# Start with a basic response structure:
response = defaultdict(int)
# Get the list of existing branches from TCAOFF:
tcaoff_branches = await tcaoff_client.branch_list()
tcaoff_branches = [_["branch_name"].lower() for _ in tcaoff_branches]
tcaoff_branches = list(set(tcaoff_branches))
# Get only the branch names from Cosec:
cosec_branches = [tcaoff_client.remove_special_chars(_["Branch Name"]) for _ in cosec_muster_roll["report"]]
cosec_branches = list(set(cosec_branches))
# Loop through the data from Cosec and add missing branches to TCAOFF:
for cosec_branch in cosec_branches:
if cosec_branch.lower() not in tcaoff_branches:
success = await tcaoff_client.branch_add(branch_name = cosec_branch)
response["total"] += 1
if success: response["success"] += 1
else: response["fail"] += 1
# Done here:
return response
# ---------------------------------------------------------------------------------------------------------------------
async def sync_departments_to_tcaoff(
tcaoff_client: AsyncTheCAOffice,
cosec_muster_roll: dict,
) -> Dict[str, int]:
# Start with a basic response structure:
response = defaultdict(int)
# Get the list of existing departments from TCAOFF:
tcaoff_depts = await tcaoff_client.department_list()
tcaoff_depts = [_["department_name"].lower() for _ in tcaoff_depts]
tcaoff_depts = list(set(tcaoff_depts))
# Get only the department names from Cosec:
cosec_depts = [tcaoff_client.remove_special_chars(_["Department Name"]) for _ in cosec_muster_roll["report"]]
cosec_depts = list(set(cosec_depts))
# Loop through the data from Cosec and add missing departments to TCAOFF:
for cosec_dept in cosec_depts:
if cosec_dept.lower() not in tcaoff_depts:
success = tcaoff_client.department_add(department_name = cosec_dept)
response["total"] += 1
if success: response["success"] += 1
else: response["fail"] += 1
# Done here:
return response
# ---------------------------------------------------------------------------------------------------------------------
async def sync_teams_to_tcaoff(
tcaoff_client: AsyncTheCAOffice,
cosec_muster_roll: dict,
) -> Dict[str, int]:
# Start with a basic response structure:
response = defaultdict(int)
# Map out branch ids:
tcaoff_branches = await tcaoff_client.branch_list()
tcaoff_branches_lookup = {d["branch_name"].lower(): d["branch_id"] for d in tcaoff_branches}
# Map out dept. ids:
tcaoff_depts = await tcaoff_client.department_list()
tcaoff_depts_lookup = {d["department_name"].lower(): d["department_id"] for d in tcaoff_depts}
# 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()
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:
app_notes = {}
cosec_notes = app_notes.get("cosec", {})
print("COSEC NOTES:", cosec_notes)
if cosec_notes:
user_id = cosec_notes.get("User ID")
if not user_id: user_id = cosec_notes["UserID"]
synced_team_ids.append(user_id)
tcaoff_teams = [_["pseudonym"] for _ in tcaoff_teams]
tcaoff_teams = list(set(tcaoff_teams))
# Loop through the data from Cosec and add missing team-members 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
# Remove special chars:
branch_name = tcaoff_client.remove_special_chars(cosec_team["Branch Name"]).lower()
dept_name = tcaoff_client.remove_special_chars(cosec_team["Branch Name"]).lower()
# Check the branch id and department id:
branch_id = tcaoff_branches_lookup.get(branch_name)
dept_id = tcaoff_depts_lookup.get(dept_name)
if branch_id is None:
err_printer("TEAM SYNC ERR", "Branch Not Found in TCAOFF", branch_name)
response["fail"] += 1
continue
if dept_id is None:
err_printer("TEAM SYNC ERR", "Dept. Not Found in TCAOFF", dept_name)
response["fail"] += 1
continue
# Add the team:
tcaoff_username = regex.replace(
text = cosec_team["User Name"],
pattern = r"[^\w\d\._]",
substitute_text = ""
).strip().lower()
# print("Need to Add:", json.to_string(team_json))
success = await tcaoff_client.team_add(
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",
applicant_notes = {"cosec": cosec_team}
)
if success: response["success"] += 1
else:
err_printer("COULDN'T ADD", cosec_team)
response["fail"] += 1
# Done here:
printer("TCAOFF-Cosec Depts. Sync.:", json.to_string(response))
return response
# ---------------------------------------------------------------------------------------------------------------------
def get_in_out_summary(
cosec_creds: dict,
from_dt: date_time.datetime = None,
@@ -257,6 +519,13 @@ def compute_work_done(
in_out_df: pd.DataFrame
) -> 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.
:return: A list of dicts that containthe information of all the work done.
"""
# Create the structure that will be given as the output:
flattened_work_reports = []
@@ -268,6 +537,10 @@ def compute_work_done(
# and figure out the first in and last out times:
for index, row in in_out_df.iterrows():
# Ignore if the event wasn't a success:
if row["Event Status"] != "Allowed":
continue
# Extract user and date:
user_id = row["User ID"]
punch_dt = date_time.parse_date_time(row["Punch Time"], timezone = date_time.TIMEZONE_IST)
@@ -354,6 +627,18 @@ async def sync_attendance_to_tcaoff(
verbose: bool = False
) -> Dict[str, int]:
"""
: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.
:param verbose: If True, internal debugging print will be more verbose.
:return: The dict that gives you the count of the successful and failed attendance marking API calls.
"""
# Results:
results = defaultdict(int)
# Convert the In/Out data to a DataFrame:
in_out_df = pd.DataFrame(cosec_in_out_summary["report"])
@@ -385,7 +670,7 @@ async def sync_attendance_to_tcaoff(
# Match it to the TCAOFF team-member:
tcaoff_team = cosec_id_to_tcaoff_team.get(cosec_user_id)
if tcaoff_team is None:
print(f"TCAOFF SYNC ERR: Cosesc User Id '{cosec_user_id}' not found in TCAOFF")
printer(f"TCAOFF SYNC ERR: Cosesc User Id '{cosec_user_id}' not found in TCAOFF")
continue
# Create the task for this attendance:
@@ -403,23 +688,26 @@ async def sync_attendance_to_tcaoff(
)
)
print("TASKS COUNT:", len(tasks))
printer("TASKS COUNT", len(tasks))
def chunks(lst, size = 10):
for i in range(0, len(lst), size):
yield lst[i:i + size]
results = []
count = 1
for chunk in chunks(tasks[-35:]):
print("Task Chunk:", count)
_res = await asyncio.gather(*chunk)
results += _res
count = 0
for chunk in chunks(tasks[:]):
count += 1
printer("Task Chunk:", count)
_res = await asyncio.gather(*chunk)
success = sum(_res)
failure = len(_res) - success
results["success"] += success
results["failure"] += failure
results["total"] += len(_res)
print(f"RESULTS ({len(results)}):", results)
print("SUCCESS:", sum(results))
# Done here:
printer(results)
return results
# *****************************************************************************************************************