GiteadandClaude Sonnet 5
d357edd4a6
CCA step1 was treating the harmless "password set to expire" banner as a fatal error, closing the browser before the eligibility popup/report could be captured. Now it short-circuits to success once the patient popup renders and ignores password-expiration notices. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
705 lines
35 KiB
Python
705 lines
35 KiB
Python
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 deltaindiv_browser_manager import get_browser_manager
|
||
|
||
class AutomationDeltaIndivEligibilityCheck:
|
||
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.deltaindiv_username = self.data.get("deltaindivUsername", "")
|
||
self.deltaindiv_password = self.data.get("deltaindivPassword", "")
|
||
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("[Delta Indiv 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("[Delta Indiv login] Clicked logout button")
|
||
time.sleep(2)
|
||
break
|
||
except TimeoutException:
|
||
continue
|
||
except Exception as e:
|
||
print(f"[Delta Indiv login] Could not click logout button: {e}")
|
||
|
||
# Clear cookies as backup
|
||
try:
|
||
self.driver.delete_all_cookies()
|
||
print("[Delta Indiv login] Cleared all cookies")
|
||
except Exception as e:
|
||
print(f"[Delta Indiv login] Error clearing cookies: {e}")
|
||
|
||
browser_manager.clear_credentials_hash()
|
||
print("[Delta Indiv login] Logout complete")
|
||
return True
|
||
except Exception as e:
|
||
print(f"[Delta Indiv 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.deltaindiv_username and browser_manager.credentials_changed(self.deltaindiv_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.deltaindiv_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.deltaindiv_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.deltaindiv_username:
|
||
browser_manager.save_credentials_hash(self.deltaindiv_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"[Delta Indiv 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"[Delta Indiv 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"[Delta Indiv 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"[Delta Indiv 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"[Delta Indiv 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"[Delta Indiv step1] JS click failed: {js_err}")
|
||
print(f"[Delta Indiv step1] Current URL: {self.driver.current_url}")
|
||
return "ERROR: Could not open Group/Health Connector search page"
|
||
else:
|
||
print(f"[Delta Indiv 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"[Delta Indiv 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"[Delta Indiv 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"[Delta Indiv step1] Filled Birthdate: {month}/{day}/{year}")
|
||
except Exception as e:
|
||
print(f"[Delta Indiv 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"[Delta Indiv step1] Filled Subscriber ID: {self.memberId}")
|
||
except Exception as e:
|
||
print(f"[Delta Indiv 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("[Delta Indiv step1] Clicked Search button")
|
||
|
||
# Wait for either the Members results table to render, or (when the
|
||
# Subscriber ID + DOB match exactly one family member) Delta Indiv 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("[Delta Indiv 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"[Delta Indiv step1] Search navigated directly to detail page: {self.driver.current_url}")
|
||
|
||
print("[Delta Indiv step1] Search completed successfully")
|
||
return "Success"
|
||
|
||
except Exception as e:
|
||
print(f"[Delta Indiv 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 Delta Indiv 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)
|
||
|
||
# Delta Indiv 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("[Delta Indiv 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"[Delta Indiv step2] Found {len(rows)} family member row(s)")
|
||
else:
|
||
print(f"[Delta Indiv 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"[Delta Indiv 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("[Delta Indiv 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"[Delta Indiv 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"[Delta Indiv step2] Found eligibility status: {eligibilityText}")
|
||
except Exception as e:
|
||
print(f"[Delta Indiv 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"[Delta Indiv step2] Clicked into detail page for: {patientName}")
|
||
except Exception as e:
|
||
print(f"[Delta Indiv step2] Warning: Could not click patient name link: {e}")
|
||
else:
|
||
print("[Delta Indiv 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"[Delta Indiv step2] Navigated to detail page: {self.driver.current_url}")
|
||
except Exception:
|
||
print(f"[Delta Indiv 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"[Delta Indiv 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"[Delta Indiv 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
|
||
|
||
# Preferred: the "Benefits for JOSHUA ZHANG" heading names the actual
|
||
# patient. "Subscriber name:" is the policy holder (e.g. a parent), so
|
||
# it is only used below if this heading is missing.
|
||
for line in page_lines:
|
||
heading = re.match(r'^benefits\s+for\s+(.+)$', line, re.IGNORECASE)
|
||
if heading and looks_like_name(heading.group(1)):
|
||
patientName = heading.group(1).strip()
|
||
print(f"[Delta Indiv step2] Extracted name from 'Benefits for' heading: '{patientName}'")
|
||
break
|
||
|
||
# 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):
|
||
if patientName:
|
||
break
|
||
# Case 1: "Name : Value" or "Name: Value" on the same line
|
||
# (Delta Indiv'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"[Delta Indiv 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"[Delta Indiv 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("[Delta Indiv 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. Delta Indiv
|
||
# 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"[Delta Indiv step2] Extracted eligibility from 'Coverage is:' label: '{eligibilityText}'")
|
||
except Exception as e:
|
||
print(f"[Delta Indiv 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("[Delta Indiv 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"[Delta Indiv step2] PDF saved at: {pdf_path}")
|
||
|
||
print(f"[Delta Indiv 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
|