GiteadandClaude Sonnet 5
f8235787db
DDMA now often skips the family-members results table and navigates
straight to the single-patient detail page, and labels the patient's
name "Subscriber name:" instead of "Name:". Update step1/step2 to
detect the detail-page URL correctly (it's nested under the search
path, so a plain substring check couldn't tell them apart), extract
the name from the new label, and fall back to parsing "Coverage is:"
for eligibility status when there's no results row to read it from.
Also fix a status-text substring bug ("active" matching inside
"inactive") and stop the backend from collapsing an unknown
eligibility result into INACTIVE. Skip name extraction entirely when
the caller already knows the patient (existing-patient recheck).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
701 lines
34 KiB
Python
Executable File
701 lines
34 KiB
Python
Executable File
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 either the Members results table to render, or (when the
|
||
# Subscriber ID + DOB match exactly one family member) DDMA to skip
|
||
# the results list and navigate straight to the patient detail page.
|
||
# Note: the detail page URL is nested UNDER the search page path
|
||
# (".../check-patient-benefits-search/check-patient-benefit-details"),
|
||
# so "check-patient-benefits-search" alone can't distinguish them —
|
||
# only the detail page's own "-benefit-details" segment can.
|
||
try:
|
||
WebDriverWait(self.driver, 15).until(
|
||
lambda d: d.find_elements(By.XPATH, "//table//tbody//tr")
|
||
or "check-patient-benefit-details" in d.current_url.lower()
|
||
)
|
||
time.sleep(2)
|
||
except TimeoutException:
|
||
print("[DDMA step1] Warning: No results table appeared within timeout")
|
||
return "ERROR: INVALID SEARCH CRITERIA"
|
||
|
||
if "check-patient-benefit-details" in self.driver.current_url.lower():
|
||
print(f"[DDMA step1] Search navigated directly to detail page: {self.driver.current_url}")
|
||
|
||
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
|
||
|
||
eligibilityText = "unknown"
|
||
foundMemberId = self.memberId or ""
|
||
# If the caller already knows this patient (e.g. an existing patient record
|
||
# being rechecked), use that name directly — no need to scrape/parse it off
|
||
# the DDMA page, only the eligibility status is actually needed.
|
||
patientName = f"{self.firstName} {self.lastName}".strip() if self.firstName and self.lastName else ""
|
||
known_patient_name = bool(patientName)
|
||
|
||
# DDMA now sometimes skips the family-members results table entirely and
|
||
# navigates straight to the single-patient detail page (when Subscriber ID +
|
||
# DOB match exactly one person). In that case there's no members table to
|
||
# parse here — skip straight to the detail-page fallback extraction below.
|
||
already_on_detail_page = "check-patient-benefit-details" in self.driver.current_url.lower()
|
||
|
||
rows = []
|
||
target_row = None
|
||
|
||
if not already_on_detail_page:
|
||
# 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")
|
||
|
||
# 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)")
|
||
else:
|
||
print(f"[DDMA step2] Already on detail page: {self.driver.current_url}")
|
||
|
||
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 and not known_patient_name:
|
||
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 "inactive" in status_text or "termed" in status_text:
|
||
eligibilityText = "inactive"
|
||
elif "active" in status_text:
|
||
eligibilityText = "active"
|
||
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 gains the "-benefit-details" segment)
|
||
try:
|
||
WebDriverWait(self.driver, 15).until(
|
||
lambda d: "check-patient-benefit-details" in d.current_url.lower()
|
||
)
|
||
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 2–60 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
|
||
# (DDMA's detail page labels this "Subscriber name:", so that
|
||
# prefix must be accepted rather than excluded)
|
||
same_line = re.match(
|
||
r'^(?:member\s+|subscriber\s+)?name\s*[:\-]\s*(.+)$', line, re.IGNORECASE
|
||
)
|
||
if same_line and not re.search(
|
||
r'(provider|group|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+|subscriber\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")
|
||
|
||
# Fallback: refine eligibility status from the detail page's "Coverage is:"
|
||
# label if the family-table row extraction didn't set it (e.g. DDMA
|
||
# navigated straight to the detail page, skipping the members table).
|
||
if eligibilityText == "unknown":
|
||
try:
|
||
page_text = self.driver.execute_script("return document.body.innerText;")
|
||
coverage_match = re.search(
|
||
r'coverage\s+is\s*[:\-]?\s*([A-Za-z]+)', page_text, re.IGNORECASE
|
||
)
|
||
if coverage_match:
|
||
status_text = coverage_match.group(1).strip().lower()
|
||
if "inactive" in status_text or "termed" in status_text:
|
||
eligibilityText = "inactive"
|
||
elif "active" in status_text:
|
||
eligibilityText = "active"
|
||
else:
|
||
eligibilityText = status_text
|
||
print(f"[DDMA step2] Extracted eligibility from 'Coverage is:' label: '{eligibilityText}'")
|
||
except Exception as e:
|
||
print(f"[DDMA step2] Could not extract eligibility from 'Coverage is:' label: {e}")
|
||
|
||
# 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
|