""" NSW 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 # API libraries for web requests import requests from rapidfuzz import fuzz import pandas as pd from dateutil import parser # ================================================================================================ # CONFIGURATION - Change these settings if needed # ================================================================================================ @dataclass class SimpleConfig: """Simple configuration""" # API URLs - Don't change unless NSW government changes their API token_url: str = "https://api.onegov.nsw.gov.au/oauth/client_credential/accesstoken" verify_url: str = "https://api.onegov.nsw.gov.au/securityregister/v1/verify" # API credentials basic_auth_header: str = "Basic NmtISWVtWG8zR010VWFzR2JWNzE4Y2VUSW1kbkdvRms6blZYNWRVR05yVWc3d3NRMQ==" api_key: str = "6kHIemXo3GMtUasGbV718ceTImdnGoFk" # How similar names should be to match (85 = 85% similar) name_similarity_required: int = 85 # How many times to retry if API is slow max_retry_attempts: int = 3 # How long to wait for API to respond (seconds) request_timeout_seconds: int = 60 # Delay between requests to be nice to the API (seconds) delay_between_requests: float = 1.0 # How many parallel API calls to run (2 is optimal for rate-limited APIs) parallel_workers: int = 2 # Test mode - only process first 10 records for testing test_mode: bool = True 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_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 NSW API""" found_name: str = "Not Found" license_type: str = "Not Found" license_expiry: str = "Not Found" license_status: 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 # ================================================================================================ # API MANAGER # ================================================================================================ class APIManager: """Manages API connections and rate limiting""" def __init__(self, config: SimpleConfig): self.config = config self.session_pool = queue.Queue() self.all_sessions = [] self.setup_complete = False self.lock = threading.Lock() self.access_token = None def setup_sessions(self): """Create HTTP sessions for parallel processing""" if self.setup_complete: return with self.lock: if self.setup_complete: return print(f"Setting up {self.config.parallel_workers} API sessions...") # Get access token first if not self._get_access_token(): raise Exception("Failed to get API access token") for i in range(self.config.parallel_workers): session = self._create_session() self.all_sessions.append(session) self.session_pool.put(session) print(f" Session {i + 1} ready") self.setup_complete = True print(f"All API sessions ready for parallel processing!") def _create_session(self) -> requests.Session: """Create one HTTP session with optimal settings""" session = requests.Session() session.headers.update({ 'User-Agent': 'NSW-License-Verifier/1.0', 'content-type': 'application/json', 'authorization': f'Bearer {self.access_token}', 'apikey': self.config.api_key }) return session def _get_access_token(self) -> bool: """Get access token from NSW API""" try: headers = { "content-type": "application/json", "authorization": self.config.basic_auth_header } params = {"grant_type": "client_credentials"} response = requests.get( self.config.token_url, headers=headers, params=params, timeout=30 ) response.raise_for_status() self.access_token = response.json().get("access_token") if not self.access_token: print("ERROR: No access token received") return False print("Successfully obtained access token") return True except requests.RequestException as e: print(f"ERROR: Failed to get access token: {e}") return False def get_session(self): """Get a session from the pool (thread-safe)""" if not self.setup_complete: self.setup_sessions() return self.session_pool.get() # Wait for available session def return_session(self, session): """Return session to pool when done""" self.session_pool.put(session) def cleanup_all_sessions(self): """Close all sessions when program ends""" failed_cleanups = 0 for session in self.all_sessions: try: session.close() except: failed_cleanups += 1 if failed_cleanups > 0: print(f"WARNING: {failed_cleanups} sessions failed to close properly") # ================================================================================================ # LICENSE SEARCHER # ================================================================================================ class APILicenseSearcher: """Searches NSW API using HTTP requests""" def __init__(self, config: SimpleConfig, api_manager: APIManager): self.config = config self.api_manager = api_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, api_name: str) -> str: """Compare names with enhanced cleaning and return match status""" if api_name == "Not Found": return "Not Found" if not csv_name or not api_name: return "No Match" # Clean both names before comparison cleaned_csv = self._clean_name_for_matching(csv_name) cleaned_api = self._clean_name_for_matching(api_name) # Normalize both names to standard format norm_csv = self.normalize_employee_name(cleaned_csv) norm_api = self.normalize_employee_name(cleaned_api) # Quick exact match check on cleaned names if norm_csv == norm_api: return "Yes" # Try direct comparison of cleaned names (without format conversion) if cleaned_csv.upper() == cleaned_api.upper(): return "Yes" # Calculate similarity percentage using multiple methods for better accuracy similarity_methods = [ fuzz.token_set_ratio(norm_api, norm_csv), fuzz.token_sort_ratio(cleaned_api.upper(), cleaned_csv.upper()), fuzz.partial_ratio(cleaned_api.upper(), cleaned_csv.upper()), fuzz.ratio(cleaned_api.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, api_expiry: str) -> str: """Calculate expiry status comparison between CSV and API dates""" if not csv_expiry or not api_expiry or api_expiry == "Not Found": return "No License" try: # Parse dates - NSW uses various formats, parser handles them automatically csv_date = parser.parse(csv_expiry, dayfirst=True) api_date = parser.parse(api_expiry, dayfirst=True) is_expired = api_date < datetime.today() dates_match = (csv_date.date() == api_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 API""" # 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: session = self.api_manager.get_session() try: start_time = time.time() result = self._do_api_search(session, 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.api_manager.return_session(session) except Exception as e: last_error = str(e) if attempt < self.config.max_retry_attempts - 1: # Exponential backoff for rate limiting wait_time = self.config.delay_between_requests * (2 ** attempt) time.sleep(wait_time) # 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_api_search(self, session: requests.Session, employee: EmployeeRecord) -> SearchResult: """Actually search the NSW API using HTTP requests""" try: params = {"licenceNumber": employee.license_number.strip()} response = session.get( self.config.verify_url, params=params, timeout=self.config.request_timeout_seconds ) if response.status_code == 200: data = response.json() if isinstance(data, list) and data: result = data[0] return SearchResult( found_name=result.get("licenceName", "Not Found"), license_type=result.get("licenceType", "Not Found"), license_expiry=result.get("expiryDate", "Not Found"), license_status=result.get("status", "Not Found") ) else: return SearchResult() elif response.status_code == 408: # Traffic limit exceeded raise Exception(f"API rate limit exceeded (408)") elif response.status_code == 429: # Too Many Requests retry_after = response.headers.get('Retry-After', '2') raise Exception(f"Rate limited (429), retry after {retry_after}s") else: response.raise_for_status() except requests.Timeout: raise Exception(f"API timeout for license: {employee.license_number}") except requests.RequestException as e: raise Exception(f"API request error: {str(e)}") except Exception as e: raise Exception(f"API 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"NSW 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', '') database_name = result.get('Database Name', '') license_type = result.get('Type', '') expiry_status = result.get('Expiry Status', '') 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%)" database_name == "Not Found" or database_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 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', 'Database Name': f'{len(results)} records', 'Name Match': 'No exceptions', 'Type': 'All found', 'Rolecall Expiry': 'All dates match', 'Database Expiry': 'All valid', 'Expiry Status': 'All active', 'Status': 'All valid' } exception_results = [summary_result] excel_path = original_path.parent / f"{original_path.stem}_NSW_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 NSWLicenseChecker: """Main application class - using NSW API""" def __init__(self): self.config = SimpleConfig() self.api_manager = APIManager(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 your internet connection") print("3. API credentials might be expired or invalid") print("4. Try reducing parallel_workers to 1 if getting rate limit errors") finally: self.api_manager.cleanup_all_sessions() def _show_welcome(self): """Show welcome message""" print("NSW 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_workers} parallel API sessions") 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 = APILicenseSearcher(self.config, self.api_manager) all_results = [] # Silent processing - no verbose startup messages with ThreadPoolExecutor(max_workers=self.config.parallel_workers) as executor: # Submit all search tasks with staggered delays to avoid API burst future_to_employee = {} for i, employee in enumerate(employees): # Small delay between submissions to avoid burst requests time.sleep(0.1) future = executor.submit(searcher.search_single_license, employee) future_to_employee[future] = employee # 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, 'Database Name': search_result.found_name, 'Name Match': search_result.name_matches, 'Rolecall Expiry': employee.csv_expiry, 'Database Expiry': search_result.license_expiry, 'Expiry Status': search_result.expiry_status, '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 delay between API calls to respect rate limits 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, 'Database Name': "Error", 'Name Match': "Error", 'Rolecall Expiry': employee.csv_expiry, 'Database Expiry': "Error", 'Expiry Status': "Error", '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("API request") return " | ".join(notes) if notes else "Processed successfully" # ================================================================================================ # HELPER FUNCTIONS # ================================================================================================ def test_api_connection(): try: config = SimpleConfig() # Test getting access token headers = { "content-type": "application/json", "authorization": config.basic_auth_header } params = {"grant_type": "client_credentials"} response = requests.get( config.token_url, headers=headers, params=params, timeout=30 ) response.raise_for_status() access_token = response.json().get("access_token") if access_token: print("API connection test PASSED!") return True else: print("API token test failed - no access token received") return False except Exception as e: print(f"API connection test failed: {str(e)}") print("Common fixes:") print("1. Check internet connection") print("2. API credentials might be expired") print("3. NSW API service might be down") print("4. Try again later if rate limited") return False # ================================================================================================ # API COMPATIBLE CLASS FOR FLASK ENDPOINTS # ================================================================================================ class NSWLicenseCheckerAPI: """API-compatible wrapper for NSW License Checker""" def __init__(self): self.config = SimpleConfig() self.api_manager = APIManager(self.config) self.searcher = APILicenseSearcher(self.config, self.api_manager) def process_csv_file(self, file_path: str) -> List[Dict]: """Process CSV file and return results as list of dictionaries""" try: # Load employee data employees = SimpleCSVHandler.load_employee_data(file_path, self.config) # Setup API sessions self.api_manager.setup_sessions() # Process all employees results = [] 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, 'Rolecall Name': employee.employee_name, 'Database Name': search_result.found_name, 'Name Match': search_result.name_matches, 'Rolecall Expiry': employee.csv_expiry, 'Database Expiry': search_result.license_expiry, 'Expiry Status': search_result.expiry_status, 'Type': search_result.license_type, 'Status': search_result.license_status, 'Error Message': search_result.error_message if search_result.error_message else None, 'Search Time': search_result.search_time_seconds, 'Retries': search_result.how_many_retries, 'Was Cached': search_result.was_cached } results.append(result_dict) # Add delay between API calls time.sleep(self.config.delay_between_requests) except Exception as e: # Handle individual employee processing errors error_result = { 'Payroll Number': employee.payroll_number, 'License Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'Database Name': "Error", 'Name Match': "Error", 'Rolecall Expiry': employee.csv_expiry, 'Database Expiry': "Error", 'Expiry Status': "Error", 'Type': "Error", 'Status': "Error", 'Error Message': str(e), 'Search Time': 0.0, 'Retries': 0, 'Was Cached': False } results.append(error_result) return results except Exception as e: raise Exception(f"NSW processing failed: {str(e)}") finally: # Cleanup API sessions self.api_manager.cleanup_all_sessions() def get_processing_stats(self, results: List[Dict]) -> Dict: """Calculate processing statistics from results""" if not results: return {} total = len(results) successful = sum(1 for r in results if r.get('Database Name') not in ['Error', 'Not Found']) failed = sum(1 for r in results if r.get('Database Name') in ['Error', 'Not Found']) cached = sum(1 for r in results if r.get('Was Cached', False)) # Calculate name match statistics name_matches = {} for result in results: match_status = result.get('Name Match', 'Unknown') name_matches[match_status] = name_matches.get(match_status, 0) + 1 # Calculate expiry status statistics expiry_statuses = {} for result in results: expiry_status = result.get('Expiry Status', 'Unknown') expiry_statuses[expiry_status] = expiry_statuses.get(expiry_status, 0) + 1 return { 'total_records': total, 'successful_searches': successful, 'failed_searches': failed, 'cached_results': cached, 'success_rate': (successful / total * 100) if total > 0 else 0, 'name_match_breakdown': name_matches, 'expiry_status_breakdown': expiry_statuses } 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 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_expiry=str(record_data.get('Expiry/Update Date', '')).strip(), csv_row_number=i + 1 ) employee.clean_data() employees.append(employee) # Setup API sessions self.api_manager.setup_sessions() # Process all employees results = [] 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, 'Rolecall Name': employee.employee_name, 'Database Name': search_result.found_name, 'Name Match': search_result.name_matches, 'Rolecall Expiry': employee.csv_expiry, 'Database Expiry': search_result.license_expiry, 'Expiry Status': search_result.expiry_status, 'Type': search_result.license_type, 'Status': search_result.license_status, 'Error Message': search_result.error_message if search_result.error_message else None, 'Search Time': search_result.search_time_seconds, 'Retries': search_result.how_many_retries, 'Was Cached': search_result.was_cached } results.append(result_dict) # Add delay between API calls time.sleep(self.config.delay_between_requests) except Exception as e: # Handle individual employee processing errors error_result = { 'Payroll Number': employee.payroll_number, 'License Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'Database Name': "Error", 'Name Match': "Error", 'Rolecall Expiry': employee.csv_expiry, 'Database Expiry': "Error", 'Expiry Status': "Error", 'Type': "Error", 'Status': "Error", 'Error Message': str(e), 'Search Time': 0.0, 'Retries': 0, 'Was Cached': False } results.append(error_result) return results except Exception as e: raise Exception(f"NSW processing failed: {str(e)}") finally: # Cleanup API sessions self.api_manager.cleanup_all_sessions() # ================================================================================================ # 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_api_connection() return # Run the main application try: print("Starting NSW License Checker...") # Quick connection test print("\nTesting API connection to NSW OneGov...") if not test_api_connection(): print("Connection issues detected, but continuing anyway...") print("If all searches fail, the API credentials may be invalid or expired") app = NSWLicenseChecker() app.run() except KeyboardInterrupt: print("\nAPI application stopped by user") except Exception as e: print(f"\nFATAL ERROR: {str(e)}") print("\nAPI TROUBLESHOOTING:") print("1. Check your CSV file has the right columns") print("2. Check your internet connection") print("3. API credentials might be expired or invalid") print("4. Try reducing parallel_workers to 1 if getting rate limit errors") print("5. Run with --test to check API connectivity") # ================================================================================================ # RUN THE PROGRAM # ================================================================================================ if __name__ == "__main__": main()