#!/usr/bin/env python3 """ HUB System Batch Employee Inactive Script with Auto-Login ========================================================== Description: Uses Playwright to auto-login and batch process employee accounts Requirements: Python 3.7+, playwright, openpyxl Installation: pip install playwright openpyxl playwright install chromium """ import os import json import time import logging import argparse import csv from datetime import datetime from pathlib import Path try: from playwright.sync_api import sync_playwright, Browser, Page, BrowserContext except ImportError: print("ERROR: Playwright is not installed!") print("Please run: pip install playwright") print("Then run: playwright install chromium") exit(1) try: import openpyxl from openpyxl import load_workbook except ImportError: print("ERROR: openpyxl is not installed!") print("Please run: pip install openpyxl") exit(1) import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # ============================================ # Configuration # ============================================ BASE_URL = "https://ballyhub.macausjm-glp.com" LOGIN_USERNAME = "166340" LOGIN_PASSWORD = "Gfds@456456782" SITE_ID = "110000002" # Debug options KEEP_BROWSER_OPEN_ON_ERROR = True # Keep browser open if login fails (for debugging) HEADLESS_MODE = False # Set to True to run browser in headless mode (no window) # File paths SCRIPT_DIR = Path(__file__).parent EXCEL_PATH = SCRIPT_DIR / "AdTermination.xlsx" LOG_PATH = SCRIPT_DIR / f"BatchInactive_Auto_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log" # ============================================ # Logging Setup # ============================================ class ColoredFormatter(logging.Formatter): """Custom formatter with colors""" COLORS = { 'ERROR': '\033[91m', # Red 'WARNING': '\033[93m', # Yellow 'SUCCESS': '\033[92m', # Green 'INFO': '\033[97m', # White 'RESET': '\033[0m' } def format(self, record): levelname = record.levelname if levelname in self.COLORS: record.levelname = f"{self.COLORS[levelname]}{levelname}{self.COLORS['RESET']}" return super().format(record) # Create logger logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) # Custom SUCCESS level SUCCESS_LEVEL = 35 # Between WARNING (30) and ERROR (40) logging.addLevelName(SUCCESS_LEVEL, 'SUCCESS') def success(self, message, *args, **kwargs): if self.isEnabledFor(SUCCESS_LEVEL): self._log(SUCCESS_LEVEL, message, args, **kwargs) logging.Logger.success = success # File handler (no colors) file_handler = logging.FileHandler(LOG_PATH, encoding='utf-8') file_handler.setFormatter(logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S')) logger.addHandler(file_handler) # Console handler (with colors) console_handler = logging.StreamHandler() console_handler.setFormatter(ColoredFormatter('[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S')) logger.addHandler(console_handler) # ============================================ # Functions # ============================================ def read_excel_data(file_path): """Read employee data from Excel file""" logger.info(f"Reading Excel file: {file_path}") try: wb = load_workbook(file_path, read_only=True) ws = wb.active # Log headers for debugging headers = [cell.value for cell in ws[1]] logger.info(f"Excel headers: {headers}") # Create header to index mapping header_map = {header: idx for idx, header in enumerate(headers) if header} logger.info(f"Header mapping: {header_map}") employees = [] # Skip header row, start from row 2 for row_num, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2): # Skip empty rows if not any(row): continue # Log row data for debugging (first few rows) if row_num <= 4: logger.info(f"Row {row_num} data: {list(row)}") # Extract data based on header names employee = { 'Backup': str(row[header_map.get('Backup', 5)]) if len(row) > header_map.get('Backup', 5) and row[header_map.get('Backup', 5)] else '', 'Domain': str(row[header_map.get('Domain', 0)]) if len(row) > header_map.get('Domain', 0) and row[header_map.get('Domain', 0)] else '', 'EmployeeID': str(row[header_map.get('EmployeeID', 1)]) if len(row) > header_map.get('EmployeeID', 1) and row[header_map.get('EmployeeID', 1)] else '', 'SamAccountName': str(row[header_map.get('SamAccountName', 2)]) if len(row) > header_map.get('SamAccountName', 2) and row[header_map.get('SamAccountName', 2)] else '', 'TerminationDay': row[header_map.get('TerminationDay', 3)] if len(row) > header_map.get('TerminationDay', 3) else None, 'Status': str(row[header_map.get('Status', 4)]) if len(row) > header_map.get('Status', 4) and row[header_map.get('Status', 4)] else '' } # Only add if EmployeeID exists if employee['EmployeeID']: employees.append(employee) wb.close() logger.info(f"Successfully read {len(employees)} employee records") # Log summary of what was read for emp in employees[:3]: # First 3 logger.info(f" - EmployeeID={emp['EmployeeID']}, SamAccountName={emp['SamAccountName']}, Domain={emp['Domain']}") return employees except Exception as e: logger.error(f"Failed to read Excel: {e}") raise def get_cookies_via_playwright(username, password): """Auto-login using Playwright and extract cookies""" logger.info("Starting auto-login using Playwright...") with sync_playwright() as p: # Launch browser browser = p.chromium.launch( headless=HEADLESS_MODE, slow_mo=100 ) context = browser.new_context() page = context.new_page() try: # Navigate to homepage logger.info("Navigating to homepage...") page.goto(f"{BASE_URL}/") time.sleep(1) # Step 1: Fill username logger.info("Step 1: Filling username...") try: username_field = page.get_by_role("textbox", name="Username...") username_field.click() username_field.fill(username) logger.info(f"Filled username: {username}") except Exception as e: logger.warning(f"Could not fill username: {e}") try: page.fill('input[name="UserName"]', username) except: logger.error("Failed to fill username") browser.close() return None time.sleep(0.5) # Step 2: Fill password and press Tab logger.info("Step 2: Filling password and pressing Tab...") try: password_field = page.get_by_role("textbox", name="Password...") password_field.fill(password) password_field.press("Tab") logger.info("Filled password and pressed Tab") except Exception as e: logger.warning(f"Could not fill password: {e}") try: page.fill('input[name="Password"]', password) page.keyboard.press("Tab") except: logger.error("Failed to fill password") browser.close() return None time.sleep(0.5) # Step 3: Check for "Maximum login session reached" message logger.info("Step 3: Checking for 'Maximum login session reached' message...") page_content = page.content() page_content_lower = page_content.lower() # Check for various forms of the message needs_wait = ( 'maximum login session reached' in page_content_lower or 'kindly wait' in page_content_lower or ('till other user responds' in page_content_lower and 'sec' in page_content_lower) ) logger.info(f"Wait message detected: {needs_wait}") if needs_wait: logger.warning("Detected 'wait 20 seconds' message, waiting...") # Wait for the message dialog to appear (20 seconds countdown) logger.info("Waiting for session timeout message to complete (20s)...") time.sleep(21) # Now click OK button to dismiss the message logger.info("Clicking OK button to dismiss message...") try: # Try multiple selectors for OK button ok_clicked = False try: page.get_by_role("button", name="OK").click(timeout=5000) ok_clicked = True logger.success("Clicked OK button (by role)") except: pass if not ok_clicked: try: page.click('button:has-text("OK")', timeout=5000) ok_clicked = True logger.success("Clicked OK button (by text)") except: pass if not ok_clicked: try: page.click('input[type="button"][value="OK"]', timeout=5000) ok_clicked = True logger.success("Clicked OK button (input)") except: pass if not ok_clicked: logger.warning("Could not click OK button, trying Enter key...") page.keyboard.press("Enter") time.sleep(0.5) except Exception as e: logger.warning(f"Could not click OK button: {e}") # Re-fill password after message is dismissed logger.info("Re-filling password after wait...") try: page.get_by_role("textbox", name="Password...").fill(password) page.get_by_role("textbox", name="Password...").press("Tab") except: try: page.fill('input[name="Password"]', password) page.keyboard.press("Tab") except: logger.warning("Could not re-fill password") time.sleep(0.5) else: logger.info("No 'wait 20 seconds' message detected") # Step 4: Select site logger.info("Step 4: Selecting site...") try: page.select_option('select#ddlSite', str(SITE_ID), timeout=5000) logger.info(f"Selected site {SITE_ID}") except Exception as e: logger.warning(f"Could not select site: {e}") time.sleep(0.5) # Step 5: Click "Submit" button logger.info("Step 5: Clicking 'Submit' button...") try: submit_button = page.get_by_role("button", name="Submit") submit_button.click() logger.info("Clicked 'Submit' button") except Exception as e: logger.warning(f"Could not click 'Submit' button: {e}") try: page.click('button:has-text("Submit")') logger.info("Clicked 'Submit' using text selector") except: logger.error("Failed to click 'Submit' button") browser.close() return None # Step 6: Wait for login to complete logger.info("Step 6: Waiting for login to complete...") for i in range(25): try: current_url = page.url if '/User/Login' not in current_url: logger.success("Login successful! Redirected to: " + current_url) break time.sleep(1) except: time.sleep(1) time.sleep(2) # Get cookies logger.info("Extracting cookies from browser...") cookies = context.cookies() asp_session = next((c for c in cookies if c['name'] == 'ASP.NET_SessionId'), None) hub_session = next((c for c in cookies if c['name'] == 'HUBUserSessionId'), None) if not asp_session: logger.error("Could not find ASP.NET_SessionId cookie!") logger.error("Login may have failed. Please check the browser window.") if KEEP_BROWSER_OPEN_ON_ERROR: logger.warning("Keeping browser open for debugging. Press Enter to close...") input() browser.close() return None logger.success("Successfully extracted cookies:") logger.info(f" ASP.NET_SessionId: {asp_session['value']}") if hub_session: logger.info(f" HUBUserSessionId: {hub_session['value']}") else: logger.warning("HUBUserSessionId not found") browser.close() return { 'ASPNET_SessionId': asp_session['value'], 'HUBUserSessionId': hub_session['value'] if hub_session else '' } except Exception as e: logger.error(f"Auto-login failed: {e}") import traceback logger.error(traceback.format_exc()) if KEEP_BROWSER_OPEN_ON_ERROR: logger.warning("Keeping browser open for debugging. Press Enter to close...") input() browser.close() return None def create_session_with_cookies(asp_session_id, hub_session_id): """Create a requests session with cookies""" session = requests.Session() # Disable SSL verification (for self-signed certificates) session.verify = False # Set headers session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36 Edg/144.0.0.0' }) # Add cookies domain = 'ballyhub.macausjm-glp.com' session.cookies.set('_culture', 'en-us', domain=domain) session.cookies.set('ASP.NET_SessionId', asp_session_id, domain=domain) if hub_session_id: session.cookies.set('HUBUserSessionId', hub_session_id, domain=domain) # Setup retry strategy retry_strategy = Retry( total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504] ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount('http://', adapter) session.mount('https://', adapter) # Suppress SSL warnings import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) return session def get_employee_info(session, staff_id): """Get employee information""" try: url = f"{BASE_URL}/Employee/GetEmpInfo" body = "sort=&group=&filter=" logger.info(f"Request URL: {url}") logger.info(f"Request Body: {body}") logger.info(f"Looking for StaffId/DomainName: {staff_id}") response = session.post( url, headers={ 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate, br, zstd', 'Accept-Language': 'en-US,en;q=0.9,zh-HK;q=0.8,zh;q=0.7', 'Cache-Control': 'no-cache', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Origin': BASE_URL, 'Pragma': 'no-cache', 'Referer': f'{BASE_URL}/Employee/EmployeeInfo', 'X-Requested-With': 'XMLHttpRequest' }, data=body ) logger.info(f"Response Status Code: {response.status_code}") logger.info(f"Response Content-Length: {len(response.content)} bytes") # Log raw response content for debugging content_text = response.text preview_length = min(1000, len(content_text)) logger.info(f"========== API Response Start (first {preview_length} chars) ==========") logger.info(content_text[:preview_length]) logger.info(f"========== API Response End ==========") if response.status_code != 200: logger.error(f"Unexpected status code: {response.status_code}") return None # Try to parse JSON try: data = response.json() # Log the parsed JSON structure if isinstance(data, dict): logger.info(f"Response is a dict with keys: {list(data.keys())}") elif isinstance(data, list): logger.info(f"Response is a list with {len(data)} items") if len(data) > 0: logger.info(f"First item keys: {list(data[0].keys()) if isinstance(data[0], dict) else 'not a dict'}") except Exception as e: logger.error(f"Failed to parse JSON response: {e}") logger.error(f"Response content: {content_text[:500]}") return None # Check different response formats employees = None if isinstance(data, dict): if 'Data' in data: employees = data['Data'] logger.info(f"Found 'Data' property with {len(employees) if isinstance(employees, list) else 1} items") elif 'EmpId' in data: employees = [data] logger.info("Found single employee object with EmpId") else: # The whole response might be the employee data employees = [data] logger.info("Treating entire response as employee object") elif isinstance(data, list): employees = data logger.info(f"Response is a list with {len(employees)} items") if not employees: logger.warning("No employee data found in response") logger.warning(f"Full response data: {data}") return None logger.info(f"Total employees in response: {len(employees) if isinstance(employees, list) else 1}") # Find matching employee matched = None # Try to find by DomainName (matching EmployeeID from Excel) if isinstance(employees, list): logger.info(f"Searching in {len(employees)} employees for DomainName={staff_id}") # Log all employees for debugging for idx, emp in enumerate(employees): logger.info(f" [{idx}] EmpId={emp.get('EmpId')}, DomainName={emp.get('DomainName')}, FirstName={emp.get('FirstName')}, isInActive={emp.get('isInActive')}") matched = next((e for e in employees if str(e.get('DomainName', '')) == str(staff_id)), None) if not matched: # Also try matching by EmpId (in case Excel has EmpId instead of DomainName) logger.info(f"No match found by DomainName, trying EmpId={staff_id}") matched = next((e for e in employees if str(e.get('EmpId', '')) == str(staff_id)), None) if matched: logger.success(f"Found employee: EmpId={matched.get('EmpId')}, DomainName={matched.get('DomainName')}, isInActive={matched.get('isInActive')}") return matched logger.error(f"No match found! Looking for DomainName='{staff_id}' or EmpId='{staff_id}'") domain_names = [e.get('DomainName') for e in employees[:5]] emp_ids = [e.get('EmpId') for e in employees[:5]] logger.error(f"Available DomainNames: {domain_names}") logger.error(f"Available EmpIds: {emp_ids}") elif isinstance(employees, dict): logger.info(f"Single employee: DomainName={employees.get('DomainName')}, EmpId={employees.get('EmpId')}") if str(employees.get('DomainName', '')) == str(staff_id) or str(employees.get('EmpId', '')) == str(staff_id): matched = employees logger.success(f"Matched employee: EmpId={matched.get('EmpId')}") else: logger.error(f"Single employee doesn't match! Looking for DomainName='{staff_id}' or EmpId='{staff_id}'") logger.error(f"Employee has: DomainName={employees.get('DomainName')}, EmpId={employees.get('EmpId')}") if matched: logger.success(f"Found employee: EmpId={matched.get('EmpId')}, DomainName={matched.get('DomainName')}") return matched logger.warning(f"No employee found matching: {staff_id}") return None except Exception as e: logger.error(f"Failed to get employee info (StaffId: {staff_id}): {e}") import traceback logger.error(traceback.format_exc()) return None def set_employee_inactive(session, employee_info): """Set employee to inactive status""" try: url = f"{BASE_URL}/Employee/SaveEmpInfo/" # Build request data save_body = { 'FirstName': employee_info.get('FirstName'), 'LastName': employee_info.get('LastName'), 'Email': employee_info.get('Email'), 'PhoneNumber': employee_info.get('PhoneNumber'), 'CultureName': employee_info.get('CultureName'), 'MaxParallelLogin': employee_info.get('MaxParallelLogin'), 'RoleId': employee_info.get('RoleId'), 'TeamId': employee_info.get('TeamId'), 'languageId': employee_info.get('languageId'), 'DomainName': employee_info.get('DomainName'), 'DataRowVersion': employee_info.get('DataRowVersion'), 'isLock': employee_info.get('isLock'), 'isInActive': True, # Key: set to inactive 'EmpId': employee_info.get('EmpId'), 'isReloadForNewConfig': False, 'StaffId': employee_info.get('StaffId'), 'Password': employee_info.get('Password'), 'isADAccount': employee_info.get('isADAccount'), 'PasswordPolicyId': employee_info.get('PasswordPolicyId'), 'lastPassChangeDTM': employee_info.get('lastPassChangeDTM') } response = session.post( url, headers={ 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate, br, zstd', 'Accept-Language': 'en-US,en;q=0.9', 'Content-Type': 'application/json; charset=UTF-8', 'X-Requested-With': 'XMLHttpRequest', 'Origin': BASE_URL, 'Referer': f"{BASE_URL}/Employee/EmployeeCrud?id={employee_info.get('EmpId')}&keepThis=true&" }, json=save_body ) return response.status_code == 200 except Exception as e: logger.error(f"Failed to set employee inactive: {e}") return False def get_company_accounts(session, employee_id): """Get company accounts (Opera External IDs) for an employee""" try: url = f"{BASE_URL}/OperaExternalId/GetEmployeeInfo" body = "sort=&group=&filter=" logger.info(f"Fetching company accounts for EmployeeID: {employee_id}") response = session.post( url, headers={ 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate, br, zstd', 'Accept-Language': 'en-US,en;q=0.9,zh-HK;q=0.8,zh;q=0.7', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Origin': BASE_URL, 'Referer': f'{BASE_URL}/OperaExternalId/ExternalIDInfo', 'X-Requested-With': 'XMLHttpRequest' }, data=body ) logger.info(f"Company Accounts Response Status: {response.status_code}") if response.status_code != 200: logger.error(f"Unexpected status code: {response.status_code}") return None # Parse response - it returns a JSON object with a Data array content_text = response.text.strip() accounts = [] logger.info(f"Response preview (first 200 chars): {content_text[:200]}") try: data = json.loads(content_text) # Check if response has Data key if isinstance(data, dict) and 'Data' in data: accounts = data['Data'] logger.info(f"Found 'Data' array with {len(accounts)} accounts") elif isinstance(data, list): accounts = data logger.info(f"Response is a list with {len(accounts)} accounts") else: logger.error(f"Unexpected response format") logger.error(f"Response type: {type(data)}") return None except json.JSONDecodeError as e: logger.error(f"Failed to parse JSON response: {e}") logger.error(f"Response content (first 500 chars): {content_text[:500]}") return None # Log first few accounts for debugging for idx, acc in enumerate(accounts[:3]): logger.info(f" [{idx}] ExternalCode={acc.get('ExternalCode')}, IsInactive={acc.get('IsInactive')}, EmployeeName={acc.get('EmployeeName')}") logger.info(f"Total company accounts in response: {len(accounts)}") # Filter accounts matching the employee_id (ExternalCode starts with employee_id) matched_accounts = [] for acc in accounts: external_code = acc.get('ExternalCode', '') # Strict match: ExternalCode must start with employee_id # e.g., EmployeeID="114764" matches ExternalCode="114764G", "114764H", etc. if external_code and str(external_code).startswith(str(employee_id)): matched_accounts.append(acc) logger.info(f" ✓ Matched: ExternalCode={external_code}, IsInactive={acc.get('IsInactive')}, ExternalId={acc.get('ExternalId')}") logger.info(f"Found {len(matched_accounts)} accounts for EmployeeID={employee_id}") return matched_accounts except Exception as e: logger.error(f"Failed to get company accounts: {e}") import traceback logger.error(traceback.format_exc()) return None def set_company_account_inactive_playwright(context, account): """Set a company account (Opera External ID) to inactive using Playwright""" page = None try: page = context.new_page() external_id = account.get('ExternalId') external_code = account.get('ExternalCode') # Navigate to EmployeeCRUD page crud_url = f"{BASE_URL}/OperaExternalId/EmployeeCRUD?KeyId={external_id}&SkipParentReload=true&keepThis=true&" logger.info(f"[Playwright] Navigating to EmployeeCRUD page for {external_code}...") page.goto(crud_url, wait_until='domcontentloaded') page.wait_for_load_state('networkidle', timeout=5000) # Wait for the page to load time.sleep(0.5) # Try to find and click the IsInactive checkbox or toggle logger.info(f"[Playwright] Looking for IsInactive element...") # Try multiple selectors for the IsInactive element selectors_to_try = [ 'input[name="IsInactive"]', 'input[type="checkbox"]', '#IsInactive', 'input[id*="IsInactive"]' ] checkbox_found = False for selector in selectors_to_try: try: checkbox = page.locator(selector).first if checkbox.count() > 0: # Check current state is_checked = checkbox.is_checked() logger.info(f"[Playwright] Found checkbox with selector: {selector}, current state: {is_checked}") if not is_checked: # Click to check (set to inactive) checkbox.click(timeout=5000) logger.success(f"[Playwright] Clicked checkbox for {external_code}") checkbox_found = True time.sleep(0.3) else: logger.info(f"[Playwright] {external_code} is already inactive") checkbox_found = True break except Exception as e: continue if not checkbox_found: logger.warning(f"[Playwright] Could not find IsInactive checkbox for {external_code}") page.close() return False # Look for and click Save button logger.info(f"[Playwright] Looking for Save button...") save_selectors = [ 'button:has-text("Save")', 'input[type="submit"][value="Save"]', 'button[type="submit"]', '#btnSave', 'button:has-text("Submit")' ] save_clicked = False for selector in save_selectors: try: save_button = page.locator(selector).first if save_button.count() > 0: save_button.click(timeout=5000) logger.success(f"[Playwright] Clicked Save button for {external_code}") save_clicked = True break except Exception as e: continue if not save_clicked: logger.warning(f"[Playwright] Could not find Save button, trying JavaScript submit...") try: page.evaluate('() => { $("form").submit(); }') save_clicked = True except: pass # Wait for response time.sleep(1) page.close() return save_clicked except Exception as e: logger.error(f"[Playwright] Failed to set {account.get('ExternalCode')} inactive: {e}") import traceback logger.error(traceback.format_exc()) if page: try: page.close() except: pass return False def set_company_account_inactive(session, account): """Set a company account (Opera External ID) to inactive""" logger.warning("Direct API save is not supported. Please use Playwright mode.") return False def get_all_employees(session): """Get all employees from GetEmpInfo API""" try: url = f"{BASE_URL}/Employee/GetEmpInfo" body = "sort=&group=&filter=" logger.info(f"Fetching all employees from: {url}") response = session.post( url, headers={ 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate, br, zstd', 'Accept-Language': 'en-US,en;q=0.9,zh-HK;q=0.8,zh;q=0.7', 'Cache-Control': 'no-cache', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Origin': BASE_URL, 'Pragma': 'no-cache', 'Referer': f'{BASE_URL}/Employee/EmployeeInfo', 'X-Requested-With': 'XMLHttpRequest' }, data=body ) logger.info(f"Response Status Code: {response.status_code}") if response.status_code != 200: logger.error(f"Unexpected status code: {response.status_code}") return None data = response.json() # Extract employees from Data key if isinstance(data, dict) and 'Data' in data: employees = data['Data'] logger.info(f"Found {len(employees)} employees") return employees else: logger.error(f"Unexpected response format") logger.error(f"Response keys: {list(data.keys()) if isinstance(data, dict) else 'not a dict'}") return None except Exception as e: logger.error(f"Failed to get all employees: {e}") import traceback logger.error(traceback.format_exc()) return None def save_employees_to_csv(employees, filepath): """Save employee data to CSV file""" try: if not employees: logger.warning("No employees to save") return False # Define CSV columns - using all fields from employee data fieldnames = [ 'EmpId', 'DomainId', 'DomainName', 'FirstName', 'LastName', 'Email', 'PhoneNumber', 'isLock', 'isInActive', 'RoleId', 'TeamId', 'CultureName', 'MaxParallelLogin' ] # Add any additional fields from the first employee that aren't in our list first_emp = employees[0] for key in first_emp.keys(): if key not in fieldnames: fieldnames.append(key) with open(filepath, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore') writer.writeheader() writer.writerows(employees) logger.success(f"Saved {len(employees)} employees to: {filepath}") return True except Exception as e: logger.error(f"Failed to save CSV: {e}") import traceback logger.error(traceback.format_exc()) return False def get_all_comp_users(session): """Get all company accounts (Opera External IDs) and export to CSV""" try: url = f"{BASE_URL}/OperaExternalId/GetEmployeeInfo" body = "sort=&group=&filter=" logger.info(f"Fetching all company accounts from: {url}") response = session.post( url, headers={ 'Accept': '*/*', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Origin': BASE_URL, 'Referer': f'{BASE_URL}/OperaExternalId/ExternalIDInfo', 'X-Requested-With': 'XMLHttpRequest' }, data=body ) logger.info(f"Response Status Code: {response.status_code}") if response.status_code != 200: logger.error(f"Unexpected status code: {response.status_code}") return None # Parse JSON response data = response.json() # Extract accounts from Data key if isinstance(data, dict) and 'Data' in data: accounts = data['Data'] logger.info(f"Found {len(accounts)} company accounts") return accounts else: logger.error(f"Unexpected response format") logger.error(f"Response keys: {list(data.keys()) if isinstance(data, dict) else 'not a dict'}") return None except Exception as e: logger.error(f"Failed to get all company accounts: {e}") import traceback logger.error(traceback.format_exc()) return None def save_comp_users_to_csv(accounts, filepath): """Save company accounts to CSV file""" try: if not accounts: logger.warning("No company accounts to save") return False # Define CSV columns fieldnames = [ 'EmployeeId', 'EmployeeName', 'ExternalId', 'ExternalCode', 'IdDescription', 'IsInactive', 'DataRowVersion' ] with open(filepath, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore') writer.writeheader() writer.writerows(accounts) logger.success(f"Saved {len(accounts)} company accounts to: {filepath}") return True except Exception as e: logger.error(f"Failed to save CSV: {e}") import traceback logger.error(traceback.format_exc()) return False # ============================================ # Main Program # ============================================ def main(get_all_users=False, get_all_comp=False, disable_comp=False, disable_user=False, disable_both=False): # -disablecompanduser mode: First disable comp accounts, then disable user if disable_both: logger.info("=" * 50) logger.info("Complete Termination Mode") logger.info("Step 1: Disable Company Accounts") logger.info("Step 2: Disable HUB Users") logger.info("=" * 50) # Step 1: Disable company accounts logger.info("=== Phase 1: Disabling Company Accounts ===") disable_company_accounts_playwright(LOGIN_USERNAME, LOGIN_PASSWORD) # Step 2: Disable HUB users logger.info("=== Phase 2: Disabling HUB Users ===") # Re-login for the second phase cookies = get_cookies_via_playwright(LOGIN_USERNAME, LOGIN_PASSWORD) if not cookies or not cookies.get('ASPNET_SessionId'): logger.error("Failed to re-login for phase 2") return session = create_session_with_cookies( cookies['ASPNET_SessionId'], cookies.get('HUBUserSessionId', '') ) # Initialize page and process users try: session.get(f"{BASE_URL}/Employee/EmployeeInfo", headers={'Referer': f'{BASE_URL}/'}) except: pass time.sleep(2) # Read Excel and disable users employees = read_excel_data(EXCEL_PATH) if not employees: logger.warning("No employee data found") return # Process each employee (disable user) success_count = 0 fail_count = 0 results = [] for emp in employees: logger.info("-" * 50) logger.info(f"Processing: EmployeeID={emp['EmployeeID']}") emp_info = get_employee_info(session, emp['EmployeeID']) if not emp_info: logger.error("Unable to get employee info") fail_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'FAILED', 'Message': 'Unable to get employee info' }) continue logger.info(f"Found employee: EmpId={emp_info.get('EmpId')}, isInActive={emp_info.get('isInActive')}") # Check if already inactive if emp_info.get('isInActive') is True: logger.info("Employee already inactive") success_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'SKIPPED', 'Message': 'Already inactive' }) continue # Set to inactive result = set_employee_inactive(session, emp_info) if result: logger.success(f"Successfully set {emp['SamAccountName']} to Inactive") success_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'SUCCESS', 'Message': 'Successfully set to Inactive' }) else: logger.error(f"Failed to set {emp['SamAccountName']} Inactive") fail_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'FAILED', 'Message': 'Failed to set Inactive' }) time.sleep(2) # Export phase 2 results logger.info("=" * 50) logger.info("Phase 2 Completed!") logger.info(f"Success: {success_count} | Failed: {fail_count} | Total: {len(employees)}") logger.info("=" * 50) csv_path = SCRIPT_DIR / f"DisableUser_Result_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" with open(csv_path, 'w', newline='', encoding='utf-8') as f: if results: writer = csv.DictWriter(f, fieldnames=results[0].keys()) writer.writeheader() writer.writerows(results) logger.success(f"Phase 2 Results exported to: {csv_path}") logger.success(f"Complete termination finished!") return # -disablecomp mode: Use dedicated Playwright function if disable_comp: disable_company_accounts_playwright(LOGIN_USERNAME, LOGIN_PASSWORD) return # -getallcompuser mode: Export all company accounts if get_all_comp: logger.info("=" * 50) logger.info("Get All Company Accounts Mode") logger.info("=" * 50) cookies = get_cookies_via_playwright(LOGIN_USERNAME, LOGIN_PASSWORD) if not cookies or not cookies.get('ASPNET_SessionId'): logger.error("Failed to login") return session = create_session_with_cookies( cookies['ASPNET_SessionId'], cookies.get('HUBUserSessionId', '') ) # Initialize ExternalIDInfo page try: session.get(f"{BASE_URL}/OperaExternalId/ExternalIDInfo", headers={'Referer': f'{BASE_URL}/'}) except: pass time.sleep(1) # Get all company accounts all_comp = get_all_comp_users(session) if all_comp: csv_path = SCRIPT_DIR / f"AllCompUsers_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" if save_comp_users_to_csv(all_comp, csv_path): logger.success("=" * 50) logger.success(f"Successfully exported {len(all_comp)} company accounts") logger.success(f"CSV file: {csv_path}") logger.success(f"Log file: {LOG_PATH}") logger.success("=" * 50) return else: logger.error("Failed to fetch company accounts") return # -getalluser mode or -disableuser mode or default if get_all_users: logger.info("=" * 50) logger.info("Get All Users Mode") logger.info("=" * 50) elif disable_user: logger.info("=" * 50) logger.info("Disable HUB Users Mode") logger.info("=" * 50) else: logger.info("=" * 50) logger.info("Batch Employee Inactive Processing Started (Default)") logger.info("=" * 50) # Step 1: Auto-login and get cookies logger.info("Step 1: Auto-login to get session cookies...") cookies = get_cookies_via_playwright(LOGIN_USERNAME, LOGIN_PASSWORD) if not cookies or not cookies.get('ASPNET_SessionId'): logger.error("ERROR: Failed to get cookies via auto-login") logger.error("Please check if:") logger.error(" 1. Playwright is installed correctly") logger.error(" 2. Username and password are correct") logger.error(" 3. Network connection is stable") return # Step 2: Create session with cookies logger.success("Step 2: Creating session from cookies...") session = create_session_with_cookies( cookies['ASPNET_SessionId'], cookies.get('HUBUserSessionId', '') ) # Step 3: Initialize EmployeeInfo page logger.info("Step 3: Initializing EmployeeInfo page...") try: response = session.get( f"{BASE_URL}/Employee/EmployeeInfo", headers={ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8', 'Referer': f'{BASE_URL}/' } ) logger.success("EmployeeInfo page initialized successfully") except Exception as e: logger.warning(f"Warning: Could not initialize EmployeeInfo page: {e}") time.sleep(2) # If -getalluser mode, fetch all users and save to CSV if get_all_users: logger.info("Step 4: Fetching all employees...") all_employees = get_all_employees(session) if all_employees: # Save to CSV with timestamp csv_path = SCRIPT_DIR / f"AllEmployees_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" if save_employees_to_csv(all_employees, csv_path): logger.success("=" * 50) logger.success(f"Successfully exported {len(all_employees)} employees") logger.success(f"CSV file: {csv_path}") logger.success(f"Log file: {LOG_PATH}") logger.success("=" * 50) return else: logger.error("Failed to save employees to CSV") return else: logger.error("Failed to fetch employees") return # If -disablecomp mode, disable company accounts if disable_comp: logger.info("Step 4: Disabling company accounts mode...") # Initialize ExternalIDInfo page try: response = session.get( f"{BASE_URL}/OperaExternalId/ExternalIDInfo", headers={ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Referer': f'{BASE_URL}/' } ) logger.success("ExternalIDInfo page initialized successfully") except Exception as e: logger.warning(f"Warning: Could not initialize ExternalIDInfo page: {e}") time.sleep(1) # Step 5: Read Excel data logger.info("Step 5: Reading Excel file...") employees = read_excel_data(EXCEL_PATH) if not employees: logger.warning("No employee data found to process") return logger.info(f"Total {len(employees)} employee records to process") # Step 6: Process each employee's company accounts success_count = 0 fail_count = 0 skipped_count = 0 results = [] for emp in employees: logger.info("-" * 50) logger.info(f"Processing: EmployeeID={emp['EmployeeID']}, SamAccountName={emp['SamAccountName']}") # Get company accounts for this employee accounts = get_company_accounts(session, emp['EmployeeID']) if not accounts: logger.warning(f"No company accounts found for EmployeeID={emp['EmployeeID']}") fail_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'FAILED', 'Message': 'No company accounts found' }) continue logger.info(f"Found {len(accounts)} company accounts") # Count accounts that need to be disabled active_accounts = [acc for acc in accounts if acc.get('IsInactive') is False] already_inactive = [acc for acc in accounts if acc.get('IsInactive') is True] logger.info(f" Active accounts (need disable): {len(active_accounts)}") logger.info(f" Already inactive: {len(already_inactive)}") if not active_accounts: logger.info(f"All accounts are already inactive for EmployeeID={emp['EmployeeID']}") skipped_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'SKIPPED', 'Message': f'All {len(accounts)} accounts already inactive' }) continue # Disable each active account account_success = 0 account_fail = 0 for acc in active_accounts: logger.info(f" Processing account: {acc.get('ExternalCode')}") if set_company_account_inactive(session, acc): account_success += 1 else: account_fail += 1 time.sleep(0.5) # Small delay between requests # Record results if account_fail == 0: logger.success(f"Successfully disabled {account_success}/{len(active_accounts)} accounts for {emp['SamAccountName']}") success_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'SUCCESS', 'Message': f'Disabled {account_success} accounts' }) else: logger.error(f"Partial failure for {emp['SamAccountName']}: {account_success} success, {account_fail} failed") fail_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'PARTIAL', 'Message': f'{account_success} success, {account_fail} failed' }) time.sleep(1) # Delay between employees # Step 7: Export results logger.info("=" * 50) logger.info("Processing completed!") logger.info(f"Success: {success_count} | Skipped: {skipped_count} | Failed: {fail_count} | Total: {len(employees)}") logger.info("=" * 50) # Export to CSV csv_path = SCRIPT_DIR / f"DisableComp_Result_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" with open(csv_path, 'w', newline='', encoding='utf-8') as f: if results: writer = csv.DictWriter(f, fieldnames=results[0].keys()) writer.writeheader() writer.writerows(results) logger.success(f"Results exported to: {csv_path}") logger.success(f"Log file: {LOG_PATH}") return # Normal mode: Read Excel data and process employee inactive # Step 5: Read Excel data logger.info("Step 5: Reading Excel file...") employees = read_excel_data(EXCEL_PATH) if not employees: logger.warning("No employee data found to process") return logger.info(f"Total {len(employees)} employee records to process") # Step 6: Process each employee success_count = 0 fail_count = 0 results = [] for emp in employees: logger.info("-" * 50) logger.info(f"Processing: EmployeeID={emp['EmployeeID']}, SamAccountName={emp['SamAccountName']}") # Get employee info emp_info = get_employee_info(session, emp['EmployeeID']) if not emp_info: logger.error("Unable to get employee info, skipping this employee") fail_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'FAILED', 'Message': 'Unable to get employee info' }) continue logger.info(f"Found employee: EmpId={emp_info.get('EmpId')}, Name={emp_info.get('FirstName', '')} {emp_info.get('LastName', '')}, isInActive={emp_info.get('isInActive')}") # Check if already inactive if emp_info.get('isInActive') is True: logger.info(f"Employee is already inactive (isInActive=True), skipping...") success_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'SKIPPED', 'Message': 'Already inactive' }) continue # Set to inactive result = set_employee_inactive(session, emp_info) if result: logger.success(f"Employee {emp['SamAccountName']} successfully set to Inactive") success_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'SUCCESS', 'Message': 'Successfully set to Inactive' }) else: logger.error(f"Employee {emp['SamAccountName']} failed to set Inactive") fail_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'FAILED', 'Message': 'Failed to set Inactive' }) # Delay to avoid rate limiting time.sleep(2) # Step 6: Export results logger.info("=" * 50) logger.info("Processing completed!") logger.info(f"Success: {success_count} | Failed: {fail_count} | Total: {len(employees)}") logger.info("=" * 50) # Export to CSV csv_path = SCRIPT_DIR / f"BatchInactive_Result_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" with open(csv_path, 'w', newline='', encoding='utf-8') as f: if results: writer = csv.DictWriter(f, fieldnames=results[0].keys()) writer.writeheader() writer.writerows(results) logger.success(f"Results exported to: {csv_path}") logger.success(f"Log file: {LOG_PATH}") def disable_company_accounts_playwright(username, password): """Disable company accounts using Playwright (for -disablecomp mode)""" logger.info("=" * 50) logger.info("Disable Company Accounts Mode (Playwright)") logger.info("=" * 50) with sync_playwright() as p: # Launch browser browser = p.chromium.launch( headless=HEADLESS_MODE, slow_mo=100 ) context = browser.new_context() page = context.new_page() try: # ===== LOGIN PROCESS ===== logger.info("Step 1: Auto-login using Playwright...") logger.info("Navigating to homepage...") page.goto(f"{BASE_URL}/") time.sleep(1) # Fill username logger.info("Filling username...") try: page.get_by_role("textbox", name="Username...").fill(username) except: page.fill('input[name="UserName"]', username) time.sleep(0.5) # Fill password logger.info("Filling password...") try: page.get_by_role("textbox", name="Password...").fill(password) page.get_by_role("textbox", name="Password...").press("Tab") except: page.fill('input[name="Password"]', password) page.keyboard.press("Tab") time.sleep(0.5) # Check for wait message logger.info("Checking for wait message...") page_content = page.content().lower() if 'maximum login session reached' in page_content or 'kindly wait' in page_content: logger.warning("Detected wait message, waiting 21 seconds...") time.sleep(21) # Click OK try: page.get_by_role("button", name="OK").click(timeout=5000) except: try: page.click('button:has-text("OK")', timeout=5000) except: page.keyboard.press("Enter") time.sleep(0.5) # Re-fill password logger.info("Re-filling password...") try: page.get_by_role("textbox", name="Password...").fill(password) page.get_by_role("textbox", name="Password...").press("Tab") except: page.fill('input[name="Password"]', password) page.keyboard.press("Tab") time.sleep(0.5) # Select site logger.info("Selecting site...") try: page.select_option('select#ddlSite', str(SITE_ID), timeout=5000) except: pass time.sleep(0.5) # Click Submit logger.info("Clicking Submit...") try: page.get_by_role("button", name="Submit").click() except: page.click('button:has-text("Submit")') # Wait for login logger.info("Waiting for login to complete...") for i in range(25): if '/User/Login' not in page.url: logger.success("Login successful!") break time.sleep(1) time.sleep(2) # ===== READ EXCEL ===== logger.info("Step 2: Reading Excel file...") employees = read_excel_data(EXCEL_PATH) if not employees: logger.warning("No employee data found") browser.close() return logger.info(f"Total {len(employees)} employees to process") # ===== PROCESS EACH EMPLOYEE ===== success_count = 0 fail_count = 0 skipped_count = 0 results = [] for emp in employees: logger.info("-" * 50) logger.info(f"Processing: EmployeeID={emp['EmployeeID']}") # Navigate to ExternalIDInfo page logger.info("Navigating to ExternalIDInfo page...") page.goto(f"{BASE_URL}/OperaExternalId/ExternalIDInfo") page.wait_for_load_state('networkidle', timeout=5000) time.sleep(1) # Use requests session to get company accounts # First, get cookies from Playwright context cookies_dict = {c['name']: c['value'] for c in context.cookies()} asp_session_id = cookies_dict.get('ASP.NET_SessionId', '') hub_session_id = cookies_dict.get('HUBUserSessionId', '') session = create_session_with_cookies(asp_session_id, hub_session_id) accounts = get_company_accounts(session, emp['EmployeeID']) if not accounts: logger.warning(f"No company accounts found") fail_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'FAILED', 'Message': 'No company accounts found' }) continue # Count active accounts active_accounts = [acc for acc in accounts if acc.get('IsInactive') is False] already_inactive = [acc for acc in accounts if acc.get('IsInactive') is True] logger.info(f" Active: {len(active_accounts)}, Already inactive: {len(already_inactive)}") if not active_accounts: logger.info("All accounts already inactive") skipped_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'SKIPPED', 'Message': 'All accounts already inactive' }) continue # Disable each active account using Playwright account_success = 0 account_fail = 0 # Stay on ExternalIDInfo page and try to open edit dialog for each account for acc in active_accounts: external_id = acc.get('ExternalId') external_code = acc.get('ExternalCode') logger.info(f" Processing: {external_code}") # Try to open the edit dialog using JavaScript (simulating ThickBox) try: # Use JavaScript to open the ThickBox dialog dialog_url = f"{BASE_URL}/OperaExternalId/EmployeeCRUD?KeyId={external_id}&SkipParentReload=true&keepThis=true&TB_iframe=true&height=400&width=600" js_code = f''' tb_show("{external_code}", "{dialog_url}", false); ''' page.evaluate(js_code) logger.info(f" Opened dialog using JavaScript") time.sleep(1) # Now wait for iframe iframe = page.locator('iframe[name^="TB_iframeContent"]').first iframe.wait_for(state='visible', timeout=5000) frame = iframe.content_frame logger.info(f" Got iframe content") # Find and check IsInactive checkbox = frame.locator('#IsInactive').first is_checked = checkbox.is_checked() if not is_checked: checkbox.check(timeout=3000) logger.info(f" ✓ Checked IsInactive") time.sleep(0.5) else: logger.info(f" Already inactive") account_success += 1 # Close dialog try: page.evaluate('tb_remove()') except: pass continue # Click Save save_btn = frame.get_by_role("button", name="Save").first save_btn.click(timeout=3000) logger.info(f" ✓ Clicked Save") account_success += 1 time.sleep(0.5) # Click Ok if present try: ok_btn = frame.get_by_role("button", name="Ok") if ok_btn.count() > 0: ok_btn.click(timeout=2000) logger.info(f" ✓ Clicked Ok") except: pass # Close dialog try: page.evaluate('tb_remove()') except: pass except Exception as e: logger.warning(f" Error: {e}") import traceback logger.warning(traceback.format_exc()) account_fail += 1 time.sleep(0.5) # Disable each active account using Playwright account_success = 0 account_fail = 0 for acc in active_accounts: external_id = acc.get('ExternalId') external_code = acc.get('ExternalCode') logger.info(f" Processing: {external_code}") # Navigate to EmployeeCRUD page crud_url = f"{BASE_URL}/OperaExternalId/EmployeeCRUD?KeyId={external_id}&SkipParentReload=true&keepThis=true&" logger.info(f" Navigating to: {crud_url}") page.goto(crud_url, wait_until='domcontentloaded', timeout=10000) time.sleep(1) # Check if there's an iframe or direct page success = False frame = None # Debug: Save screenshot before processing try: screenshot_path = SCRIPT_DIR / f"debug_{external_code}_{datetime.now().strftime('%H%M%S')}.png" page.screenshot(path=str(screenshot_path)) logger.info(f" Screenshot saved: {screenshot_path}") except: pass # First, try to find iframe try: iframe = page.locator('iframe[name^="TB_iframeContent"]').first if iframe.count() > 0: logger.info(f" Found iframe, waiting for it...") iframe.wait_for(state='visible', timeout=3000) frame = iframe.content_frame logger.info(f" Switched to iframe") except: logger.info(f" No iframe found, using main page") # Use frame if found, otherwise use main page work_page = frame if frame else page # Try to find and check IsInactive checkbox try: # Try multiple selectors checkbox = work_page.locator('#IsInactive').first if checkbox.count() == 0: checkbox = work_page.locator('input[name="IsInactive"]').first if checkbox.count() > 0: is_checked = checkbox.is_checked() logger.info(f" Checkbox found, state: {is_checked}") if not is_checked: checkbox.check(timeout=3000) logger.info(f" ✓ Checked IsInactive for {external_code}") time.sleep(0.5) else: logger.info(f" {external_code} already inactive") account_success += 1 success = True # Close dialog if in iframe if frame: try: frame.locator('body').click() time.sleep(0.3) ok_btn = frame.get_by_role("button", name="Ok") if ok_btn.count() > 0: ok_btn.click(timeout=2000) logger.info(f" Closed dialog") except: pass continue else: logger.warning(f" Checkbox not found") except Exception as e: logger.warning(f" Error with checkbox: {e}") # If not already inactive, try to save if not success: # Try to find and click Save button save_clicked = False save_selectors = [ 'button:has-text("Save")', 'input[type="submit"][value="Save"]', 'button[type="submit"]', '#btnSave', '[id*="Save"]' ] for selector in save_selectors: try: save_btn = work_page.locator(selector).first if save_btn.count() > 0: save_btn.click(timeout=3000) logger.info(f" ✓ Clicked Save button for {external_code}") save_clicked = True account_success += 1 time.sleep(0.5) # Click Ok if dialog appears try: ok_btn = work_page.get_by_role("button", name="Ok") if ok_btn.count() > 0: ok_btn.click(timeout=2000) logger.info(f" ✓ Clicked Ok button") except: pass break except: continue if save_clicked: success = True else: logger.warning(f" Save button not found") account_fail += 1 if not success and account_success == 0: account_fail += 1 time.sleep(0.5) # Record results if account_fail == 0: logger.success(f"Disabled {account_success}/{len(active_accounts)} accounts") success_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'SUCCESS', 'Message': f'Disabled {account_success} accounts' }) else: logger.error(f"Partial failure: {account_success} success, {account_fail} failed") fail_count += 1 results.append({ 'EmployeeID': emp['EmployeeID'], 'SamAccountName': emp['SamAccountName'], 'Status': 'PARTIAL', 'Message': f'{account_success} success, {account_fail} failed' }) time.sleep(1) # ===== EXPORT RESULTS ===== logger.info("=" * 50) logger.info("Processing completed!") logger.info(f"Success: {success_count} | Skipped: {skipped_count} | Failed: {fail_count} | Total: {len(employees)}") logger.info("=" * 50) csv_path = SCRIPT_DIR / f"DisableComp_Result_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" with open(csv_path, 'w', newline='', encoding='utf-8') as f: if results: writer = csv.DictWriter(f, fieldnames=results[0].keys()) writer.writeheader() writer.writerows(results) logger.success(f"Results exported to: {csv_path}") logger.success(f"Log file: {LOG_PATH}") except Exception as e: logger.error(f"Error in disable_company_accounts_playwright: {e}") import traceback logger.error(traceback.format_exc()) finally: browser.close() if __name__ == "__main__": # Parse command line arguments parser = argparse.ArgumentParser( description='HUB System Batch Employee Inactive Script', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=''' Examples: # Export modes python batch_inactive_auto.py -getalluser # Export all HUB employees to CSV python batch_inactive_auto.py -getallcompuser # Export all company accounts to CSV # Disable modes python batch_inactive_auto.py # Disable HUB users (default) python batch_inactive_auto.py -disableuser # Disable HUB users (explicit) python batch_inactive_auto.py -disablecomp # Disable company accounts only python batch_inactive_auto.py -disablecompanduser # Complete: disable comp accounts, then HUB users ''' ) # Export parameters parser.add_argument('-getalluser', action='store_true', help='Export all HUB employees to CSV') parser.add_argument('-getallcompuser', action='store_true', help='Export all company accounts (Opera External IDs) to CSV') # Disable parameters parser.add_argument('-disableuser', action='store_true', help='Disable HUB users only') parser.add_argument('-disablecomp', action='store_true', help='Disable company accounts only') parser.add_argument('-disablecompanduser', action='store_true', help='Complete process: disable company accounts, then disable HUB users') args = parser.parse_args() # Count how many modes are selected export_modes = sum([args.getalluser, args.getallcompuser]) disable_modes = sum([args.disableuser, args.disablecomp, args.disablecompanduser]) total_modes = export_modes + disable_modes # Validate: only one mode can be selected at a time if total_modes > 1: parser.error("Can only use one mode at a time. Please choose only one of: -getalluser, -getallcompuser, -disableuser, -disablecomp, -disablecompanduser") # Run main with appropriate mode main( get_all_users=args.getalluser, get_all_comp=args.getallcompuser, disable_comp=args.disablecomp, disable_user=args.disableuser, disable_both=args.disablecompanduser )