diff --git a/apps/SeleniumService/ddma_browser_manager.py b/apps/SeleniumService/ddma_browser_manager.py new file mode 100644 index 0000000..57f165b --- /dev/null +++ b/apps/SeleniumService/ddma_browser_manager.py @@ -0,0 +1,102 @@ +""" +Minimal browser manager for DDMA - only handles persistent profile and keeping browser alive. +Does NOT modify any login/OTP logic. +""" +import os +import threading +from selenium import webdriver +from selenium.webdriver.chrome.service import Service +from webdriver_manager.chrome import ChromeDriverManager + + +class DDMABrowserManager: + """ + Singleton that manages a persistent Chrome browser instance. + - Uses --user-data-dir for persistent profile (device trust tokens, cookies) + - Keeps browser alive between patient runs + """ + _instance = None + _lock = threading.Lock() + + def __new__(cls): + with cls._lock: + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._driver = None + cls._instance.profile_dir = os.path.abspath("chrome_profile_ddma") + cls._instance.download_dir = os.path.abspath("seleniumDownloads") + os.makedirs(cls._instance.profile_dir, exist_ok=True) + os.makedirs(cls._instance.download_dir, exist_ok=True) + return cls._instance + + def get_driver(self, headless=False): + """Get or create the persistent browser instance.""" + with self._lock: + if self._driver is None: + print("[BrowserManager] Driver is None, creating new driver") + self._create_driver(headless) + elif not self._is_alive(): + print("[BrowserManager] Driver not alive, recreating") + self._create_driver(headless) + else: + print("[BrowserManager] Reusing existing driver") + return self._driver + + def _is_alive(self): + """Check if browser is still responsive.""" + try: + url = self._driver.current_url + print(f"[BrowserManager] Driver alive, current URL: {url[:50]}...") + return True + except Exception as e: + print(f"[BrowserManager] Driver not alive: {e}") + return False + + def _create_driver(self, headless=False): + """Create browser with persistent profile.""" + if self._driver: + try: + self._driver.quit() + except: + pass + + options = webdriver.ChromeOptions() + if headless: + options.add_argument("--headless") + + # Persistent profile - THIS IS THE KEY for device trust + options.add_argument(f"--user-data-dir={self.profile_dir}") + options.add_argument("--no-sandbox") + options.add_argument("--disable-dev-shm-usage") + + prefs = { + "download.default_directory": self.download_dir, + "plugins.always_open_pdf_externally": True, + "download.prompt_for_download": False, + "download.directory_upgrade": True + } + options.add_experimental_option("prefs", prefs) + + service = Service(ChromeDriverManager().install()) + self._driver = webdriver.Chrome(service=service, options=options) + self._driver.maximize_window() + + def quit_driver(self): + """Quit browser (only call on shutdown).""" + with self._lock: + if self._driver: + try: + self._driver.quit() + except: + pass + self._driver = None + + +# Singleton accessor +_manager = None + +def get_browser_manager(): + global _manager + if _manager is None: + _manager = DDMABrowserManager() + return _manager diff --git a/apps/SeleniumService/helpers_ddma_eligibility.py b/apps/SeleniumService/helpers_ddma_eligibility.py index 0177013..c9db23c 100644 --- a/apps/SeleniumService/helpers_ddma_eligibility.py +++ b/apps/SeleniumService/helpers_ddma_eligibility.py @@ -60,14 +60,8 @@ async def cleanup_session(sid: str, message: str | None = None): except Exception: pass - # Attempt to quit driver (may already be dead) - driver = s.get("driver") - if driver: - try: - driver.quit() - except Exception: - # ignore errors from quit (session already gone) - pass + # NOTE: Do NOT quit driver - keep browser alive for next patient + # Browser manager handles the persistent browser instance finally: # Remove session entry from map @@ -126,8 +120,15 @@ async def start_ddma_run(sid: str, data: dict, url: str): await cleanup_session(sid, s["message"]) return {"status": "error", "message": s["message"]} + # Already logged in - session persisted from profile, skip to step1 + if isinstance(login_result, str) and login_result == "ALREADY_LOGGED_IN": + print("[start_ddma_run] Session persisted - skipping OTP") + s["status"] = "running" + s["message"] = "Session persisted" + # Continue to step1 below + # OTP required path - if isinstance(login_result, str) and login_result == "OTP_REQUIRED": + elif isinstance(login_result, str) and login_result == "OTP_REQUIRED": s["status"] = "waiting_for_otp" s["message"] = "OTP required for login" s["last_activity"] = time.time() @@ -147,10 +148,20 @@ async def start_ddma_run(sid: str, data: dict, url: str): await cleanup_session(sid) return {"status": "error", "message": "OTP missing after event"} - # Submit OTP in the same Selenium window + # Submit OTP - check if it's in a popup window try: driver = s["driver"] wait = WebDriverWait(driver, 30) + + # Check if there's a popup window and switch to it + original_window = driver.current_window_handle + all_windows = driver.window_handles + if len(all_windows) > 1: + for window in all_windows: + if window != original_window: + driver.switch_to.window(window) + print(f"[OTP] Switched to popup window for OTP entry") + break otp_input = wait.until( EC.presence_of_element_located( @@ -169,6 +180,11 @@ async def start_ddma_run(sid: str, data: dict, url: str): submit_btn.click() except Exception: otp_input.send_keys("\n") + + # Wait for verification and switch back to main window if needed + await asyncio.sleep(2) + if len(driver.window_handles) > 0: + driver.switch_to.window(driver.window_handles[0]) s["status"] = "otp_submitted" s["last_activity"] = time.time() diff --git a/apps/SeleniumService/selenium_DDMA_eligibilityCheckWorker.py b/apps/SeleniumService/selenium_DDMA_eligibilityCheckWorker.py index d7dc0ea..1f3a010 100644 --- a/apps/SeleniumService/selenium_DDMA_eligibilityCheckWorker.py +++ b/apps/SeleniumService/selenium_DDMA_eligibilityCheckWorker.py @@ -10,6 +10,8 @@ import time import os import base64 +from ddma_browser_manager import get_browser_manager + class AutomationDeltaDentalMAEligibilityCheck: def __init__(self, data): self.headless = False @@ -24,31 +26,139 @@ class AutomationDeltaDentalMAEligibilityCheck: self.massddma_username = self.data.get("massddmaUsername", "") self.massddma_password = self.data.get("massddmaPassword", "") - self.download_dir = os.path.abspath("seleniumDownloads") + # Use browser manager's download dir + self.download_dir = get_browser_manager().download_dir os.makedirs(self.download_dir, exist_ok=True) def config_driver(self): - options = webdriver.ChromeOptions() - if self.headless: - options.add_argument("--headless") - - # Add PDF download preferences - prefs = { - "download.default_directory": self.download_dir, - "plugins.always_open_pdf_externally": True, - "download.prompt_for_download": False, - "download.directory_upgrade": True - } - options.add_experimental_option("prefs", prefs) - - s = Service(ChromeDriverManager().install()) - driver = webdriver.Chrome(service=s, options=options) - self.driver = driver + # Use persistent browser from manager (keeps device trust tokens) + self.driver = get_browser_manager().get_driver(self.headless) def login(self, url): wait = WebDriverWait(self.driver, 30) try: + # First check if we're already on a logged-in page (from previous run) + try: + current_url = self.driver.current_url + print(f"[login] Current URL: {current_url}") + + # Check if we're on any logged-in page (dashboard, member pages, etc.) + logged_in_patterns = ["member", "dashboard", "eligibility", "search", "patients"] + is_logged_in_url = any(pattern in current_url.lower() for pattern in logged_in_patterns) + + if is_logged_in_url and "onboarding" not in current_url.lower(): + print(f"[login] Already on logged-in page - skipping login entirely") + # Navigate directly to member search if not already there + if "member" not in current_url.lower(): + # Try to find a link to member search or just check for search input + try: + member_search = WebDriverWait(self.driver, 5).until( + EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]')) + ) + print("[login] Found member search input - returning ALREADY_LOGGED_IN") + return "ALREADY_LOGGED_IN" + except TimeoutException: + # Try navigating to members page + members_url = "https://providers.deltadentalma.com/members" + print(f"[login] Navigating to members page: {members_url}") + self.driver.get(members_url) + time.sleep(2) + + # Verify we have the member search input + try: + member_search = WebDriverWait(self.driver, 5).until( + EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]')) + ) + print("[login] Member search found - ALREADY_LOGGED_IN") + return "ALREADY_LOGGED_IN" + except TimeoutException: + print("[login] Could not find member search, will try login") + except Exception as e: + print(f"[login] Error checking current state: {e}") + + # Navigate to login URL self.driver.get(url) + time.sleep(2) # Wait for page to load and any redirects + + # Check if we got redirected to member search (session still valid) + try: + current_url = self.driver.current_url + print(f"[login] URL after navigation: {current_url}") + + if "onboarding" not in current_url.lower(): + member_search = WebDriverWait(self.driver, 3).until( + EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]')) + ) + if member_search: + print("[login] Session valid - skipping login") + return "ALREADY_LOGGED_IN" + except TimeoutException: + print("[login] Proceeding with login") + + # Dismiss any "Authentication flow continued in another tab" modal + modal_dismissed = False + try: + ok_button = WebDriverWait(self.driver, 3).until( + EC.element_to_be_clickable((By.XPATH, "//button[normalize-space(text())='Ok' or normalize-space(text())='OK']")) + ) + ok_button.click() + print("[login] Dismissed authentication modal") + modal_dismissed = True + time.sleep(2) + + # Check if a popup window opened for authentication + all_windows = self.driver.window_handles + print(f"[login] Windows after modal dismiss: {len(all_windows)}") + + if len(all_windows) > 1: + # Switch to the auth popup + original_window = self.driver.current_window_handle + for window in all_windows: + if window != original_window: + self.driver.switch_to.window(window) + print(f"[login] Switched to auth popup window") + break + + # Look for OTP input in the popup + try: + otp_candidate = WebDriverWait(self.driver, 10).until( + EC.presence_of_element_located( + (By.XPATH, "//input[contains(@aria-lable,'Verification code') or contains(@placeholder,'Enter your verification code') or contains(@aria-label,'Verification code')]") + ) + ) + if otp_candidate: + print("[login] OTP input found in popup -> OTP_REQUIRED") + return "OTP_REQUIRED" + except TimeoutException: + print("[login] No OTP in popup, checking main window") + self.driver.switch_to.window(original_window) + + except TimeoutException: + pass # No modal present + + # If modal was dismissed but no popup, page might have changed - wait and check + if modal_dismissed: + time.sleep(2) + # Check if we're now on member search page (already authenticated) + try: + member_search = WebDriverWait(self.driver, 5).until( + EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]')) + ) + if member_search: + print("[login] Already authenticated after modal dismiss") + return "ALREADY_LOGGED_IN" + except TimeoutException: + pass + + # Try to fill login form + try: + email_field = WebDriverWait(self.driver, 10).until( + EC.element_to_be_clickable((By.XPATH, "//input[@name='username' and @type='text']")) + ) + except TimeoutException: + print("[login] Could not find login form - page may have changed") + return "ERROR: Login form not found" + email_field = wait.until(EC.presence_of_element_located((By.XPATH, "//input[@name='username' and @type='text']"))) email_field.clear() email_field.send_keys(self.massddma_username) @@ -226,6 +336,15 @@ class AutomationDeltaDentalMAEligibilityCheck: pass print("Screenshot saved at:", screenshot_path) + + # Close the browser window after screenshot (session preserved in profile) + try: + from ddma_browser_manager import get_browser_manager + get_browser_manager().quit_driver() + print("[step2] Browser closed - session preserved in profile") + except Exception as e: + print(f"[step2] Error closing browser: {e}") + output = { "status": "success", "eligibility": eligibilityText, @@ -254,14 +373,7 @@ class AutomationDeltaDentalMAEligibilityCheck: print(f"[cleanup] unexpected error while cleaning downloads dir: {cleanup_exc}") return {"status": "error", "message": str(e)} - finally: - # Keep your existing quit behavior; if you want the driver to remain open for further - # actions, remove or change this. - if self.driver: - try: - self.driver.quit() - except Exception: - pass + # NOTE: Do NOT quit driver here - keep browser alive for next patient def main_workflow(self, url): try: @@ -289,10 +401,4 @@ class AutomationDeltaDentalMAEligibilityCheck: "status": "error", "message": e } - - finally: - try: - if self.driver: - self.driver.quit() - except Exception: - pass + # NOTE: Do NOT quit driver - keep browser alive for next patient