GiteadandClaude Opus 5.5
0082b7bd53
Eligibility: rebuilt the Delta Indiv worker for my.deltadentalcoversme.com — Dentist tab -> Universal Delta Sign In -> username/Next -> password -> SIGN IN, confirmation-code (MFA) page detected by URL so the app's OTP popup shows, then Search for Patient -> Last Name + DOB -> Submit. On the "Benefits for" details page it reads the patient name, Subscriber # and Eligible through (active while that date is today or later) and saves the PDF downloaded by the Print link. Session persists: login is skipped while Search for Patient is reachable, and Delta Indiv cookies are no longer wiped on agent startup. The button needs Last Name + DOB (member ID optional; backend falls back to the scraped Subscriber #). Chatbot gains a Delta Indiv option and aliases. Claims: Delta Indiv uses the same 2024 ADA form as BCBS with the payer address Delta Dental, P.O. Box 103, Stevens Point, WI 54481-0103. Adds Delta Indiv Claim/PreAuth (ADA Form) buttons, Claim All and AI queue routing, Print ADA Form in the claim viewer, a Delta Indiv fee schedule (copied from BCBS), and Delta Indiv as its own insurance type so it is no longer treated as Deltains. Also renames eligibility buttons: Delta Indiv (Last name, DOB or ID), Metlife Dental (ID or SSN), Altus (ID, First name, DOB). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
639 lines
29 KiB
Python
639 lines
29 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 datetime import datetime, date
|
|
|
|
from deltaindiv_browser_manager import get_browser_manager
|
|
|
|
# Provider Dashboard "Search for Patient" button (Individual and Family plans)
|
|
SEARCH_FOR_PATIENT_XPATH = "//a[@data-test-id='SearchForPatientButton']"
|
|
SEARCH_DASHBOARD_URL = "https://my.deltadentalcoversme.com/private/provider/dashboard"
|
|
|
|
# deltadentalma.com/self-service/mfa page: "Confirmation Code" input + CONFIRM button
|
|
OTP_INPUT_XPATH = (
|
|
"//input[@data-test-id='MfaCode']"
|
|
" | //label[normalize-space()='Confirmation Code']/preceding::input[not(@type='hidden')][1]"
|
|
" | //label[normalize-space()='Confirmation Code']/following::input[not(@type='hidden')][1]"
|
|
" | //label[normalize-space()='Confirmation Code']/..//input[not(@type='hidden')]"
|
|
)
|
|
OTP_CONFIRM_XPATH = (
|
|
"//button[@data-test-id='ConfirmSubmitButton']"
|
|
" | //button[translate(normalize-space(), 'confirm', 'CONFIRM')='CONFIRM']"
|
|
)
|
|
|
|
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://my.deltadentalcoversme.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)
|
|
|
|
# Persistent session: if the dashboard still shows "Search for Patient",
|
|
# we're logged in — skip the sign-in / OTP flow entirely.
|
|
print(f"[login] Current URL: {self.driver.current_url}")
|
|
if self.driver.find_elements(By.XPATH, SEARCH_FOR_PATIENT_XPATH):
|
|
print("[login] Search for Patient already visible - ALREADY_LOGGED_IN")
|
|
return "ALREADY_LOGGED_IN"
|
|
try:
|
|
print(f"[login] Checking session via dashboard: {SEARCH_DASHBOARD_URL}")
|
|
self.driver.get(SEARCH_DASHBOARD_URL)
|
|
WebDriverWait(self.driver, 8).until(
|
|
EC.presence_of_element_located((By.XPATH, SEARCH_FOR_PATIENT_XPATH))
|
|
)
|
|
print("[login] Session still valid - ALREADY_LOGGED_IN")
|
|
return "ALREADY_LOGGED_IN"
|
|
except TimeoutException:
|
|
print(f"[login] Not logged in (landed on {self.driver.current_url}) - signing in")
|
|
|
|
# Navigate to login URL
|
|
self.driver.get(url)
|
|
time.sleep(2)
|
|
print(f"[login] URL after navigation: {self.driver.current_url}")
|
|
|
|
# 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, OTP_INPUT_XPATH)
|
|
)
|
|
)
|
|
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[@data-test-id='SearchForPatientButton' or contains(@href, 'check-patient-benefits-search')]"))
|
|
)
|
|
if member_search:
|
|
print("[login] Already authenticated after modal dismiss")
|
|
return "ALREADY_LOGGED_IN"
|
|
except TimeoutException:
|
|
pass
|
|
|
|
# coversme sign-in page may open on the "Member" tab — click "Dentist" first
|
|
# so the "Universal Delta Sign In" link appears
|
|
try:
|
|
dentist_tab = WebDriverWait(self.driver, 10).until(EC.element_to_be_clickable((
|
|
By.XPATH, "//span[contains(@class,'mainLabel') and normalize-space()='Dentist']"
|
|
)))
|
|
dentist_tab.click()
|
|
print("[login] Clicked Dentist tab")
|
|
time.sleep(1)
|
|
except TimeoutException:
|
|
print("[login] Dentist tab not found - continuing")
|
|
|
|
# coversme sign-in page: click "Universal Delta Sign In" to reach the DDPA login form
|
|
try:
|
|
universal_btn = WebDriverWait(self.driver, 10).until(
|
|
EC.element_to_be_clickable((By.XPATH, "//a[@data-test-id='DdpaLoginLink']"))
|
|
)
|
|
universal_btn.click()
|
|
print("[login] Clicked Universal Delta Sign In")
|
|
time.sleep(2)
|
|
except TimeoutException:
|
|
if "deltadentalcoversme.com/sign-in" in self.driver.current_url.lower():
|
|
# Still on coversme — don't fall through to the Member tab's username form
|
|
print("[login] Universal Delta Sign In link not found on coversme sign-in page")
|
|
return "ERROR:LOGIN FAILED: Universal Delta Sign In link not found"
|
|
print("[login] Universal Delta Sign In link not found - assuming already on login form")
|
|
|
|
# auth.deltadental.com "Have we met?" page: enter username, then click Next
|
|
username_xpath = (
|
|
"//input[@id='usernameInput' or @name='username' or @data-test-id='Username']"
|
|
)
|
|
try:
|
|
username_field = WebDriverWait(self.driver, 15).until(
|
|
EC.element_to_be_clickable((By.XPATH, username_xpath))
|
|
)
|
|
except TimeoutException:
|
|
print(f"[login] Could not find username field - page may have changed. URL: {self.driver.current_url}")
|
|
return "ERROR: Login form not found"
|
|
|
|
username_field.clear()
|
|
username_field.send_keys(self.deltaindiv_username)
|
|
print("[login] Entered username")
|
|
|
|
# Next is an <a id="verify-user">; aria-disabled flips to "false" once a username is typed
|
|
next_button = wait.until(EC.element_to_be_clickable((
|
|
By.XPATH, "//a[@id='verify-user' and @aria-disabled='false']"
|
|
)))
|
|
next_button.click()
|
|
print("[login] Clicked Next")
|
|
time.sleep(2)
|
|
|
|
# dp-prod-ddma auth0 page: username is prefilled, enter password, then SIGN IN
|
|
password_field = wait.until(EC.element_to_be_clickable((By.XPATH, "//input[@id='password']")))
|
|
password_field.clear()
|
|
password_field.send_keys(self.deltaindiv_password)
|
|
print("[login] Entered password")
|
|
|
|
try:
|
|
login_button = WebDriverWait(self.driver, 5).until(EC.element_to_be_clickable((
|
|
By.XPATH,
|
|
"//button[translate(normalize-space(), 'sign', 'SIGN')='SIGN IN' or @type='submit']"
|
|
)))
|
|
login_button.click()
|
|
print("[login] Clicked SIGN IN")
|
|
except TimeoutException:
|
|
password_field.send_keys(Keys.RETURN)
|
|
print("[login] SIGN IN button not found - submitted with Enter")
|
|
|
|
# Save credentials hash after login attempt
|
|
if self.deltaindiv_username:
|
|
browser_manager.save_credentials_hash(self.deltaindiv_username)
|
|
|
|
# After SIGN IN wait (up to 30s) for whichever page comes next:
|
|
# confirmation-code (MFA) page or the dashboard.
|
|
def next_page(d):
|
|
url = d.current_url.lower()
|
|
if "self-service/mfa" in url or d.find_elements(By.XPATH, OTP_INPUT_XPATH):
|
|
return "otp"
|
|
if d.find_elements(By.XPATH, SEARCH_FOR_PATIENT_XPATH):
|
|
return "dashboard"
|
|
return False
|
|
|
|
try:
|
|
landed = WebDriverWait(self.driver, 30).until(next_page)
|
|
except TimeoutException:
|
|
landed = None
|
|
print(f"[login] After SIGN IN: {landed or 'nothing recognised'} — URL: {self.driver.current_url}")
|
|
|
|
if landed == "otp":
|
|
print("[login] Confirmation code page detected -> OTP_REQUIRED")
|
|
return "OTP_REQUIRED"
|
|
|
|
if landed == "dashboard":
|
|
print("[login] Login successful - Search for Patient visible")
|
|
return "SUCCESS"
|
|
|
|
# Only a visible, non-empty error message counts as a failed login
|
|
for el in self.driver.find_elements(
|
|
By.XPATH,
|
|
"//*[contains(@class,'error') or contains(@class,'alert-danger') or @role='alert']"
|
|
):
|
|
try:
|
|
text = el.text.strip()
|
|
if text and el.is_displayed():
|
|
print(f"[login] Login failed - error detected: {text}")
|
|
return f"ERROR:LOGIN FAILED: {text}"
|
|
except Exception:
|
|
continue
|
|
|
|
if "sign-in" in self.driver.current_url.lower() or self.driver.find_elements(
|
|
By.XPATH, "//input[@id='password']"
|
|
):
|
|
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):
|
|
"""Click Search for Patient on the dashboard and search by Last Name + Date of Birth."""
|
|
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)}")
|
|
|
|
# Provider Dashboard -> "Search for Patient" (Individual and Family plans).
|
|
# Look for the button wherever login landed; only navigate if it isn't there.
|
|
search_link = None
|
|
for nav_url in (None, SEARCH_DASHBOARD_URL, "https://my.deltadentalcoversme.com/"):
|
|
if nav_url:
|
|
print(f"[Delta Indiv step1] Search for Patient not found — navigating to {nav_url}")
|
|
self.driver.get(nav_url)
|
|
time.sleep(2)
|
|
try:
|
|
search_link = WebDriverWait(self.driver, 5 if nav_url is None else 15).until(
|
|
EC.presence_of_element_located((By.XPATH, SEARCH_FOR_PATIENT_XPATH))
|
|
)
|
|
break
|
|
except TimeoutException:
|
|
print(f"[Delta Indiv step1] No Search for Patient button at: {self.driver.current_url}")
|
|
|
|
if search_link is None:
|
|
if "check-patient-benefits-search" not in self.driver.current_url.lower():
|
|
return "ERROR: Could not open patient search page"
|
|
else:
|
|
print(f"[Delta Indiv step1] Found Search for Patient on: {self.driver.current_url}")
|
|
href = search_link.get_attribute("href")
|
|
try:
|
|
self.driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", search_link)
|
|
search_link.click()
|
|
except Exception as click_err:
|
|
print(f"[Delta Indiv step1] Normal click failed ({click_err}) — using JS click")
|
|
self.driver.execute_script("arguments[0].click();", search_link)
|
|
print("[Delta Indiv step1] Clicked Search for Patient")
|
|
|
|
try:
|
|
WebDriverWait(self.driver, 15).until(
|
|
lambda d: "check-patient-benefits-search" in d.current_url.lower()
|
|
)
|
|
except TimeoutException:
|
|
if not href:
|
|
print(f"[Delta Indiv step1] Click did not navigate. Current URL: {self.driver.current_url}")
|
|
return "ERROR: Could not open patient search page"
|
|
print(f"[Delta Indiv step1] Click did not navigate — opening link directly: {href}")
|
|
self.driver.get(href)
|
|
try:
|
|
wait.until(lambda d: "check-patient-benefits-search" in d.current_url.lower())
|
|
except TimeoutException:
|
|
print(f"[Delta Indiv step1] Could not reach search page. Current URL: {self.driver.current_url}")
|
|
return "ERROR: Could not open patient search page"
|
|
|
|
print(f"[Delta Indiv step1] On patient search page: {self.driver.current_url}")
|
|
|
|
# "Search by Patient Details" tab (default): Last Name + Date of Birth, then Submit
|
|
if not self.lastName or not self.dateOfBirth:
|
|
return "ERROR: Last name and date of birth are required for Delta Indiv search"
|
|
|
|
last_name_input = wait.until(EC.element_to_be_clickable(
|
|
(By.XPATH, "//input[@data-test-id='LastName']")
|
|
))
|
|
last_name_input.clear()
|
|
last_name_input.send_keys(self.lastName)
|
|
print(f"[Delta Indiv step1] Filled Last Name: {self.lastName}")
|
|
|
|
# Masked mm/dd/yyyy input — send digits only, mask inserts slashes
|
|
dob_parts = self.dateOfBirth.split("-")
|
|
year = dob_parts[0]
|
|
month = dob_parts[1].zfill(2)
|
|
day = dob_parts[2].zfill(2)
|
|
|
|
dob_input = wait.until(EC.element_to_be_clickable(
|
|
(By.XPATH, "//input[@data-test-id='DateOfBirth']")
|
|
))
|
|
dob_input.click()
|
|
dob_input.send_keys(Keys.CONTROL, "a")
|
|
dob_input.send_keys(Keys.BACKSPACE)
|
|
dob_input.send_keys(f"{month}{day}{year}")
|
|
print(f"[Delta Indiv step1] Filled Date of Birth: {month}/{day}/{year}")
|
|
|
|
time.sleep(0.3)
|
|
|
|
submit_btn = wait.until(EC.element_to_be_clickable(
|
|
(By.XPATH, "//button[@data-test-id='SearchByPatientButton']")
|
|
))
|
|
submit_btn.click()
|
|
print("[Delta Indiv step1] Clicked Submit")
|
|
|
|
# Submit normally lands straight on the "Benefits for <NAME>" details page;
|
|
# a results list (multiple matches) is handled in step2.
|
|
try:
|
|
WebDriverWait(self.driver, 20).until(
|
|
lambda d: self._on_detail_page()
|
|
or d.find_elements(By.XPATH, "//table//tbody//tr//a")
|
|
)
|
|
time.sleep(2)
|
|
except TimeoutException:
|
|
print("[Delta Indiv step1] Warning: No patient details or results appeared within timeout")
|
|
return "ERROR: INVALID SEARCH CRITERIA"
|
|
|
|
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 _on_detail_page(self):
|
|
"""True when the "Benefits for <NAME>" patient details page is showing."""
|
|
try:
|
|
return bool(self.driver.find_elements(
|
|
By.XPATH,
|
|
"//h1[starts-with(normalize-space(), 'Benefits for')]"
|
|
" | //*[normalize-space()='Subscriber #:']"
|
|
))
|
|
except Exception:
|
|
return False
|
|
|
|
def _open_patient_from_results(self):
|
|
"""If the search returned a list instead of the details page, click the matching patient."""
|
|
rows = self.driver.find_elements(By.XPATH, "//table//tbody//tr[.//a]")
|
|
print(f"[Delta Indiv step2] Search returned {len(rows)} result row(s)")
|
|
if not rows:
|
|
return False
|
|
|
|
target_row = None
|
|
if self.dateOfBirth:
|
|
try:
|
|
y, m, d = self.dateOfBirth.split("-")
|
|
dob_variants = {
|
|
f"{calendar.month_abbr[int(m)]} {int(d)}, {y}".lower(),
|
|
f"{m.zfill(2)}/{d.zfill(2)}/{y}",
|
|
}
|
|
target_row = next(
|
|
(r for r in rows if any(v in r.text.lower() for v in dob_variants)), None
|
|
)
|
|
except Exception as e:
|
|
print(f"[Delta Indiv step2] DOB matching error (non-fatal): {e}")
|
|
if target_row is None and self.firstName:
|
|
target_row = next(
|
|
(r for r in rows if self.firstName.strip().lower() in r.text.lower()), None
|
|
)
|
|
if target_row is None:
|
|
target_row = rows[0]
|
|
print("[Delta Indiv step2] No DOB/name match — using first result row")
|
|
|
|
target_row.find_element(By.XPATH, ".//a").click()
|
|
WebDriverWait(self.driver, 20).until(lambda d: self._on_detail_page())
|
|
return True
|
|
|
|
def _write_full_page_pdf(self, pdf_path):
|
|
"""Print the current window as ONE long PDF page (no page breaks)."""
|
|
dims = self.driver.execute_script(
|
|
"return {w: Math.max(document.documentElement.scrollWidth, document.body.scrollWidth),"
|
|
" h: Math.max(document.documentElement.scrollHeight, document.body.scrollHeight)};"
|
|
)
|
|
# 8.5in wide; shrink to fit the page width, then make the paper as tall as the content
|
|
margin_in = 0.3
|
|
page_width_in = 8.5
|
|
scale = max(0.1, min(1.0, (page_width_in - 2 * margin_in) / (dims["w"] / 96)))
|
|
page_height_in = dims["h"] / 96 * scale + 2 * margin_in + 0.2
|
|
result = self.driver.execute_cdp_cmd("Page.printToPDF", {
|
|
"landscape": False,
|
|
"displayHeaderFooter": False,
|
|
"printBackground": True,
|
|
"preferCSSPageSize": False,
|
|
"paperWidth": page_width_in,
|
|
"paperHeight": page_height_in,
|
|
"marginTop": margin_in,
|
|
"marginBottom": margin_in,
|
|
"marginLeft": margin_in,
|
|
"marginRight": margin_in,
|
|
"scale": scale,
|
|
})
|
|
with open(pdf_path, "wb") as f:
|
|
f.write(base64.b64decode(result.get("data", "")))
|
|
|
|
def _save_print_pdf(self, pdf_path):
|
|
"""
|
|
Click the details page's Print link — it downloads a PDF of the page — and move
|
|
that download to pdf_path. Falls back to printing the page itself if no PDF arrives.
|
|
"""
|
|
files_before = set(os.listdir(self.download_dir))
|
|
|
|
try:
|
|
print_link = WebDriverWait(self.driver, 10).until(
|
|
EC.presence_of_element_located((By.XPATH, "//a[@data-test-id='PrintCpdIcon']"))
|
|
)
|
|
self.driver.execute_script("arguments[0].click();", print_link)
|
|
print("[Delta Indiv step2] Clicked Print")
|
|
|
|
def downloaded_pdf(_):
|
|
new = set(os.listdir(self.download_dir)) - files_before
|
|
# Chrome writes .crdownload until the download finishes
|
|
if any(f.endswith(".crdownload") for f in new):
|
|
return False
|
|
pdfs = [f for f in new if f.lower().endswith(".pdf")]
|
|
return pdfs[0] if pdfs else False
|
|
|
|
name = WebDriverWait(self.driver, 60).until(downloaded_pdf)
|
|
os.replace(os.path.join(self.download_dir, name), pdf_path)
|
|
print(f"[Delta Indiv step2] Print downloaded PDF: {name}")
|
|
except Exception as e:
|
|
print(f"[Delta Indiv step2] Print download failed ({e}) — printing the page itself")
|
|
self._write_full_page_pdf(pdf_path)
|
|
|
|
print(f"[Delta Indiv step2] PDF saved at: {pdf_path}")
|
|
|
|
def step2(self):
|
|
"""On the patient details page: extract Subscriber #, name and eligibility, then save a PDF."""
|
|
try:
|
|
if not self._on_detail_page():
|
|
if not self._open_patient_from_results():
|
|
return {"status": "error", "message": "Patient details page not found"}
|
|
|
|
print(f"[Delta Indiv step2] On patient details page: {self.driver.current_url}")
|
|
|
|
try:
|
|
WebDriverWait(self.driver, 30).until(
|
|
lambda d: d.execute_script("return document.readyState") == "complete"
|
|
)
|
|
except Exception:
|
|
pass
|
|
time.sleep(3) # Let the benefit maximum sections finish rendering
|
|
|
|
page_text = self.driver.execute_script("return document.body.innerText;") or ""
|
|
|
|
def label_value(pattern):
|
|
m = re.search(pattern, page_text, re.IGNORECASE)
|
|
return m.group(1).strip() if m else ""
|
|
|
|
# "Benefits for KAOUTER MOUSSAID" heading names the patient
|
|
# ("Subscriber name:" is the policy holder, which may be a parent)
|
|
patientName = label_value(r"(?m)^\s*Benefits\s+for\s+(.+?)\s*$")
|
|
if not patientName:
|
|
patientName = f"{self.firstName} {self.lastName}".strip()
|
|
firstName = patientName.split()[0] if patientName else self.firstName
|
|
|
|
foundMemberId = label_value(r"Subscriber\s*#\s*:\s*([A-Za-z0-9\-]+)") or self.memberId or ""
|
|
coverageIs = label_value(r"Coverage\s+is\s*:\s*([A-Za-z]+)")
|
|
eligibleThrough = label_value(
|
|
r"Eligible\s+through\s*:\s*([A-Za-z]{3,9}\.?\s+\d{1,2},\s*\d{4})"
|
|
)
|
|
|
|
# Eligibility: active while "Eligible through" is today or later;
|
|
# fall back to "Coverage is:" when the date can't be read.
|
|
eligibilityText = "unknown"
|
|
through_date = None
|
|
for fmt in ("%b %d, %Y", "%B %d, %Y"):
|
|
try:
|
|
through_date = datetime.strptime(eligibleThrough.replace(".", ""), fmt).date()
|
|
break
|
|
except ValueError:
|
|
continue
|
|
if through_date:
|
|
eligibilityText = "active" if through_date >= date.today() else "inactive"
|
|
elif coverageIs:
|
|
c = coverageIs.lower()
|
|
eligibilityText = "inactive" if ("inactive" in c or "termed" in c) else ("active" if "active" in c else c)
|
|
|
|
print(
|
|
f"[Delta Indiv step2] Extracted — Name: '{patientName}', First: '{firstName}', "
|
|
f"Subscriber #: '{foundMemberId}', Coverage is: '{coverageIs}', "
|
|
f"Eligible through: '{eligibleThrough}' -> {eligibilityText}"
|
|
)
|
|
|
|
pdf_id = foundMemberId or "unknown"
|
|
pdf_path = os.path.join(self.download_dir, f"eligibility_{pdf_id}.pdf")
|
|
self._save_print_pdf(pdf_path)
|
|
|
|
return {
|
|
"status": "success",
|
|
"eligibility": eligibilityText,
|
|
"eligibleThrough": eligibleThrough,
|
|
"coverageIs": coverageIs,
|
|
"ss_path": pdf_path, # kept for backward compatibility
|
|
"pdf_path": pdf_path,
|
|
"patientName": patientName,
|
|
"firstName": firstName,
|
|
"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
|