1330 lines
50 KiB
Python
1330 lines
50 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
CREATED: Thu, 13th Nov, 2025
|
|
UPDATED: Thu, 13th Nov, 2025
|
|
|
|
OBJECTIVE:
|
|
|
|
To automate actions on Cosec's HR system via web browser automation.
|
|
|
|
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
|
|
import psutil
|
|
|
|
# To work with date and time:
|
|
import time
|
|
import datetime
|
|
|
|
# To work with tabulated data:
|
|
import pandas as pd
|
|
|
|
# My utils:
|
|
from utils_v2.string import json
|
|
from utils_v2.system import files
|
|
from utils_v2.system import pfinfo
|
|
from utils_v2.date_time import date_time
|
|
|
|
# For browser automation:
|
|
from selenium import webdriver
|
|
from selenium.webdriver.chrome.service import Service
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.webdriver.support.ui import Select
|
|
|
|
# For working with various datatypes:
|
|
from typing import List, Literal, Any
|
|
|
|
# For debugging:
|
|
from icecream import IceCreamDebugger
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
class CosecWeb:
|
|
|
|
def __init__(
|
|
self,
|
|
cosec_url: str,
|
|
username: str,
|
|
password: str,
|
|
driver_dir: str,
|
|
user_data_dir: str,
|
|
downloads_dir: str,
|
|
debug: bool = True,
|
|
debug_prefix: str = "Cosec-Web | "
|
|
):
|
|
|
|
"""
|
|
Creates an instance of a web-driver that automates browser actions to get data from Cosec Web's portal.
|
|
:param cosec_url: The URL to hit to go to Cosec's web portal. Typically hosted on-prem.
|
|
:param username: The username of the account to log in with.
|
|
:param password: The password of the account to log in with.
|
|
:param driver_dir: The directory where the driver is placed.
|
|
:param user_data_dir: The directory where you want the web-driver to store user data.
|
|
:param downloads_dir: The directory where you want to hold the downloaded files.
|
|
:param debug: Whether, or not, to show debug messages.
|
|
:param debug_prefix: The prefix to show before the debug messages.
|
|
"""
|
|
|
|
# For debugging:
|
|
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
|
|
self._nc_printer = IceCreamDebugger(prefix = debug_prefix, includeContext = False)
|
|
if not debug: self.enable_debug()
|
|
|
|
# Hold the credentials internally:
|
|
self.cosec_url = cosec_url
|
|
self.username = username
|
|
self._password = password
|
|
|
|
# Hold other variables:
|
|
self.driver_dir = driver_dir
|
|
self.user_data_dir = user_data_dir
|
|
self.downloads_dir = downloads_dir
|
|
|
|
# Figure out which driver to use:
|
|
self.driver_path = None
|
|
driver_map = {
|
|
"win64": r"win64/chromedriver.exe",
|
|
"linux64": r"linux64/chromedriver",
|
|
}
|
|
if pfinfo.is_windows() and pfinfo.is_64_bit(): self.driver_path = os.path.join(self.driver_dir, driver_map["win64"])
|
|
elif pfinfo.is_linux() and pfinfo.is_64_bit(): self.driver_path = os.path.join(self.driver_dir, driver_map["linux64"])
|
|
|
|
# If we don't have a matching driver:
|
|
if self.driver_path is None: raise NotImplementedError("OS and/or CPU Architecture Not Supported.")
|
|
|
|
# Else, we initialize the driver:
|
|
self._nc_printer("Using driver:", self.driver_path)
|
|
self._service = Service(self.driver_path)
|
|
self._options = webdriver.ChromeOptions()
|
|
self._options.add_argument(f"--user-data-dir={user_data_dir}")
|
|
self._options.add_argument("--safebrowsing-disable-downloads-protection")
|
|
self._options.add_argument("--disable-popup-blocking")
|
|
self._options.add_experimental_option(
|
|
"prefs",
|
|
{
|
|
"downloads.default_directory": downloads_dir, # ... The custom downloads path.
|
|
"downloads.prompt_for_download": False, # ......... Do not show save dialog.
|
|
"downloads.directory_upgrade": True, # ............ If path doesn't exist, try to create it.
|
|
"safebrowsing.enabled": True, # .................. Allow safe downloads.
|
|
"profile.default_content_settings.popups": 0,
|
|
"profile.default_content_setting_values.automatic_downloads": 1,
|
|
"profile.content_settings.exceptions.automatic_downloads.*.setting": 1,
|
|
}
|
|
)
|
|
self._driver = webdriver.Chrome(service = self._service, options = self._options)
|
|
self._driver.set_page_load_timeout(300)
|
|
self._driver.execute_cdp_cmd("Page.setDownloadBehavior", {
|
|
"behavior": "allow",
|
|
"downloadPath": downloads_dir
|
|
})
|
|
|
|
# For later use, we will need variables to hold window information:
|
|
self._original_window = None
|
|
self._current_window = None
|
|
|
|
# ┳┓ ┓ •
|
|
# ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓
|
|
# ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫
|
|
# ┛ ┛ ┛
|
|
|
|
def enable_debug(self) -> None:
|
|
|
|
"""
|
|
To quickly shut off debugging messages.
|
|
:return: None.
|
|
"""
|
|
|
|
self._printer.enable()
|
|
self._nc_printer.enable()
|
|
|
|
def disable_debug(self) -> None:
|
|
|
|
"""
|
|
To quickly start showing debugging messages.
|
|
:return: None.
|
|
"""
|
|
|
|
self._printer.disable()
|
|
self._nc_printer.disable()
|
|
|
|
# ┏┓ ┓┏┓•┓┓
|
|
# ┃┃┏┓┏┓┏┏┓┏┏ ┃┫ ┓┃┃
|
|
# ┣┛┛ ┗┛┗┗ ┛┛ ┛┗┛┗┗┗
|
|
|
|
@staticmethod
|
|
def kill_chrome_processes() -> int:
|
|
|
|
"""
|
|
To kill previous Chrome-driver processes. Use this when you think that older processes are going to interfere
|
|
with new ones.
|
|
:return: The count of the processes that were killed.
|
|
"""
|
|
|
|
kill_count = 0
|
|
|
|
# Enlist all the processes:
|
|
for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
|
|
|
|
# Try to kill the process:
|
|
try:
|
|
name = proc.info['name']
|
|
cmd = proc.info['cmdline']
|
|
if name and ('chrome' in name.lower() or 'chromedriver' in name.lower()):
|
|
# self._nc_printer("Killing Chrome Proc.", name, proc.pid, cmd)
|
|
proc.kill()
|
|
kill_count += 1
|
|
|
|
# In case of exceptions:
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied): pass
|
|
|
|
# Done here:
|
|
# self._nc_printer("Killed Chrome Procs.", kill_count)
|
|
return kill_count
|
|
|
|
# ┳┳┓•
|
|
# ┃┃┃┓┏┏
|
|
# ┛ ┗┗┛┗
|
|
|
|
@staticmethod
|
|
def norm_list(
|
|
v: Any | List[Any],
|
|
target_length: int
|
|
) -> List[Any]:
|
|
|
|
"""
|
|
To normalize a variable which is expected to be a list.
|
|
:param v: The variable that needs to be a list.
|
|
:param target_length: The length of items expected in the list.
|
|
:return: The normalized list.
|
|
"""
|
|
|
|
if not isinstance(v, list): v = [v]
|
|
if len(v) > target_length: v = v[:target_length]
|
|
while len(v) < target_length: v.append(v[-1])
|
|
return v
|
|
|
|
# ┏┓•┓ ┳┓ ┓•
|
|
# ┣ ┓┃┏┓ ┣┫┏┓┏┓┏┫┓┏┓┏┓
|
|
# ┻ ┗┗┗ ┛┗┗ ┗┻┗┻┗┛┗┗┫
|
|
# ┛
|
|
|
|
@staticmethod
|
|
def read_muster_roll_xls(file_path) -> pd.DataFrame:
|
|
|
|
"""
|
|
When the muster-roll is downloaded as an XLS file, it will have extra junk in the header section. This method
|
|
removes that junk and keeps only a clean dataframe.
|
|
:param file_path: The path to the XLS report file.
|
|
:return: A clean pandas dataframe.
|
|
"""
|
|
|
|
# Simple file read:
|
|
df = pd.read_excel(file_path)
|
|
|
|
# Drop junk rows, rename the columns:
|
|
df = df[3:] # ............... drop the co. name and other info
|
|
new_header = df.iloc[0] # ... extract the desired header
|
|
new_header.name = None # .... no name for the index column
|
|
df = df[1:] # ............... drop more rows till the actual data starts
|
|
df.columns = new_header # ... the actual header row becomes the DF's header
|
|
|
|
# Drop fully null rows:
|
|
df = df.dropna(how = "all", axis = 0) # ......... cleans rows
|
|
df = df.dropna(how = "all", axis = 1) # ......... cleans cols
|
|
df.reset_index(drop = True, inplace = True) # ... cleans the index
|
|
|
|
# Done here:
|
|
return df
|
|
|
|
@staticmethod
|
|
def read_in_out_summary_xls(file_path) -> pd.DataFrame:
|
|
|
|
"""
|
|
When the in/out report is downloaded as an XLS file, it will have extra junk in the header section. This method
|
|
removes that junk and keeps only a clean dataframe.
|
|
:param file_path: The path to the XLS report file.
|
|
:return: A clean pandas dataframe.
|
|
"""
|
|
|
|
# Simple file read:
|
|
df = pd.read_excel(file_path)
|
|
|
|
# Drop junk rows, rename the columns:
|
|
df = df[2:] # ............... drop the co. name and other info
|
|
new_header = df.iloc[0] # ... extract the desired header
|
|
new_header.name = None # .... no name for the index column
|
|
df = df[3:] # ............... drop more rows till the actual data starts
|
|
df.columns = new_header # ... the actual header row becomes the DF's header
|
|
|
|
# If you request for data that spans over multiple days, you will get a row with datetime in to visually mark
|
|
# date change. You don't need this, drop it:
|
|
df = df[~df["User ID"].apply(lambda x: isinstance(x, (datetime.datetime, pd.Timestamp)))]
|
|
|
|
# Drop fully null rows:
|
|
df = df.dropna(how = "all", axis = 0) # ......... cleans rows
|
|
df = df.dropna(how = "all", axis = 1) # ......... cleans cols
|
|
df.reset_index(drop = True, inplace = True) # ... cleans the index
|
|
|
|
# Done here:
|
|
return df
|
|
|
|
# ┏┓┓• ┓ • ┏┓ •
|
|
# ┃ ┃┓┏┃┏┓┏┓┏┓ ┣┫┏╋┓┏┓┏┓┏
|
|
# ┗┛┗┗┗┛┗┗┛┗┗┫ ┛┗┗┗┗┗┛┛┗┛
|
|
# ┛
|
|
|
|
def click_one(
|
|
self,
|
|
element_ref: str,
|
|
timeout: float = 10.0,
|
|
initial_sleep: float | None = 0.0,
|
|
by: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID
|
|
) -> None:
|
|
|
|
"""
|
|
To click one item on the portal.
|
|
:param element_ref: The id, path, or other reference to the target element.
|
|
:param timeout: The timeout to wait after which the process raises an exception.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:param by: The kind of reference to the element (id, path, etc.).
|
|
:return: None.
|
|
"""
|
|
|
|
# Initial sleep:
|
|
time.sleep(initial_sleep or 0.0)
|
|
wait = WebDriverWait(self._driver, timeout = timeout)
|
|
|
|
# Perform the click:
|
|
element = wait.until(EC.element_to_be_clickable((by, element_ref)))
|
|
element.click()
|
|
self._printer("Clicked", element_ref, by)
|
|
|
|
def click_many(
|
|
self,
|
|
element_refs: List[str],
|
|
timeouts: List[float] = 10.0,
|
|
initial_sleeps: List[float] = None,
|
|
bys: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID
|
|
) -> None:
|
|
|
|
"""
|
|
To click many items sequentially on the portal.
|
|
:param element_refs: The id, path, or other reference to the target element.
|
|
:param timeouts: The timeout to wait after which the process raises an exception.
|
|
:param initial_sleeps: How many seconds to wait before the first action is taken.
|
|
:param bys: The kind of reference to the element (id, path, etc.).
|
|
:return: None.
|
|
"""
|
|
|
|
# Parse the list-inputs well:
|
|
target_length = len(element_refs)
|
|
timeouts = self.norm_list(timeouts, target_length)
|
|
initial_sleeps = self.norm_list(initial_sleeps, target_length)
|
|
bys = self.norm_list(bys, target_length)
|
|
|
|
# Execute each click:
|
|
for element_ref, timeout, initial_sleep, by in zip(element_refs, timeouts, initial_sleeps, bys):
|
|
self.click_one(element_ref, timeout, initial_sleep, by = by)
|
|
|
|
# ┏┳┓ • ┏┓ •
|
|
# ┃ ┓┏┏┓┓┏┓┏┓ ┣┫┏╋┓┏┓┏┓┏
|
|
# ┻ ┗┫┣┛┗┛┗┗┫ ┛┗┗┗┗┗┛┛┗┛
|
|
# ┛┛ ┛
|
|
|
|
def type_one(
|
|
self,
|
|
element_ref: str,
|
|
text: str,
|
|
clear_first: bool = True,
|
|
timeout: float = 10.0,
|
|
initial_sleep: float | None = 0.0,
|
|
by: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID
|
|
) -> None:
|
|
|
|
"""
|
|
To perform one typing action on the portal.
|
|
:param element_ref: The id, path, or other reference to the target element.
|
|
:param text: The content that you want to type out in the field.
|
|
:param clear_first: Whether you want to clear out any default content from the field before you start typing.
|
|
:param timeout: The timeout to wait after which the process raises an exception.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:param by: The kind of reference to the element (id, path, etc.).
|
|
:return: None
|
|
"""
|
|
|
|
# Initial sleep:
|
|
time.sleep(initial_sleep or 0.0)
|
|
wait = WebDriverWait(self._driver, timeout=timeout)
|
|
|
|
# Perform the click:
|
|
element = wait.until(EC.visibility_of_element_located((by, element_ref)))
|
|
if clear_first: element.clear()
|
|
element.send_keys(text)
|
|
self._printer("Typed", text, element_ref, by)
|
|
|
|
def type_many(
|
|
self,
|
|
element_refs: List[str],
|
|
texts: List[str],
|
|
clear_firsts: List[bool] = True,
|
|
timeouts: List[float] = 10.0,
|
|
initial_sleeps: List[float] = None,
|
|
bys: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID
|
|
) -> None:
|
|
|
|
"""
|
|
To perform many typing actions on the portal.
|
|
:param element_refs: The id, path, or other reference to the target element.
|
|
:param texts: The content that you want to type out in the field.
|
|
:param clear_firsts: Whether you want to clear out any default content from the field before you start typing.
|
|
:param timeouts: The timeout to wait after which the process raises an exception.
|
|
:param initial_sleeps: How many seconds to wait before the first action is taken.
|
|
:param bys: The kind of reference to the element (id, path, etc.).
|
|
:return: None
|
|
"""
|
|
|
|
# Parse the list-inputs well:
|
|
target_length = len(element_refs)
|
|
texts = self.norm_list(texts, target_length)
|
|
clear_firsts = self.norm_list(clear_firsts, target_length)
|
|
timeouts = self.norm_list(timeouts, target_length)
|
|
initial_sleeps = self.norm_list(initial_sleeps, target_length)
|
|
bys = self.norm_list(bys, target_length)
|
|
|
|
# Execute each click:
|
|
for element_ref, text, clear_first, timeout, initial_sleep, by in zip(element_refs, texts, clear_firsts, timeouts, initial_sleeps, bys):
|
|
self.type_one(element_ref, text, clear_first, timeout, initial_sleep, by = by)
|
|
|
|
# ┳┓ ┓ ┏┓ •
|
|
# ┃┃┏┓┏┓┏┓┏┫┏┓┓┏┏┏┓ ┣┫┏╋┓┏┓┏┓┏
|
|
# ┻┛┛ ┗┛┣┛┗┻┗┛┗┻┛┛┗ ┛┗┗┗┗┗┛┛┗┛
|
|
# ┛
|
|
|
|
def dropdown_one(
|
|
self,
|
|
element_ref: str,
|
|
selection: str,
|
|
timeout: float = 10.0,
|
|
initial_sleep: float | None = 0.0,
|
|
dropdown_sleep: float | None = 1.0,
|
|
by: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID
|
|
) -> None:
|
|
|
|
"""
|
|
To select one option on one dropdown selector on the portal.
|
|
:param element_ref: The id, path, or other reference to the target element.
|
|
:param selection: The selection that you want to make from the options.
|
|
:param timeout: The timeout to wait after which the process raises an exception.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:param dropdown_sleep: How many seconds you would like to wait to let the dropdown load its options.
|
|
:param by: The kind of reference to the element (id, path, etc.).
|
|
:return: None.
|
|
"""
|
|
|
|
# Initial sleep:
|
|
time.sleep(initial_sleep or 0.0)
|
|
wait = WebDriverWait(self._driver, timeout = timeout)
|
|
|
|
# Lock-in the dropdown,
|
|
# and select the option:
|
|
dropdown = wait.until(
|
|
EC.element_to_be_clickable((
|
|
by,
|
|
element_ref,
|
|
))
|
|
)
|
|
select = Select(dropdown)
|
|
time.sleep(dropdown_sleep)
|
|
select.select_by_visible_text(selection)
|
|
self._printer("Selected", selection, element_ref, by)
|
|
|
|
def dropdown_many(
|
|
self,
|
|
element_refs: List[str],
|
|
selections: List[str],
|
|
timeouts: List[float] = 10.0,
|
|
initial_sleeps: List[float] = None,
|
|
dropdown_sleeps: List[float] = None,
|
|
bys: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID
|
|
) -> None:
|
|
|
|
"""
|
|
To select one option each on many dropdown selectors on the portal.
|
|
:param element_refs: The id, path, or other reference to the target element.
|
|
:param selections: The selection that you want to make from the options.
|
|
:param timeouts: The timeout to wait after which the process raises an exception.
|
|
:param initial_sleeps: How many seconds to wait before the first action is taken.
|
|
:param dropdown_sleeps: How many seconds you would like to wait to let the dropdown load its options.
|
|
:param bys: The kind of reference to the element (id, path, etc.).
|
|
:return: None.
|
|
"""
|
|
|
|
# Parse the list-inputs well:
|
|
target_length = len(element_refs)
|
|
selections = self.norm_list(selections, target_length)
|
|
dropdown_sleeps = self.norm_list(dropdown_sleeps, target_length)
|
|
timeouts = self.norm_list(timeouts, target_length)
|
|
initial_sleeps = self.norm_list(initial_sleeps, target_length)
|
|
bys = self.norm_list(bys, target_length)
|
|
|
|
# Execute each click:
|
|
for element_ref, selection, timeout, initial_sleep, dropdown_sleep, by in zip(element_refs, selections, timeouts, initial_sleeps, dropdown_sleeps, bys):
|
|
self.dropdown_one(element_ref, selection, timeout, initial_sleep, dropdown_sleep, by = by)
|
|
|
|
# ┏┓ ┳ ┓ ┏┓ •
|
|
# ┃┓┏┓┏┓┓┏┏┓ ┃┏┫ ┣┫┏╋┓┏┓┏┓┏
|
|
# ┗┛┛ ┗┛┗┻┣┛ ┻┗┻ ┛┗┗┗┗┗┛┛┗┛
|
|
# ┛
|
|
|
|
def group_id_one(
|
|
self,
|
|
element_ref: str,
|
|
selection: str,
|
|
timeout: float = 10.0,
|
|
initial_sleep: float | None = 0.0,
|
|
by: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID
|
|
) -> None:
|
|
|
|
"""
|
|
The portal has custom group-id selectors on various pages that let you request for reports. This method allows
|
|
you to pick on group id from said UI element. Call it multiple times on the same selector to select many group
|
|
ids.
|
|
:param element_ref: The id, path, or other reference to the target element.
|
|
:param selection: The selection that you want to make from the options.
|
|
:param timeout: The timeout to wait after which the process raises an exception.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:param by: The kind of reference to the element (id, path, etc.).
|
|
:return: None.
|
|
"""
|
|
|
|
# Initial sleep:
|
|
time.sleep(initial_sleep or 0.0)
|
|
wait = WebDriverWait(self._driver, timeout = timeout)
|
|
|
|
# Get the text-field where you must type the group id,
|
|
# then type in the desired group id:
|
|
selection_box = wait.until(
|
|
EC.element_to_be_clickable((
|
|
by,
|
|
element_ref
|
|
))
|
|
)
|
|
selection_box.clear()
|
|
selection_box.send_keys(selection)
|
|
|
|
# Now locate the pop-up that opens:
|
|
ul_locator = (By.XPATH, "//ul[@class='dropdown-menu ng-isolate-scope' and contains(@class,'ng-hide') = false]")
|
|
ul_elem = wait.until(EC.visibility_of_element_located(ul_locator))
|
|
|
|
# Now select the list item that has your exact group id:
|
|
li_locator = (By.XPATH, f"//ul[@class='dropdown-menu ng-isolate-scope' and contains(@style,'display: block')]//li[normalize-space(.)='{selection}']")
|
|
li_elem = wait.until(EC.element_to_be_clickable(li_locator))
|
|
li_elem.click()
|
|
|
|
# Done here:
|
|
self._printer("Selected Group Id", selection, element_ref, by)
|
|
|
|
def group_id_many(
|
|
self,
|
|
element_refs: List[str],
|
|
selections: List[str],
|
|
timeouts: List[float] = 10.0,
|
|
initial_sleeps: List[float] = None,
|
|
bys: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID
|
|
) -> None:
|
|
|
|
"""
|
|
The portal has custom group-id selectors on various pages that let you request for reports. This method allows
|
|
you to pick group ids from the target UI component. Typically, you will call this on the same UI component again
|
|
and again.
|
|
:param element_refs: The id, path, or other reference to the target element.
|
|
:param selections: The selection that you want to make from the options.
|
|
:param timeouts: The timeout to wait after which the process raises an exception.
|
|
:param initial_sleeps: How many seconds to wait before the first action is taken.
|
|
:param bys: The kind of reference to the element (id, path, etc.).
|
|
:return: None.
|
|
"""
|
|
|
|
# Parse the list-inputs well:
|
|
target_length = len(element_refs)
|
|
selections = self.norm_list(selections, target_length)
|
|
timeouts = self.norm_list(timeouts, target_length)
|
|
initial_sleeps = self.norm_list(initial_sleeps, target_length)
|
|
bys = self.norm_list(bys, target_length)
|
|
|
|
# Parse the list-inputs well:
|
|
for v in [selections, timeouts, initial_sleeps, bys]:
|
|
if not isinstance(v, list): v = [v]
|
|
if len(v) > len(element_refs): v = v[:len(element_refs)]
|
|
while len(v) < len(element_refs): v.append(v[-1])
|
|
|
|
# Execute each click:
|
|
for element_ref, selection, timeout, initial_sleep, by in zip(element_refs, selections, timeouts, initial_sleeps, bys):
|
|
self.group_id_one(element_ref, selection, timeout, initial_sleep, by = by)
|
|
|
|
# ┳┓ ┳┓ ┓ ┓ ┏┓ •
|
|
# ┣┫┏┓┏┓┏┓┏┓╋ ┃┃┏┓┓┏┏┏┓┃┏┓┏┓┏┫ ┣┫┏╋┓┏┓┏┓┏
|
|
# ┛┗┗ ┣┛┗┛┛ ┗ ┻┛┗┛┗┻┛┛┗┗┗┛┗┻┗┻ ┛┗┗┗┗┗┛┛┗┛
|
|
# ┛
|
|
|
|
def report_download_one(
|
|
self,
|
|
iframe_ref: str,
|
|
progress_bar_ref: str = "//*[@id=\"ReportViewer\"]/div/div/div/div[2]/div[1]/div[2]",
|
|
export_to_ref: str = "//*[@id=\"ReportViewer\"]/div/div/div/div[1]/div[1]/div/div[1]/div[2]/div/div[13]/div[1]",
|
|
format_ref: str = "//div[contains(@class,'dxrd-preview-export-menu-item') and @title='XLS']",
|
|
report_gen_timeout: float | None = 120.0,
|
|
timeout: float = 10.0,
|
|
initial_sleep: float | None = 0.0,
|
|
iframe_by: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.ID,
|
|
progress_bar_by: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.XPATH,
|
|
export_to_by: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.XPATH,
|
|
format_by: Literal[By.ID, By.XPATH, By.CSS_SELECTOR] = By.XPATH
|
|
) -> None:
|
|
|
|
"""
|
|
To downloads one report after you hit the "Generate Report" button. The report is first displayed in an iframe,
|
|
then you must click on an "Export To" button and select your preferred format (typically "XLS"). This method
|
|
waits for the iframe, then waits for the report to be ready, then clicks through the downloads process, and goes
|
|
back to the parent context (exits the iframe).
|
|
:param iframe_ref: The id of the iframe. Do NOT pass the path.
|
|
:param progress_bar_ref: The reference to the progress bar element. Can be the id or the xpath. Tells you
|
|
whether, or not, the report is ready.
|
|
:param export_to_ref: The reference to the "Export To" button. Can be the id or the xpath.
|
|
:param format_ref: The reference to the option that allows you to select the format. Can be the id or the xpath.
|
|
:param report_gen_timeout: How long you want to wait for the report to get ready.
|
|
:param timeout: A general timeout for simpler actions.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:param iframe_by: The kind of reference to the element (id, path, etc.).
|
|
:param progress_bar_by: The kind of reference to the element (id, path, etc.).
|
|
:param export_to_by: The kind of reference to the element (id, path, etc.).
|
|
:param format_by: The kind of reference to the element (id, path, etc.).
|
|
:return: None.
|
|
"""
|
|
|
|
# Initial sleep:
|
|
time.sleep(initial_sleep or 0.0)
|
|
wait = WebDriverWait(self._driver, timeout = timeout)
|
|
|
|
# Wait for the iframe to become available,
|
|
# then switch to it:
|
|
self._printer("Waiting for iframe.", iframe_ref)
|
|
wait.until(EC.visibility_of_element_located((iframe_by, iframe_ref)))
|
|
self._driver.switch_to.frame(iframe_ref)
|
|
self._printer("Switched to iframe.", iframe_ref)
|
|
|
|
# Wait for the report to get generated by detecting the presence of the progress bar:
|
|
wait.until(EC.presence_of_element_located((progress_bar_by, progress_bar_ref)))
|
|
self._printer("⏱️ Report Generation Started")
|
|
WebDriverWait(
|
|
self._driver,
|
|
timeout = report_gen_timeout
|
|
).until(EC.invisibility_of_element_located((progress_bar_by, progress_bar_ref)))
|
|
self._printer("✅ Report Ready")
|
|
|
|
# Click all the buttons to downloads the report:
|
|
self._nc_printer("Waiting for downloads.")
|
|
self.click_many(
|
|
element_refs = [
|
|
export_to_ref, # ... "Export To" selector.
|
|
format_ref, # ...... "XLS" or other export option.
|
|
],
|
|
timeouts = [
|
|
timeout,
|
|
timeout,
|
|
],
|
|
initial_sleeps = [
|
|
initial_sleep,
|
|
initial_sleep,
|
|
],
|
|
bys = [
|
|
export_to_by,
|
|
format_by,
|
|
]
|
|
)
|
|
|
|
# Go back to the parent of the iframe:
|
|
self._driver.switch_to.default_content()
|
|
|
|
# ┳┓ •
|
|
# ┣┫┏┓┏┓┏
|
|
# ┻┛┗┻┛┗┗
|
|
|
|
def login(
|
|
self,
|
|
initial_sleep: float = 1.0
|
|
) -> None:
|
|
|
|
"""
|
|
Performs the login process.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:return: None.
|
|
"""
|
|
|
|
# Initial sleep:
|
|
time.sleep(initial_sleep)
|
|
wait = WebDriverWait(self._driver, timeout = 10.0)
|
|
|
|
# Open the base page:
|
|
self._driver.get(self.cosec_url)
|
|
|
|
# On login, Cosec open a new popup.
|
|
# We must switch our context to that popup:
|
|
self._original_window = self._driver.current_window_handle
|
|
self._current_window = self._driver.current_window_handle
|
|
while len(self._driver.window_handles) < 2: pass
|
|
for window in self._driver.window_handles:
|
|
if window != self._original_window: self._current_window = window
|
|
self._driver.switch_to.window(self._current_window)
|
|
self._printer("Context switched.", self._current_window)
|
|
|
|
# Make the window full-screen for standardized behaviour:
|
|
self._driver.maximize_window()
|
|
|
|
# The username text field:
|
|
element = wait.until(EC.presence_of_element_located((By.ID, "loginid")))
|
|
element.send_keys(self.username)
|
|
self._printer("Username typed.")
|
|
|
|
# The password text field:
|
|
element = wait.until(EC.presence_of_element_located((By.ID, "pwd")))
|
|
element.send_keys(self._password)
|
|
self._printer("Password typed.")
|
|
|
|
# Click the login button:
|
|
element = wait.until(EC.presence_of_element_located((By.ID, "btnlogin")))
|
|
element.click()
|
|
self._printer("Login clicked.")
|
|
|
|
def logout(
|
|
self,
|
|
initial_sleep: float = 1.0
|
|
) -> None:
|
|
|
|
"""
|
|
Logs out of the current session.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:return: None.
|
|
"""
|
|
|
|
# Click on the logout button:
|
|
self.click_one(
|
|
element_ref = "//a[@key='lnkLogout' and @title='Logout']",
|
|
timeout = 10.0,
|
|
initial_sleep = initial_sleep,
|
|
by = By.XPATH
|
|
)
|
|
|
|
def return_home(
|
|
self,
|
|
initial_sleep: float = 1.0
|
|
) -> None:
|
|
|
|
"""
|
|
Logs out of the current session.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:return: None.
|
|
"""
|
|
|
|
# Click on the logout button:
|
|
self.click_one(
|
|
element_ref = "//a[@key='lnkHome' and @title='Home']",
|
|
timeout = 10.0,
|
|
initial_sleep = initial_sleep,
|
|
by = By.XPATH
|
|
)
|
|
|
|
def quit(self) -> None:
|
|
|
|
"""
|
|
Just closes the full browser window.
|
|
:return: None.
|
|
"""
|
|
|
|
self._nc_printer("Closing browser.")
|
|
self._driver.quit()
|
|
self._nc_printer("Browser closed.")
|
|
|
|
# ┳┳┓
|
|
# ┃┃┃┏┓┏╋┏┓┏┓┏
|
|
# ┛ ┗┗┻┛┗┗ ┛ ┛
|
|
|
|
def get_muster_roll(
|
|
self,
|
|
on_date: datetime.datetime,
|
|
group_ids: List[str],
|
|
download_timeout: float = 60.0,
|
|
initial_sleep: float = 1.0
|
|
) -> str | None:
|
|
|
|
"""
|
|
Goes through the full process of clicking and filling out the UI to get the Muster Roll as a file.
|
|
WARNING: All previously downloaded files will be erased.
|
|
:param on_date: The date on which you need the muster roll. Only the month and year are picked from it.
|
|
:param group_ids: The ids of the groups for which you want the information. Each entity/company is recognized by
|
|
one group id.
|
|
:param download_timeout: The time to wait for the report to get downloaded.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:return: The path to the downloaded report file.
|
|
"""
|
|
|
|
# Empty out the past downloads:
|
|
for file_name in files.list_files(
|
|
self.downloads_dir,
|
|
full_path = False
|
|
):
|
|
self._nc_printer("Deleting", file_name)
|
|
files.delete_file(os.path.join(self.downloads_dir, file_name))
|
|
|
|
# Initial sleep:
|
|
time.sleep(initial_sleep)
|
|
wait = WebDriverWait(self._driver, timeout = 10.0)
|
|
|
|
# Click all the buttons till the menu that allows you to specify the report specs:
|
|
self.click_many(
|
|
element_refs = [
|
|
"Link_3", # ................................................ The "Time and Attendance" page.
|
|
"Reports", # ............................................... The "Reports" menu.
|
|
"//a[@data-parent='#3032' and @data-target='#30101']", # ... The "Customized Reports" sub-menu.
|
|
"30106", # ................................................. The "Muster Roll" item.
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
10.0,
|
|
10.0,
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
1.0,
|
|
0.0,
|
|
0.0,
|
|
0.0,
|
|
],
|
|
bys = [
|
|
By.ID,
|
|
By.ID,
|
|
By.XPATH,
|
|
By.ID,
|
|
]
|
|
)
|
|
|
|
self.dropdown_many(
|
|
element_refs = [
|
|
"FromMonth", # .................................................... "Month-Year" field for the month.
|
|
"FromYear", # ..................................................... "Month-Year" field for the year.
|
|
"grpddl", # ....................................................... "Select Users" dropdown.
|
|
"//select[@type='dropdown' and @dropdownsource='groupList']", # ... "Select Group" dropdown.
|
|
],
|
|
selections = [
|
|
on_date.strftime("%B"),
|
|
on_date.strftime("%Y"),
|
|
"Group Wise",
|
|
"Organization",
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
10.0,
|
|
10.0,
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
0.0,
|
|
0.0,
|
|
0.0,
|
|
0.0,
|
|
],
|
|
dropdown_sleeps = [
|
|
1.0,
|
|
1.0,
|
|
1.0,
|
|
1.0,
|
|
],
|
|
bys = [
|
|
By.ID,
|
|
By.ID,
|
|
By.ID,
|
|
By.XPATH,
|
|
]
|
|
)
|
|
|
|
# Select all the groups:
|
|
self.group_id_many(
|
|
element_refs = ["grpid"] * len(group_ids),
|
|
selections = group_ids,
|
|
timeouts = [10.0] * len(group_ids),
|
|
initial_sleeps = [0.0] * len(group_ids),
|
|
bys = [By.ID] * len(group_ids),
|
|
)
|
|
|
|
# Select which reports you want:
|
|
self.dropdown_many(
|
|
element_refs = [
|
|
"GenerateRptFor", # ... "Generate Report For" dropdown.
|
|
],
|
|
selections = [
|
|
"Active Users",
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
0.0,
|
|
],
|
|
dropdown_sleeps = [
|
|
0.0,
|
|
],
|
|
bys = [
|
|
By.ID,
|
|
]
|
|
)
|
|
|
|
# Click all the buttons to generate the report:
|
|
self.click_many(
|
|
element_refs = [
|
|
"//input[@type='button' and @default='Generate Report']", # ... "Generate Report" button.
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
0.0,
|
|
],
|
|
bys = [
|
|
By.XPATH,
|
|
]
|
|
)
|
|
|
|
# Get the report:
|
|
old_files = files.list_files(self.downloads_dir, full_path = True)
|
|
old_file_count = len(old_files)
|
|
self.report_download_one(
|
|
iframe_ref = "report1",
|
|
progress_bar_ref = "//*[@id=\"ReportViewer\"]/div/div/div/div[2]/div[1]/div[2]",
|
|
export_to_ref = "//*[@id=\"ReportViewer\"]/div/div/div/div[1]/div[1]/div/div[1]/div[2]/div/div[13]/div[1]",
|
|
format_ref = "//div[contains(@class,'dxrd-preview-export-menu-item') and @title='XLS']",
|
|
timeout = 10.0,
|
|
report_gen_timeout = 180.0,
|
|
initial_sleep = 0.0,
|
|
iframe_by = By.ID,
|
|
progress_bar_by = By.XPATH,
|
|
export_to_by = By.XPATH,
|
|
format_by = By.XPATH,
|
|
)
|
|
|
|
# Wait for the downloads to finish (or time to run out):
|
|
report_file_path = None
|
|
start_t = time.time()
|
|
while True:
|
|
new_files = [f for f in files.list_files(self.downloads_dir, full_path = True) if f.lower().find(".crdownload") < 0]
|
|
new_file_count = len(new_files)
|
|
if new_file_count > old_file_count: break
|
|
if time.time() - start_t > download_timeout: break
|
|
|
|
# The new file will be our desired report:
|
|
for f in new_files:
|
|
if f not in old_files:
|
|
report_file_path = f
|
|
|
|
# Done here:
|
|
self._nc_printer("Download complete.")
|
|
self._printer(report_file_path)
|
|
return report_file_path
|
|
|
|
# ┳┳ ┳ ┏┓ ┏┓
|
|
# ┃┃┏┏┓┏┓ ┃┏┓━━┃┃┓┏╋ ┣ ┓┏┏┓┏┓╋┏
|
|
# ┗┛┛┗ ┛ ┻┛┗ ┗┛┗┻┗ ┗┛┗┛┗ ┛┗┗┛
|
|
|
|
def get_in_out_summary(
|
|
self,
|
|
from_date: datetime.datetime,
|
|
to_date: datetime.datetime,
|
|
group_ids: List[str],
|
|
download_timeout: float = 60.0,
|
|
initial_sleep: float = 1.0,
|
|
timezone: str = None,
|
|
) -> str | None:
|
|
|
|
"""
|
|
Goes through the full process of clicking and filling out the UI to get the In/Out Summary as a file.
|
|
WARNING: All previously downloaded files will be erased.
|
|
:param from_date: The starting date of the required information.
|
|
:param to_date: The ending date of the required information.
|
|
:param group_ids: The ids of the groups for which you want the information. Each entity/company is recognized by
|
|
one group id.
|
|
:param download_timeout: The time to wait for the report to get downloaded.
|
|
:param initial_sleep: How many seconds to wait before the first action is taken.
|
|
:param timezone: A timezone to apply to the given date-time objects.
|
|
:return: The path to the downloaded report file.
|
|
"""
|
|
|
|
# Apply the timezone if given:
|
|
if timezone:
|
|
from_date = date_time.to_timezone(from_date, timezone)
|
|
to_date = date_time.to_timezone(to_date, timezone)
|
|
|
|
# Empty out the past downloads:
|
|
for file_name in files.list_files(
|
|
self.downloads_dir,
|
|
full_path = False
|
|
):
|
|
self._nc_printer("Deleting", file_name)
|
|
files.delete_file(os.path.join(self.downloads_dir, file_name))
|
|
|
|
# Initial sleep:
|
|
time.sleep(initial_sleep)
|
|
wait = WebDriverWait(self._driver, timeout = 10.0)
|
|
|
|
# Click all the buttons till the menu that allows you to specify the report specs:
|
|
self.click_many(
|
|
element_refs = [
|
|
"Link_7", # ............................................... The "Users" page.
|
|
"Reports", # .............................................. The "Reports" menu.
|
|
"//a[@data-parent='#7018' and @data-target='#7039']", # ... The "User Events" sub-menu.
|
|
"7040", # ................................................. The "In/Out Event" item.
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
10.0,
|
|
10.0,
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
1.0,
|
|
0.0,
|
|
0.0,
|
|
0.0,
|
|
],
|
|
bys = [
|
|
By.ID,
|
|
By.ID,
|
|
By.XPATH,
|
|
By.ID,
|
|
]
|
|
)
|
|
|
|
# Type out all the date-time fields:
|
|
self.type_many(
|
|
element_refs = [
|
|
"_calFromDate", # ... The starting date.
|
|
"_calToDate", # ..... The ending date.
|
|
"txtFromTime", # .... The starting time.
|
|
"txtToTime", # ...... The ending time.
|
|
],
|
|
texts = [
|
|
from_date.strftime("%d/%m/%Y"),
|
|
to_date.strftime("%d/%m/%Y"),
|
|
from_date.strftime("%H:%M"),
|
|
to_date.strftime("%H:%M"),
|
|
],
|
|
clear_firsts = [
|
|
True,
|
|
True,
|
|
True,
|
|
True,
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
10.0,
|
|
10.0,
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
0.0,
|
|
0.0,
|
|
0.0,
|
|
0.0,
|
|
],
|
|
bys = [
|
|
By.ID,
|
|
By.ID,
|
|
By.ID,
|
|
By.ID,
|
|
]
|
|
)
|
|
|
|
self.dropdown_many(
|
|
element_refs = [
|
|
"grpddl", # ....................................................... "Select Users" dropdown.
|
|
"//select[@type='dropdown' and @dropdownsource='groupList']", # ... "Select Group" dropdown.
|
|
],
|
|
selections = [
|
|
"Group Wise",
|
|
"Organization",
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
0.0,
|
|
0.0,
|
|
],
|
|
dropdown_sleeps = [
|
|
1.0,
|
|
1.0,
|
|
],
|
|
bys = [
|
|
By.ID,
|
|
By.XPATH,
|
|
]
|
|
)
|
|
|
|
# Select all the groups:
|
|
self.group_id_many(
|
|
element_refs = ["grpid"] * len(group_ids),
|
|
selections = group_ids,
|
|
timeouts = [10.0] * len(group_ids),
|
|
initial_sleeps = [0.0] * len(group_ids),
|
|
bys = [By.ID] * len(group_ids),
|
|
)
|
|
|
|
# Select which reports you want:
|
|
self.dropdown_many(
|
|
element_refs = [
|
|
"cboUserSel", # ... "Generate Report For" dropdown.
|
|
],
|
|
selections = [
|
|
"Active Users",
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
0.0,
|
|
],
|
|
dropdown_sleeps = [
|
|
0.0,
|
|
],
|
|
bys = [
|
|
By.ID,
|
|
]
|
|
)
|
|
|
|
# Click all the buttons to generate the report:
|
|
self.click_many(
|
|
element_refs = [
|
|
"//input[@type='button' and @value='Generate Report']", # ... "Generate Report" button.
|
|
],
|
|
timeouts = [
|
|
10.0,
|
|
],
|
|
initial_sleeps = [
|
|
0.0,
|
|
],
|
|
bys = [
|
|
By.XPATH,
|
|
]
|
|
)
|
|
|
|
# Get the report:
|
|
old_files = files.list_files(self.downloads_dir, full_path = True)
|
|
old_file_count = len(old_files)
|
|
self.report_download_one(
|
|
iframe_ref = "report1",
|
|
progress_bar_ref = "//*[@id=\"ReportViewer\"]/div/div/div/div[2]/div[1]/div[2]",
|
|
export_to_ref = "//*[@id=\"ReportViewer\"]/div/div/div/div[1]/div[1]/div/div[1]/div[2]/div/div[13]/div[1]",
|
|
format_ref = "//div[contains(@class,'dxrd-preview-export-menu-item') and @title='XLS']",
|
|
timeout = 10.0,
|
|
report_gen_timeout = 120.0,
|
|
initial_sleep = 0.0,
|
|
iframe_by = By.ID,
|
|
progress_bar_by = By.XPATH,
|
|
export_to_by = By.XPATH,
|
|
format_by = By.XPATH,
|
|
)
|
|
|
|
# Wait for the downloads to finish (or time to run out):
|
|
report_file_path = None
|
|
start_t = time.time()
|
|
while True:
|
|
new_files = [f for f in files.list_files(self.downloads_dir, full_path = True) if f.lower().find(".crdownload") < 0]
|
|
new_file_count = len(new_files)
|
|
if new_file_count > old_file_count: break
|
|
if time.time() - start_t > download_timeout: break
|
|
|
|
# The new file will be our desired report:
|
|
for f in new_files:
|
|
if f not in old_files:
|
|
report_file_path = f
|
|
|
|
# Done here:
|
|
self._nc_printer("Download complete.")
|
|
self._printer(report_file_path)
|
|
return report_file_path
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
# Figure out the base_directory:
|
|
base_dir = files.get_parent_directory(
|
|
files.get_file_directory(include_filename = False),
|
|
depth = 1
|
|
)
|
|
|
|
# Put together the directory for the drivers, the downloads, etc.:
|
|
chrome_driver_dir = os.path.join(base_dir, r"drivers/chrome")
|
|
user_data_dir = os.path.join(base_dir, r"browser/user_data")
|
|
downloads_dir = os.path.join(base_dir, r"downloads")
|
|
|
|
# Show all the paths for debugging:
|
|
print("PATHS:")
|
|
print("Driver :", chrome_driver_dir)
|
|
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,
|
|
# )
|
|
|
|
# # 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 : ")
|
|
in_out_report_df = CosecWeb.read_in_out_summary_xls(in_out_report_path)
|
|
in_out_report_df = in_out_report_df[:35]
|
|
print(in_out_report_df.to_string())
|
|
print("\n\n---\n\n")
|
|
print(in_out_report_df.info())
|
|
|
|
# # Go back to the home page:
|
|
# cosec.return_home()
|
|
|
|
# # Get the muster roll:
|
|
# muster_roll_path = cosec.get_muster_roll(
|
|
# on_date = datetime.datetime.now(),
|
|
# group_ids = [
|
|
# "2", # ... Velankani Information Systems Pvt Ltd
|
|
# "3", # ... Velankani Bydesign India Pvt Ltd
|
|
# "4", # ... Velankani Electronics & Automotive Pvt Ltd
|
|
# ],
|
|
# initial_sleep = 1.0,
|
|
# download_timeout = 60.0
|
|
# )
|
|
|
|
# Convert the Muster Roll to a Pandas DF:
|
|
muster_roll_path = input("File Path : ")
|
|
muster_roll_df = CosecWeb.read_muster_roll_xls(muster_roll_path)
|
|
muster_roll_df = muster_roll_df[[
|
|
"User ID", "User Name", "Category Name",
|
|
"Grade Name", "Branch Name", "Department Name",
|
|
"Direct Reporting", "Level-1"
|
|
]]
|
|
muster_roll_df = muster_roll_df[:20]
|
|
print(muster_roll_df.to_string())
|
|
print("\n\n---\n\n")
|
|
print(muster_roll_df.info())
|
|
|
|
# # Get unique combinations:
|
|
# unique_branches = muster_roll_df["Branch Name"].unique().tolist()
|
|
# unique_depts = muster_roll_df[["Branch Name", "Department Name"]].drop_duplicates().to_dict(orient = "records")
|
|
# unique_reportees = muster_roll_df[["Branch Name", "Department Name", "Direct Reporting"]].drop_duplicates().to_dict(orient = "records")
|
|
# unique_combos = {
|
|
# "Branch Name": unique_branches,
|
|
# "Department Name": unique_depts,
|
|
# "Direct Reporting": unique_reportees
|
|
# }
|
|
# print("UNIQUE COMBOS:", json.to_string(unique_combos))
|
|
|
|
# # Log out to end the cycle:
|
|
# cosec.logout()
|
|
#
|
|
# # Close the browser window:
|
|
# cosec.quit()
|
|
#
|
|
# # Just to see what's going on,
|
|
# # doesn't add to the operational requirements:
|
|
# time.sleep(10.0)
|