376 lines
13 KiB
Python
Executable File
376 lines
13 KiB
Python
Executable File
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.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from webdriver_manager.chrome import ChromeDriverManager
|
|
import time
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import base64
|
|
|
|
class AutomationMassHealthEligibilityCheck:
|
|
def __init__(self, data):
|
|
self.headless = False
|
|
self.driver = None
|
|
self.extracted_data = {} # Store extracted data
|
|
|
|
self.data = data.get("data")
|
|
|
|
# Flatten values for convenience
|
|
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", "")
|
|
|
|
# Convert dateOfBirth from YYYY-MM-DD to MMDDYYYY format
|
|
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)
|
|
|
|
def config_driver(self):
|
|
options = webdriver.ChromeOptions()
|
|
if self.headless:
|
|
options.add_argument("--headless")
|
|
|
|
# Add PDF download preferences
|
|
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())
|
|
driver = webdriver.Chrome(service=s, options=options)
|
|
self.driver = driver
|
|
|
|
def login(self):
|
|
wait = WebDriverWait(self.driver, 30)
|
|
|
|
try:
|
|
# Step 1: Click the SIGN IN button on the initial page
|
|
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()
|
|
|
|
# Wait for the new page to load
|
|
time.sleep(3)
|
|
|
|
# Step 2: Enter email on the new login page
|
|
email_field = wait.until(EC.presence_of_element_located((By.ID, "User")))
|
|
email_field.clear()
|
|
email_field.send_keys(self.massdhp_username)
|
|
|
|
# Step 3: Enter password
|
|
password_field = wait.until(EC.presence_of_element_located((By.ID, "Password")))
|
|
password_field.clear()
|
|
password_field.send_keys(self.massdhp_password)
|
|
|
|
# Step 4: Click login button
|
|
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"Error while logging in: {e}")
|
|
return "ERROR:LOGIN FAILED"
|
|
|
|
def step1(self):
|
|
wait = WebDriverWait(self.driver, 30)
|
|
|
|
try:
|
|
# Skip My Account - go directly to Patient Management
|
|
# Step 1: Click Patient Management directly
|
|
patient_mgmt = wait.until(
|
|
EC.presence_of_element_located(
|
|
(By.XPATH, "//strong[@translate='Patient Management']")
|
|
)
|
|
)
|
|
|
|
# Scroll + JS click = fixes most dropdown issues
|
|
self.driver.execute_script("arguments[0].scrollIntoView(true);", patient_mgmt)
|
|
self.driver.execute_script("arguments[0].click();", patient_mgmt)
|
|
time.sleep(2)
|
|
|
|
# Step 3: Click Member Eligibility
|
|
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)
|
|
|
|
# Step 4: Wait for Provider dropdown
|
|
provider_dropdown = wait.until(
|
|
EC.presence_of_element_located((By.NAME, "provider"))
|
|
)
|
|
|
|
# Use Select class (important!)
|
|
select_provider = Select(provider_dropdown)
|
|
|
|
# Step 5: Select correct option
|
|
select_provider.select_by_visible_text(
|
|
"Summit Dental Care - Framingham - Kai Gao"
|
|
)
|
|
time.sleep(2)
|
|
|
|
# Step 6: Member Date Of Birth (SECOND dateInput)
|
|
member_dob = wait.until(
|
|
EC.presence_of_all_elements_located((By.NAME, "dateInput"))
|
|
)[1]
|
|
member_dob.clear()
|
|
member_dob.send_keys(self.dateOfBirth)
|
|
|
|
# Step 7: Member Number
|
|
member_number = wait.until(
|
|
EC.presence_of_element_located((By.NAME, "memberNumber"))
|
|
)
|
|
member_number.clear()
|
|
member_number.send_keys(self.memberId)
|
|
|
|
# Step 8: Click Search
|
|
search_button = wait.until(
|
|
EC.element_to_be_clickable((By.XPATH, "//button[contains(@class,'btn-primary')]"))
|
|
)
|
|
search_button.click()
|
|
|
|
# ✅ CRITICAL WAIT FOR RESULTS
|
|
wait.until(
|
|
EC.presence_of_element_located(
|
|
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']")
|
|
)
|
|
)
|
|
|
|
# ✅ EXTRACT DATA FROM RESULTS PAGE
|
|
self.extracted_data = self._extract_data_from_page()
|
|
print(f"Data extracted in step1: {self.extracted_data}")
|
|
|
|
return "Success"
|
|
except Exception as e:
|
|
print(f"Error while step1: {e}")
|
|
return "ERROR:STEP1"
|
|
|
|
def _extract_data_from_page(self):
|
|
wait = WebDriverWait(self.driver, 5)
|
|
extracted = {}
|
|
|
|
try:
|
|
# Wait until Eligible or Ineligible header appears
|
|
wait.until(
|
|
EC.presence_of_element_located(
|
|
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']")
|
|
)
|
|
)
|
|
|
|
eligible_rows = self.driver.find_elements(
|
|
By.XPATH,
|
|
"//h4[text()='Eligible']/following::table[1]/tbody/tr"
|
|
)
|
|
|
|
if eligible_rows:
|
|
for row in eligible_rows:
|
|
cells = row.find_elements(By.TAG_NAME, "td")
|
|
|
|
if len(cells) < 6:
|
|
continue
|
|
|
|
member_number = cells[2].text.strip()
|
|
|
|
if str(self.memberId) in member_number:
|
|
full_name = cells[4].text.strip()
|
|
plan_name = cells[6].text.strip() if len(cells) > 6 else cells[-1].text.strip()
|
|
|
|
name_parts = full_name.split()
|
|
first_name = name_parts[0] if name_parts else ""
|
|
last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else ""
|
|
|
|
extracted = {
|
|
"eligibility": "Y",
|
|
"firstName": first_name,
|
|
"lastName": last_name,
|
|
"insurance": plan_name
|
|
}
|
|
|
|
return extracted
|
|
|
|
ineligible_rows = self.driver.find_elements(
|
|
By.XPATH,
|
|
"//h4[text()='Ineligible']/following::table[1]/tbody/tr"
|
|
)
|
|
|
|
if ineligible_rows:
|
|
for row in ineligible_rows:
|
|
cells = row.find_elements(By.TAG_NAME, "td")
|
|
|
|
if len(cells) < 5:
|
|
continue
|
|
|
|
member_number = cells[2].text.strip()
|
|
|
|
if str(self.memberId) in member_number:
|
|
full_name = cells[4].text.strip()
|
|
plan_name = cells[5].text.strip()
|
|
|
|
name_parts = full_name.split()
|
|
first_name = name_parts[0] if name_parts else ""
|
|
last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else ""
|
|
|
|
extracted = {
|
|
"eligibility": "N",
|
|
"firstName": first_name,
|
|
"lastName": last_name,
|
|
"insurance": plan_name
|
|
}
|
|
|
|
return extracted
|
|
|
|
return {"eligibility": None}
|
|
|
|
except Exception as e:
|
|
print("Extraction error:", e)
|
|
return {"eligibility": None}
|
|
|
|
def step2(self):
|
|
wait = WebDriverWait(self.driver, 30)
|
|
|
|
try:
|
|
# ✅ USE STORED EXTRACTED DATA
|
|
print(f"Using stored extracted data: {self.extracted_data}")
|
|
|
|
# Step 1: Click Printer Friendly Format
|
|
download_button = wait.until(
|
|
EC.element_to_be_clickable(
|
|
(By.XPATH, "//button[contains(.,'Printer Friendly Format')]")
|
|
)
|
|
)
|
|
original_tab = self.driver.current_window_handle
|
|
download_button.click()
|
|
|
|
# Wait for NEW TAB
|
|
wait.until(lambda d: len(d.window_handles) > 1)
|
|
|
|
# Switch ONLY to new tab
|
|
new_tabs = [tab for tab in self.driver.window_handles if tab != original_tab]
|
|
self.driver.switch_to.window(new_tabs[0])
|
|
|
|
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
|
|
|
# Wait for page to render (CRITICAL)
|
|
wait.until(EC.presence_of_element_located((By.TAG_NAME, "body")))
|
|
time.sleep(2)
|
|
|
|
print("Capturing PDF from:", self.driver.current_url)
|
|
|
|
# Extract data from page BEFORE generating PDF
|
|
extracted_data = self._extract_data_from_page()
|
|
print(f"Extracted data: {extracted_data}")
|
|
|
|
# Step 2: Save PDF using Chrome DevTools (REAL solution)
|
|
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("PDF saved at:", pdf_path)
|
|
|
|
# Return PDF path along with extracted data
|
|
result = {
|
|
"status": "success",
|
|
"pdf_path": pdf_path,
|
|
"file_type": "pdf",
|
|
"message": "PDF captured successfully"
|
|
}
|
|
|
|
# Add stored extracted data to result
|
|
if self.extracted_data:
|
|
result.update(self.extracted_data)
|
|
|
|
return result
|
|
except Exception as e:
|
|
print("PDF capture failed:", e)
|
|
|
|
except Exception as e:
|
|
print("PDF capture failed:", e)
|
|
|
|
# Fallback to screenshot (always keep this)
|
|
try:
|
|
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
|
screenshot_path = os.path.join(self.download_dir, f"eligibility_{safe_member}.png")
|
|
|
|
self.driver.save_screenshot(screenshot_path)
|
|
|
|
print("Screenshot saved at:", screenshot_path)
|
|
|
|
result = {
|
|
"status": "success",
|
|
"pdf_path": screenshot_path,
|
|
"file_type": "screenshot",
|
|
"message": "Screenshot captured (PDF failed)"
|
|
}
|
|
|
|
# Add stored extracted data to result even for screenshots
|
|
if self.extracted_data:
|
|
result.update(self.extracted_data)
|
|
|
|
return result
|
|
|
|
except Exception as ss_error:
|
|
return {
|
|
"status": "error",
|
|
"message": f"PDF + Screenshot failed: {ss_error}"
|
|
}
|
|
|
|
finally:
|
|
if self.driver:
|
|
self.driver.quit()
|
|
|
|
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}
|
|
|
|
step2_result = self.step2()
|
|
if step2_result.get("status") == "error":
|
|
return {"status": "error", "message": step2_result.get("message")}
|
|
|
|
return step2_result
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"message": e
|
|
}
|
|
|
|
finally:
|
|
self.driver.quit()
|
|
|