""" NT License Checker ================================================================================================ """ import json import time from pathlib import Path from dataclasses import dataclass from typing import List, Dict, Optional from datetime import datetime import sys from concurrent.futures import ThreadPoolExecutor, as_completed import threading import queue # Selenium libraries for web scraping from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.common.exceptions import TimeoutException, WebDriverException, NoSuchElementException # Data processing libraries from rapidfuzz import fuzz import pandas as pd # ================================================================================================ # CONFIGURATION - Change these settings if needed # ================================================================================================ @dataclass class SimpleConfig: """Simple configuration""" # Website URL - Don't change unless NT government changes their site website_url: str = "https://licensingnt.nt.gov.au/PublicRegister/" # How similar names should be to match (85 = 85% similar) name_similarity_required: int = 85 # How many times to retry if website is slow max_retry_attempts: int = 2 # How long to wait for website to respond (seconds) request_timeout_seconds: int = 15 # Delay between requests to be nice to the website (seconds) delay_between_requests: float = 0.3 # How many parallel browsers to run (3 is optimal for Selenium) parallel_browsers: int = 3 # Test mode - only process first 10 records for testing test_mode: bool = False test_record_limit: int = 10 # ================================================================================================ # DATA CLASSES # ================================================================================================ @dataclass class EmployeeRecord: """Holds one employee's information from CSV""" payroll_number: str employee_name: str license_number: str csv_row_number: int = 0 def clean_data(self): """Remove extra spaces and clean up the data""" self.payroll_number = str(self.payroll_number).strip() self.employee_name = str(self.employee_name).strip() self.license_number = str(self.license_number).strip() def has_required_data(self) -> bool: """Check if this record has the minimum required information""" return bool(self.license_number and self.employee_name) @dataclass class SearchResult: """Holds the result of searching NT website""" found_name: str = "Not Found" license_type: str = "Not Found" license_status: str = "Not Found" name_matches: str = "No License" error_message: str = "" how_many_retries: int = 0 search_time_seconds: float = 0.0 was_cached: bool = False def is_successful(self) -> bool: """Did we successfully find license information?""" return self.found_name != "Not Found" and not self.error_message # ================================================================================================ # BROWSER MANAGER # ================================================================================================ class OptimizedBrowserManager: """Manages multiple Chrome browsers for reliable scraping""" def __init__(self, config: SimpleConfig): self.config = config self.browser_pool = queue.Queue() self.all_browsers = [] self.setup_complete = False self.lock = threading.Lock() def setup_browsers(self): """Create Chrome browsers for parallel processing""" if self.setup_complete: return with self.lock: if self.setup_complete: return print(f"Setting up {self.config.parallel_browsers} Chrome browsers...") for i in range(self.config.parallel_browsers): browser = self._create_chrome_browser() self.all_browsers.append(browser) self.browser_pool.put(browser) print(f" Browser {i + 1} ready") self.setup_complete = True print(f"All browsers ready for parallel processing!") def _create_chrome_browser(self) -> webdriver.Chrome: """Create one Chrome browser with optimal settings""" chrome_options = Options() # Optimized settings for speed and reliability chrome_options.add_argument("--headless") # No GUI needed chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--disable-gpu") chrome_options.add_argument("--disable-extensions") chrome_options.add_argument("--disable-logging") chrome_options.add_argument("--disable-web-security") chrome_options.add_argument("--window-size=1280,720") chrome_options.add_argument("--disable-background-timer-throttling") chrome_options.add_argument("--disable-renderer-backgrounding") chrome_options.add_argument("--disable-backgrounding-occluded-windows") # Additional stability options chrome_options.add_argument("--disable-ipc-flooding-protection") chrome_options.add_argument("--disable-features=TranslateUI") chrome_options.add_argument("--disable-default-apps") chrome_options.add_argument("--no-first-run") chrome_options.add_argument("--memory-pressure-off") return webdriver.Chrome(options=chrome_options) def get_browser(self): """Get a browser from the pool (thread-safe)""" if not self.setup_complete: self.setup_browsers() browser = self.browser_pool.get() # Wait for available browser wait_helper = WebDriverWait(browser, self.config.request_timeout_seconds) return browser, wait_helper def return_browser(self, browser): """Return browser to pool when done""" self.browser_pool.put(browser) def cleanup_all_browsers(self): """Close all browsers when program ends""" # Silent cleanup - only show errors failed_cleanups = 0 for browser in self.all_browsers: try: browser.quit() except: failed_cleanups += 1 if failed_cleanups > 0: print(f"WARNING: {failed_cleanups} browsers failed to close properly") # ================================================================================================ # LICENSE SEARCHER # ================================================================================================ class SeleniumLicenseSearcher: """Searches NT government website using Selenium """ def __init__(self, config: SimpleConfig, browser_manager: OptimizedBrowserManager): self.config = config self.browser_manager = browser_manager self.search_cache = {} # Store results to avoid duplicate searches self.cache_lock = threading.Lock() self.name_cache = {} # Cache normalized names for speed def normalize_employee_name(self, name: str) -> str: """Convert name to standard format for comparison with enhanced cleaning""" if not name: return "" # Check if we already normalized this name if name in self.name_cache: return self.name_cache[name] # Clean and normalize the name cleaned_name = self._clean_name_for_matching(name) # Convert "First Last" to "LAST, FIRST" format parts = cleaned_name.strip().upper().split() if len(parts) >= 2: normalized = f"{parts[-1]}, {' '.join(parts[:-1])}" else: normalized = cleaned_name.strip().upper() # Cache the result self.name_cache[name] = normalized return normalized def _clean_name_for_matching(self, name: str) -> str: """Clean name by removing common variations and formatting issues""" if not name: return "" # Convert to string and strip whitespace cleaned = str(name).strip() # Remove extra whitespace and normalize spacing cleaned = ' '.join(cleaned.split()) # Split into words for processing words = cleaned.upper().split() # Remove common middle name abbreviations and single letters filtered_words = [] for word in words: # Skip single letters (middle initials) unless it's the only remaining word if len(word) == 1 and len(words) > 1: continue # Skip common middle name patterns if word.endswith('.') and len(word) <= 3: continue filtered_words.append(word) # Handle hyphenated names processed_words = [] for word in filtered_words: if '-' in word: # Split hyphenated names and take both parts hyphen_parts = word.split('-') processed_words.extend(hyphen_parts) else: processed_words.append(word) # Remove empty strings processed_words = [word for word in processed_words if word.strip()] # Rejoin the cleaned words return ' '.join(processed_words) if processed_words else cleaned def check_name_similarity(self, csv_name: str, website_name: str) -> str: """Compare names with enhanced cleaning and return match status""" if website_name == "Not Found": return "Not Found" if not csv_name or not website_name: return "No Match" # Clean both names before comparison cleaned_csv = self._clean_name_for_matching(csv_name) cleaned_website = self._clean_name_for_matching(website_name) # Normalize both names to standard format norm_csv = self.normalize_employee_name(cleaned_csv) norm_website = self.normalize_employee_name(cleaned_website) # Quick exact match check on cleaned names if norm_csv == norm_website: return "Yes" # Try direct comparison of cleaned names (without format conversion) if cleaned_csv.upper() == cleaned_website.upper(): return "Yes" # Calculate similarity percentage using multiple methods for better accuracy similarity_methods = [ fuzz.token_set_ratio(norm_website, norm_csv), fuzz.token_sort_ratio(cleaned_website.upper(), cleaned_csv.upper()), fuzz.partial_ratio(cleaned_website.upper(), cleaned_csv.upper()), fuzz.ratio(cleaned_website.upper(), cleaned_csv.upper()) ] # Use the highest similarity score similarity = max(similarity_methods) if similarity >= self.config.name_similarity_required: return "Yes" else: return f"No ({similarity:.1f}%)" def search_single_license(self, employee: EmployeeRecord) -> SearchResult: """Search for one employee's license using Selenium""" # Check cache first to avoid duplicate searches if employee.license_number in self.search_cache: cached_result = self.search_cache[employee.license_number] # Update name match for this specific employee cached_result.name_matches = self.check_name_similarity( employee.employee_name, cached_result.found_name ) cached_result.was_cached = True return cached_result # Do the actual search with retries result = self._search_with_retries(employee) # Cache successful results if result.is_successful(): with self.cache_lock: self.search_cache[employee.license_number] = result return result def _search_with_retries(self, employee: EmployeeRecord) -> SearchResult: """Try searching multiple times if it fails""" if not employee.license_number.strip(): return SearchResult() last_error = "" for attempt in range(self.config.max_retry_attempts): try: browser, wait_helper = self.browser_manager.get_browser() try: start_time = time.time() result = self._do_selenium_search(browser, wait_helper, employee.license_number) search_time = time.time() - start_time if result.is_successful(): result.name_matches = self.check_name_similarity( employee.employee_name, result.found_name ) result.how_many_retries = attempt result.search_time_seconds = search_time return result finally: self.browser_manager.return_browser(browser) except Exception as e: last_error = str(e) if attempt < self.config.max_retry_attempts - 1: time.sleep(0.5 * (attempt + 1)) # Wait longer each retry # All attempts failed return SearchResult( error_message=f"Failed after {self.config.max_retry_attempts} attempts: {last_error}", how_many_retries=self.config.max_retry_attempts ) def _do_selenium_search(self, browser: webdriver.Chrome, wait_helper: WebDriverWait, license_number: str) -> SearchResult: """Actually search the NT government website using Selenium""" try: # Navigate to the search page browser.get(self.config.website_url) # Find and clear the license number input field license_input = wait_helper.until( EC.element_to_be_clickable((By.ID, "Licencenumber")) ) license_input.clear() license_input.send_keys(license_number.strip()) license_input.send_keys(Keys.RETURN) # Wait for results table to load results_table = wait_helper.until( EC.presence_of_element_located((By.ID, "ctl00_searchpage_ResultBox_ResultGrid")) ) # Small delay to ensure DOM is fully updated time.sleep(0.5) # Parse results from the table rows = browser.find_elements(By.CSS_SELECTOR, "#ctl00_searchpage_ResultBox_ResultGrid tr") if len(rows) > 1: # More than just header row cells = rows[1].find_elements(By.TAG_NAME, "td") if len(cells) >= 5: return SearchResult( found_name=cells[0].text.strip() or "Not Found", license_type=cells[1].text.strip() or "Not Found", license_status=cells[4].text.strip() or "Not Found" ) # No results found return SearchResult() except TimeoutException: raise Exception(f"Website timeout for license: {license_number}") except NoSuchElementException as e: raise Exception(f"Page element not found: {str(e)}") except WebDriverException as e: raise Exception(f"WebDriver error: {str(e)}") except Exception as e: raise Exception(f"Selenium search error: {str(e)}") # ================================================================================================ # CSV FILE HANDLER - Reads employee data from Excel/CSV files # ================================================================================================ class SimpleCSVHandler: """Reads employee data from CSV files""" @staticmethod def load_employee_data(file_path: str, config: SimpleConfig) -> List[EmployeeRecord]: """Load employee data from CSV file""" try: # Silent CSV reading - only show errors df = pd.read_csv(file_path, dtype=str, na_filter=False) df.columns = df.columns.str.strip().str.replace('\ufeff', '') # Check if required columns exist required_columns = ['Payroll Number', 'Employee Name', 'License Number'] missing_columns = [col for col in required_columns if col not in df.columns] if missing_columns: raise ValueError(f""" ERROR: Missing required columns: {', '.join(missing_columns)} Your CSV file MUST have these exact column names: - Payroll Number - Employee Name - License Number Current columns in your file: {list(df.columns)} """) # Convert to employee records employees = [] for index, row in df.iterrows(): employee = EmployeeRecord( payroll_number=row['Payroll Number'], employee_name=row['Employee Name'], license_number=row['License Number'], csv_row_number=index + 2 ) employee.clean_data() if employee.has_required_data(): employees.append(employee) if config.test_mode and len(employees) >= config.test_record_limit: print(f"TEST MODE: Processing only first {config.test_record_limit} records") break if not employees: raise ValueError("ERROR: No valid employee records found in CSV file") # Remove duplicates unique_employees = [] seen_licenses = set() duplicates_removed = 0 for employee in employees: if employee.license_number not in seen_licenses: seen_licenses.add(employee.license_number) unique_employees.append(employee) else: duplicates_removed += 1 # Only show duplicate message if there were duplicates if duplicates_removed > 0: print(f"NOTE: Removed {duplicates_removed} duplicate license numbers") return unique_employees except Exception as e: print(f"ERROR reading CSV file: {str(e)}") raise # ================================================================================================ # PROGRESS TRACKER # ================================================================================================ class SimpleProgressTracker: """Shows nice progress bar and processing statistics""" def __init__(self, total_employees: int): self.total_employees = total_employees self.completed = 0 self.successful = 0 self.failed = 0 self.cached = 0 self.start_time = time.time() self.last_update = 0 self.lock = threading.Lock() def show_header(self): """Show the processing header""" print(f"\n{'=' * 85}") print(f"NT LICENSE CHECKER - PROCESSING {self.total_employees} EMPLOYEES") print(f"{'=' * 85}") print(f"{'Progress':<15} {'Employee':<25} {'License':<15} {'Status':<12} {'Speed'}") print(f"{'-' * 85}") def update_progress(self, employee: EmployeeRecord, result: SearchResult): """Update progress display""" with self.lock: self.completed += 1 if result.was_cached: self.cached += 1 elif result.is_successful(): self.successful += 1 else: self.failed += 1 # Only update display every 0.5 seconds for smoother experience current_time = time.time() if current_time - self.last_update < 0.5: return self.last_update = current_time # Calculate progress percent = (self.completed / self.total_employees) * 100 progress_bar = self._make_progress_bar(percent) # Calculate speed elapsed = max(current_time - self.start_time, 0.01) speed = (self.completed / elapsed) * 60 # Records per minute # Prepare display data display_name = employee.employee_name[:23] + ".." if len( employee.employee_name) > 25 else employee.employee_name display_license = employee.license_number[:13] + ".." if len( employee.license_number) > 15 else employee.license_number # Determine status if result.was_cached: status = "Cached" elif result.is_successful(): status = "Success" else: status = "Failed" # Show progress line print(f"\r{progress_bar} {display_name:<25} {display_license:<15} {status:<12} {speed:.0f}/min", end="", flush=True) def _make_progress_bar(self, percent: float, width: int = 20) -> str: """Create ASCII progress bar""" filled = int(width * percent / 100) bar = 'X' * filled + '-' * (width - filled) return f"[{bar}] {percent:5.1f}%" def show_final_summary(self, elapsed_time: float, excel_file_path: str): """Show final processing summary""" print(f"\n\n{'=' * 85}") print("PROCESSING COMPLETED!") print(f"{'=' * 85}") print(f"Total employees processed: {self.completed}") print(f"Successful searches: {self.successful}") print(f"Failed searches: {self.failed}") print(f"Cached results: {self.cached}") print(f"Success rate: {(self.successful / max(self.completed, 1)) * 100:.1f}%") print(f"Total time: {self._format_time(elapsed_time)}") print(f"Processing speed: {(self.completed / elapsed_time) * 60:.1f} employees/minute") print(f"Excel report saved: {excel_file_path}") print(f"{'=' * 85}") def _format_time(self, seconds: float) -> str: """Convert seconds to readable format""" if seconds < 60: return f"{seconds:.0f} seconds" elif seconds < 3600: minutes = int(seconds // 60) secs = int(seconds % 60) return f"{minutes}m {secs}s" else: hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) return f"{hours}h {minutes}m" # ================================================================================================ # EXCEL REPORT GENERATOR # ================================================================================================ class SimpleExcelGenerator: """Creates Excel reports with color coding""" @staticmethod def create_excel_report(results: List[Dict], original_file_path: str) -> str: """Create Excel report - EXCEPTIONS ONLY (problems that need attention)""" try: original_path = Path(original_file_path) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") excel_path = original_path.parent / f"{original_path.stem}_Exceptions_{timestamp}.xlsx" # Filter results to only include exceptions (problems) exception_results = [] for result in results: name_match = result.get('Name Match', '') nt_register_name = result.get('NT Register Name', '') license_type = result.get('License Type', '') status = result.get('Status', '') # Include row if ANY of these conditions are true (exceptions): is_exception = ( name_match == "Not Found" or name_match == "No License" or name_match == "Error" or name_match.startswith("No (") or # Partial matches like "No (75.2%)" nt_register_name == "Not Found" or nt_register_name == "Error" or license_type == "Not Found" or license_type == "Error" or status == "Not Found" or status == "Error" ) if is_exception: exception_results.append(result) # If no exceptions found, create a summary message if not exception_results: summary_result = { 'Payroll Number': 'N/A', 'License Number': 'N/A', 'Rolecall Name': 'ALL RECORDS PROCESSED SUCCESSFULLY', 'NT Register Name': f'{len(results)} records', 'Name Match': 'No exceptions', 'License Type': 'All found', 'Status': 'All valid' } exception_results = [summary_result] excel_path = original_path.parent / f"{original_path.stem}_NT_ALL_SUCCESS_{timestamp}.xlsx" # Create DataFrame with exceptions only df = pd.DataFrame(exception_results) with pd.ExcelWriter(excel_path, engine='xlsxwriter') as writer: sheet_name = 'Exceptions' if len(exception_results) > 1 or 'ALL RECORDS' not in str( exception_results[0].get('Employee Name', '')) else 'Processing Summary' df.to_excel(writer, index=False, sheet_name=sheet_name) workbook = writer.book worksheet = writer.sheets[sheet_name] # Define header color only header_color = workbook.add_format({ 'bold': True, 'bg_color': '#8C1E31', # Custom burgundy background 'font_color': 'white', # White text 'border': 1, 'align': 'center', 'valign': 'vcenter', 'text_wrap': True }) # Format headers with enhanced styling for col_num, header in enumerate(df.columns): worksheet.write(0, col_num, header, header_color) # Auto-adjust column width with minimum and maximum limits max_length = max( len(str(header)), df[header].astype(str).str.len().max() if not df.empty else 0 ) # Set column width with reasonable limits col_width = min(max(max_length + 3, 12), 50) # Min 12, Max 50 characters worksheet.set_column(col_num, col_num, col_width) # Set header row height for better visibility worksheet.set_row(0, 20) # 20 pixels height for header row # Clean white rows for all data - no conditional formatting # Focus is on content, not colors # Add filters and freeze panes for better usability worksheet.autofilter(0, 0, len(df), len(df.columns) - 1) # Freeze the header row and first column (Payroll Number) worksheet.freeze_panes(1, 1) # Freeze row 1 (header) and column 1 (Payroll Number) return str(excel_path) except Exception as e: print(f"ERROR creating Excel report: {str(e)}") raise # ================================================================================================ # MAIN APPLICATION # ================================================================================================ class SeleniumNTLicenseChecker: """Main application class - using reliable""" def __init__(self): self.config = SimpleConfig() self.browser_manager = OptimizedBrowserManager(self.config) self.progress_tracker = None def run(self): """Main function - this is where everything happens""" try: # Show welcome message self._show_welcome() # Get CSV file from user csv_file_path = self._get_csv_file_from_user() # Read employee data employees = SimpleCSVHandler.load_employee_data(csv_file_path, self.config) # Ask user if they want to continue if not self._ask_user_to_continue(len(employees)): print("Process cancelled by user") return # Process all employees results = self._process_all_employees(employees) # Create Excel report excel_path = SimpleExcelGenerator.create_excel_report(results, csv_file_path) # Show final summary elapsed_time = time.time() - self.progress_tracker.start_time self.progress_tracker.show_final_summary(elapsed_time, excel_path) except KeyboardInterrupt: print("\n\nProcess stopped by user (Ctrl+C pressed)") except Exception as e: print(f"\nERROR: {str(e)}") print("\nTROUBLESHOOTING TIPS:") print( "1. Make sure your CSV file has the exact column names: 'Payroll Number', 'Employee Name', 'License Number'") print("2. Check that Chrome browser is installed") print("3. Make sure you have internet connection") print("4. Try running as administrator if you get permission errors") finally: self.browser_manager.cleanup_all_browsers() def _show_welcome(self): """Show welcome message""" def _get_csv_file_from_user(self) -> str: """Get CSV file path from user""" # Check if file path was provided as command line argument if len(sys.argv) > 1: file_path = sys.argv[1].strip().strip('"\'') print(f"Using file from command line: {file_path}") else: # Ask user for file path print("\nPlease provide your CSV file:") print(" You can either:") print(" 1. Type the full file path") print(" 2. Drag and drop the file into this window") print(" 3. Copy and paste the file path") file_path = input("\nEnter CSV file path: ").strip().strip('"\'') # Check if file exists if not file_path: raise Exception("No file path provided") if not Path(file_path).exists(): raise Exception(f"File not found: {file_path}") print(f"File found: {Path(file_path).name}") return file_path def _ask_user_to_continue(self, employee_count: int) -> bool: """Ask user if they want to process the employees""" if self.config.test_mode: print(f"\nTEST MODE: Will process {min(employee_count, self.config.test_record_limit)} employees") return True print(f"\nReady to process {employee_count} employees") print(f"Will use {self.config.parallel_browsers} parallel Chrome browsers") while True: response = input("\nContinue with processing? (Y/N): ").strip().lower() if response in ['y', 'yes']: return True elif response in ['n', 'no']: return False else: print("Please enter 'y' for yes or 'n' for no") def _process_all_employees(self, employees: List[EmployeeRecord]) -> List[Dict]: """Process all employees with minimal output unless there are issues""" # Setup progress tracking (silent mode) self.progress_tracker = SimpleProgressTracker(len(employees)) # Setup license searcher searcher = SeleniumLicenseSearcher(self.config, self.browser_manager) all_results = [] # Silent processing - no verbose startup messages with ThreadPoolExecutor(max_workers=self.config.parallel_browsers) as executor: # Submit all search tasks future_to_employee = { executor.submit(searcher.search_single_license, employee): employee for employee in employees } # Process results as they complete for future in as_completed(future_to_employee): employee = future_to_employee[future] try: # Get search result search_result = future.result() # Create result dictionary for Excel result_dict = { 'Payroll Number': employee.payroll_number, 'License Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'NT Register Name': search_result.found_name, 'Name Match': search_result.name_matches, 'License Type': search_result.license_type, 'Status': search_result.license_status } all_results.append(result_dict) # Update progress (only shows failures and periodic updates) self.progress_tracker.update_progress(employee, search_result) # Add small delay to be nice to the website time.sleep(self.config.delay_between_requests) except Exception as e: # Show processing errors immediately print(f"ERROR processing {employee.employee_name}: {str(e)}") # Handle any processing errors error_result = { 'Payroll Number': employee.payroll_number, 'License Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'NT Register Name': "Error", 'Name Match': "Error", 'License Type': "Error", 'Status': "Error" } all_results.append(error_result) # Create error result for progress tracking error_search_result = SearchResult(error_message=str(e)) self.progress_tracker.update_progress(employee, error_search_result) return all_results def _create_notes_for_result(self, result: SearchResult) -> str: """Create informative notes for the Excel report""" notes = [] if result.was_cached: notes.append("Cached result (duplicate license)") if result.search_time_seconds > 0: notes.append(f"Search time: {result.search_time_seconds:.2f}s") if result.how_many_retries > 0: notes.append(f"Retries: {result.how_many_retries}") if result.error_message: notes.append(f"Error: {result.error_message}") else: notes.append("Selenium scraping") return " | ".join(notes) if notes else "Processed successfully" # ================================================================================================ # HELPER FUNCTIONS # ================================================================================================ def test_selenium_connection(): try: from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC # Create test browser chrome_options = Options() chrome_options.add_argument("--headless") chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") driver = webdriver.Chrome(options=chrome_options) wait = WebDriverWait(driver, 15) try: # Test navigation to website driver.get("https://licensingnt.nt.gov.au/PublicRegister/") # Test finding the license input field license_input = wait.until( EC.presence_of_element_located((By.ID, "Licencenumber")) ) if license_input: print("Selenium connection test PASSED!") return True else: print("License input field not found - website may have changed") return False finally: driver.quit() except Exception as e: print(f"Selenium connection test failed: {str(e)}") print("Common fixes:") print("1. Install Google Chrome browser") print("2. Update Chrome to latest version") print("3. Run as administrator") print("4. Check internet connection") return False # ================================================================================================ # API CLASS FOR FLASK INTEGRATION # ================================================================================================ class NTLicenseCheckerAPI: """API class for NT License Checker that can be called from Flask""" def __init__(self): self.config = SimpleConfig() self.browser_manager = None self.searcher = None def _ensure_browser_manager(self): """Ensure browser manager is initialized and ready""" if self.browser_manager is None or not self.browser_manager.setup_complete: self.browser_manager = OptimizedBrowserManager(self.config) self.browser_manager.setup_browsers() def process_csv_file(self, csv_file_path: str) -> list: """Process CSV file and return results as JSON-compatible list""" try: # Ensure browser manager is ready self._ensure_browser_manager() # Load employee data employees = SimpleCSVHandler.load_employee_data(csv_file_path, self.config) # Initialize searcher self.searcher = SeleniumLicenseSearcher(self.config, self.browser_manager) # Process all employees results = self._process_employees_for_api(employees) return results except Exception as e: raise Exception(f"Failed to process CSV file: {str(e)}") # Note: Don't cleanup browsers here - keep them for reuse def _process_employees_for_api(self, employees: List[EmployeeRecord]) -> List[Dict]: """Process employees and return JSON-compatible results""" results = [] # Process employees sequentially for API (avoiding threading complexity) for employee in employees: try: # Search for license search_result = self.searcher.search_single_license(employee) # Create result dictionary result_dict = { 'payroll_number': employee.payroll_number, 'license_number': employee.license_number, 'employee_name': employee.employee_name, 'nt_register_name': search_result.found_name, 'name_match': search_result.name_matches, 'license_type': search_result.license_type, 'license_status': search_result.license_status, 'was_cached': search_result.was_cached, 'search_time_seconds': search_result.search_time_seconds, 'error_message': search_result.error_message, 'csv_row_number': employee.csv_row_number } results.append(result_dict) # Add small delay to be nice to the website time.sleep(self.config.delay_between_requests) except Exception as e: # Handle processing errors error_result = { 'payroll_number': employee.payroll_number, 'license_number': employee.license_number, 'employee_name': employee.employee_name, 'nt_register_name': "Error", 'name_match': "Error", 'license_type': "Error", 'license_status': "Error", 'was_cached': False, 'search_time_seconds': 0.0, 'error_message': str(e), 'csv_row_number': employee.csv_row_number } results.append(error_result) return results def get_processing_stats(self, results: List[Dict]) -> Dict: """Calculate processing statistics""" total = len(results) successful = sum(1 for r in results if r.get('nt_register_name') != "Not Found" and not r.get('error_message')) failed = sum(1 for r in results if r.get('nt_register_name') == "Not Found" or r.get('error_message')) cached = sum(1 for r in results if r.get('was_cached', False)) errors = sum(1 for r in results if r.get('error_message')) return { 'total_records': total, 'successful_searches': successful, 'failed_searches': failed, 'cached_results': cached, 'errors': errors, 'success_rate': (successful / total * 100) if total > 0 else 0 } def process_records(self, records_data: List[Dict]) -> List[Dict]: """Process list of records and return results as list of dictionaries""" try: # Ensure browser manager is ready self._ensure_browser_manager() # Convert dictionary records to EmployeeRecord objects employees = [] for i, record_data in enumerate(records_data): employee = EmployeeRecord( payroll_number=str(record_data.get('Payroll Number', '')).strip(), employee_name=str(record_data.get('Employee Name', '')).strip(), license_number=str(record_data.get('License Number', '')).strip(), csv_row_number=i + 1 ) employee.clean_data() employees.append(employee) # Initialize searcher self.searcher = SeleniumLicenseSearcher(self.config, self.browser_manager) # Process all employees results = self._process_employees_for_api(employees) return results except Exception as e: raise Exception(f"Failed to process records: {str(e)}") # Note: Don't cleanup browsers here - keep them for reuse def cleanup(self): """Clean up browsers when completely done with the API instance""" if self.browser_manager: self.browser_manager.cleanup_all_browsers() self.browser_manager = None self.searcher = None # ================================================================================================ # MAIN PROGRAM ENTRY POINT # ================================================================================================ def main(): """Main entry point""" # Handle command line arguments if len(sys.argv) > 1: arg = sys.argv[1].lower() if arg in ['--test', '-t', 'test']: test_selenium_connection() return # Run the main application try: print("Starting NT License Checker...") # Quick connection test print("\nTesting Selenium connection to NT website...") if not test_selenium_connection(): print("Connection issues detected, but continuing anyway...") print("If all searches fail, the website may have changed or Chrome needs updating") app = SeleniumNTLicenseChecker() app.run() except KeyboardInterrupt: print("\nSelenium application stopped by user") except Exception as e: print(f"\nFATAL ERROR: {str(e)}") print("\nSELENIUM TROUBLESHOOTING:") print("1. Check your CSV file has the right columns") print("2. Install/update Google Chrome browser") print("3. Run as administrator") print("4. Check your internet connection") print("5. Try reducing parallel_browsers to 1 if getting errors") print("6. Run with --test to check Selenium connectivity") print("7. Run with --help for more information") # ================================================================================================ # RUN THE PROGRAM # ================================================================================================ if __name__ == "__main__": main()