diff --git a/cosec_web/cosec_web.py b/cosec_web/cosec_web.py index f97bb98..c768330 100644 --- a/cosec_web/cosec_web.py +++ b/cosec_web/cosec_web.py @@ -1245,31 +1245,31 @@ if __name__ == "__main__": print("User Data :", user_data_dir) print("Downloads :", downloads_dir) - # # Create an instance of the automation object: - # cosec = CosecWeb( - # cosec_url = "http://103.89.8.21/COSEC", - # username = input("Username : "), - # password = input("Password : "), - # driver_dir = chrome_driver_dir, - # user_data_dir = user_data_dir, - # downloads_dir = downloads_dir, - # ) + # Create an instance of the automation object: + cosec = CosecWeb( + cosec_url = "http://103.89.8.21/COSEC", + username = input("Username : "), + password = input("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: - # in_out_report_path = cosec.get_in_out_summary( - # initial_sleep = 1.0, - # from_date = datetime.datetime.now() - datetime.timedelta(days = 1), - # to_date = datetime.datetime.now(), - # group_ids = [ - # "2", # ... Velankani Information Systems Pvt Ltd - # "3", # ... Velankani Bydesign India Pvt Ltd - # "4", # ... Velankani Electronics & Automotive Pvt Ltd - # ], - # download_timeout = 60.0 - # ) + # Perform the login: + cosec.login(initial_sleep = 2.5) + + # Get the in/out report: + in_out_report_path = cosec.get_in_out_summary( + initial_sleep = 1.0, + from_date = datetime.datetime.now() - datetime.timedelta(days = 1), + to_date = datetime.datetime.now(), + group_ids = [ + "2", # ... Velankani Information Systems Pvt Ltd + "3", # ... Velankani Bydesign India Pvt Ltd + "4", # ... Velankani Electronics & Automotive Pvt Ltd + ], + download_timeout = 60.0 + ) # Convert the In/Out Summary to a Pandas DF: in_out_report_path = input("File Path : ") diff --git a/helpers/async_tcaoff.py b/helpers/async_tcaoff.py new file mode 100644 index 0000000..2dda0b4 --- /dev/null +++ b/helpers/async_tcaoff.py @@ -0,0 +1,614 @@ +""" + + 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: + LOGIN_URL = r"https://api.thecaoffice.com/ca/login" + LOGOUT_URL = r"https://api.thecaoffice.com/user/logout" + BRANCH_LIST_URL = r"https://api.thecaoffice.com/commons/branch/list" + BRANCH_ADD_URL = r"https://api.thecaoffice.com/commons/branch/add" + DEPT_LIST_URL = r"https://api.thecaoffice.com/commons/departments/list" + DEPT_ADD_URL = r"https://api.thecaoffice.com/commons/departments/add" + TEAM_LIST_URL = r"https://api.thecaoffice.com/team/list" + TEAM_ADD_URL = r"https://api.thecaoffice.com/team/add" + TEAM_UPDATE_URL = r"https://api.thecaoffice.com/team/update" + ATTENDANCE_LIST_URL = r"https://api.thecaoffice.com/user/attendance/register" + ATTENDANCE_MARK_URL = r"https://api.thecaoffice.com/user/attendance/mark" + + 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. + """ + + # # Ensure that JSON notes are converted to string: + # if isinstance(applicant_notes, (list, dict)): + # applicant_notes = json.to_string(applicant_notes, no_space = True) + + # Make the API call: + response = requests.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 = requests.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() + print("TCAOFF Attendance-Mark:", json.to_string(response_json)) + except Exception as e: + print("TCAOFF Attendance-Mark:", e) + print("TCAOFF Attendance-Mark:", 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 + + +# ***************************************************************************************************************** +# ***** **** +# *** 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() + branches = await client.branch_list() + await client.logout() + + asyncio.run(main())