""" QLD 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 QLD government changes their site website_url: str = "https://www.qld.gov.au/law/laws-regulated-industries-and-accountability/queensland-laws-and-regulations/check-a-licence-association-charity-or-register/check-a-licence" # 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.5 # 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 = True test_record_limit: int = 1 # Start with just 1 record for debugging # ================================================================================================ # 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 QLD website""" found_name: str = "Not Found" license_activities: List[str] = None license_expiry: 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 __post_init__(self): if self.license_activities is None: self.license_activities = [] def is_successful(self) -> bool: """Did we successfully find license information?""" return self.found_name != "Not Found" and not self.error_message def get_formatted_activities(self) -> str: """Return formatted activities string with security officer prioritized.""" if not self.license_activities: return "Not Found" # Sort activities with Security Officer (Unarmed) first sorted_activities = sorted( self.license_activities, key=lambda x: (0, x.upper()) if "SECURITY OFFICER (UNARMED" in x.upper() else (1, x.upper()) ) return ", ".join(sorted_activities) def get_primary_license(self) -> str: """Return the primary license type.""" if not self.license_activities: return "Not Found" sorted_activities = sorted( self.license_activities, key=lambda x: (0, x.upper()) if "SECURITY OFFICER (UNARMED" in x.upper() else (1, x.upper()) ) return sorted_activities[0] # ================================================================================================ # 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") chrome_options.add_argument("--allow-running-insecure-content") chrome_options.add_argument("--disable-blink-features=AutomationControlled") chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"]) chrome_options.add_experimental_option('useAutomationExtension', False) driver = webdriver.Chrome(options=chrome_options) driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") return driver 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 QLD 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 calculate_expiry_status(self, csv_expiry: str, web_expiry: str) -> str: """Calculate expiry status comparison between CSV and web dates""" print(f"DEBUG: calculate_expiry_status called with csv_expiry='{csv_expiry}' (type: {type(csv_expiry)}, len: {len(str(csv_expiry))}) and web_expiry='{web_expiry}'") # If web expiry is not found, return "No License" if not web_expiry or web_expiry == "Not Found": print("DEBUG: Web expiry not found, returning 'No License'") return "No License" # If CSV expiry is empty but web has valid expiry, just check if web license is active if not csv_expiry or csv_expiry.strip() == "": print("DEBUG: CSV expiry is empty, checking web license status only") try: # Clean and fix web expiry date before parsing cleaned_web_expiry = self._clean_web_date(web_expiry) if not cleaned_web_expiry: return "Invalid Date" # Parse web date - QLD uses DD-MMM-YYYY format web_date = datetime.strptime(cleaned_web_expiry, "%d-%b-%Y") # Check if web license is active if web_date >= datetime.today(): return "Active" else: return "Expired" except Exception as e: return f"Date Error: {str(e)[:15]}" try: # Parse CSV date - QLD uses DD/MM/YYYY format csv_date = datetime.strptime(csv_expiry, "%d/%m/%Y") # Clean and fix web expiry date before parsing cleaned_web_expiry = self._clean_web_date(web_expiry) if not cleaned_web_expiry: return "Invalid Date" # Parse web date - QLD uses DD-MMM-YYYY format web_date = datetime.strptime(cleaned_web_expiry, "%d-%b-%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"Date Error: {str(e)[:15]}" def _clean_web_date(self, web_date: str) -> str: """Clean and fix malformed web dates from QLD website""" if not web_date or web_date == "Not Found": return "" # Remove extra whitespace cleaned = web_date.strip() # Fix common malformed date patterns # Pattern: "07-01-202" (missing last digit) -> try to fix if len(cleaned) == 9 and cleaned.count('-') == 2: parts = cleaned.split('-') if len(parts) == 3 and len(parts[2]) == 3: # Year has only 3 digits # Assume it's missing the last digit and it should be current decade current_year = datetime.now().year decade = str(current_year)[:3] # e.g., "202" for 2020s if parts[2].startswith(decade[:2]): # e.g., starts with "20" # Add the decade's last digit parts[2] = decade + parts[2][-1] cleaned = "-".join(parts) # Pattern: "07-Jan-24" -> "07-Jan-2024" elif len(cleaned) == 9 and cleaned.count('-') == 2: parts = cleaned.split('-') if len(parts) == 3 and len(parts[2]) == 2 and parts[2].isdigit(): year_suffix = int(parts[2]) # Assume 00-30 = 2000s, 31-99 = 1900s if year_suffix <= 30: parts[2] = f"20{parts[2]}" else: parts[2] = f"19{parts[2]}" cleaned = "-".join(parts) return cleaned def _select_best_record(self, matching_records: List[Dict], employee: EmployeeRecord) -> Dict: """Select the best record from multiple matching records based on priority criteria""" if not matching_records: return {} if len(matching_records) == 1: return matching_records[0] print(f"DEBUG: Evaluating {len(matching_records)} matching records to find the best one...") # Priority criteria (in order of importance): # 1. Exact match over partial match # 2. Active/current license (not expired) # 3. Most recent expiry date # 4. Security Officer license type (if available) # 5. First occurrence (as fallback) def calculate_priority_score(record): score = 0 # 1. Exact match gets higher priority if record['match_type'] == 'exact': score += 1000 else: # partial match score += 500 # 2. Check if license is active (not expired) try: if record['expiry'] and record['expiry'] != "Not Found": cleaned_expiry = self._clean_web_date(record['expiry']) if cleaned_expiry: expiry_date = datetime.strptime(cleaned_expiry, "%d-%b-%Y") if expiry_date >= datetime.today(): score += 200 # Active license # 3. More recent expiry dates get higher priority days_until_expiry = (expiry_date - datetime.today()).days score += min(days_until_expiry, 365) # Cap at 1 year else: score += 50 # Expired but still a match except: score += 10 # Can't parse date, but still a match # 4. Security Officer license type gets priority license_type = record.get('license_type', '').upper() if 'SECURITY OFFICER' in license_type: score += 100 elif 'SECURITY' in license_type: score += 50 # 5. Earlier row index (first occurrence) as tiebreaker score += (100 - record['row_index']) return score # Calculate scores for all records scored_records = [] for record in matching_records: score = calculate_priority_score(record) scored_records.append((score, record)) print(f"DEBUG: Record {record['row_index']} - Name: {record['name']}, Type: {record['license_type']}, Expiry: {record['expiry']}, Score: {score}") # Sort by score (highest first) scored_records.sort(key=lambda x: x[0], reverse=True) best_record = scored_records[0][1] best_score = scored_records[0][0] print(f"DEBUG: Selected record with score {best_score}: {best_record['name']} - {best_record['license_type']} - {best_record['expiry']}") return best_record 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) 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 QLD government website using Selenium with enhanced debugging""" try: print(f"DEBUG: Searching for license: {employee.license_number}") # Navigate to the search page browser.get(self.config.website_url) print(f"DEBUG: Loaded page: {browser.title}") # Handle iframe if present try: iframe = wait_helper.until(EC.presence_of_element_located((By.TAG_NAME, "iframe"))) browser.switch_to.frame(iframe) print("DEBUG: Switched to iframe") except TimeoutException: print("DEBUG: No iframe found, continuing with main page") pass # Wait for and interact with search input try: search_input = wait_helper.until(EC.element_to_be_clickable((By.ID, "LicenceNumber"))) print("DEBUG: Found license input field") except TimeoutException: # Try alternative selectors print("DEBUG: Trying alternative selectors for license input...") alternative_selectors = [ "input[name='LicenceNumber']", "input[placeholder*='licence']", "input[placeholder*='license']", "input[type='text']" ] search_input = None for selector in alternative_selectors: try: search_input = wait_helper.until(EC.element_to_be_clickable((By.CSS_SELECTOR, selector))) print(f"DEBUG: Found input with selector: {selector}") break except TimeoutException: continue if not search_input: raise Exception("Could not find license input field with any selector") # Clear and set the license number search_input.clear() search_input.send_keys(employee.license_number.strip()) print(f"DEBUG: Entered license number: {employee.license_number.strip()}") # Try multiple ways to submit the form try: search_input.send_keys(Keys.RETURN) print("DEBUG: Submitted form with Enter key") except: try: # Try clicking a submit button submit_button = browser.find_element(By.CSS_SELECTOR, "button[type='submit'], input[type='submit'], button:contains('Search')") submit_button.click() print("DEBUG: Submitted form with submit button") except: # Try JavaScript submission browser.execute_script("arguments[0].form.submit();", search_input) print("DEBUG: Submitted form with JavaScript") # Wait for results with multiple possible selectors print("DEBUG: Waiting for results...") results_found = False table = None # Try different selectors for the results table table_selectors = [ "table#licenceList", "table[id*='licence']", "table[id*='license']", "table[class*='data']", "table" ] for selector in table_selectors: try: table = wait_helper.until(EC.presence_of_element_located((By.CSS_SELECTOR, selector))) print(f"DEBUG: Found results table with selector: {selector}") results_found = True break except TimeoutException: continue if not results_found: # Check if there's a "no results" message try: no_results = browser.find_element(By.CSS_SELECTOR, "*:contains('No results'), *:contains('not found'), *:contains('no data')") print(f"DEBUG: Found 'no results' message: {no_results.text}") return SearchResult( found_name="Not Found", license_activities=[], license_expiry="Not Found" ) except: pass # Take a screenshot for debugging browser.save_screenshot(f"debug_qld_search_{employee.license_number}.png") print(f"DEBUG: Screenshot saved as debug_qld_search_{employee.license_number}.png") raise Exception("Could not find results table") # Set dropdown to show more entries if available try: browser.execute_script(""" let dropdown = document.querySelector('select[name="licenceList_length"], select[class*="length"]'); if (dropdown) { dropdown.value = '50'; dropdown.dispatchEvent(new Event('change', { bubbles: true })); console.log('Set dropdown to 50 entries'); } """) time.sleep(2) # Allow table to reload print("DEBUG: Set dropdown to show 50 entries") except Exception as e: print(f"DEBUG: Could not set dropdown: {e}") # Parse results with enhanced debugging rows = table.find_elements(By.TAG_NAME, "tr")[1:] # Skip header print(f"DEBUG: Found {len(rows)} result rows") # Collect all matching records instead of stopping at first match matching_records = [] for i, row in enumerate(rows): cells = row.find_elements(By.TAG_NAME, "td") if not cells or len(cells) < 4: print(f"DEBUG: Row {i} has insufficient cells: {len(cells) if cells else 0}") continue row_license = cells[0].text.strip() print(f"DEBUG: Row {i} license: '{row_license}' vs search: '{employee.license_number.strip()}'") # Check for exact match if row_license == employee.license_number.strip(): record = { 'license_number': row_license, 'name': cells[1].text.strip(), 'license_type': cells[2].text.strip(), 'expiry': cells[3].text.strip(), 'match_type': 'exact', 'row_index': i } matching_records.append(record) print(f"DEBUG: Exact match found - Name: {record['name']}, Type: {record['license_type']}, Expiry: {record['expiry']}") # Check for partial match (license number contains search term) elif employee.license_number.strip() in row_license: record = { 'license_number': row_license, 'name': cells[1].text.strip(), 'license_type': cells[2].text.strip(), 'expiry': cells[3].text.strip(), 'match_type': 'partial', 'row_index': i } matching_records.append(record) print(f"DEBUG: Partial match found - Name: {record['name']}, Type: {record['license_type']}, Expiry: {record['expiry']}") print(f"DEBUG: Found {len(matching_records)} matching records") # Show all matching records for debugging if matching_records: print("DEBUG: All matching records found:") for i, record in enumerate(matching_records, 1): print(f" {i}. License: {record['license_number']}, Name: {record['name']}, Type: {record['license_type']}, Expiry: {record['expiry']}, Match: {record['match_type']}") # Select the best record from all matches web_name = "Not Found" web_activities = [] web_expiry = "Not Found" license_found = False if matching_records: # Select the best record based on priority criteria best_record = self._select_best_record(matching_records, employee) web_name = best_record['name'] web_expiry = best_record['expiry'] if best_record['license_type']: web_activities.append(best_record['license_type']) license_found = True print(f"DEBUG: Selected best record - Name: {web_name}, Type: {best_record['license_type']}, Expiry: {web_expiry}") else: print(f"DEBUG: No matching license found for {employee.license_number}") # Print all available license numbers for debugging print("DEBUG: Available license numbers in results:") for i, row in enumerate(rows[:5]): # Show first 5 rows cells = row.find_elements(By.TAG_NAME, "td") if cells: print(f" Row {i}: {cells[0].text.strip()}") return SearchResult( found_name=web_name, license_activities=web_activities, license_expiry=web_expiry ) except TimeoutException as e: print(f"DEBUG: Timeout error: {e}") raise Exception(f"Website timeout for license: {employee.license_number}") except NoSuchElementException as e: print(f"DEBUG: Element not found: {e}") raise Exception(f"Page element not found: {str(e)}") except WebDriverException as e: print(f"DEBUG: WebDriver error: {e}") raise Exception(f"WebDriver error: {str(e)}") except Exception as e: print(f"DEBUG: General error: {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"QLD 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', '') licence_name = result.get('Licence Name', '') licence_types = result.get('Licence Type(s)', '') 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%)" licence_name == "Not Found" or licence_name == "Error" or licence_types == "Not Found" or licence_types == "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', 'Licence Number': 'N/A', 'Rolecall Name': 'ALL RECORDS PROCESSED SUCCESSFULLY', 'Licence Name': f'{len(results)} records', 'Name Match': 'No exceptions', 'Licence Type(s)': 'All found', 'Licence Expiry': 'All valid', 'Rolecall Expiry': 'All dates match', 'Expiry Status': 'All active' } exception_results = [summary_result] excel_path = original_path.parent / f"{original_path.stem}_QLD_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 SeleniumQLDLicenseChecker: """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("QLD 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, 'Licence Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'Licence Name': search_result.found_name, 'Name Match': search_result.name_matches, 'Licence Type(s)': search_result.get_formatted_activities(), 'Licence Expiry': search_result.license_expiry, 'Rolecall Expiry': employee.csv_expiry, 'Expiry Status': search_result.expiry_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, 'Licence Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'Licence Name': "Error", 'Name Match': "Error", 'Licence Type(s)': "Error", 'Licence Expiry': "Error", 'Rolecall Expiry': employee.csv_expiry, 'Expiry 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 QLD website print("Testing connection to QLD website...") driver.get( "https://www.qld.gov.au/law/laws-regulated-industries-and-accountability/queensland-laws-and-regulations/check-a-licence-association-charity-or-register/check-a-licence") print(f"Page title: {driver.title}") print(f"Current URL: {driver.current_url}") # Check if page loads (look for license search elements) try: # Try to find iframe first iframe = wait.until(EC.presence_of_element_located((By.TAG_NAME, "iframe"))) driver.switch_to.frame(iframe) print("Found iframe and switched to it") except: print("No iframe found, continuing with main page") pass # Test finding the license input field with multiple selectors license_input = None selectors_to_try = [ (By.ID, "LicenceNumber"), (By.CSS_SELECTOR, "input[name='LicenceNumber']"), (By.CSS_SELECTOR, "input[placeholder*='licence']"), (By.CSS_SELECTOR, "input[placeholder*='license']"), (By.CSS_SELECTOR, "input[type='text']") ] for by, selector in selectors_to_try: try: license_input = wait.until(EC.presence_of_element_located((by, selector))) print(f"Found license input with selector: {selector}") break except TimeoutException: continue if license_input: print("Selenium connection test PASSED!") return True else: print("License input field not found - website may have changed") # Take a screenshot for debugging driver.save_screenshot("debug_connection_test.png") print("Screenshot saved as debug_connection_test.png") 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 WRAPPER FOR FLASK ROUTES # ================================================================================================ class QLDLicenseCheckerAPI: """API wrapper for Flask routes to process CSV files and return JSON results""" def __init__(self): self.config = SimpleConfig() # Disable test mode for API usage self.config.test_mode = False 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() self.searcher = SeleniumLicenseSearcher(self.config, self.browser_manager) def process_csv_file(self, csv_file_path: str) -> List[Dict]: """Process CSV file and return results as JSON-compatible list of dictionaries""" try: # Ensure browser manager is ready self._ensure_browser_manager() # Load employee data employees = SimpleCSVHandler.load_employee_data(csv_file_path, self.config) if not employees: return [] # Setup progress tracking (silent mode for API) progress_tracker = SimpleProgressTracker(len(employees)) all_results = [] # Process all employees with ThreadPoolExecutor(max_workers=self.config.parallel_browsers) as executor: # Submit all search tasks future_to_employee = { executor.submit(self.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 JSON response result_dict = { 'payroll_number': employee.payroll_number, 'license_number': employee.license_number, 'employee_name': employee.employee_name, 'found_name': search_result.found_name, 'name_match': search_result.name_matches, 'license_types': search_result.get_formatted_activities(), 'primary_license': search_result.get_primary_license(), 'license_expiry': search_result.license_expiry, 'csv_expiry': employee.csv_expiry, 'expiry_status': search_result.expiry_status, 'search_time_seconds': search_result.search_time_seconds, 'retries': search_result.how_many_retries, 'was_cached': search_result.was_cached, 'error_message': search_result.error_message, 'is_successful': search_result.is_successful() } all_results.append(result_dict) # Update progress (silent for API) 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: # Handle any processing errors error_result = { 'payroll_number': employee.payroll_number, 'license_number': employee.license_number, 'employee_name': employee.employee_name, 'found_name': "Error", 'name_match': "Error", 'license_types': "Error", 'primary_license': "Error", 'license_expiry': "Error", 'csv_expiry': employee.csv_expiry, 'expiry_status': "Error", 'search_time_seconds': 0.0, 'retries': 0, 'was_cached': False, 'error_message': str(e), 'is_successful': False } all_results.append(error_result) # Create error result for progress tracking error_search_result = SearchResult(error_message=str(e)) progress_tracker.update_progress(employee, error_search_result) return all_results except Exception as e: # Return error result return [{ 'error': 'CSV processing failed', 'message': str(e), 'is_successful': False }] def test_connection(self) -> Dict: """Test connection to QLD website and return status""" try: if test_selenium_connection(): return { 'status': 'success', 'message': 'QLD website connection test passed' } else: return { 'status': 'warning', 'message': 'QLD website connection test failed - website may have changed' } except Exception as e: return { 'status': 'error', 'message': f'Connection test failed: {str(e)}' } def get_config(self) -> Dict: """Get current configuration settings""" return { 'website_url': self.config.website_url, 'name_similarity_required': self.config.name_similarity_required, 'max_retry_attempts': self.config.max_retry_attempts, 'request_timeout_seconds': self.config.request_timeout_seconds, 'delay_between_requests': self.config.delay_between_requests, 'parallel_browsers': self.config.parallel_browsers, 'test_mode': self.config.test_mode } def process_records(self, records_data: List[Dict]) -> List[Dict]: """Process list of records and return results as JSON-compatible list of dictionaries""" try: # Ensure browser manager is ready self._ensure_browser_manager() # Debug: Show what keys are available in the first record if records_data: print(f"DEBUG: Available keys in first record: {list(records_data[0].keys())}") print(f"DEBUG: First record data: {records_data[0]}") # Convert dictionary records to EmployeeRecord objects employees = [] for i, record_data in enumerate(records_data): # Try multiple possible key names for expiry date expiry_key = None possible_keys = [ 'Expiry/Update Date', # Original with two spaces 'Expiry/Update Date', # With one space 'Expiry Date', # Shorter version 'Expiry', # Even shorter 'csv_expiry', # API format 'expiry_date' # Another common format ] for key in possible_keys: if key in record_data: expiry_key = key break if not expiry_key: print(f"DEBUG: No expiry key found in record {i}, available keys: {list(record_data.keys())}") expiry_value = "" else: expiry_value = str(record_data.get(expiry_key, '')).strip() print(f"DEBUG: Using key '{expiry_key}' with value '{expiry_value}' for record {i}") 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_expiry=expiry_value, csv_row_number=i + 1 ) employee.clean_data() employees.append(employee) if not employees: return [] # Setup progress tracking (silent mode for API) progress_tracker = SimpleProgressTracker(len(employees)) all_results = [] # Process all employees with ThreadPoolExecutor(max_workers=self.config.parallel_browsers) as executor: # Submit all search tasks future_to_employee = { executor.submit(self.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 JSON response result_dict = { 'payroll_number': employee.payroll_number, 'license_number': employee.license_number, 'employee_name': employee.employee_name, 'found_name': search_result.found_name, 'name_match': search_result.name_matches, 'license_types': search_result.get_formatted_activities(), 'primary_license': search_result.get_primary_license(), 'license_expiry': search_result.license_expiry, 'csv_expiry': employee.csv_expiry, 'expiry_status': search_result.expiry_status, 'search_time_seconds': search_result.search_time_seconds, 'retries': search_result.how_many_retries, 'was_cached': search_result.was_cached, 'error_message': search_result.error_message, 'is_successful': search_result.is_successful() } all_results.append(result_dict) # Update progress (silent for API) 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: # Handle any processing errors error_result = { 'payroll_number': employee.payroll_number, 'license_number': employee.license_number, 'employee_name': employee.employee_name, 'found_name': "Error", 'name_match': "Error", 'license_types': "Error", 'primary_license': "Error", 'license_expiry': "Error", 'csv_expiry': employee.csv_expiry, 'expiry_status': "Error", 'search_time_seconds': 0.0, 'retries': 0, 'was_cached': False, 'error_message': str(e), 'is_successful': False } all_results.append(error_result) # Create error result for progress tracking error_search_result = SearchResult(error_message=str(e)) progress_tracker.update_progress(employee, error_search_result) return all_results except Exception as e: # Return error result return [{ 'error': 'Records processing failed', 'message': str(e), 'is_successful': False }] 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 QLD License Checker...") # Quick connection test print("\nTesting Selenium connection to QLD 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 = SeleniumQLDLicenseChecker() 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()