""" 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, window_width: int = 1920, window_height: int = 1080, headless: bool = True, 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 window_width: The width of the browser window. :param window_height: The height of the browser window. :param headless: Whether, or not, to use Chrome browser in headless (No UI) mode. :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 self.headless = headless self.window_width = window_width self.window_height = window_height # 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 start constructing the preferred configuration:: 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. "safebrowsing.disable_download_protection": True, # ... Don't show "keep" button before downloading. "profile.default_content_settings.popups": 0, "profile.default_content_setting_values.automatic_downloads": 1, "profile.content_settings.exceptions.automatic_downloads.*.setting": 1, } ) # For headless mode: if headless: self._printer("HEADLESS MODE!") self._options.add_argument("--headless=new") # .................................. Modern headless mode. self._options.add_argument("--disable-gpu") # ................................... Optional (mostly for Windows). self._options.add_argument(f"--window-size={window_width},{window_height}") # ... The internally simulated screen size. # Initialize the driver: 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 }) self._printer("Driver ready.") # For later use, we will need variables to hold window information: self._original_window = None self._current_window = None def __del__(self): self._printer("Deinitializing!") try: self.quit() except Exception as exception: self._printer(exception) # ┳┓ ┓ • # ┃┃┏┓┣┓┓┏┏┓┏┓┓┏┓┏┓ # ┻┛┗ ┗┛┗┻┗┫┗┫┗┛┗┗┫ # ┛ ┛ ┛ 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) # Safety rename in case the column names change in the future: df = df.rename( columns = { "User ID": "User ID", "User Name": "User Name", "Category Name": "Category Name", "Grade Name": "Grade Name", "Branch Name": "Branch Name", "Department Name": "Department Name", "Direct Reporting": "Direct Reporting", "Level-1": "Level-1" } ) # 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 download.") 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) # Set the size of the window so that you are certain that all the elements will be visible: self._driver.set_window_size(self.window_width, self.window_height) # 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. """ self._printer("MUSTER ROLL") # 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. """ self._printer("IN-OUT SUMMARY") # 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 = 240.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)