Files
DentalManagementMH07/apps/SeleniumService/selenium_DDMA_eligibilityCheckWorker.py
Gitead 877feb32fa fix: update DDMA eligibility check worker for new deltadentalma.com sign-in and search flow
Delta Dental MA moved their provider login from providers.deltadentalma.com/onboarding
to www.deltadentalma.com/sign-in with new form field selectors, and moved eligibility
search behind a "Group/Health Connector" search page keyed by subscriber ID + birthdate
instead of the old member-search form. Updates login, OTP/MFA handling, and member
search/selection across the eligibility and claim DDMA workers and their session helpers.
2026-07-04 22:57:59 -04:00

654 lines
31 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from selenium import webdriver
from selenium.common.exceptions import WebDriverException, TimeoutException
from selenium.webdriver.chrome.service import Service
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 re
import base64
import calendar
from ddma_browser_manager import get_browser_manager
class AutomationDeltaDentalMAEligibilityCheck:
def __init__(self, data):
self.headless = False
self.driver = None
self.data = data.get("data", {}) if isinstance(data, dict) else {}
# Flatten values for convenience
self.memberId = self.data.get("memberId", "")
self.dateOfBirth = self.data.get("dateOfBirth", "")
self.firstName = self.data.get("firstName", "")
self.lastName = self.data.get("lastName", "")
self.massddma_username = self.data.get("massddmaUsername", "")
self.massddma_password = self.data.get("massddmaPassword", "")
self.providerName = self.data.get("providerName", "")
# 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):
# Use persistent browser from manager (keeps device trust tokens)
self.driver = get_browser_manager().get_driver(self.headless)
def _force_logout(self):
"""Force logout by clearing cookies when credentials change."""
try:
print("[DDMA login] Forcing logout due to credential change...")
browser_manager = get_browser_manager()
# Try to click logout button if visible
try:
self.driver.get("https://www.deltadentalma.com/")
time.sleep(2)
logout_selectors = [
"//button[contains(text(), 'Log out') or contains(text(), 'Logout') or contains(text(), 'Sign out')]",
"//a[contains(text(), 'Log out') or contains(text(), 'Logout') or contains(text(), 'Sign out')]",
"//button[@aria-label='Log out' or @aria-label='Logout' or @aria-label='Sign out']",
"//*[contains(@class, 'logout') or contains(@class, 'signout')]"
]
for selector in logout_selectors:
try:
logout_btn = WebDriverWait(self.driver, 3).until(
EC.element_to_be_clickable((By.XPATH, selector))
)
logout_btn.click()
print("[DDMA login] Clicked logout button")
time.sleep(2)
break
except TimeoutException:
continue
except Exception as e:
print(f"[DDMA login] Could not click logout button: {e}")
# Clear cookies as backup
try:
self.driver.delete_all_cookies()
print("[DDMA login] Cleared all cookies")
except Exception as e:
print(f"[DDMA login] Error clearing cookies: {e}")
browser_manager.clear_credentials_hash()
print("[DDMA login] Logout complete")
return True
except Exception as e:
print(f"[DDMA login] Error during forced logout: {e}")
return False
def login(self, url):
wait = WebDriverWait(self.driver, 30)
browser_manager = get_browser_manager()
try:
# Check if credentials changed — force logout first
if self.massddma_username and browser_manager.credentials_changed(self.massddma_username):
self._force_logout()
self.driver.get(url)
time.sleep(2)
# Check if already on a logged-in page (persistent session from profile)
try:
current_url = self.driver.current_url
print(f"[login] Current URL: {current_url}")
logged_in_patterns = ["member", "dashboard", "eligibility", "search", "patients", "private/provider"]
is_logged_in_url = any(pattern in current_url.lower() for pattern in logged_in_patterns)
if is_logged_in_url and "sign-in" not in current_url.lower():
print("[login] Already on logged-in page - skipping login entirely")
if "member" not in current_url.lower():
try:
member_search = WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
print("[login] Found member search input - returning ALREADY_LOGGED_IN")
return "ALREADY_LOGGED_IN"
except TimeoutException:
members_url = "https://www.deltadentalma.com/private/provider/dashboard"
print(f"[login] Navigating to members page: {members_url}")
self.driver.get(members_url)
time.sleep(2)
try:
member_search = WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
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)
# Check if session redirected us straight to member search
try:
current_url = self.driver.current_url
print(f"[login] URL after navigation: {current_url}")
if "sign-in" not in current_url.lower():
member_search = WebDriverWait(self.driver, 3).until(
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
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)
all_windows = self.driver.window_handles
print(f"[login] Windows after modal dismiss: {len(all_windows)}")
if len(all_windows) > 1:
original_window = self.driver.current_window_handle
for window in all_windows:
if window != original_window:
self.driver.switch_to.window(window)
print("[login] Switched to auth popup window")
break
try:
otp_candidate = WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located(
(By.XPATH, "//input[@data-test-id='MfaCode']")
)
)
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_dismissed:
time.sleep(2)
try:
member_search = WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
if member_search:
print("[login] Already authenticated after modal dismiss")
return "ALREADY_LOGGED_IN"
except TimeoutException:
pass
# Fill login form
try:
email_field = WebDriverWait(self.driver, 10).until(
EC.element_to_be_clickable((By.XPATH, "//input[@data-test-id='Username']"))
)
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[@data-test-id='Username']")))
email_field.clear()
email_field.send_keys(self.massddma_username)
password_field = wait.until(EC.presence_of_element_located((By.XPATH, "//input[@data-test-id='Password']")))
password_field.clear()
password_field.send_keys(self.massddma_password)
login_button = wait.until(EC.element_to_be_clickable((By.XPATH, "//button[@data-test-id='SigninSubmitButton']")))
login_button.click()
# Save credentials hash after login attempt
if self.massddma_username:
browser_manager.save_credentials_hash(self.massddma_username)
# OTP detection — wait up to 30 seconds for OTP input
try:
otp_candidate = WebDriverWait(self.driver, 30).until(
EC.presence_of_element_located(
(By.XPATH, "//input[@data-test-id='MfaCode']")
)
)
if otp_candidate:
print("[login] OTP input detected -> OTP_REQUIRED")
return "OTP_REQUIRED"
except TimeoutException:
print("[login] No OTP input detected in allowed time.")
# Check if we're now on the member search page (login succeeded without OTP)
try:
current_url = self.driver.current_url.lower()
if "member" in current_url or "dashboard" in current_url:
member_search = WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
print("[login] Login successful - now on member search page")
return "SUCCESS"
except TimeoutException:
pass
# Check for error messages
try:
error_elem = WebDriverWait(self.driver, 3).until(
EC.presence_of_element_located((By.XPATH, "//*[contains(@class,'error') or contains(text(),'invalid') or contains(text(),'failed')]"))
)
print(f"[login] Login failed - error detected: {error_elem.text}")
return f"ERROR:LOGIN FAILED: {error_elem.text}"
except TimeoutException:
pass
if "sign-in" in self.driver.current_url.lower():
print("[login] Login failed - still on login page")
return "ERROR:LOGIN FAILED: Still on login page"
print("[login] Assuming login succeeded (no errors detected)")
return "SUCCESS"
except Exception as e:
print("[login] Exception during login:", e)
return f"ERROR:LOGIN FAILED: {e}"
def step1(self):
"""Open the Group/Health Connector search page and search by Subscriber ID + Birthdate."""
wait = WebDriverWait(self.driver, 30)
try:
fields = []
if self.memberId:
fields.append(f"SubscriberID: {self.memberId}")
if self.dateOfBirth:
fields.append(f"DOB: {self.dateOfBirth}")
print(f"[DDMA step1] Starting search with: {', '.join(fields)}")
search_page_url = "https://www.deltadentalma.com/private/provider/check-patient-benefits-search"
# Try navigating straight to the search page first — avoids any click/visibility issues
print(f"[DDMA step1] Navigating directly to search page: {search_page_url}")
self.driver.get(search_page_url)
time.sleep(2)
if "check-patient-benefits-search" not in self.driver.current_url.lower():
# Direct deep-link didn't stick (e.g. bounced back to sign-in/dashboard) —
# fall back to clicking the Group/Health Connector link from the dashboard.
print(f"[DDMA step1] Direct navigation redirected to: {self.driver.current_url} — falling back to clicking the link")
if "private/provider/dashboard" not in self.driver.current_url.lower():
dashboard_url = "https://www.deltadentalma.com/private/provider/dashboard"
print(f"[DDMA step1] Navigating to dashboard: {dashboard_url}")
self.driver.get(dashboard_url)
time.sleep(2)
link_xpath = "//a[contains(@href, 'check-patient-benefits-search')]"
try:
connector_link = WebDriverWait(self.driver, 20).until(
EC.element_to_be_clickable((By.XPATH, link_xpath))
)
connector_link.click()
except TimeoutException:
# Element may exist but be hidden (e.g. a responsive duplicate) — try any match via JS click
candidates = self.driver.find_elements(By.XPATH, link_xpath)
print(f"[DDMA step1] element_to_be_clickable timed out; found {len(candidates)} matching link(s) in DOM")
if candidates:
try:
self.driver.execute_script("arguments[0].click();", candidates[0])
except Exception as js_err:
print(f"[DDMA step1] JS click failed: {js_err}")
print(f"[DDMA step1] Current URL: {self.driver.current_url}")
return "ERROR: Could not open Group/Health Connector search page"
else:
print(f"[DDMA step1] No Group/Health Connector link found in DOM. Current URL: {self.driver.current_url}")
return "ERROR: Could not open Group/Health Connector search page"
try:
wait.until(lambda d: "check-patient-benefits-search" in d.current_url)
except TimeoutException:
print(f"[DDMA step1] Click did not navigate to search page. Current URL: {self.driver.current_url}")
return "ERROR: Could not open Group/Health Connector search page"
print(f"[DDMA step1] On Group/Health Connector search page: {self.driver.current_url}")
# Fill Birthdate (masked mm/dd/yyyy input — send digits only, mask inserts slashes)
if self.dateOfBirth:
try:
dob_parts = self.dateOfBirth.split("-")
year = dob_parts[0]
month = dob_parts[1].zfill(2)
day = dob_parts[2].zfill(2)
birthdate_input = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//input[@data-test-id='InStateSearchBirthdate']")
))
birthdate_input.click()
birthdate_input.send_keys(Keys.CONTROL, "a")
birthdate_input.send_keys(Keys.BACKSPACE)
birthdate_input.send_keys(f"{month}{day}{year}")
print(f"[DDMA step1] Filled Birthdate: {month}/{day}/{year}")
except Exception as e:
print(f"[DDMA step1] Warning: Could not fill Birthdate: {e}")
# Fill Subscriber ID (== member ID)
if self.memberId:
try:
subscriber_input = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//input[@data-test-id='InStateSearchSubscriberId']")
))
subscriber_input.clear()
subscriber_input.send_keys(self.memberId)
print(f"[DDMA step1] Filled Subscriber ID: {self.memberId}")
except Exception as e:
print(f"[DDMA step1] Warning: Could not fill Subscriber ID: {e}")
time.sleep(0.3)
# Click Search button
search_btn = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//button[@data-test-id='InStateSearchButton']")
))
search_btn.click()
print("[DDMA step1] Clicked Search button")
# Wait for the Members results table to render
try:
WebDriverWait(self.driver, 15).until(
EC.presence_of_element_located((By.XPATH, "//table//tbody//tr"))
)
time.sleep(2)
except TimeoutException:
print("[DDMA step1] Warning: No results table appeared within timeout")
return "ERROR: INVALID SEARCH CRITERIA"
print("[DDMA step1] Search completed successfully")
return "Success"
except Exception as e:
print(f"[DDMA step1] Exception: {e}")
return f"ERROR:STEP1 - {e}"
def step2(self):
"""Navigate to patient detail page and generate PDF."""
wait = WebDriverWait(self.driver, 90)
try:
import re
# Wait for results table, then pause for full render
try:
WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.XPATH, "//tbody//tr"))
)
time.sleep(2) # Let the row content fully render after table appears
except TimeoutException:
print("[DDMA step2] Warning: Results table not found within timeout")
eligibilityText = "unknown"
foundMemberId = self.memberId or ""
patientName = ""
# The Members table lists the whole family sharing this subscriber ID —
# find the row matching this patient's DOB (falls back to last name, then first row).
rows = self.driver.find_elements(By.XPATH, "//table//tbody//tr")
print(f"[DDMA step2] Found {len(rows)} family member row(s)")
target_row = None
if self.dateOfBirth and rows:
try:
dob_parts = self.dateOfBirth.split("-")
dob_year, dob_month, dob_day = dob_parts[0], int(dob_parts[1]), int(dob_parts[2])
expected_dob = f"{calendar.month_abbr[dob_month]} {dob_day}, {dob_year}".lower()
for row in rows:
cells = row.find_elements(By.XPATH, "./td")
if len(cells) >= 3 and cells[2].text.strip().lower() == expected_dob:
target_row = row
break
except Exception as e:
print(f"[DDMA step2] DOB matching error (non-fatal): {e}")
if target_row is None and self.lastName and rows:
for row in rows:
if self.lastName.strip().lower() in row.text.lower():
target_row = row
break
if target_row is None and rows:
target_row = rows[0]
print("[DDMA step2] No DOB/name match — defaulting to first family member row")
if target_row is not None:
try:
cells = target_row.find_elements(By.XPATH, "./td")
if cells:
patientName = cells[0].text.strip()
print(f"[DDMA step2] Extracted patient name from row: '{patientName}'")
if len(cells) >= 2 and cells[1].text.strip():
foundMemberId = cells[1].text.strip()
if len(cells) >= 7:
status_text = cells[6].text.strip().lower()
if "active" in status_text:
eligibilityText = "active"
elif "inactive" in status_text or "termed" in status_text:
eligibilityText = "inactive"
else:
eligibilityText = status_text
print(f"[DDMA step2] Found eligibility status: {eligibilityText}")
except Exception as e:
print(f"[DDMA step2] Error reading matched row cells: {e}")
# Click into the member's detail page
try:
name_link = target_row.find_element(By.XPATH, ".//a")
name_link.click()
print(f"[DDMA step2] Clicked into detail page for: {patientName}")
except Exception as e:
print(f"[DDMA step2] Warning: Could not click patient name link: {e}")
else:
print("[DDMA step2] Warning: No family member rows found in search results")
# Wait for the detail page to load (URL moves away from the search page)
try:
WebDriverWait(self.driver, 15).until(
lambda d: "check-patient-benefits-search" not in d.current_url
)
print(f"[DDMA step2] Navigated to detail page: {self.driver.current_url}")
except Exception:
print(f"[DDMA step2] Warning: Still on search page: {self.driver.current_url}")
# Wait for meaningful content on the detail page
for selector in [
"//*[contains(text(),'Date of Birth') or contains(text(),'Address') or contains(text(),'Member ID')]",
"//table", "//h1", "//h2",
]:
try:
WebDriverWait(self.driver, 15).until(
EC.presence_of_element_located((By.XPATH, selector))
)
print(f"[DDMA step2] Detail page content loaded: {selector}")
break
except Exception:
continue
time.sleep(3) # Let JavaScript finish rendering all sections
# Fallback: refine patient name from the detail page text if row extraction failed
if not patientName:
try:
page_lines = self.driver.execute_script(
"return document.body.innerText;"
).split('\n')
page_lines = [l.strip() for l in page_lines if l.strip()]
except Exception as e:
print(f"[DDMA step2] Could not get page text: {e}")
page_lines = []
# UI noise words that appear in accessibility/sort-button text or field labels
ui_noise = ['click', 'sort', 'activate', 'direction', 'enter', 'space',
'current', 'use ', 'table', 'column', 'filter', 'search',
'date', 'birth', 'relationship', 'subscriber', 'coverage',
'status', 'period', 'network', 'plan', 'deductible']
def looks_like_name(text):
"""Return True if text is a plausible patient name."""
t = text.strip()
# Must be 260 chars, letters/spaces/hyphens/apostrophes only (no periods)
if not t or not (2 <= len(t) <= 60):
return False
if not re.match(r"^[A-Za-z\s\-']+$", t):
return False
# Must not be UI noise
if any(w in t.lower() for w in ui_noise):
return False
return True
# Scan lines for the "Name" label. When "Name" appears alone we scan
# forward past accessibility text to the first plausible name value.
for i, line in enumerate(page_lines):
# Case 1: "Name : Value" or "Name: Value" on the same line
same_line = re.match(
r'^(?:member\s+)?name\s*[:\-]\s*(.+)$', line, re.IGNORECASE
)
if same_line and not re.search(
r'(provider|group|subscriber|plan)\s+name', line, re.IGNORECASE
):
candidate = same_line.group(1).strip()
if looks_like_name(candidate):
patientName = candidate
print(f"[DDMA step2] Extracted name from 'Name:' label: '{patientName}'")
break
# Case 2: "Name" or "Name:" alone on a line — scan forward for value
elif re.match(r'^(?:member\s+)?name\s*:?\s*$', line, re.IGNORECASE):
for j in range(i + 1, min(i + 6, len(page_lines))):
candidate = page_lines[j].strip()
if looks_like_name(candidate):
patientName = candidate
print(f"[DDMA step2] Extracted name from 'Name' label (line +{j-i}): '{patientName}'")
break
if patientName:
break
# Clean patient name — strip any remaining date artifacts
if patientName:
cleaned = re.sub(r'\s*DOB[:\s]*\d{1,2}/\d{1,2}/\d{2,4}\s*', '', patientName, flags=re.IGNORECASE).strip()
cleaned = re.sub(r'\s+\d{1,2}/\d{1,2}/\d{2,4}$', '', cleaned).strip()
patientName = cleaned if cleaned else patientName
if not patientName:
print("[DDMA step2] Could not extract patient name from detail page")
# Wait for the page to be fully ready before capturing PDF
try:
WebDriverWait(self.driver, 30).until(
lambda d: d.execute_script("return document.readyState") == "complete"
)
except Exception:
pass
# Extra wait for lazy-loaded sections (benefits summary, member history, etc.)
time.sleep(4)
# Generate PDF via Chrome DevTools Protocol
print("[DDMA step2] Generating PDF of patient detail page...")
pdf_options = {
"landscape": False,
"displayHeaderFooter": False,
"printBackground": True,
"preferCSSPageSize": True,
"paperWidth": 8.5,
"paperHeight": 11,
"marginTop": 0.4,
"marginBottom": 0.4,
"marginLeft": 0.4,
"marginRight": 0.4,
"scale": 0.9,
}
result = self.driver.execute_cdp_cmd("Page.printToPDF", pdf_options)
pdf_data = base64.b64decode(result.get('data', ''))
pdf_id = foundMemberId or self.memberId or "unknown"
pdf_path = os.path.join(self.download_dir, f"eligibility_{pdf_id}.pdf")
with open(pdf_path, "wb") as f:
f.write(pdf_data)
print(f"[DDMA step2] PDF saved at: {pdf_path}")
# Close the browser window after PDF (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}")
print(f"[DDMA step2] Final — PatientName: '{patientName}', MemberID: '{foundMemberId}'")
return {
"status": "success",
"eligibility": eligibilityText,
"ss_path": pdf_path, # kept for backward compatibility
"pdf_path": pdf_path, # explicit pdf_path
"patientName": patientName,
"memberId": foundMemberId,
"dateOfBirth": self.dateOfBirth,
}
except Exception as e:
print("ERROR in step2:", e)
# Cleanup download dir on error
try:
dl = os.path.abspath(self.download_dir)
if os.path.isdir(dl):
for name in os.listdir(dl):
item = os.path.join(dl, name)
try:
if os.path.isfile(item) or os.path.islink(item):
os.remove(item)
except Exception as rm_err:
print(f"[cleanup] failed to remove {item}: {rm_err}")
except Exception as cleanup_exc:
print(f"[cleanup] unexpected error: {cleanup_exc}")
return {"status": "error", "message": str(e)}
def main_workflow(self, url):
try:
self.config_driver()
self.driver.maximize_window()
time.sleep(3)
login_result = self.login(url)
if login_result.startswith("ERROR"):
return {"status": "error", "message": login_result}
if login_result == "OTP_REQUIRED":
return {"status": "otp_required", "message": "OTP required after login"}
step1_result = self.step1()
if step1_result.startswith("ERROR"):
return {"status": "error", "message": step1_result}
step2_result = self.step2()
if step2_result.get("status") == "error":
return {"status": "error", "message": step2_result.get("message")}
return step2_result
except Exception as e:
return {"status": "error", "message": str(e)}
# NOTE: Do NOT quit driver — keep browser alive for next patient