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

(20260202) Made TCAOFF APIs async.

Browse Source
This commit is contained in:
Khushal P Soonderji
2026-02-02 15:18:20 +05:30
parent e2049b683f
commit 1e98e53092
2 changed files with 638 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
helpers/async_tcaoff.py
+614
View File
@@ -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())