864 lines
28 KiB
Python
864 lines
28 KiB
Python
"""
|
|
|
|
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 = 60.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 | Any) -> str:
|
|
|
|
# If the input is not a string,
|
|
# we convert that to a string:
|
|
if not isinstance(s, str):
|
|
s = str(s)
|
|
|
|
# If the input is a string,
|
|
# we ensure we remove unsupported chars:
|
|
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,
|
|
raise_exception: bool = False,
|
|
) -> List[Dict[str, Any]] | None:
|
|
|
|
"""
|
|
List the existing branches.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: A list of dictionaries where each dictionary describes on branch. None if the API call fails.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
branches = None
|
|
|
|
try:
|
|
|
|
# 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.")
|
|
|
|
# If the call failed:
|
|
else:
|
|
self._err_printer("Branch-list failed.")
|
|
branches = None
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer("Branch-list failed.", exception)
|
|
if raise_exception: raise
|
|
branches = None
|
|
|
|
# Done here:
|
|
return branches
|
|
|
|
async def branch_add(
|
|
self,
|
|
branch_name: str,
|
|
raise_exception: bool = False,
|
|
) -> bool:
|
|
|
|
"""
|
|
Add a new branch.
|
|
:param branch_name: The name of the branch to add.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: True if logged out, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
|
|
try:
|
|
|
|
# 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
|
|
)
|
|
success = True
|
|
|
|
# If the call failed:
|
|
else:
|
|
response_json = response.json()
|
|
self._err_printer(
|
|
"Branch NOT added.",
|
|
branch_name,
|
|
response_json
|
|
)
|
|
success = False
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer("Branch NOT added.", exception, branch_name)
|
|
if raise_exception: raise
|
|
success = False
|
|
|
|
# Done here:
|
|
return success
|
|
|
|
# ┳┓
|
|
# ┃┃┏┓┏┓┏┓┏┓╋┏┳┓┏┓┏┓╋
|
|
# ┻┛┗ ┣┛┗┻┛ ┗┛┗┗┗ ┛┗┗
|
|
# ┛
|
|
|
|
async def department_list(
|
|
self,
|
|
raise_exception: bool = False,
|
|
) -> List[Dict[str, Any]] | None:
|
|
|
|
"""
|
|
List the existing departments.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: A list of dictionaries where each dictionary describes one department. None if the API call fails.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
depts = None
|
|
|
|
try:
|
|
|
|
# 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.")
|
|
|
|
# If the call failed:
|
|
else:
|
|
self._err_printer("Dept.-list failed.")
|
|
depts = None
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer("Dept.-list failed.", exception)
|
|
if raise_exception: raise
|
|
depts = None
|
|
|
|
# Done here:
|
|
return depts
|
|
|
|
async def department_add(
|
|
self,
|
|
department_name: str,
|
|
raise_exception: bool = False,
|
|
) -> bool:
|
|
|
|
"""
|
|
Add a new department.
|
|
:param department_name: The name of the department to add.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: True if logged out, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
|
|
try:
|
|
|
|
# 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
|
|
)
|
|
success = True
|
|
|
|
# If the call failed:
|
|
else:
|
|
response_json = response.json()
|
|
self._err_printer(
|
|
"Dept. NOT added.",
|
|
department_name,
|
|
response_json
|
|
)
|
|
success = False
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer("Dept. NOT added.", exception, department_name)
|
|
if raise_exception: raise
|
|
success = False
|
|
|
|
# Done here:
|
|
return success
|
|
|
|
# ┏┳┓
|
|
# ┃ ┏┓┏┓┏┳┓
|
|
# ┻ ┗ ┗┻┛┗┗
|
|
|
|
async def team_list(
|
|
self,
|
|
raise_exception: bool = False,
|
|
) -> List[Dict[str, Any]] | None:
|
|
|
|
"""
|
|
List the existing team members.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: A list of dictionaries where each dictionary describes one team-member. None if the API call fails.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
teams = None
|
|
|
|
try:
|
|
|
|
# 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.")
|
|
|
|
# If the call failed:
|
|
else:
|
|
self._err_printer("Team-list failed.")
|
|
teams = None
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer("Team-list filed.", exception)
|
|
if raise_exception: raise
|
|
teams = None
|
|
|
|
# Done here:
|
|
return teams
|
|
|
|
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 = None,
|
|
raise_exception: bool = False,
|
|
) -> 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.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: True if successful, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
|
|
try:
|
|
|
|
# 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,
|
|
)
|
|
success = True
|
|
|
|
# If the call failed:
|
|
else:
|
|
response_json = response.json()
|
|
self._err_printer(
|
|
"Team NOT added.",
|
|
team_name,
|
|
username,
|
|
response_json
|
|
)
|
|
success = False
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer("Team NOT added.", exception)
|
|
if raise_exception: raise
|
|
success = False
|
|
|
|
# Done here:
|
|
return success
|
|
|
|
async def team_update(
|
|
self,
|
|
user_id: int,
|
|
dept_id: int,
|
|
team_name: str,
|
|
email: str,
|
|
phone_no: str,
|
|
role: str,
|
|
applicant_notes: dict | list | str = None,
|
|
raise_exception: bool = False,
|
|
) -> bool:
|
|
|
|
"""
|
|
Update an existing team member.
|
|
:param user_id: The id of the team member.
|
|
:param dept_id: The id of the department that this team member is working in.
|
|
: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 applicant_notes: Optional notes about the team member.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: True if successful, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
|
|
try:
|
|
|
|
# Make the API call:
|
|
response = await self._http_client.post(
|
|
url = self.TEAM_UPDATE_URL,
|
|
headers = {"X-Session-Token": self.__session_token},
|
|
json = {
|
|
"idUser": user_id,
|
|
"idDepartment": dept_id,
|
|
"name": team_name,
|
|
"email": email,
|
|
"phoneNo": phone_no,
|
|
"role": role,
|
|
"hierarchy": 1,
|
|
"applicantNotes": applicant_notes
|
|
}
|
|
)
|
|
|
|
# If the call succeeded:
|
|
if response.status_code in [200]:
|
|
self._printer(
|
|
"Team updated.",
|
|
user_id,
|
|
team_name,
|
|
)
|
|
success = True
|
|
|
|
# If the call failed:
|
|
else:
|
|
response_json = response.json()
|
|
self._err_printer(
|
|
"Team NOT updated.",
|
|
user_id,
|
|
team_name,
|
|
response_json
|
|
)
|
|
success = False
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer("Team NOT updated.", exception)
|
|
if raise_exception: raise
|
|
success = False
|
|
|
|
# Done here:
|
|
return success
|
|
|
|
# ┏┓ ┓
|
|
# ┣┫╋╋┏┓┏┓┏┫┏┓┏┓┏┏┓
|
|
# ┛┗┗┗┗ ┛┗┗┻┗┻┛┗┗┗
|
|
|
|
async def attendance_mark(
|
|
self,
|
|
user_id: int,
|
|
status: Literal["A", "H", "HD1", "HD2", "P", "OT"],
|
|
over_time: int | float,
|
|
attendance_date: datetime.datetime | None = None,
|
|
json_notes: dict = None,
|
|
raise_exception: bool = False
|
|
) -> 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. "A" for absent,
|
|
2. "H" for holiday,
|
|
3. "HD1" for half-day (1st half),
|
|
4. "HD2" for half-day (2nd half),
|
|
5. "P" for present,
|
|
6. "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.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: True if the attendance was marked, else False.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
success = False
|
|
|
|
try:
|
|
|
|
# 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,
|
|
timeout = httpx.Timeout(
|
|
pool = 60.0,
|
|
connect = 5.0,
|
|
write = 15.0,
|
|
read = 10.0
|
|
)
|
|
)
|
|
|
|
# If the call succeeded:
|
|
if response.status_code in [200]:
|
|
self._printer("Attendance marked.", user_id, status, json_notes)
|
|
success = 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)
|
|
success = False
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer("Attendance NOT added.", exception, user_id)
|
|
if raise_exception: raise
|
|
success = False
|
|
|
|
# Done here:
|
|
return success
|
|
|
|
async def attendance_register(
|
|
self,
|
|
target_month: datetime.datetime,
|
|
mode: Literal["month", "date"] = "month",
|
|
raise_exception: bool = False,
|
|
) -> 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.
|
|
:param mode: The mode of the attendance filtering.
|
|
:param raise_exception: Whether to raise any exceptions, or to suppress them.
|
|
:return: The attendance register records if successful, else None.
|
|
"""
|
|
|
|
# Start by assuming failure:
|
|
attendance_register = None
|
|
|
|
try:
|
|
|
|
# Make the API call:
|
|
response = await self._http_client.post(
|
|
url = self.ATTENDANCE_REGISTER_URL,
|
|
headers = {"X-Session-Token": self.__session_token},
|
|
json = {
|
|
"month": target_month.strftime("%Y-%m-%d"),
|
|
"mode": mode
|
|
},
|
|
)
|
|
|
|
# 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
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
self._err_printer(exception)
|
|
if raise_exception: raise
|
|
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())
|