feat: rename DDMA to ClaimMD across selenium service, remove MH/CMSP eligibility feature
Rename ddma_browser_manager/helpers/worker files to claimmd_* naming in SeleniumService. Remove the MassHealth (MH) and CMSP eligibility-check feature end to end: buttons and dead placeholder buttons on the insurance eligibility page, their handlers and auto-trigger wiring, the chatbot's MassHealth-only manual paste flow, backend routes/queue job types/processors, and the corresponding Python Selenium worker endpoints. Multi-provider eligibility checking (DDMA, Delta, Tufts SCO, United SCO, CCA, BCBS MA) and the Twilio SMS eligibility flow are untouched. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Summit Dental Care
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
00fd213291
commit
a3d2960c98
@@ -6,15 +6,13 @@ import uvicorn
|
||||
import asyncio
|
||||
from selenium_claimSubmitWorker import AutomationMassHealthClaimsLogin
|
||||
from selenium_eligibilityCheckWorker import AutomationMassHealthEligibilityCheck
|
||||
from selenium_MH_eligibilityHistoryCheckWorker import AutomationMassHealthEligibilityHistoryCheck
|
||||
from selenium_CMSP_eligibilityHistoryRemainingCheckWorker import AutomationCMSPEligibilityHistoryRemainingCheck
|
||||
from selenium_claimStatusCheckWorker import AutomationMassHealthClaimStatusCheck
|
||||
from selenium_preAuthWorker import AutomationMassHealthPreAuth
|
||||
from selenium_MHSinglePaymentCheckWorker import AutomationMassHealthSinglePaymentCheck
|
||||
from selenium_MHBatchPaymentCheckWorker import AutomationMassHealthBatchPaymentCheck
|
||||
import os
|
||||
import time
|
||||
import helpers_ddma_eligibility as hddma
|
||||
import helpers_claimmd_eligibility as hddma
|
||||
import helpers_deltains_eligibility as hdeltains
|
||||
import helpers_unitedsco_eligibility as hunitedsco
|
||||
import helpers_dentaquest_eligibility as hdentaquest
|
||||
@@ -29,7 +27,7 @@ import helpers_tuftssco_preauth as htuftssco_preauth
|
||||
import helpers_bcbs_ma_eligibility as hbcbs_ma
|
||||
|
||||
# Import startup session-clear functions
|
||||
from ddma_browser_manager import clear_ddma_session_on_startup
|
||||
from claimmd_browser_manager import clear_ddma_session_on_startup
|
||||
from deltains_browser_manager import clear_deltains_session_on_startup
|
||||
from unitedsco_browser_manager import clear_unitedsco_session_on_startup
|
||||
from dentaquest_browser_manager import clear_dentaquest_session_on_startup
|
||||
@@ -146,83 +144,6 @@ async def start_workflow(request: Request):
|
||||
async with lock:
|
||||
active_jobs -= 1
|
||||
|
||||
# Endpoint: 2a — MH Eligibility + Service History
|
||||
@app.post("/mh-eligibility-history-check")
|
||||
async def mh_eligibility_history_check(request: Request):
|
||||
global active_jobs, waiting_jobs
|
||||
data = await request.json()
|
||||
|
||||
async with lock:
|
||||
waiting_jobs += 1
|
||||
|
||||
async with semaphore:
|
||||
async with lock:
|
||||
waiting_jobs -= 1
|
||||
active_jobs += 1
|
||||
try:
|
||||
bot = AutomationMassHealthEligibilityHistoryCheck(data)
|
||||
result = bot.main_workflow("https://provider.masshealth-dental.org/mh_provider_login")
|
||||
|
||||
if result.get("status") == "error":
|
||||
return {"status": "error", "message": result.get("message")}
|
||||
|
||||
port = os.getenv("PORT", "5002")
|
||||
url_host = os.getenv("HOST", "localhost")
|
||||
|
||||
if result.get("pdf_path"):
|
||||
filename = os.path.basename(result["pdf_path"])
|
||||
result["pdf_url"] = f"http://{url_host}:{port}/downloads/{filename}"
|
||||
|
||||
if result.get("history_pdf_path"):
|
||||
filename = os.path.basename(result["history_pdf_path"])
|
||||
result["history_pdf_url"] = f"http://{url_host}:{port}/downloads/{filename}"
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
finally:
|
||||
async with lock:
|
||||
active_jobs -= 1
|
||||
|
||||
# Endpoint: 2b — CMSP Eligibility + Service History + Accumulator (Remaining)
|
||||
@app.post("/cmsp-eligibility-history-remaining-check")
|
||||
async def cmsp_eligibility_history_remaining_check(request: Request):
|
||||
global active_jobs, waiting_jobs
|
||||
data = await request.json()
|
||||
|
||||
async with lock:
|
||||
waiting_jobs += 1
|
||||
|
||||
async with semaphore:
|
||||
async with lock:
|
||||
waiting_jobs -= 1
|
||||
active_jobs += 1
|
||||
try:
|
||||
bot = AutomationCMSPEligibilityHistoryRemainingCheck(data)
|
||||
result = bot.main_workflow("https://provider.masshealth-dental.org/mh_provider_login")
|
||||
|
||||
if result.get("status") == "error":
|
||||
return {"status": "error", "message": result.get("message")}
|
||||
|
||||
port = os.getenv("PORT", "5002")
|
||||
url_host = os.getenv("HOST", "localhost")
|
||||
|
||||
for key, url_key in [
|
||||
("pdf_path", "pdf_url"),
|
||||
("history_pdf_path", "history_pdf_url"),
|
||||
("accumulator_pdf_path", "accumulator_pdf_url"),
|
||||
]:
|
||||
if result.get(key):
|
||||
filename = os.path.basename(result[key])
|
||||
result[url_key] = f"http://{url_host}:{port}/downloads/{filename}"
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
finally:
|
||||
async with lock:
|
||||
active_jobs -= 1
|
||||
|
||||
# Endpoint: 2.1 — Start the automation for Claims login (open browser and log in)
|
||||
@app.post("/claims-login")
|
||||
async def start_claims_login(request: Request):
|
||||
@@ -410,7 +331,7 @@ async def ddma_eligibility(request: Request):
|
||||
waiting_jobs += 1
|
||||
|
||||
# run in background (queued under semaphore)
|
||||
asyncio.create_task(_ddma_worker_wrapper(sid, data, url="https://providers.deltadentalma.com/onboarding/start/"))
|
||||
asyncio.create_task(_ddma_worker_wrapper(sid, data, url="https://www.deltadentalma.com/sign-in"))
|
||||
|
||||
return {"status": "started", "session_id": sid}
|
||||
|
||||
|
||||
+9
-13
@@ -7,7 +7,7 @@ from selenium.webdriver.support.ui import WebDriverWait
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
from selenium.common.exceptions import WebDriverException, TimeoutException
|
||||
|
||||
from selenium_DDMA_eligibilityCheckWorker import AutomationDeltaDentalMAEligibilityCheck
|
||||
from selenium_ClaimMD_eligibilityCheckWorker import AutomationDeltaDentalMAEligibilityCheck
|
||||
|
||||
# In-memory session store
|
||||
sessions: Dict[str, Dict[str, Any]] = {}
|
||||
@@ -146,13 +146,11 @@ async def start_ddma_run(sid: str, data: dict, url: str):
|
||||
if otp_value:
|
||||
print(f"[OTP poll {poll+1}] OTP received from app, typing it in...")
|
||||
try:
|
||||
otp_input = driver.find_element(By.XPATH,
|
||||
"//input[contains(@aria-label,'Verification') or contains(@placeholder,'verification') or @type='tel' or contains(@aria-lable,'Verification code') or contains(@placeholder,'Enter your verification code')]"
|
||||
)
|
||||
otp_input = driver.find_element(By.XPATH, "//input[@data-test-id='MfaCode']")
|
||||
otp_input.clear()
|
||||
otp_input.send_keys(otp_value)
|
||||
try:
|
||||
verify_btn = driver.find_element(By.XPATH, "//button[@type='button' and @aria-label='Verify']")
|
||||
verify_btn = driver.find_element(By.XPATH, "//button[@data-test-id='ConfirmSubmitButton']")
|
||||
verify_btn.click()
|
||||
except Exception:
|
||||
otp_input.send_keys("\n")
|
||||
@@ -169,7 +167,7 @@ async def start_ddma_run(sid: str, data: dict, url: str):
|
||||
if "member" in current_url or "dashboard" in current_url or "eligibility" in current_url:
|
||||
try:
|
||||
member_search = WebDriverWait(driver, 5).until(
|
||||
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]'))
|
||||
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
|
||||
)
|
||||
print("[OTP] Member search found — login successful!")
|
||||
login_success = True
|
||||
@@ -179,16 +177,14 @@ async def start_ddma_run(sid: str, data: dict, url: str):
|
||||
|
||||
# Check if OTP input is still visible (user hasn't finished)
|
||||
try:
|
||||
otp_input_elem = driver.find_element(By.XPATH,
|
||||
"//input[contains(@aria-label,'Verification') or contains(@placeholder,'verification') or @type='tel']"
|
||||
)
|
||||
otp_input_elem = driver.find_element(By.XPATH, "//input[@data-test-id='MfaCode']")
|
||||
print(f"[OTP poll {poll+1}] OTP input still visible - waiting...")
|
||||
except Exception:
|
||||
# OTP input gone — may mean login is completing; try members page
|
||||
if "onboarding" in current_url or "start" in current_url:
|
||||
if "sign-in" in current_url:
|
||||
print("[OTP] OTP input gone, trying to navigate to members page...")
|
||||
try:
|
||||
driver.get("https://providers.deltadentalma.com/members")
|
||||
driver.get("https://www.deltadentalma.com/private/provider/dashboard")
|
||||
await asyncio.sleep(2)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -200,10 +196,10 @@ async def start_ddma_run(sid: str, data: dict, url: str):
|
||||
# Final check — navigate directly to members page
|
||||
try:
|
||||
print("[OTP] Final attempt - navigating to members page...")
|
||||
driver.get("https://providers.deltadentalma.com/members")
|
||||
driver.get("https://www.deltadentalma.com/private/provider/dashboard")
|
||||
await asyncio.sleep(3)
|
||||
member_search = WebDriverWait(driver, 10).until(
|
||||
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]'))
|
||||
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
|
||||
)
|
||||
print("[OTP] Member search found — login successful!")
|
||||
login_success = True
|
||||
@@ -8,7 +8,7 @@ from selenium.webdriver.support import expected_conditions as EC
|
||||
from selenium.common.exceptions import WebDriverException, TimeoutException
|
||||
|
||||
from selenium_DDMA_claimSubmitWorker import AutomationDDMAClaimSubmit
|
||||
from ddma_browser_manager import get_browser_manager
|
||||
from claimmd_browser_manager import get_browser_manager
|
||||
|
||||
sessions: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
@@ -56,7 +56,7 @@ async def _remove_session_later(sid: str, delay: int = 20):
|
||||
|
||||
async def start_ddma_claim_run(sid: str, data: dict, url: str):
|
||||
"""
|
||||
Run the DDMA claim workflow — mirrors helpers_ddma_eligibility exactly for
|
||||
Run the DDMA claim workflow — mirrors helpers_claimmd_eligibility exactly for
|
||||
login/OTP, then continues into claim-specific steps.
|
||||
|
||||
OTP handling uses two complementary strategies:
|
||||
@@ -321,7 +321,7 @@ async def start_ddma_claim_run(sid: str, data: dict, url: str):
|
||||
|
||||
# Close the browser window after claim (session preserved in profile)
|
||||
try:
|
||||
from ddma_browser_manager import get_browser_manager
|
||||
from claimmd_browser_manager import get_browser_manager
|
||||
get_browser_manager().quit_driver()
|
||||
print("[DDMA Claim] Browser closed - session preserved in profile")
|
||||
except Exception as close_err:
|
||||
|
||||
@@ -1,453 +0,0 @@
|
||||
from selenium import webdriver
|
||||
from selenium.common import TimeoutException
|
||||
from selenium.webdriver.chrome.service import Service
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.support.ui import WebDriverWait, Select
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
from webdriver_manager.chrome import ChromeDriverManager
|
||||
import time
|
||||
import os
|
||||
import base64
|
||||
|
||||
|
||||
class AutomationCMSPEligibilityHistoryRemainingCheck:
|
||||
def __init__(self, data):
|
||||
self.headless = False
|
||||
self.driver = None
|
||||
self.extracted_data = {}
|
||||
self.eligibility_tab = None
|
||||
self.member_details_tab = None
|
||||
self.accumulator_tab = None
|
||||
|
||||
self.data = data.get("data")
|
||||
|
||||
self.massdhp_username = self.data.get("massdhpUsername", "")
|
||||
self.massdhp_password = self.data.get("massdhpPassword", "")
|
||||
self.dateOfBirth = self.data.get("dateOfBirth", "")
|
||||
self.memberId = self.data.get("memberId", "")
|
||||
self.firstName = self.data.get("firstName", "")
|
||||
self.lastName = self.data.get("lastName", "")
|
||||
|
||||
# Convert dateOfBirth from YYYY-MM-DD to MMDDYYYY
|
||||
if self.dateOfBirth and "-" in self.dateOfBirth:
|
||||
parts = self.dateOfBirth.split("-")
|
||||
if len(parts) == 3:
|
||||
year, month, day = parts
|
||||
self.dateOfBirth = f"{month.zfill(2)}{day.zfill(2)}{year}"
|
||||
|
||||
self.download_dir = os.path.abspath("downloads")
|
||||
os.makedirs(self.download_dir, exist_ok=True)
|
||||
|
||||
# ── driver ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def config_driver(self):
|
||||
options = webdriver.ChromeOptions()
|
||||
if self.headless:
|
||||
options.add_argument("--headless")
|
||||
|
||||
prefs = {
|
||||
"download.default_directory": self.download_dir,
|
||||
"plugins.always_open_pdf_externally": False,
|
||||
"download.prompt_for_download": False,
|
||||
"download.directory_upgrade": True,
|
||||
}
|
||||
options.add_experimental_option("prefs", prefs)
|
||||
|
||||
s = Service(ChromeDriverManager().install())
|
||||
self.driver = webdriver.Chrome(service=s, options=options)
|
||||
|
||||
# ── login (same MassHealth credentials) ─────────────────────────────────────
|
||||
|
||||
def login(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
try:
|
||||
signin_button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "a.btn.btn-block.btn-primary[href='https://connectsso.masshealth-dental.org/mhprovider/index.html']")
|
||||
)
|
||||
)
|
||||
signin_button.click()
|
||||
time.sleep(3)
|
||||
|
||||
email_field = wait.until(EC.presence_of_element_located((By.ID, "User")))
|
||||
email_field.clear()
|
||||
email_field.send_keys(self.massdhp_username)
|
||||
|
||||
password_field = wait.until(EC.presence_of_element_located((By.ID, "Password")))
|
||||
password_field.clear()
|
||||
password_field.send_keys(self.massdhp_password)
|
||||
|
||||
login_button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "input[type='submit'][name='submit'][value='Login']")
|
||||
)
|
||||
)
|
||||
login_button.click()
|
||||
return "Success"
|
||||
except Exception as e:
|
||||
print(f"[login] Error: {e}")
|
||||
return "ERROR:LOGIN FAILED"
|
||||
|
||||
# ── step 1 — search member, extract eligibility data ────────────────────────
|
||||
|
||||
def step1(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
substep = "init"
|
||||
try:
|
||||
substep = "patient_management"
|
||||
patient_mgmt = wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//strong[@translate='Patient Management']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView(true);", patient_mgmt)
|
||||
self.driver.execute_script("arguments[0].click();", patient_mgmt)
|
||||
time.sleep(2)
|
||||
|
||||
substep = "member_eligibility_link"
|
||||
eligibility_link = wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//a[@translate='Member Eligibility']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].click();", eligibility_link)
|
||||
time.sleep(2)
|
||||
|
||||
substep = "provider_dropdown"
|
||||
provider_dropdown = wait.until(
|
||||
EC.presence_of_element_located((By.NAME, "provider"))
|
||||
)
|
||||
select_provider = Select(provider_dropdown)
|
||||
first_option = next(
|
||||
(o for o in select_provider.options if o.get_attribute("value").strip()),
|
||||
select_provider.options[0]
|
||||
)
|
||||
select_provider.select_by_value(first_option.get_attribute("value"))
|
||||
time.sleep(2)
|
||||
|
||||
substep = "member_dob"
|
||||
member_dob = wait.until(
|
||||
EC.presence_of_all_elements_located((By.NAME, "dateInput"))
|
||||
)[1]
|
||||
member_dob.clear()
|
||||
member_dob.send_keys(self.dateOfBirth)
|
||||
|
||||
substep = "member_number"
|
||||
member_number = wait.until(
|
||||
EC.presence_of_element_located((By.NAME, "memberNumber"))
|
||||
)
|
||||
member_number.clear()
|
||||
member_number.send_keys(self.memberId)
|
||||
|
||||
substep = "search_button"
|
||||
search_button = wait.until(
|
||||
EC.element_to_be_clickable((By.XPATH, "//button[contains(@class,'btn-primary')]"))
|
||||
)
|
||||
search_button.click()
|
||||
|
||||
substep = "wait_results"
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']")
|
||||
)
|
||||
)
|
||||
|
||||
self.extracted_data = self._extract_data_from_page()
|
||||
print(f"[step1] data extracted: {self.extracted_data}")
|
||||
return "Success"
|
||||
except Exception as e:
|
||||
print(f"[step1] FAILED at substep='{substep}': {e}")
|
||||
return f"ERROR:STEP1:{substep}"
|
||||
|
||||
# ── helpers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _cell_text(self, cell):
|
||||
text = cell.text.strip()
|
||||
if not text:
|
||||
try:
|
||||
text = (self.driver.execute_script("return arguments[0].innerText;", cell) or "").strip()
|
||||
except Exception:
|
||||
pass
|
||||
return text
|
||||
|
||||
def _normalize_id(self, s):
|
||||
return "".join(c for c in str(s) if c.isalnum()).lower()
|
||||
|
||||
def _extract_data_from_page(self):
|
||||
wait = WebDriverWait(self.driver, 15)
|
||||
try:
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']/following::table[1]/tbody/tr")
|
||||
)
|
||||
)
|
||||
for status_label, elig_flag in [("Eligible", "Y"), ("Ineligible", "N")]:
|
||||
rows = self.driver.find_elements(
|
||||
By.XPATH,
|
||||
f"//h4[text()='{status_label}']/following::table[1]/tbody/tr"
|
||||
)
|
||||
for row in rows:
|
||||
cells = row.find_elements(By.TAG_NAME, "td")
|
||||
if len(cells) < 3:
|
||||
continue
|
||||
norm_cell = self._normalize_id(self._cell_text(cells[2]))
|
||||
norm_self = self._normalize_id(self.memberId)
|
||||
if norm_self and norm_cell and (norm_self in norm_cell or norm_cell in norm_self):
|
||||
full_name = self._cell_text(cells[4]) if len(cells) > 4 else ""
|
||||
plan_name = (
|
||||
self._cell_text(cells[6]) if len(cells) > 6
|
||||
else (self._cell_text(cells[-1]) if len(cells) > 4 else "")
|
||||
)
|
||||
name_parts = full_name.split()
|
||||
return {
|
||||
"eligibility": elig_flag,
|
||||
"firstName": name_parts[0] if name_parts else "",
|
||||
"lastName": " ".join(name_parts[1:]) if len(name_parts) > 1 else "",
|
||||
"insurance": plan_name,
|
||||
}
|
||||
return {"eligibility": None}
|
||||
except Exception as e:
|
||||
print("Extraction error:", e)
|
||||
return {"eligibility": None}
|
||||
|
||||
# ── step 2 — print eligibility PDF, return to results tab ───────────────────
|
||||
|
||||
def step2_eligibility_pdf(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
self.eligibility_tab = self.driver.current_window_handle
|
||||
try:
|
||||
download_button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.XPATH, "//button[contains(.,'Printer Friendly Format')]")
|
||||
)
|
||||
)
|
||||
download_button.click()
|
||||
|
||||
try:
|
||||
WebDriverWait(self.driver, 10).until(lambda d: len(d.window_handles) > 1)
|
||||
new_tabs = [t for t in self.driver.window_handles if t != self.eligibility_tab]
|
||||
self.driver.switch_to.window(new_tabs[0])
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
wait.until(EC.presence_of_element_located((By.TAG_NAME, "body")))
|
||||
time.sleep(2)
|
||||
except TimeoutException:
|
||||
print("No new tab for eligibility; printing current page directly")
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"cmsp_eligibility_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
pdf_bytes = base64.b64decode(pdf_data["data"])
|
||||
pdf_path = os.path.join(self.download_dir, pdf_filename)
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_bytes)
|
||||
print("Eligibility PDF saved at:", pdf_path)
|
||||
|
||||
if self.driver.current_window_handle != self.eligibility_tab:
|
||||
self.driver.close()
|
||||
self.driver.switch_to.window(self.eligibility_tab)
|
||||
time.sleep(1)
|
||||
|
||||
return pdf_path
|
||||
except Exception as e:
|
||||
print(f"[step2_eligibility_pdf] failed: {e}")
|
||||
if self.eligibility_tab and self.driver.current_window_handle != self.eligibility_tab:
|
||||
try:
|
||||
self.driver.close()
|
||||
self.driver.switch_to.window(self.eligibility_tab)
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
|
||||
# ── step 3 — click member ID link → member details ──────────────────────────
|
||||
|
||||
def step3_click_member_id(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
substep = "ng_bind_link"
|
||||
try:
|
||||
member_link = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "a[ng-bind='member.memberNumber']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].click();", member_link)
|
||||
time.sleep(2)
|
||||
print(f"[step3] clicked member ID link, URL: {self.driver.current_url}")
|
||||
return "Success"
|
||||
except Exception as e:
|
||||
print(f"[step3] FAILED at substep='{substep}': {e}")
|
||||
return f"ERROR:STEP3:{substep}"
|
||||
|
||||
# ── step 3b — print member details page as PDF via CDP ───────────────────────
|
||||
|
||||
def step3b_member_details_pdf(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
try:
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
time.sleep(2)
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"cmsp_member_details_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
pdf_bytes = base64.b64decode(pdf_data["data"])
|
||||
pdf_path = os.path.join(self.download_dir, pdf_filename)
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_bytes)
|
||||
print("Member details PDF saved at:", pdf_path)
|
||||
return pdf_path
|
||||
except Exception as e:
|
||||
print(f"[step3b_member_details_pdf] failed: {e}")
|
||||
return None
|
||||
|
||||
# ── step 4 — duplicate member details tab ───────────────────────────────────
|
||||
# Opens the same member details URL in a new tab so that Tab A can navigate
|
||||
# to service history while Tab B navigates to accumulator independently.
|
||||
|
||||
def step4_open_duplicate_tab(self):
|
||||
try:
|
||||
self.member_details_tab = self.driver.current_window_handle
|
||||
member_details_url = self.driver.current_url
|
||||
|
||||
self.driver.execute_script("window.open(arguments[0]);", member_details_url)
|
||||
time.sleep(3)
|
||||
|
||||
all_tabs = self.driver.window_handles
|
||||
self.accumulator_tab = next(
|
||||
t for t in all_tabs
|
||||
if t not in [self.eligibility_tab, self.member_details_tab]
|
||||
)
|
||||
print(f"[step4] duplicate tab opened: {self.accumulator_tab}")
|
||||
return "Success"
|
||||
except Exception as e:
|
||||
print(f"[step4] FAILED: {e}")
|
||||
return f"ERROR:STEP4:open_duplicate_tab"
|
||||
|
||||
# ── step 5 — Tab A: service history → CDP print ──────────────────────────────
|
||||
|
||||
def step5_service_history_pdf(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
try:
|
||||
self.driver.switch_to.window(self.member_details_tab)
|
||||
|
||||
history_link = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "a.btn.btn-primary[href*='service-history']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView(true);", history_link)
|
||||
self.driver.execute_script("arguments[0].click();", history_link)
|
||||
time.sleep(2)
|
||||
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
time.sleep(3)
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"cmsp_history_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
pdf_bytes = base64.b64decode(pdf_data["data"])
|
||||
pdf_path = os.path.join(self.download_dir, pdf_filename)
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_bytes)
|
||||
print("History PDF saved at:", pdf_path)
|
||||
return pdf_path
|
||||
except Exception as e:
|
||||
print(f"[step5_service_history_pdf] failed: {e}")
|
||||
return None
|
||||
|
||||
# ── step 6 — Tab B: accumulator → wait for data → CDP print ─────────────────
|
||||
|
||||
def step6_accumulator_pdf(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
try:
|
||||
self.driver.switch_to.window(self.accumulator_tab)
|
||||
|
||||
# Wait for the duplicate member details page to fully load
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
time.sleep(2)
|
||||
|
||||
# Click View Accumulator
|
||||
accumulator_link = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "a.btn.btn-primary[ng-if='vm.showAccumulator']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView(true);", accumulator_link)
|
||||
self.driver.execute_script("arguments[0].click();", accumulator_link)
|
||||
time.sleep(2)
|
||||
|
||||
# Wait for accumulator page to load then print whatever is on screen —
|
||||
# member info is always present even when there are no dollar amounts
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
time.sleep(4)
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"cmsp_accumulator_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
pdf_bytes = base64.b64decode(pdf_data["data"])
|
||||
pdf_path = os.path.join(self.download_dir, pdf_filename)
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_bytes)
|
||||
print("Accumulator PDF saved at:", pdf_path)
|
||||
return pdf_path
|
||||
except Exception as e:
|
||||
print(f"[step6_accumulator_pdf] failed: {e}")
|
||||
return None
|
||||
|
||||
# ── main workflow ────────────────────────────────────────────────────────────
|
||||
|
||||
def main_workflow(self, url):
|
||||
try:
|
||||
self.config_driver()
|
||||
self.driver.maximize_window()
|
||||
self.driver.get(url)
|
||||
time.sleep(3)
|
||||
|
||||
login_result = self.login()
|
||||
if login_result.startswith("ERROR"):
|
||||
return {"status": "error", "message": login_result}
|
||||
|
||||
step1_result = self.step1()
|
||||
if step1_result.startswith("ERROR"):
|
||||
return {"status": "error", "message": step1_result}
|
||||
|
||||
eligibility_pdf_path = self.step2_eligibility_pdf()
|
||||
|
||||
step3_result = self.step3_click_member_id()
|
||||
if step3_result.startswith("ERROR"):
|
||||
return {"status": "partial", "message": step3_result,
|
||||
"pdf_path": eligibility_pdf_path, "file_type": "pdf"}
|
||||
|
||||
member_details_pdf_path = self.step3b_member_details_pdf()
|
||||
|
||||
# Duplicate the member details tab so service history and accumulator
|
||||
# can be opened independently without back-navigation
|
||||
step4_result = self.step4_open_duplicate_tab()
|
||||
if step4_result.startswith("ERROR"):
|
||||
return {"status": "partial", "message": step4_result,
|
||||
"pdf_path": eligibility_pdf_path,
|
||||
"member_details_pdf_path": member_details_pdf_path,
|
||||
"file_type": "pdf"}
|
||||
|
||||
# Tab A: service history PDF
|
||||
history_pdf_path = self.step5_service_history_pdf()
|
||||
|
||||
# Tab B: accumulator PDF
|
||||
accumulator_pdf_path = self.step6_accumulator_pdf()
|
||||
|
||||
result = {
|
||||
"status": "success",
|
||||
"pdf_path": eligibility_pdf_path,
|
||||
"member_details_pdf_path": member_details_pdf_path,
|
||||
"history_pdf_path": history_pdf_path,
|
||||
"accumulator_pdf_path": accumulator_pdf_path,
|
||||
"file_type": "pdf",
|
||||
"message": "Eligibility, member details, service history, and accumulator PDFs captured",
|
||||
}
|
||||
if self.extracted_data:
|
||||
result.update(self.extracted_data)
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
finally:
|
||||
if self.driver:
|
||||
self.driver.quit()
|
||||
+238
-247
@@ -10,8 +10,9 @@ import time
|
||||
import os
|
||||
import re
|
||||
import base64
|
||||
import calendar
|
||||
|
||||
from ddma_browser_manager import get_browser_manager
|
||||
from claimmd_browser_manager import get_browser_manager
|
||||
|
||||
class AutomationDeltaDentalMAEligibilityCheck:
|
||||
def __init__(self, data):
|
||||
@@ -45,7 +46,7 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
|
||||
# Try to click logout button if visible
|
||||
try:
|
||||
self.driver.get("https://providers.deltadentalma.com/")
|
||||
self.driver.get("https://www.deltadentalma.com/")
|
||||
time.sleep(2)
|
||||
|
||||
logout_selectors = [
|
||||
@@ -99,27 +100,27 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
current_url = self.driver.current_url
|
||||
print(f"[login] Current URL: {current_url}")
|
||||
|
||||
logged_in_patterns = ["member", "dashboard", "eligibility", "search", "patients"]
|
||||
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 "onboarding" not in current_url.lower():
|
||||
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, '//input[@placeholder="Search by member ID"]'))
|
||||
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://providers.deltadentalma.com/members"
|
||||
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, '//input[@placeholder="Search by member ID"]'))
|
||||
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"
|
||||
@@ -137,9 +138,9 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
current_url = self.driver.current_url
|
||||
print(f"[login] URL after navigation: {current_url}")
|
||||
|
||||
if "onboarding" not in current_url.lower():
|
||||
if "sign-in" not in current_url.lower():
|
||||
member_search = WebDriverWait(self.driver, 3).until(
|
||||
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]'))
|
||||
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
|
||||
)
|
||||
if member_search:
|
||||
print("[login] Session valid - skipping login")
|
||||
@@ -172,7 +173,7 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
try:
|
||||
otp_candidate = WebDriverWait(self.driver, 10).until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//input[contains(@aria-lable,'Verification code') or contains(@placeholder,'Enter your verification code') or contains(@aria-label,'Verification code')]")
|
||||
(By.XPATH, "//input[@data-test-id='MfaCode']")
|
||||
)
|
||||
)
|
||||
if otp_candidate:
|
||||
@@ -189,7 +190,7 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
time.sleep(2)
|
||||
try:
|
||||
member_search = WebDriverWait(self.driver, 5).until(
|
||||
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]'))
|
||||
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
|
||||
)
|
||||
if member_search:
|
||||
print("[login] Already authenticated after modal dismiss")
|
||||
@@ -200,30 +201,21 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
# Fill login form
|
||||
try:
|
||||
email_field = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.XPATH, "//input[@name='username' and @type='text']"))
|
||||
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[@name='username' and @type='text']")))
|
||||
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[@name='password' and @type='password']")))
|
||||
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)
|
||||
|
||||
# Remember me
|
||||
try:
|
||||
remember_me_checkbox = wait.until(EC.element_to_be_clickable(
|
||||
(By.XPATH, "//label[.//span[contains(text(),'Remember me')]]")
|
||||
))
|
||||
remember_me_checkbox.click()
|
||||
except Exception:
|
||||
print("[login] Remember me checkbox not found (continuing).")
|
||||
|
||||
login_button = wait.until(EC.element_to_be_clickable((By.XPATH, "//button[@type='submit' and @aria-label='Sign in']")))
|
||||
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
|
||||
@@ -234,7 +226,7 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
try:
|
||||
otp_candidate = WebDriverWait(self.driver, 30).until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//input[contains(@aria-lable,'Verification code') or contains(@placeholder,'Enter your verification code')]")
|
||||
(By.XPATH, "//input[@data-test-id='MfaCode']")
|
||||
)
|
||||
)
|
||||
if otp_candidate:
|
||||
@@ -247,7 +239,7 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
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, '//input[@placeholder="Search by member ID"]'))
|
||||
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"
|
||||
@@ -264,7 +256,7 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
except TimeoutException:
|
||||
pass
|
||||
|
||||
if "onboarding" in self.driver.current_url.lower() or "login" in self.driver.current_url.lower():
|
||||
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"
|
||||
|
||||
@@ -275,103 +267,66 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
print("[login] Exception during login:", e)
|
||||
return f"ERROR:LOGIN FAILED: {e}"
|
||||
|
||||
def _select_provider_dropdown(self):
|
||||
"""
|
||||
Click the provider dropdown on the member search page and select the option
|
||||
matching self.providerName (case-insensitive). Falls back to the first option.
|
||||
The button data-testid is 'member-search_provider_select-btn'.
|
||||
"""
|
||||
try:
|
||||
short_wait = WebDriverWait(self.driver, 5)
|
||||
try:
|
||||
provider_btn = short_wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.XPATH, '//button[@data-testid="member-search_provider_select-btn"]')
|
||||
)
|
||||
)
|
||||
except TimeoutException:
|
||||
print("[DDMA step1] No provider dropdown found — skipping")
|
||||
return
|
||||
|
||||
provider_btn.click()
|
||||
time.sleep(0.5)
|
||||
|
||||
try:
|
||||
WebDriverWait(self.driver, 5).until(
|
||||
EC.presence_of_element_located((By.XPATH, "//*[@role='option']"))
|
||||
)
|
||||
except TimeoutException:
|
||||
print("[DDMA step1] Provider listbox did not open")
|
||||
return
|
||||
|
||||
options = self.driver.find_elements(By.XPATH, "//*[@role='option']")
|
||||
print(f"[DDMA step1] Provider options: {[o.text.strip() for o in options]}")
|
||||
|
||||
target = (self.providerName or "").strip().lower()
|
||||
selected = False
|
||||
|
||||
if target:
|
||||
for opt in options:
|
||||
if target in opt.text.lower():
|
||||
opt.click()
|
||||
print(f"[DDMA step1] Selected provider: '{opt.text.strip()}'")
|
||||
selected = True
|
||||
break
|
||||
|
||||
if not selected and options:
|
||||
options[0].click()
|
||||
print(f"[DDMA step1] No match for '{self.providerName}', selected first: '{options[0].text.strip()}'")
|
||||
|
||||
time.sleep(0.3)
|
||||
|
||||
except Exception as e:
|
||||
print(f"[DDMA step1] Provider selection error (non-fatal): {e}")
|
||||
|
||||
def step1(self):
|
||||
"""Fill search form with all available fields (flexible search)."""
|
||||
"""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"ID: {self.memberId}")
|
||||
if self.firstName:
|
||||
fields.append(f"FirstName: {self.firstName}")
|
||||
if self.lastName:
|
||||
fields.append(f"LastName: {self.lastName}")
|
||||
fields.append(f"SubscriberID: {self.memberId}")
|
||||
if self.dateOfBirth:
|
||||
fields.append(f"DOB: {self.dateOfBirth}")
|
||||
print(f"[DDMA step1] Starting search with: {', '.join(fields)}")
|
||||
|
||||
# Wait for the search page to be fully loaded before interacting
|
||||
wait.until(EC.presence_of_element_located(
|
||||
(By.XPATH, '//input[@placeholder="Search by member ID"]')
|
||||
))
|
||||
time.sleep(1)
|
||||
search_page_url = "https://www.deltadentalma.com/private/provider/check-patient-benefits-search"
|
||||
|
||||
# Select provider from dropdown based on NPI settings
|
||||
self._select_provider_dropdown()
|
||||
# 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)
|
||||
|
||||
def replace_with_sendkeys(el, value):
|
||||
el.click()
|
||||
el.send_keys(Keys.CONTROL, "a")
|
||||
el.send_keys(Keys.BACKSPACE)
|
||||
el.send_keys(value)
|
||||
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")
|
||||
|
||||
# 1. Fill Member ID if provided
|
||||
if self.memberId:
|
||||
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:
|
||||
member_id_input = wait.until(EC.presence_of_element_located(
|
||||
(By.XPATH, '//input[@placeholder="Search by member ID"]')
|
||||
))
|
||||
member_id_input.clear()
|
||||
member_id_input.send_keys(self.memberId)
|
||||
print(f"[DDMA step1] Entered Member ID: {self.memberId}")
|
||||
time.sleep(0.2)
|
||||
except Exception as e:
|
||||
print(f"[DDMA step1] Warning: Could not fill Member ID: {e}")
|
||||
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"
|
||||
|
||||
# 2. Fill DOB if provided
|
||||
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("-")
|
||||
@@ -379,54 +334,57 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
month = dob_parts[1].zfill(2)
|
||||
day = dob_parts[2].zfill(2)
|
||||
|
||||
dob_container = wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//div[@data-testid='member-search_date-of-birth']")
|
||||
)
|
||||
)
|
||||
|
||||
month_elem = dob_container.find_element(By.XPATH, ".//span[@data-type='month' and @contenteditable='true']")
|
||||
day_elem = dob_container.find_element(By.XPATH, ".//span[@data-type='day' and @contenteditable='true']")
|
||||
year_elem = dob_container.find_element(By.XPATH, ".//span[@data-type='year' and @contenteditable='true']")
|
||||
|
||||
replace_with_sendkeys(month_elem, month)
|
||||
time.sleep(0.05)
|
||||
replace_with_sendkeys(day_elem, day)
|
||||
time.sleep(0.05)
|
||||
replace_with_sendkeys(year_elem, year)
|
||||
print(f"[DDMA step1] Filled DOB: {month}/{day}/{year}")
|
||||
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 DOB: {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
|
||||
continue_btn = wait.until(EC.element_to_be_clickable(
|
||||
(By.XPATH, '//button[@data-testid="member-search_search-button"]')
|
||||
search_btn = wait.until(EC.element_to_be_clickable(
|
||||
(By.XPATH, "//button[@data-test-id='InStateSearchButton']")
|
||||
))
|
||||
continue_btn.click()
|
||||
search_btn.click()
|
||||
print("[DDMA step1] Clicked Search button")
|
||||
|
||||
# Wait for either results row or no-results message (up to 15s)
|
||||
# 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(
|
||||
EC.any_of(
|
||||
EC.presence_of_element_located((By.XPATH, "//tbody//tr")),
|
||||
EC.presence_of_element_located((By.XPATH, '//div[@data-testid="member-search-result-no-results"]')),
|
||||
)
|
||||
lambda d: d.find_elements(By.XPATH, "//table//tbody//tr")
|
||||
or "check-patient-benefit-details" in d.current_url.lower()
|
||||
)
|
||||
time.sleep(3)
|
||||
time.sleep(2)
|
||||
except TimeoutException:
|
||||
pass # proceed and let step2 handle missing results
|
||||
print("[DDMA step1] Warning: No results table appeared within timeout")
|
||||
return "ERROR: INVALID SEARCH CRITERIA"
|
||||
|
||||
# Check for no-results error
|
||||
try:
|
||||
error_msg = self.driver.find_element(By.XPATH, '//div[@data-testid="member-search-result-no-results"]')
|
||||
if error_msg:
|
||||
print("[DDMA step1] Error: No results found")
|
||||
return "ERROR: INVALID SEARCH CRITERIA"
|
||||
except Exception:
|
||||
pass
|
||||
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"
|
||||
@@ -442,112 +400,120 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
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 = ""
|
||||
# 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)
|
||||
|
||||
# Extract patient name from search results row
|
||||
try:
|
||||
first_row = self.driver.find_element(By.XPATH, "(//tbody//tr)[1]")
|
||||
row_text = first_row.text.strip()
|
||||
print(f"[DDMA step2] First row text: {row_text[:150]}...")
|
||||
# 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()
|
||||
|
||||
lines = row_text.split('\n') if row_text else []
|
||||
for line in lines:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
if re.match(r"^[A-Za-z\s\-']{2,60}$", line):
|
||||
patientName = line
|
||||
print(f"[DDMA step2] Extracted patient name from row: '{patientName}'")
|
||||
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
|
||||
except Exception as e:
|
||||
print(f"[DDMA step2] Error reading first row: {e}")
|
||||
|
||||
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:
|
||||
short_wait = WebDriverWait(self.driver, 3)
|
||||
status_link = short_wait.until(EC.presence_of_element_located((
|
||||
By.XPATH,
|
||||
"(//tbody//tr)[1]//a[contains(@href, 'member-eligibility-search')]"
|
||||
)))
|
||||
eligibilityText = status_link.text.strip().lower()
|
||||
print(f"[DDMA step2] Found eligibility status: {eligibilityText}")
|
||||
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:
|
||||
try:
|
||||
alt_status = self.driver.find_element(By.XPATH, "//*[contains(text(),'Active') or contains(text(),'Inactive') or contains(text(),'Eligible')]")
|
||||
eligibilityText = alt_status.text.strip().lower()
|
||||
if "active" in eligibilityText or "eligible" in eligibilityText:
|
||||
eligibilityText = "active"
|
||||
elif "inactive" in eligibilityText:
|
||||
eligibilityText = "inactive"
|
||||
print(f"[DDMA step2] Found eligibility via alternative: {eligibilityText}")
|
||||
except Exception:
|
||||
pass
|
||||
print(f"[DDMA step2] Warning: Still on search page: {self.driver.current_url}")
|
||||
|
||||
# Find the member-details URL from the first row
|
||||
print("[DDMA step2] Looking for patient detail link...")
|
||||
detail_url = None
|
||||
|
||||
link_selectors = [
|
||||
"(//table//tbody//tr)[1]//td[1]//a",
|
||||
"(//tbody//tr)[1]//a[contains(@href, 'member-details')]",
|
||||
"(//tbody//tr)[1]//a[contains(@href, 'member')]",
|
||||
"//a[contains(@href, 'member-details')]",
|
||||
]
|
||||
for selector in link_selectors:
|
||||
# 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:
|
||||
link_el = WebDriverWait(self.driver, 5).until(
|
||||
WebDriverWait(self.driver, 15).until(
|
||||
EC.presence_of_element_located((By.XPATH, selector))
|
||||
)
|
||||
href = link_el.get_attribute("href")
|
||||
if href and "member-details" in href:
|
||||
detail_url = href
|
||||
print(f"[DDMA step2] Found detail URL: {href}")
|
||||
break
|
||||
print(f"[DDMA step2] Detail page content loaded: {selector}")
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if detail_url:
|
||||
# Always navigate via driver.get() — this blocks until the new page loads,
|
||||
# unlike click() which returns immediately and causes a race condition.
|
||||
print(f"[DDMA step2] Navigating to detail page: {detail_url}")
|
||||
self.driver.get(detail_url)
|
||||
time.sleep(3) # Let JavaScript finish rendering all sections
|
||||
|
||||
# Confirm we actually landed on the detail page (not redirected away)
|
||||
try:
|
||||
WebDriverWait(self.driver, 15).until(
|
||||
lambda d: "member-details" in d.current_url
|
||||
)
|
||||
print(f"[DDMA step2] Confirmed on detail page: {self.driver.current_url}")
|
||||
except Exception:
|
||||
print(f"[DDMA step2] Warning: URL after navigation: {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
|
||||
|
||||
# Get full page text as lines
|
||||
# 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;"
|
||||
@@ -576,15 +542,29 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
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"[DDMA 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
|
||||
# (DDMA's detail page labels this "Subscriber name:", so that
|
||||
# prefix must be accepted rather than excluded)
|
||||
same_line = re.match(
|
||||
r'^(?:member\s+)?name\s*[:\-]\s*(.+)$', line, re.IGNORECASE
|
||||
r'^(?:member\s+|subscriber\s+)?name\s*[:\-]\s*(.+)$', line, re.IGNORECASE
|
||||
)
|
||||
if same_line and not re.search(
|
||||
r'(provider|group|subscriber|plan)\s+name', line, re.IGNORECASE
|
||||
r'(provider|group|plan)\s+name', line, re.IGNORECASE
|
||||
):
|
||||
candidate = same_line.group(1).strip()
|
||||
if looks_like_name(candidate):
|
||||
@@ -593,7 +573,7 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
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):
|
||||
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):
|
||||
@@ -602,8 +582,6 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
break
|
||||
if patientName:
|
||||
break
|
||||
else:
|
||||
print("[DDMA step2] Warning: Could not find patient detail link")
|
||||
|
||||
# Clean patient name — strip any remaining date artifacts
|
||||
if patientName:
|
||||
@@ -614,6 +592,27 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
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(
|
||||
@@ -650,14 +649,6 @@ class AutomationDeltaDentalMAEligibilityCheck:
|
||||
|
||||
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 {
|
||||
@@ -11,7 +11,7 @@ import json
|
||||
import base64
|
||||
from datetime import date
|
||||
|
||||
from ddma_browser_manager import get_browser_manager
|
||||
from claimmd_browser_manager import get_browser_manager
|
||||
|
||||
MEMBERS_URL = "https://providers.deltadentalma.com/members"
|
||||
|
||||
|
||||
@@ -1,471 +0,0 @@
|
||||
from selenium import webdriver
|
||||
from selenium.common import TimeoutException
|
||||
from selenium.webdriver.chrome.service import Service
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.support.ui import WebDriverWait, Select
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
from webdriver_manager.chrome import ChromeDriverManager
|
||||
import time
|
||||
import os
|
||||
import base64
|
||||
|
||||
|
||||
class AutomationMassHealthEligibilityHistoryCheck:
|
||||
def __init__(self, data):
|
||||
self.headless = False
|
||||
self.driver = None
|
||||
self.extracted_data = {}
|
||||
self.eligibility_tab = None # handle to the eligibility results tab
|
||||
|
||||
self.data = data.get("data")
|
||||
|
||||
self.massdhp_username = self.data.get("massdhpUsername", "")
|
||||
self.massdhp_password = self.data.get("massdhpPassword", "")
|
||||
self.dateOfBirth = self.data.get("dateOfBirth", "")
|
||||
self.memberId = self.data.get("memberId", "")
|
||||
self.firstName = self.data.get("firstName", "")
|
||||
self.lastName = self.data.get("lastName", "")
|
||||
|
||||
# Convert dateOfBirth from YYYY-MM-DD to MMDDYYYY
|
||||
if self.dateOfBirth and "-" in self.dateOfBirth:
|
||||
parts = self.dateOfBirth.split("-")
|
||||
if len(parts) == 3:
|
||||
year, month, day = parts
|
||||
self.dateOfBirth = f"{month.zfill(2)}{day.zfill(2)}{year}"
|
||||
|
||||
self.download_dir = os.path.abspath("downloads")
|
||||
os.makedirs(self.download_dir, exist_ok=True)
|
||||
|
||||
# ── driver ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def config_driver(self):
|
||||
options = webdriver.ChromeOptions()
|
||||
if self.headless:
|
||||
options.add_argument("--headless")
|
||||
|
||||
prefs = {
|
||||
"download.default_directory": self.download_dir,
|
||||
"plugins.always_open_pdf_externally": False,
|
||||
"download.prompt_for_download": False,
|
||||
"download.directory_upgrade": True,
|
||||
}
|
||||
options.add_experimental_option("prefs", prefs)
|
||||
|
||||
s = Service(ChromeDriverManager().install())
|
||||
self.driver = webdriver.Chrome(service=s, options=options)
|
||||
|
||||
# ── login ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def login(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
try:
|
||||
signin_button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "a.btn.btn-block.btn-primary[href='https://connectsso.masshealth-dental.org/mhprovider/index.html']")
|
||||
)
|
||||
)
|
||||
signin_button.click()
|
||||
time.sleep(3)
|
||||
|
||||
email_field = wait.until(EC.presence_of_element_located((By.ID, "User")))
|
||||
email_field.clear()
|
||||
email_field.send_keys(self.massdhp_username)
|
||||
|
||||
password_field = wait.until(EC.presence_of_element_located((By.ID, "Password")))
|
||||
password_field.clear()
|
||||
password_field.send_keys(self.massdhp_password)
|
||||
|
||||
login_button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "input[type='submit'][name='submit'][value='Login']")
|
||||
)
|
||||
)
|
||||
login_button.click()
|
||||
|
||||
# Wait for the SSO redirect. On some machines Chrome opens the dashboard
|
||||
# in a new tab and closes the SSO tab — poll ALL window handles.
|
||||
print("[login] Waiting for SSO redirect to provider.masshealth-dental.org ...")
|
||||
deadline = time.time() + 30
|
||||
found = False
|
||||
while time.time() < deadline:
|
||||
time.sleep(0.5)
|
||||
for handle in self.driver.window_handles:
|
||||
try:
|
||||
self.driver.switch_to.window(handle)
|
||||
if self.driver.current_url.startswith("https://provider.masshealth-dental.org"):
|
||||
print(f"[login] Redirect complete. URL: {self.driver.current_url}")
|
||||
found = True
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
if found:
|
||||
break
|
||||
|
||||
if not found:
|
||||
print("[login] Redirect timeout — portal window not found in any tab.")
|
||||
return "ERROR: Login redirect timed out — check credentials or portal availability."
|
||||
|
||||
return "Success"
|
||||
except Exception as e:
|
||||
print(f"[login] Error: {e}")
|
||||
return "ERROR:LOGIN FAILED"
|
||||
|
||||
# ── step 1 — search member, extract eligibility data ────────────────────────
|
||||
|
||||
def step1(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
substep = "init"
|
||||
try:
|
||||
print(f"[step1] URL: {self.driver.current_url}")
|
||||
|
||||
substep = "patient_management"
|
||||
patient_mgmt = wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//strong[@translate='Patient Management']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView(true);", patient_mgmt)
|
||||
self.driver.execute_script("arguments[0].click();", patient_mgmt)
|
||||
time.sleep(2)
|
||||
|
||||
substep = "member_eligibility_link"
|
||||
eligibility_link = wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//a[@translate='Member Eligibility']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].click();", eligibility_link)
|
||||
time.sleep(2)
|
||||
|
||||
substep = "provider_dropdown"
|
||||
provider_dropdown = wait.until(
|
||||
EC.presence_of_element_located((By.NAME, "provider"))
|
||||
)
|
||||
select_provider = Select(provider_dropdown)
|
||||
options = [o.text for o in select_provider.options]
|
||||
print(f"[step1] provider options: {options}")
|
||||
|
||||
substep = "select_provider"
|
||||
first_option = next(
|
||||
(o for o in select_provider.options if o.get_attribute("value").strip()),
|
||||
select_provider.options[0]
|
||||
)
|
||||
print(f"[step1] selecting provider: '{first_option.text}'")
|
||||
select_provider.select_by_value(first_option.get_attribute("value"))
|
||||
time.sleep(2)
|
||||
|
||||
substep = "member_dob"
|
||||
member_dob = wait.until(
|
||||
EC.presence_of_all_elements_located((By.NAME, "dateInput"))
|
||||
)[1]
|
||||
member_dob.clear()
|
||||
member_dob.send_keys(self.dateOfBirth)
|
||||
|
||||
substep = "member_number"
|
||||
member_number = wait.until(
|
||||
EC.presence_of_element_located((By.NAME, "memberNumber"))
|
||||
)
|
||||
member_number.clear()
|
||||
member_number.send_keys(self.memberId)
|
||||
|
||||
substep = "search_button"
|
||||
search_button = wait.until(
|
||||
EC.element_to_be_clickable((By.XPATH, "//button[contains(@class,'btn-primary')]"))
|
||||
)
|
||||
search_button.click()
|
||||
|
||||
substep = "wait_results"
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']")
|
||||
)
|
||||
)
|
||||
|
||||
self.extracted_data = self._extract_data_from_page()
|
||||
print(f"[step1] data extracted: {self.extracted_data}")
|
||||
return "Success"
|
||||
except Exception as e:
|
||||
print(f"[step1] FAILED at substep='{substep}': {e}")
|
||||
print(f"[step1] URL at failure: {self.driver.current_url}")
|
||||
return f"ERROR:STEP1:{substep}"
|
||||
|
||||
# ── helpers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _cell_text(self, cell):
|
||||
text = cell.text.strip()
|
||||
if not text:
|
||||
try:
|
||||
text = (self.driver.execute_script("return arguments[0].innerText;", cell) or "").strip()
|
||||
except Exception:
|
||||
pass
|
||||
return text
|
||||
|
||||
def _normalize_id(self, s):
|
||||
return "".join(c for c in str(s) if c.isalnum()).lower()
|
||||
|
||||
def _extract_data_from_page(self):
|
||||
wait = WebDriverWait(self.driver, 15)
|
||||
extracted = {}
|
||||
try:
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']/following::table[1]/tbody/tr")
|
||||
)
|
||||
)
|
||||
|
||||
# AngularJS populates cell text after elements appear in the DOM.
|
||||
# Retry a few times until a row with non-empty cell[2] is found.
|
||||
for attempt in range(4):
|
||||
for status_label, elig_flag in [("Eligible", "Y"), ("Ineligible", "N")]:
|
||||
rows = self.driver.find_elements(
|
||||
By.XPATH,
|
||||
f"//h4[text()='{status_label}']/following::table[1]/tbody/tr"
|
||||
)
|
||||
for row in rows:
|
||||
cells = row.find_elements(By.TAG_NAME, "td")
|
||||
if len(cells) < 3:
|
||||
continue
|
||||
member_number = self._cell_text(cells[2])
|
||||
norm_cell = self._normalize_id(member_number)
|
||||
norm_self = self._normalize_id(self.memberId)
|
||||
if norm_self and norm_cell and (norm_self in norm_cell or norm_cell in norm_self):
|
||||
full_name = self._cell_text(cells[4]) if len(cells) > 4 else ""
|
||||
plan_name = (
|
||||
self._cell_text(cells[6]) if len(cells) > 6
|
||||
else (self._cell_text(cells[-1]) if len(cells) > 4 else "")
|
||||
)
|
||||
name_parts = full_name.split()
|
||||
extracted = {
|
||||
"eligibility": elig_flag,
|
||||
"firstName": name_parts[0] if name_parts else "",
|
||||
"lastName": " ".join(name_parts[1:]) if len(name_parts) > 1 else "",
|
||||
"insurance": plan_name,
|
||||
}
|
||||
print(f"[extraction] MATCHED {status_label} → name='{full_name}' plan='{plan_name}'")
|
||||
return extracted
|
||||
if attempt < 3:
|
||||
print(f"[extraction] cells not yet populated (attempt {attempt + 1}), retrying...")
|
||||
time.sleep(1.5)
|
||||
|
||||
print(f"[extraction] No matching row for memberId='{self.memberId}'")
|
||||
return {"eligibility": None}
|
||||
except Exception as e:
|
||||
print("Extraction error:", e)
|
||||
return {"eligibility": None}
|
||||
|
||||
# ── step 2 — print eligibility PDF, stay on results tab ─────────────────────
|
||||
|
||||
def step2_eligibility_pdf(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
self.eligibility_tab = self.driver.current_window_handle
|
||||
|
||||
try:
|
||||
download_button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.XPATH, "//button[contains(.,'Printer Friendly Format')]")
|
||||
)
|
||||
)
|
||||
download_button.click()
|
||||
|
||||
try:
|
||||
WebDriverWait(self.driver, 10).until(lambda d: len(d.window_handles) > 1)
|
||||
new_tabs = [t for t in self.driver.window_handles if t != self.eligibility_tab]
|
||||
self.driver.switch_to.window(new_tabs[0])
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
wait.until(EC.presence_of_element_located((By.TAG_NAME, "body")))
|
||||
time.sleep(2)
|
||||
print("Printer-friendly tab opened:", self.driver.current_url)
|
||||
except TimeoutException:
|
||||
print("No new tab for eligibility; printing current page directly")
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"eligibility_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
pdf_bytes = base64.b64decode(pdf_data["data"])
|
||||
pdf_path = os.path.join(self.download_dir, pdf_filename)
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_bytes)
|
||||
print("Eligibility PDF saved at:", pdf_path)
|
||||
|
||||
# Close printer-friendly tab and return to eligibility results
|
||||
if self.driver.current_window_handle != self.eligibility_tab:
|
||||
self.driver.close()
|
||||
self.driver.switch_to.window(self.eligibility_tab)
|
||||
time.sleep(1)
|
||||
|
||||
return pdf_path
|
||||
except Exception as e:
|
||||
print(f"[step2_eligibility_pdf] failed: {e}")
|
||||
# Ensure we are back on the eligibility results tab even on error
|
||||
if self.eligibility_tab and self.driver.current_window_handle != self.eligibility_tab:
|
||||
try:
|
||||
self.driver.close()
|
||||
self.driver.switch_to.window(self.eligibility_tab)
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
|
||||
# ── step 3 — click member ID link → member details page ─────────────────────
|
||||
|
||||
def step3_click_member_id(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
substep = "init"
|
||||
try:
|
||||
substep = "ng_bind_link"
|
||||
member_link = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "a[ng-bind='member.memberNumber']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].click();", member_link)
|
||||
time.sleep(2)
|
||||
print(f"[step3] clicked member ID link, URL: {self.driver.current_url}")
|
||||
return "Success"
|
||||
except Exception as e:
|
||||
print(f"[step3] FAILED at substep='{substep}': {e}")
|
||||
return f"ERROR:STEP3:{substep}"
|
||||
|
||||
# ── step 3b — print member details page as PDF via CDP ───────────────────────
|
||||
|
||||
def step3b_member_details_pdf(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
try:
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
time.sleep(2)
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"eligibility_member_details_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
pdf_bytes = base64.b64decode(pdf_data["data"])
|
||||
pdf_path = os.path.join(self.download_dir, pdf_filename)
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_bytes)
|
||||
print("Member details PDF saved at:", pdf_path)
|
||||
return pdf_path
|
||||
except Exception as e:
|
||||
print(f"[step3b_member_details_pdf] failed: {e}")
|
||||
return None
|
||||
|
||||
# ── step 4 — click "View Service History" on member details page ─────────────
|
||||
|
||||
def step4_view_service_history(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
substep = "init"
|
||||
try:
|
||||
# Primary: href contains /service-history and has btn-primary class
|
||||
substep = "service_history_link"
|
||||
history_link = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "a.btn.btn-primary[href*='service-history']")
|
||||
)
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView(true);", history_link)
|
||||
self.driver.execute_script("arguments[0].click();", history_link)
|
||||
time.sleep(2)
|
||||
print(f"[step4] navigated to service history, URL: {self.driver.current_url}")
|
||||
return "Success"
|
||||
except Exception as e:
|
||||
print(f"[step4] FAILED at substep='{substep}': {e}")
|
||||
return f"ERROR:STEP4:{substep}"
|
||||
|
||||
# ── step 5 — print service history PDF ──────────────────────────────────────
|
||||
|
||||
def step5_history_pdf(self):
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
current_tab = self.driver.current_window_handle
|
||||
try:
|
||||
print_button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.CSS_SELECTOR, "button.btn.btn-primary[ng-click='vm.printPage()']")
|
||||
)
|
||||
)
|
||||
print_button.click()
|
||||
|
||||
try:
|
||||
WebDriverWait(self.driver, 10).until(lambda d: len(d.window_handles) > 1)
|
||||
new_tabs = [t for t in self.driver.window_handles if t != current_tab]
|
||||
self.driver.switch_to.window(new_tabs[0])
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
wait.until(EC.presence_of_element_located((By.TAG_NAME, "body")))
|
||||
time.sleep(2)
|
||||
print("History printer-friendly tab opened:", self.driver.current_url)
|
||||
except TimeoutException:
|
||||
print("No new tab for history; printing current page directly")
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"eligibility_history_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
pdf_bytes = base64.b64decode(pdf_data["data"])
|
||||
pdf_path = os.path.join(self.download_dir, pdf_filename)
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_bytes)
|
||||
print("History PDF saved at:", pdf_path)
|
||||
return pdf_path
|
||||
except Exception as e:
|
||||
print(f"[step5_history_pdf] failed: {e}")
|
||||
return None
|
||||
|
||||
# ── main workflow ────────────────────────────────────────────────────────────
|
||||
|
||||
def main_workflow(self, url):
|
||||
try:
|
||||
self.config_driver()
|
||||
self.driver.maximize_window()
|
||||
self.driver.get(url)
|
||||
time.sleep(3)
|
||||
|
||||
login_result = self.login()
|
||||
if login_result.startswith("ERROR"):
|
||||
return {"status": "error", "message": login_result}
|
||||
|
||||
step1_result = self.step1()
|
||||
if step1_result.startswith("ERROR"):
|
||||
return {"status": "error", "message": step1_result}
|
||||
|
||||
# Print eligibility PDF; stays on results tab afterwards
|
||||
eligibility_pdf_path = self.step2_eligibility_pdf()
|
||||
|
||||
# Click the member ID link → member details
|
||||
step3_result = self.step3_click_member_id()
|
||||
if step3_result.startswith("ERROR"):
|
||||
return {
|
||||
"status": "partial",
|
||||
"message": step3_result,
|
||||
"pdf_path": eligibility_pdf_path,
|
||||
"file_type": "pdf",
|
||||
}
|
||||
|
||||
# Print member details page
|
||||
member_details_pdf_path = self.step3b_member_details_pdf()
|
||||
|
||||
# Click "View Service History"
|
||||
step4_result = self.step4_view_service_history()
|
||||
if step4_result.startswith("ERROR"):
|
||||
return {
|
||||
"status": "partial",
|
||||
"message": step4_result,
|
||||
"pdf_path": eligibility_pdf_path,
|
||||
"member_details_pdf_path": member_details_pdf_path,
|
||||
"file_type": "pdf",
|
||||
}
|
||||
|
||||
# Print history PDF
|
||||
history_pdf_path = self.step5_history_pdf()
|
||||
|
||||
result = {
|
||||
"status": "success",
|
||||
"pdf_path": eligibility_pdf_path,
|
||||
"member_details_pdf_path": member_details_pdf_path,
|
||||
"history_pdf_path": history_pdf_path,
|
||||
"file_type": "pdf",
|
||||
"message": "Eligibility, member details, and service history PDFs captured",
|
||||
}
|
||||
if self.extracted_data:
|
||||
result.update(self.extracted_data)
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
finally:
|
||||
if self.driver:
|
||||
self.driver.quit()
|
||||
Reference in New Issue
Block a user