GiteadandClaude Sonnet 5
52ddfd9f57
- Add remainingBenefit and thirdPartyInsurance columns to Patient; extract both from the MassHealth accumulator/member-detail pages during the eligibility check and surface them in the eligibility table - Add an updated-date display and a toggle to suppress the post-check PDF viewer popup (default off) on the insurance eligibility page - Rename CMSP-prefixed PDF filenames and UI copy to MH, since CMSP and MH now share the same worker - Fold CMSP into MassHealth/MH across patient/claim insurance-type dropdowns and the chatbot's eligibility routing, since they're the same insurance - Remove auto-download of saved eligibility PDFs; downloading is now manual Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
560 lines
24 KiB
Python
560 lines
24 KiB
Python
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"mh_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 extract_third_party_insurance(self):
|
|
"""Reads the "Third Party Liability" table (Policy Number | Coverage Type |
|
|
Effective Date | Termination Date | Insurer Name | Insurer Payment Order)
|
|
and returns the Insurer Name(s), or None if the table has no rows
|
|
(i.e. the member has no third-party/2nd insurance).
|
|
"""
|
|
try:
|
|
tables = self.driver.find_elements(By.TAG_NAME, "table")
|
|
for table in tables:
|
|
rows = table.find_elements(By.TAG_NAME, "tr")
|
|
if not rows:
|
|
continue
|
|
|
|
header_cells = [
|
|
c.text.strip().lower()
|
|
for c in (rows[0].find_elements(By.TAG_NAME, "th") or rows[0].find_elements(By.TAG_NAME, "td"))
|
|
]
|
|
if "insurer name" not in header_cells:
|
|
continue
|
|
|
|
insurer_idx = header_cells.index("insurer name")
|
|
insurers = []
|
|
for row in rows[1:]:
|
|
cells = row.find_elements(By.TAG_NAME, "td")
|
|
if len(cells) <= insurer_idx:
|
|
continue
|
|
name = cells[insurer_idx].text.strip()
|
|
if name:
|
|
insurers.append(name)
|
|
|
|
third_party = ", ".join(insurers) if insurers else None
|
|
self.extracted_data["third_party_insurance"] = third_party
|
|
print(f"[extract_third_party_insurance] {third_party!r}")
|
|
return third_party
|
|
|
|
print("[extract_third_party_insurance] No 'Third Party Liability' table found")
|
|
return None
|
|
except Exception as e:
|
|
print(f"[extract_third_party_insurance] failed: {e}")
|
|
return None
|
|
|
|
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)
|
|
|
|
self.extract_third_party_insurance()
|
|
|
|
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
|
pdf_filename = f"mh_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"mh_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 → extract remaining → CDP print ─────────────
|
|
|
|
def extract_remaining_amount(self):
|
|
"""Reads the accumulator benefits table and returns the Remaining amount.
|
|
|
|
Layout (see sample PDF "Accumulator Details"):
|
|
Type | Level | Network | Benefit Period | Period | Limit | Applied | Remaining
|
|
Prefers the Type="Maximum", Level="Individual" row; falls back to the
|
|
first data row with a non-empty Remaining cell if that combination
|
|
isn't found (payer may only return one row, or reorder columns).
|
|
"""
|
|
try:
|
|
wait = WebDriverWait(self.driver, 15)
|
|
wait.until(EC.presence_of_element_located((By.TAG_NAME, "table")))
|
|
tables = self.driver.find_elements(By.TAG_NAME, "table")
|
|
|
|
for table in tables:
|
|
rows = table.find_elements(By.TAG_NAME, "tr")
|
|
if not rows:
|
|
continue
|
|
|
|
header_cells = [
|
|
c.text.strip().lower()
|
|
for c in (rows[0].find_elements(By.TAG_NAME, "th") or rows[0].find_elements(By.TAG_NAME, "td"))
|
|
]
|
|
if "remaining" not in header_cells:
|
|
continue
|
|
|
|
remaining_idx = header_cells.index("remaining")
|
|
type_idx = header_cells.index("type") if "type" in header_cells else None
|
|
level_idx = header_cells.index("level") if "level" in header_cells else None
|
|
|
|
fallback_value = None
|
|
best_value = None
|
|
for row in rows[1:]:
|
|
cells = row.find_elements(By.TAG_NAME, "td")
|
|
if len(cells) <= remaining_idx:
|
|
continue
|
|
cell_texts = [c.text.strip() for c in cells]
|
|
value = cell_texts[remaining_idx]
|
|
if not value:
|
|
continue
|
|
if fallback_value is None:
|
|
fallback_value = value
|
|
|
|
type_ok = type_idx is None or "maximum" in cell_texts[type_idx].lower()
|
|
level_ok = level_idx is None or "individual" in cell_texts[level_idx].lower()
|
|
if type_ok and level_ok:
|
|
best_value = value
|
|
break
|
|
|
|
remaining_value = best_value or fallback_value
|
|
if remaining_value:
|
|
self.extracted_data["remaining_amount"] = remaining_value
|
|
print(f"[extract_remaining_amount] Remaining: {remaining_value}")
|
|
return remaining_value
|
|
|
|
print("[extract_remaining_amount] No accumulator table with a 'Remaining' column found")
|
|
return None
|
|
except Exception as e:
|
|
print(f"[extract_remaining_amount] failed: {e}")
|
|
return None
|
|
|
|
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)
|
|
|
|
self.extract_remaining_amount()
|
|
|
|
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
|
pdf_filename = f"mh_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()
|