""" VIC License Checker ================================================================================================ """ import json import time from pathlib import Path from dataclasses import dataclass, asdict 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 VIC government changes their site website_url: str = "https://www.lars.police.vic.gov.au/LARS/LARS.asp?File=/components/screens/psinfp03/psinfp03.asp" # 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 = 20 # 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 = 1 #Only use 1 VIC Police will IP BAN # Test mode - only process first 10 records for testing test_mode: bool = False test_record_limit: int = 20 # ================================================================================================ # DATA CLASSES # ================================================================================================ @dataclass class EmployeeRecord: """Holds one employee's information from CSV""" payroll_number: str employee_name: str license_number: str csv_expiry: 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() self.csv_expiry = str(self.csv_expiry).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 VIC LARS website""" found_name: str = "Not Found" license_type: str = "Not Found" license_expiry: str = "Not Found" license_activities: str = "Not Found" name_matches: str = "No License" expiry_status: 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 @dataclass class VICVerificationResult: """Data class for VIC verification results""" payroll_number: str license_number: str rolecall_name: str lars_name: str name_match: str rolecall_expiry: str lars_expiry: str expiry_status: str type: str activities: str @dataclass class VICVerificationResult: """Data class for VIC verification results""" payroll_number: str license_number: str rolecall_name: str lars_name: str name_match: str rolecall_expiry: str lars_expiry: str expiry_status: str type: str activities: str # ================================================================================================ # 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 VIC LARS 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 calculate_expiry_status(self, csv_expiry: str, web_expiry: str) -> str: """Calculate expiry status comparison between CSV and web dates""" if not csv_expiry or not web_expiry or web_expiry == "Not Found": return "No License" try: # Parse dates - VIC uses DD/MM/YYYY format csv_date = datetime.strptime(csv_expiry, "%d/%m/%Y") web_date = datetime.strptime(web_expiry, "%d/%m/%Y") is_expired = web_date < datetime.today() dates_match = (csv_date.date() == web_date.date()) if dates_match and not is_expired: return "Active" elif dates_match and is_expired: return "Expired" elif not dates_match and not is_expired: return "Active - Date Wrong" elif not dates_match and is_expired: return "Expired - Date Wrong" else: return "Unknown" except Exception as e: return f"Error: {str(e)[:20]}" 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.expiry_status = self.calculate_expiry_status( employee.csv_expiry, cached_result.license_expiry ) 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) search_time = time.time() - start_time if result.is_successful(): result.name_matches = self.check_name_similarity( employee.employee_name, result.found_name ) result.expiry_status = self.calculate_expiry_status( employee.csv_expiry, result.license_expiry ) 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, employee: EmployeeRecord) -> SearchResult: """Actually search the VIC LARS website using Selenium""" try: # Navigate to the search page browser.get(self.config.website_url) # Find the search input field and enter license number search_input = wait_helper.until( EC.presence_of_element_located((By.ID, "SearchAuthNb")) ) search_input.clear() search_input.send_keys(employee.license_number.strip()) search_input.send_keys(Keys.RETURN) # Wait for results table to load wait_helper.until(EC.presence_of_element_located((By.ID, "idResultsTable"))) time.sleep(1) # Small delay to ensure DOM is fully updated # Parse results from the table rows_html = browser.find_element(By.ID, "idResultsTable").find_elements(By.TAG_NAME, "tr") if len(rows_html) > 2: # Should have header row, possibly empty row, then data cells = rows_html[2].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_expiry=cells[3].text.strip() or "Not Found", license_activities=cells[4].text.strip() or "Not Found" ) # No results found return SearchResult() except TimeoutException: raise Exception(f"Website timeout for license: {employee.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', 'Expiry/Update Date'] 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 - Expiry/Update Date 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_expiry=row['Expiry/Update Date'], 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"VIC 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', '') lars_name = result.get('LARS Name', '') license_type = result.get('Type', '') expiry_status = result.get('Expiry 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%)" lars_name == "Not Found" or lars_name == "Error" or license_type == "Not Found" or license_type == "Error" or expiry_status == "No License" or expiry_status == "Unknown" or expiry_status.startswith("Error:") or "Date Wrong" in expiry_status or "Expired" in expiry_status ) 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', 'LARS Name': f'{len(results)} records', 'Name Match': 'No exceptions', 'Type': 'All found', 'Rolecall Expiry': 'All dates match', 'LARS Expiry': 'All valid', 'Expiry Status': 'All active', 'Activities': 'All valid' } exception_results = [summary_result] excel_path = original_path.parent / f"{original_path.stem}_VIC_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('Rolecall 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 # 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 SeleniumVICLicenseChecker: """Main application class - using reliable Selenium""" 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', 'Expiry/Update Date'") 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""" print("VIC License Checker - Starting...") 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, 'LARS Name': search_result.found_name, 'Name Match': search_result.name_matches, 'Rolecall Expiry': employee.csv_expiry, 'LARS Expiry': search_result.license_expiry, 'Expiry Status': search_result.expiry_status, 'Type': search_result.license_type, 'Activities': search_result.license_activities } 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, 'LARS Name': "Error", 'Name Match': "Error", 'Rolecall Expiry': employee.csv_expiry, 'LARS Expiry': "Error", 'Expiry Status': "Error", 'Type': "Error", 'Activities': "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" def process_records(self, records_data: List[Dict]) -> List[Dict]: """Process list of records and return results as list of dictionaries""" try: # Convert dictionary records to EmployeeRecord objects records = [] for record_data in records_data: employee = EmployeeRecord( payroll_number=record_data.get('Payroll Number', '').strip(), employee_name=record_data.get('Employee Name', '').strip(), license_number=record_data.get('License Number', '').strip(), csv_expiry=record_data.get('Expiry/Update Date', '').strip(), csv_row_number=0 ) employee.clean_data() records.append(employee) if not records: return [] output_rows = [] start_time = time.time() # Setup license searcher searcher = SeleniumLicenseSearcher(self.config, self.browser_manager) # Process records 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 records } # 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 JSON output result_dict = { 'Payroll Number': employee.payroll_number, 'License Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'LARS Name': search_result.found_name, 'Name Match': search_result.name_matches, 'Rolecall Expiry': employee.csv_expiry, 'LARS Expiry': search_result.license_expiry, 'Expiry Status': search_result.expiry_status, 'Type': search_result.license_type, 'Activities': search_result.license_activities } output_rows.append(result_dict) # Add small delay to be nice to the website time.sleep(self.config.delay_between_requests) except Exception as e: # Handle any processing errors error_result = { 'Payroll Number': employee.payroll_number, 'License Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'LARS Name': "Error", 'Name Match': "Error", 'Rolecall Expiry': employee.csv_expiry, 'LARS Expiry': "Error", 'Expiry Status': "Error", 'Type': "Error", 'Activities': "Error" } output_rows.append(error_result) # Cleanup browsers self.browser_manager.cleanup_all_browsers() total_time = time.time() - start_time print(f"Completed processing {len(records)} records in {total_time:.1f}s") return output_rows except Exception as e: print(f"Error processing records: {e}") # Cleanup browsers on error self.browser_manager.cleanup_all_browsers() raise def process_records(self, records_data: List[Dict]) -> List[Dict]: """Process list of records and return results as list of dictionaries""" try: # Convert dictionary records to EmployeeRecord objects records = [] for record_data in records_data: employee = EmployeeRecord( payroll_number=record_data.get('Payroll Number', '').strip(), employee_name=record_data.get('Employee Name', '').strip(), license_number=record_data.get('License Number', '').strip(), csv_expiry=record_data.get('Expiry/Update Date', '').strip(), csv_row_number=0 ) employee.clean_data() records.append(employee) if not records: return [] output_rows = [] start_time = time.time() # Setup license searcher searcher = SeleniumLicenseSearcher(self.config, self.browser_manager) # Process records 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 records } # 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 JSON output result_dict = { 'Payroll Number': employee.payroll_number, 'License Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'LARS Name': search_result.found_name, 'Name Match': search_result.name_matches, 'Rolecall Expiry': employee.csv_expiry, 'LARS Expiry': search_result.license_expiry, 'Expiry Status': search_result.expiry_status, 'Type': search_result.license_type, 'Activities': search_result.license_activities } output_rows.append(result_dict) # Add small delay to be nice to the website time.sleep(self.config.delay_between_requests) except Exception as e: # Handle any processing errors error_result = { 'Payroll Number': employee.payroll_number, 'License Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'LARS Name': "Error", 'Name Match': "Error", 'Rolecall Expiry': employee.csv_expiry, 'LARS Expiry': "Error", 'Expiry Status': "Error", 'Type': "Error", 'Activities': "Error" } output_rows.append(error_result) # Cleanup browsers self.browser_manager.cleanup_all_browsers() total_time = time.time() - start_time print(f"Completed processing {len(records)} records in {total_time:.1f}s") return output_rows except Exception as e: print(f"Error processing records: {e}") # Cleanup browsers on error self.browser_manager.cleanup_all_browsers() raise # ================================================================================================ # 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.common.keys import Keys 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 LARS website driver.get( "https://www.lars.police.vic.gov.au/LARS/LARS.asp?File=/components/screens/psinfp03/psinfp03.asp") # Test finding the search input field search_input = wait.until( EC.presence_of_element_located((By.ID, "SearchAuthNb")) ) if search_input: print("Selenium connection test PASSED!") return True else: print("Search 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 # ================================================================================================ # 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 VIC License Checker...") # Quick connection test print("\nTesting Selenium connection to VIC LARS 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 = SeleniumVICLicenseChecker() 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") # ================================================================================================ # RUN THE PROGRAM # ================================================================================================ if __name__ == "__main__": main()