Gitead/DentalManagement2025
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ddma_browser_manager) - added singleton browser manager for persistent Chrome instance; updated Selenium service to utilize the manager for session handling and OTP processing

Browse Source
This commit is contained in:
Emile
2026-01-05 22:01:39 -05:00
parent d3dd8077cd
commit ef18a0017c
3 changed files with 266 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
apps/SeleniumService/ddma_browser_manager.py Normal file
View File

@@ -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

36
apps/SeleniumService/helpers_ddma_eligibility.py
View File

@@ -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,11 +148,21 @@ 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(
(By.XPATH, "//input[contains(@aria-lable,'Verification code') or contains(@placeholder,'Enter your verification code')]")
@@ -170,6 +181,11 @@ async def start_ddma_run(sid: str, data: dict, url: str):
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()
await asyncio.sleep(0.5)

170
apps/SeleniumService/selenium_DDMA_eligibilityCheckWorker.py
View File

@@ -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