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

feat(singleton-selenium-driver) - by emile

Browse Source
This commit is contained in:
Potenz
2026-01-11 20:43:46 +05:30
parent 7a4ea21658
commit 656e98687f
6 changed files with 271 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -37,3 +37,5 @@ dist/
# env
*.env
*chrome_profile_ddma*

2
apps/PatientDataExtractorService/package.json
View File

@@ -3,6 +3,6 @@
"private": true,
"scripts": {
"postinstall": "pip install -r requirements.txt",
"dev": "python main.py"
"dev": "python3 main.py"
}
}

2
apps/PaymentOCRService/package.json
View File

@@ -3,6 +3,6 @@
"private": true,
"scripts": {
"postinstall": "pip install -r requirements.txt",
"dev": "python main.py"
"dev": "python3 main.py"
}
}

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)

175
apps/SeleniumService/selenium_DDMA_eligibilityCheckWorker.py
View File

@@ -1,15 +1,14 @@
from selenium import webdriver
from selenium.common.exceptions import WebDriverException, TimeoutException
from selenium.webdriver.chrome.service import Service
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from webdriver_manager.chrome import ChromeDriverManager
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 +23,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 +333,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 +370,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 +398,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