""" TAS License Checker - Web URL Integration Modified version of tas.py that captures browser URL for CAPTCHA solving in web UI """ import json import time import sys import os import threading from pathlib import Path from dataclasses import dataclass from typing import List, Dict, Optional from datetime import datetime # 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.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 from dateutil import parser # Import original TAS components from tas import ( SimpleConfig, EmployeeRecord, SearchResult, SimpleCSVHandler, SeleniumLicenseSearcher, OptimizedBrowserManager, SimpleExcelGenerator ) class WebURLBrowserManager(OptimizedBrowserManager): """Browser manager that captures URL for web UI CAPTCHA handling""" def __init__(self, config: SimpleConfig, status_file: str): super().__init__(config) self.status_file = status_file self.captcha_detected = False self.captcha_solved = False self.captcha_url = None self.captcha_lock = threading.Lock() def handle_captcha_if_needed(self): """Handle CAPTCHA if it appears - capture URL for web UI""" if self.captcha_handled: return try: # Get the current driver driver = self.browser if not driver: return # Check if browser window is still open try: current_url = driver.current_url if not current_url: return except Exception: print("Browser window not available for CAPTCHA check") return # Wait a moment for page to fully load time.sleep(2) # Check for CAPTCHA elements - be more specific captcha_elements = driver.find_elements(By.CSS_SELECTOR, "iframe[src*='recaptcha'], .g-recaptcha, #captcha, .captcha, [class*='captcha']") # Also check if search form is blocked/disabled try: search_input = driver.find_element(By.XPATH, '//*[@id="ctl00_ctl00_ctl00_ctlMainContent_ctlMainContent_MainContent_ctlOnlineSearch_txtLicenceNumberSearch"]') if search_input and not search_input.is_enabled(): print("Search form is disabled - likely CAPTCHA required") captcha_elements = [True] # Force CAPTCHA detection except: pass if captcha_elements: with self.captcha_lock: self.captcha_detected = True self.captcha_url = driver.current_url # Update status file with CAPTCHA URL self._update_captcha_status() print("\n" + "="*60) print("⚠️ CAPTCHA DETECTED!") print("="*60) print("A CAPTCHA has appeared on the TAS website.") print("Please solve the CAPTCHA in the browser window.") print("The web UI will show the CAPTCHA URL for you to solve.") print("="*60) # For web UI, we don't wait for input - just mark as detected # The web UI will handle the CAPTCHA solving with self.captcha_lock: self.captcha_solved = True self.captcha_handled = True print("✓ CAPTCHA URL captured for web UI") except Exception as e: print(f"Error checking for CAPTCHA: {e}") # Don't let CAPTCHA errors stop the process pass def _update_captcha_status(self): """Update status file with CAPTCHA information""" try: status_data = { 'status': 'captcha_required', 'message': 'CAPTCHA detected! Please solve it in the browser window.', 'captcha_detected': True, 'captcha_url': self.captcha_url, 'timestamp': datetime.now().isoformat() } with open(self.status_file, 'w') as f: json.dump(status_data, f, indent=2) except Exception as e: print(f"Error updating CAPTCHA status: {e}") def mark_captcha_solved(self): """Mark CAPTCHA as solved""" with self.captcha_lock: self.captcha_solved = True self.captcha_handled = True def wait_for_captcha_solution(self, timeout=60): """Wait for CAPTCHA to be solved (for web UI)""" if not self.captcha_detected: return True print("Waiting for CAPTCHA to be solved...") start_time = time.time() while time.time() - start_time < timeout: try: # Get the current driver driver = self.browser if not driver: print("Browser not available, waiting...") time.sleep(2) continue # Check if browser window is still open try: current_url = driver.current_url if not current_url: print("Browser window closed, continuing...") return True except Exception: print("Browser window closed, continuing...") return True # Check if we can proceed with the search (CAPTCHA solved) # Look for the search form elements to see if we can proceed try: search_input = driver.find_element(By.XPATH, '//*[@id="ctl00_ctl00_ctl00_ctlMainContent_ctlMainContent_MainContent_ctlOnlineSearch_txtLicenceNumberSearch"]') if search_input and search_input.is_enabled(): print("✓ CAPTCHA appears to be solved - search form is accessible") return True except: pass # Also check if CAPTCHA elements are still present captcha_elements = driver.find_elements(By.CSS_SELECTOR, "iframe[src*='recaptcha'], .g-recaptcha, #captcha, .captcha") if not captcha_elements: print("✓ CAPTCHA appears to be solved - no CAPTCHA elements found") return True # Check if we've been waiting too long - just proceed if time.time() - start_time > 30: # After 30 seconds, just proceed print("⚠️ CAPTCHA wait timeout - proceeding anyway") return True time.sleep(2) # Check every 2 seconds except Exception as e: print(f"Error checking CAPTCHA status: {e}") # If browser is closed, just continue if "no such window" in str(e).lower() or "web view not found" in str(e).lower(): print("Browser window closed, continuing...") return True time.sleep(2) print("⚠️ CAPTCHA timeout - continuing anyway") return False class WebURLTASLicenseChecker: """Web URL-integrated TAS License Checker""" def __init__(self, job_id: str, status_file: str, userId: str = '', entityType: str = '', userRole: str = ''): self.job_id = job_id self.status_file = status_file self.userId = userId self.entityType = entityType self.userRole = userRole self.config = SimpleConfig() self.config.show_browser_for_captcha = True self.browser_manager = None def process_file(self, csv_file_path: str) -> str: """Process CSV file and return result file path""" try: # Update status self._update_status("loading_data", "Loading employee data...", 0) # Load employee data employees = SimpleCSVHandler.load_employee_data(csv_file_path, self.config) # Update status self._update_status("processing", f"Processing {len(employees)} employees...", 0) # Initialize browser manager self.browser_manager = WebURLBrowserManager(self.config, self.status_file) self.browser_manager.setup_browsers() # Initialize searcher searcher = SeleniumLicenseSearcher(self.config, self.browser_manager) # Process employees results = [] for i, employee in enumerate(employees): try: # Update status progress = int((i / len(employees)) * 90) self._update_status("processing", f"Processing {employee.employee_name}... ({i+1}/{len(employees)})", progress) # Handle CAPTCHA if needed if self.browser_manager: self.browser_manager.handle_captcha_if_needed() if self.browser_manager.captcha_detected: # Wait for CAPTCHA to be solved with shorter timeout print("CAPTCHA detected, waiting for solution...") if not self.browser_manager.wait_for_captcha_solution(timeout=60): print("CAPTCHA timeout - continuing anyway") # Reset CAPTCHA state to continue self.browser_manager.captcha_detected = False self.browser_manager.captcha_handled = True # Search for license search_result = searcher.search_single_license(employee) # Create result dictionary 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': search_result.license_type, 'Licence Expiry': search_result.license_expiry, 'Rolecall Expiry': employee.csv_expiry, 'Expiry Status': search_result.expiry_status, 'User ID': self.userId, 'Entity Type': self.entityType, 'User Role': self.userRole } results.append(result_dict) # Add delay between requests time.sleep(self.config.delay_between_requests) except Exception as e: print(f"Error processing {employee.employee_name}: {e}") # Create error result error_result = { 'Payroll Number': employee.payroll_number, 'Licence Number': employee.license_number, 'Rolecall Name': employee.employee_name, 'Licence Name': 'ERROR', 'Name Match': False, 'Licence Type': 'ERROR', 'Licence Expiry': 'ERROR', 'Rolecall Expiry': employee.csv_expiry, 'Expiry Status': f'Error: {str(e)}', 'User ID': self.userId, 'Entity Type': self.entityType, 'User Role': self.userRole } results.append(error_result) # Generate Excel report self._update_status("generating_report", "Generating Excel report...", 95) excel_path = SimpleExcelGenerator.create_excel_report(results, csv_file_path) # Move result to results directory result_filename = f"TAS_Results_{self.job_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" result_path = os.path.join("results", result_filename) if not os.path.exists("results"): os.makedirs("results") os.rename(excel_path, result_path) # Update final status self._update_status("completed", "Processing completed successfully!", 100) return result_path except Exception as e: self._update_status("error", f"Processing failed: {str(e)}", 0) raise finally: if self.browser_manager: self.browser_manager.cleanup_all_browsers() def _update_status(self, status: str, message: str, progress: int = 0, captcha_url: str = None): """Update status file""" try: status_data = { 'job_id': self.job_id, 'status': status, 'message': message, 'progress': progress, 'captcha_url': captcha_url, 'timestamp': datetime.now().isoformat() } with open(self.status_file, 'w') as f: json.dump(status_data, f, indent=2) except Exception as e: print(f"Error updating status: {e}") def mark_captcha_solved(self): """Mark CAPTCHA as solved""" if self.browser_manager: self.browser_manager.mark_captcha_solved() def main(): """Main function for web integration""" if len(sys.argv) < 3: print("Usage: python tas_web.py [userId] [entityType] [userRole]") sys.exit(1) csv_file = sys.argv[1] job_id = sys.argv[2] status_file = sys.argv[3] userId = sys.argv[4] if len(sys.argv) > 4 else '' entityType = sys.argv[5] if len(sys.argv) > 5 else '' userRole = sys.argv[6] if len(sys.argv) > 6 else '' checker = WebURLTASLicenseChecker(job_id, status_file, userId, entityType, userRole) try: result_path = checker.process_file(csv_file) print(f"Processing completed. Result saved to: {result_path}") except Exception as e: print(f"Processing failed: {str(e)}") sys.exit(1) if __name__ == "__main__": main()