""" AUTHOR: Khushal P Soonderji DATE: CREATED: Mon., 2nd Feb, 2026 UPDATED: N/A OBJECTIVE: To 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 import datetime # To work with tabulate data: import pandas as pd # To make API calls: import requests import httpx # 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.string import regex from utils_v2.date_time import date_time # To work with datatypes: from typing import List, Dict, Any, Literal # For debugging: from icecream import IceCreamDebugger # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncTheCAOffice: # URLs: BASE_URL = r"https://api.thecaoffice.com" LOCALHOST_URL = r"http://127.0.0.1:5205" # --- LOGIN_URL = f"{BASE_URL}/ca/login" LOGOUT_URL = f"{BASE_URL}/user/logout" BRANCH_LIST_URL = f"{BASE_URL}/commons/branch/list" BRANCH_ADD_URL = f"{BASE_URL}/commons/branch/add" DEPT_LIST_URL = f"{BASE_URL}/commons/departments/list" DEPT_ADD_URL = f"{BASE_URL}/commons/departments/add" TEAM_LIST_URL = f"{BASE_URL}/team/list" TEAM_ADD_URL = f"{BASE_URL}/team/add" TEAM_UPDATE_URL = f"{BASE_URL}/team/update" ATTENDANCE_MARK_URL = f"{BASE_URL}/user/attendance/mark" ATTENDANCE_REGISTER_URL = f"{BASE_URL}/user/attendance/register" def __init__( self, username: str, password: str, debug: bool = True, debug_prefix: str = "A. TCAOFF | ", debug_only_errors: bool = False ) -> None: # Save the credentials: self.username = username self.__password = password self.__session_token = None self.__user_id = None # Create an async HTTP client: self._http_client = httpx.AsyncClient( limits = httpx.Limits( max_connections = 50, max_keepalive_connections = 25, ), timeout = httpx.Timeout( pool = 60.0, connect = 5.0, write = 15.0, read = 180.0 ), headers = None ) # Create the debugging object: self._printer = IceCreamDebugger( prefix = debug_prefix, includeContext = True ) self._err_printer = IceCreamDebugger( prefix = f"[ERR] {debug_prefix}", includeContext = True ) if debug: self.enable_debug() else: self.disable_debug() if debug_only_errors: self.debug_only_errors() else: self.debug_everything() # ┳┓ ┓ • # ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓ # ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫ # ┛ ┛ ┛ def enable_debug(self) -> None: self._printer.enable() self._err_printer.enable() self._printer("Debug enabled!") def disable_debug(self) -> None: self._printer("Debug disabled!") self._printer.disable() self._err_printer.disable() def debug_only_errors(self) -> None: self._printer.disable() self._err_printer.enable() self._err_printer("Debug only errors!") def debug_everything(self) -> None: self._printer.enable() self._err_printer.enable() self._printer("Debug everything!") # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ @staticmethod def remove_special_chars(s: str) -> str: return regex.replace( text = s, pattern = r"[^\w\d\- _]", substitute_text = "_" ) # ┏┓ • # ┗┓┏┓┏┏┓┏┓┏┓ # ┗┛┗ ┛┛┗┗┛┛┗ async def login(self) -> bool: """ Log in to TCAOFF. :return: True if logged in, else False. """ # Make the API call: response = await self._http_client.post( url = self.LOGIN_URL, json = { "username": self.username, "password": self.__password, "mode": "cosec" } ) # If login succeeded: if response.status_code in [200]: response_json = response.json() self.__session_token = response_json["sessionToken"] self.__user_id = 1234 self._printer("Logged in.") return True # If login failed: else: self._err_printer( "Log-in failed.", response.status_code ) return False async def logout(self) -> bool: """ Log out from TCAOFF. :return: True if logged out, else False. """ # Make the API call: response = await self._http_client.post( url = self.LOGOUT_URL, json = {"username": self.__session_token} ) # Clear the session details from the in-mem creds: self.__session_token = None self.__user_id = None # If logout succeeded: if response.status_code in [200]: self._printer("Logged out.") return True # If login failed: else: self._err_printer( "Log-out failed.", response.status_code ) return False # ┳┓ ┓ # ┣┫┏┓┏┓┏┓┏┣┓┏┓┏ # ┻┛┛ ┗┻┛┗┗┛┗┗ ┛ async def branch_list(self) -> List[Dict[str, Any]] | None: """ List the existing branches. :return: A list of dictionaries where each dictionary describes on branch. None if the API call fails. """ # Make the API call: response = await self._http_client.post( url = self.BRANCH_LIST_URL, headers = {"X-Session-Token": self.__session_token}, json = {"idUser": self.__user_id} ) # If the call succeeded: if response.status_code in [200]: response_json = response.json() branches = response_json["data"]["rs0"] self._printer("Branches listed.") return branches # If the call failed: else: self._err_printer("Branch-list failed.") return None async def branch_add( self, branch_name: str ) -> bool: """ Add a new branch. :param branch_name: The name of the branch to add. :return: True if logged out, else False. """ # Make the API call: response = await self._http_client.post( url = self.BRANCH_ADD_URL, headers = {"X-Session-Token": self.__session_token}, json = { "idUser": self.__user_id, "branchName": branch_name } ) # If the call succeeded: if response.status_code in [200]: self._printer( "Branch added.", branch_name ) return True # If the call failed: else: response_json = response.json() self._err_printer( "Branch NOT added.", branch_name, response_json ) return False # ┳┓ # ┃┃┏┓┏┓┏┓┏┓╋┏┳┓┏┓┏┓╋ # ┻┛┗ ┣┛┗┻┛ ┗┛┗┗┗ ┛┗┗ # ┛ async def department_list(self) -> List[Dict[str, Any]] | None: """ List the existing departments. :return: A list of dictionaries where each dictionary describes one department. None if the API call fails. """ # Make the API call: response = await self._http_client.post( url = self.DEPT_LIST_URL, headers = {"X-Session-Token": self.__session_token}, json = {"idUser": self.__user_id} ) # If the call succeeded: if response.status_code in [200]: response_json = response.json() depts = response_json["data"]["rs0"] self._printer("Depts. listed.") return depts # If the call failed: else: self._err_printer("Dept.-list failed.") return None async def department_add( self, department_name: str ) -> bool: """ Add a new department. :param department_name: The name of the department to add. :return: True if logged out, else False. """ # Make the API call: response = await self._http_client.post( url = self.DEPT_ADD_URL, headers = {"X-Session-Token": self.__session_token}, json = { "idUser": self.__user_id, "departmentName": department_name } ) # If the call succeeded: if response.status_code in [200]: self._printer( "Dept. added.", department_name ) return True # If the call failed: else: response_json = response.json() self._err_printer( "Dept. NOT added.", department_name, response_json ) return False # ┏┳┓ # ┃ ┏┓┏┓┏┳┓ # ┻ ┗ ┗┻┛┗┗ async def team_list(self) -> List[Dict[str, Any]] | None: """ List the existing team members. :return: A list of dictionaries where each dictionary describes one team-member. None if the API call fails. """ # Make the API call: response = await self._http_client.post( url = self.TEAM_LIST_URL, headers = {"X-Session-Token": self.__session_token}, json = {"idUser": self.__user_id} ) # If the call succeeded: if response.status_code in [200]: response_json = response.json() teams = response_json["data"]["rs0"] self._printer("Team listed.") return teams # If the call failed: else: self._err_printer("Team-list failed.") return None async def team_add( self, branch_id: int, dept_id: int, reporting_to: int | None, team_name: str, email: str, phone_no: str, role: str, username: str, password: str, applicant_notes: dict | list | str = None ) -> bool: """ Add a new team member. :param dept_id: The id of the department that this team member is working in. :param branch_id: The id of the branch that this team member is working in. :param reporting_to: The id of the senior to whom this team member will report. :param team_name: The name of the team member. This is the full display name. Can be the same as others. :param email: The email id of the team member. :param phone_no: The phone no. of the team member. :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. :param applicant_notes: Optional notes about the team member. :return: True if successful, else False. """ # Make the API call: response = await self._http_client.post( url = self.TEAM_ADD_URL, headers = {"X-Session-Token": self.__session_token}, json = { "branchId": branch_id, "idDepartment": dept_id, "reportingTo": reporting_to, "name": team_name, "email": email, "phoneNo": phone_no, "role": role, "username": username, "password": password, "hierarchy": 1, "applicantNotes": applicant_notes } ) # If the call succeeded: if response.status_code in [200]: self._printer( "Team added.", team_name, username, ) return True # If the call failed: else: response_json = response.json() self._err_printer( "Team NOT added.", team_name, username, response_json ) return False # ┏┓ ┓ # ┣┫╋╋┏┓┏┓┏┫┏┓┏┓┏┏┓ # ┛┗┗┗┗ ┛┗┗┻┗┻┛┗┗┗ async def attendance_mark( self, user_id: int, status: Literal["P", "H", "A", "OT"], over_time: int | float, attendance_date: datetime.datetime | None = None, json_notes: dict = None ) -> bool: """ Add a new department. :param user_id: The id of the user whose attendance is being marked. :param status: The status of the attendance. 1. "P" for present, 2. "H" for half-day, 3. "A" for absent, 4. "OT" for over-time. :param over_time: The amount of over-time work in hours. :param attendance_date: The date of the attendance. If not given, today's date will be used. :param json_notes: Optional notes about the attendance. :return: True if the attendance was marked, else False. """ # Prepare the payload: json_payload = { "date": ( date_time.parse_date_time(attendance_date) or date_time.get_current_date_time() ).strftime("%Y-%m-%d"), "idUser": user_id, "status": status if over_time <= 0.0 else "OT", "ot": over_time, } # Add JSON notes if needed: if json_notes: json_payload["jsonNotes"] = json_notes # Make the API call: response = await self._http_client.post( url = self.ATTENDANCE_MARK_URL, headers = {"X-Session-Token": self.__session_token}, json = json_payload ) # Debugging: if response.status_code not in [200]: try: response_json = response.json() self._printer("TCAOFF Attendance-Mark", response_json) except Exception as e: self._printer("TCAOFF Attendance-Mark", e, response.content) print("Payload:", json.to_string(json_payload)) # If the call succeeded: if response.status_code in [200]: self._printer( "Attendance marked.", user_id, ) return True # If the call failed: else: try: response_json = response.json() except Exception as e: response_json = response.content self._err_printer( "Attendance NOT added.", user_id, response_json ) return False async def attendance_register( self, target_month: datetime.datetime, ) -> List[dict] | None: """ To get the whole attendance register for a month for all the employees of an entity. :param target_month: The datetime which indicates the year and month in which the attendance needs to be checked. :return: The attendance register records if successful, else None. """ # Start by assuming failure: attendance_register = None # Make the API call: json_payload = {"month": target_month.strftime("%Y-%m-%d")} response = await self._http_client.post( url = self.ATTENDANCE_REGISTER_URL, headers = {"X-Session-Token": self.__session_token}, json = json_payload, ) # If the call succeeded: if response.status_code in [200]: self._printer( "Attendance register fetched.", target_month, ) attendance_register = response.json()["data"]["rs0"] # If the call failed: else: try: response_data = response.json() except Exception as e: response_data = response.content self._err_printer( "Attendance register NOT fetched.", target_month, response_data ) attendance_register = None # Done here: return attendance_register # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing yet # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": import asyncio from getpass import getpass async def main(): client = AsyncTheCAOffice( username = input("Username: "), password = input("Password: "), ) await client.login() attendance = await client.attendance_register( target_month = datetime.datetime.now() - datetime.timedelta(months = 1), ) print("ATTENDANCE:", json.to_string(attendance)) await client.logout() asyncio.run(main())