import csv import time import logging from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.common.exceptions import NoSuchElementException, TimeoutException from selenium.webdriver.support.ui import WebDriverWait from rapidfuzz import fuzz from datetime import datetime from dateutil import parser from contextlib import contextmanager from typing import Dict, List, Optional, Tuple from dataclasses import dataclass, asdict # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) @dataclass class LicenseRecord: """Data class for license records""" payroll_number: str employee_name: str license_number: str csv_expiry: str @classmethod def from_csv_row(cls, row: Dict[str, str]) -> 'LicenseRecord': return cls( payroll_number=row.get('Payroll Number', '').strip(), employee_name=row.get('Employee Name', '').strip(), license_number=row.get('License Number', '').strip(), csv_expiry=row.get('Expiry/Update Date', '').strip() ) @dataclass class ACTVerificationResult: """Data class for ACT verification results""" payroll_number: str license_number: str rolecall_name: str database_name: str name_match: str rolecall_expiry: str database_expiry: str expiry_status: str license_type: str class ACTLicenseCheckerAPI: """ACT License checking API service""" def __init__(self): self.BOM_CHAR = "\ufeff" self.SEARCH_URL_TEMPLATE = "https://services.accesscanberra.act.gov.au/s/public-registers/occupational-register?registerid=security-employee&licenceID={}" def normalize_name(self, name: str) -> str: """Normalize name format for comparison""" if not name: return "" parts = name.strip().upper().split() return f"{parts[-1]}, {' '.join(parts[:-1])}" if len(parts) >= 2 else name.strip().upper() def load_csv_file(self, file_path: str) -> List[LicenseRecord]: """Load CSV file and return records""" records = [] try: with open(file_path, newline='', encoding='utf-8') as csvfile: csv_reader = csv.DictReader(csvfile) if not csv_reader.fieldnames: raise ValueError("CSV file has no headers") # Clean BOM and whitespace from headers cleaned_headers = [ h.lstrip(self.BOM_CHAR).strip() if h.startswith(self.BOM_CHAR) else h.strip() for h in csv_reader.fieldnames ] csv_reader.fieldnames = cleaned_headers # Load records for row in csv_reader: cleaned_row = {k.strip(): v.strip() if isinstance(v, str) else v for k, v in row.items()} records.append(LicenseRecord.from_csv_row(cleaned_row)) logger.info(f"Loaded {len(records)} records from {file_path}") return records except Exception as e: logger.error(f"Error loading CSV: {e}") raise @contextmanager def setup_driver(self): """Context manager for WebDriver setup and cleanup""" chrome_options = Options() chrome_options.add_argument("--disable-gpu") chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--window-size=1920,1080") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--disable-extensions") chrome_options.add_argument("--disable-blink-features=AutomationControlled") chrome_options.add_argument("--headless") # Run in headless mode for API driver = None try: driver = webdriver.Chrome(options=chrome_options) wait = WebDriverWait(driver, 30) yield driver, wait finally: if driver: driver.quit() def get_text_or_none(self, driver: webdriver.Chrome, xpath: str) -> Optional[str]: """Helper method to safely get text from element""" try: element = driver.find_element(By.XPATH, xpath) return element.text.strip() if element.text else None except NoSuchElementException: return None def extract_license_code(self, type_str: str) -> str: """Extract license code for sorting""" try: parts = type_str.split() return parts[2] if len(parts) > 2 else type_str except (IndexError, AttributeError): return type_str def calculate_name_match(self, csv_name: str, web_name: str) -> str: """Calculate name match percentage""" if not csv_name or not web_name or web_name == "Not Found": return "No License" norm_csv = self.normalize_name(csv_name) norm_web = web_name.upper() similarity = fuzz.token_set_ratio(norm_web, norm_csv) return "Yes" if similarity >= 85 else f"No ({similarity:.1f}%)" def calculate_expiry_status(self, csv_expiry: str, web_expiry_parts: List[str]) -> str: """Calculate expiry status comparison for multiple licenses""" if not csv_expiry or not web_expiry_parts: return "Missing Date" try: csv_date = parser.parse(csv_expiry, dayfirst=True) matches = [] for date_str in web_expiry_parts: try: web_date = parser.parse(date_str.strip(), dayfirst=True) is_expired = web_date < datetime.today() dates_match = (csv_date.date() == web_date.date()) if dates_match and not is_expired: matches.append("Active") elif dates_match and is_expired: matches.append("Expired") elif not dates_match and not is_expired: matches.append("Active - Date Wrong") else: matches.append("Expired - Date Wrong") except Exception: matches.append("Parse Error") return " | ".join(matches) if matches else "Missing Date" except Exception as e: logger.warning(f"Error calculating expiry status: {e}") return f"Error: {str(e)[:20]}" def verify_license(self, driver: webdriver.Chrome, wait: WebDriverWait, license_number: str) -> Tuple[str, str, str]: """Verify a single license and return extracted data""" if not license_number: return "Not Found", "Not Found", "Not Found" try: search_url = self.SEARCH_URL_TEMPLATE.format(license_number) driver.get(search_url) # Wait for page to load - check for either table or no record message wait.until( lambda d: ( d.find_elements(By.XPATH, "//c-cxs-spf-public-register-occupational//table") or "No record" in d.page_source or "does not hold a licence" in d.page_source ) ) # Extract name try: name_element = driver.find_element(By.XPATH, "/html/body/div[3]/div[4]/div/div/div[1]/div/div/c-cxs-spf-banner/div/div/div/div/div/div/h1") web_name = name_element.text.strip() except NoSuchElementException: web_name = "Not Found" # Extract license types and expiry dates web_type_parts = [] web_expiry_parts = [] i = 1 while True: type_xpath = f"/html/body/div[3]/div[4]/div/div/div[2]/div/div/c-cxs-spf-public-register-occupational/div/div/div/div[2]/div[{i}]/table/tbody/tr[2]/td" expiry_xpath = f"/html/body/div[3]/div[4]/div/div/div[2]/div/div/c-cxs-spf-public-register-occupational/div/div/div/div[2]/div[{i}]/table/tbody/tr[4]/td" type_val = self.get_text_or_none(driver, type_xpath) expiry_val = self.get_text_or_none(driver, expiry_xpath) if type_val and expiry_val: web_type_parts.append(type_val) clean_expiry = expiry_val.replace("Date expiry:", "").strip() web_expiry_parts.append(clean_expiry) i += 1 else: break # Sort by license code if web_type_parts and web_expiry_parts: type_expiry_pairs = list(zip(web_type_parts, web_expiry_parts)) type_expiry_pairs.sort(key=lambda pair: self.extract_license_code(pair[0])) web_type_parts_sorted = [p[0] for p in type_expiry_pairs] web_expiry_parts_sorted = [p[1] for p in type_expiry_pairs] web_type = " | ".join(web_type_parts_sorted) web_expiry = " | ".join(web_expiry_parts_sorted) else: web_type = "Not Found" web_expiry = "Not Found" return web_name, web_type, web_expiry except (TimeoutException, NoSuchElementException) as e: logger.warning(f"License verification failed for {license_number}: {e}") return "Not Found", "Not Found", "Not Found" except Exception as e: logger.error(f"Unexpected error verifying license {license_number}: {e}") return "Error", "Error", "Error" def process_record(self, driver: webdriver.Chrome, wait: WebDriverWait, record: LicenseRecord) -> ACTVerificationResult: """Process a single record and return verification result""" web_name, web_type, web_expiry = self.verify_license(driver, wait, record.license_number) name_match = self.calculate_name_match(record.employee_name, web_name) # Extract expiry parts for status calculation web_expiry_parts = [] if web_expiry != "Not Found" and web_expiry != "Error": web_expiry_parts = [part.strip() for part in web_expiry.split(" | ")] expiry_status = self.calculate_expiry_status(record.csv_expiry, web_expiry_parts) return ACTVerificationResult( payroll_number=record.payroll_number, license_number=record.license_number, rolecall_name=record.employee_name, database_name=web_name, name_match=name_match, rolecall_expiry=record.csv_expiry, database_expiry=web_expiry, expiry_status=expiry_status, license_type=web_type ) def process_csv_file(self, file_path: str) -> List[Dict]: """Process CSV file and return results as list of dictionaries""" try: # Load records records = self.load_csv_file(file_path) if not records: return [] output_rows = [] start_time = time.time() # Process records with self.setup_driver() as (driver, wait): for i, record in enumerate(records): try: result = self.process_record(driver, wait, record) output_rows.append(result) # Log progress every 10 records if (i + 1) % 10 == 0: elapsed = time.time() - start_time logger.info(f"Processed {i + 1}/{len(records)} records in {elapsed:.1f}s") except Exception as e: logger.error(f"Error processing record {i + 1}: {e}") # Add error result error_result = ACTVerificationResult( payroll_number=record.payroll_number, license_number=record.license_number, rolecall_name=record.employee_name, database_name="Error", name_match="Error", rolecall_expiry=record.csv_expiry, database_expiry="Error", expiry_status="Error", license_type="Error" ) output_rows.append(error_result) # Convert to dictionaries for JSON serialization result_dicts = [] for result in output_rows: result_dicts.append(asdict(result)) total_time = time.time() - start_time logger.info(f"Completed processing {len(records)} records in {total_time:.1f}s") return result_dicts except Exception as e: logger.error(f"Error processing CSV file: {e}") 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 LicenseRecord objects records = [] for record_data in records_data: record = LicenseRecord( 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() ) records.append(record) if not records: return [] output_rows = [] start_time = time.time() # Process records with self.setup_driver() as (driver, wait): for i, record in enumerate(records): try: result = self.process_record(driver, wait, record) output_rows.append(result) # Log progress every 10 records if (i + 1) % 10 == 0: elapsed = time.time() - start_time logger.info(f"Processed {i + 1}/{len(records)} records in {elapsed:.1f}s") except Exception as e: logger.error(f"Error processing record {i + 1}: {e}") # Add error result error_result = ACTVerificationResult( payroll_number=record.payroll_number, license_number=record.license_number, rolecall_name=record.employee_name, database_name="Error", name_match="Error", rolecall_expiry=record.csv_expiry, database_expiry="Error", expiry_status="Error", license_type="Error" ) output_rows.append(error_result) # Convert to dictionaries for JSON serialization result_dicts = [] for result in output_rows: result_dicts.append(asdict(result)) total_time = time.time() - start_time logger.info(f"Completed processing {len(records)} records in {total_time:.1f}s") return result_dicts except Exception as e: logger.error(f"Error processing records: {e}") raise