1196e2afee
- Add selenium_MHPaymentCheckWorker.py: logs into MassHealth portal, navigates to Search Claims, enters claim number, extracts totalPaidAmount from results table - Register /mh-payment-check endpoint in Selenium agent - Add mhPaidAmount field to Payment model with migration - Add PATCH /api/payments/:id/mh-payment-check backend route: fetches MH credentials, calls selenium, stores result - Add Claim No. column (MassHealth claim number) as first data column in payments table - Move Payment ID and Claim ID columns to end of table - Add MH Paid column showing mhPaidAmount in green - Wire Check MH Payment button to call API for each selected payment and refresh table Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
219 lines
7.9 KiB
Python
219 lines
7.9 KiB
Python
from selenium import webdriver
|
|
from selenium.webdriver.chrome.service import Service
|
|
from selenium.webdriver.common.by import By
|
|
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 base64
|
|
|
|
|
|
class AutomationMassHealthPaymentCheck:
|
|
def __init__(self, data):
|
|
self.headless = False
|
|
self.driver = None
|
|
self.extracted_data = {}
|
|
|
|
self.data = data.get("data")
|
|
|
|
self.massdhp_username = self.data.get("massdhpUsername", "")
|
|
self.massdhp_password = self.data.get("massdhpPassword", "")
|
|
self.claim_number = self.data.get("claimNumber", "")
|
|
|
|
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")
|
|
|
|
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)
|
|
|
|
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()
|
|
|
|
time.sleep(3)
|
|
|
|
# Step 2: Enter username
|
|
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 navigate_to_payments(self):
|
|
"""
|
|
TODO: Navigate to the payments / remittance section after login.
|
|
Inspect the portal and fill in the correct selectors below.
|
|
"""
|
|
wait = WebDriverWait(self.driver, 30)
|
|
substep = "init"
|
|
|
|
try:
|
|
print(f"[navigate_to_payments] URL after login: {self.driver.current_url}")
|
|
|
|
substep = "financial_menu"
|
|
financial_menu = wait.until(
|
|
EC.presence_of_element_located(
|
|
(By.XPATH, '//*[@id="navbar-desktop"]/div/div[3]/div/strong')
|
|
)
|
|
)
|
|
self.driver.execute_script("arguments[0].scrollIntoView(true);", financial_menu)
|
|
self.driver.execute_script("arguments[0].click();", financial_menu)
|
|
time.sleep(2)
|
|
|
|
substep = "payments_link"
|
|
payments_link = wait.until(
|
|
EC.presence_of_element_located(
|
|
(By.XPATH, '//*[@id="navbar-desktop"]/div/div[3]/div/div/div[2]/div/a')
|
|
)
|
|
)
|
|
self.driver.execute_script("arguments[0].click();", payments_link)
|
|
time.sleep(2)
|
|
|
|
return "Success"
|
|
except Exception as e:
|
|
print(f"[navigate_to_payments] FAILED at substep='{substep}': {e}")
|
|
print(f"[navigate_to_payments] URL at failure: {self.driver.current_url}")
|
|
return f"ERROR:NAVIGATE_TO_PAYMENTS:{substep}"
|
|
|
|
def step1_search_claim(self):
|
|
"""Enter claim number and click SEARCH on the Search Claims/Prior Authorizations page."""
|
|
wait = WebDriverWait(self.driver, 30)
|
|
substep = "init"
|
|
|
|
try:
|
|
print(f"[step1] URL: {self.driver.current_url}")
|
|
|
|
substep = "claim_number_input"
|
|
claim_input = wait.until(
|
|
EC.presence_of_element_located(
|
|
(By.XPATH, "/html/body/div[1]/div/div/div/form/fieldset/div[4]/div[2]/input")
|
|
)
|
|
)
|
|
claim_input.clear()
|
|
claim_input.send_keys(self.claim_number)
|
|
print(f"[step1] entered claim number: {self.claim_number}")
|
|
|
|
substep = "search_button"
|
|
search_button = wait.until(
|
|
EC.element_to_be_clickable(
|
|
(By.XPATH, "/html/body/div[1]/div/div/div/form/fieldset/div[7]/div/button[2]")
|
|
)
|
|
)
|
|
self.driver.execute_script("arguments[0].click();", search_button)
|
|
print("[step1] clicked SEARCH")
|
|
time.sleep(3)
|
|
|
|
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}"
|
|
|
|
def step2_extract_paid_amount(self):
|
|
"""Read the totalPaidAmount from the search results table."""
|
|
wait = WebDriverWait(self.driver, 30)
|
|
substep = "init"
|
|
|
|
try:
|
|
substep = "wait_results_table"
|
|
wait.until(
|
|
EC.presence_of_element_located(
|
|
(By.XPATH, '//td[@ng-bind="item.totalPaidAmount | currency"]')
|
|
)
|
|
)
|
|
|
|
substep = "read_paid_amount"
|
|
paid_cell = self.driver.find_element(
|
|
By.XPATH, '//td[@ng-bind="item.totalPaidAmount | currency"]'
|
|
)
|
|
raw_text = paid_cell.text.strip()
|
|
print(f"[step2] raw paid amount text: '{raw_text}'")
|
|
|
|
# Strip currency symbol and commas, e.g. "$1,234.56" → 1234.56
|
|
numeric_str = raw_text.replace("$", "").replace(",", "").strip()
|
|
try:
|
|
paid_amount = float(numeric_str)
|
|
except ValueError:
|
|
paid_amount = 0.0
|
|
print(f"[step2] could not parse '{raw_text}' as float, defaulting to 0.0")
|
|
|
|
return {"status": "success", "mhPaidAmount": paid_amount, "mhPaidAmountRaw": raw_text}
|
|
except Exception as e:
|
|
print(f"[step2] FAILED at substep='{substep}': {e}")
|
|
return f"ERROR:STEP2:{substep}"
|
|
|
|
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}
|
|
|
|
nav_result = self.navigate_to_payments()
|
|
if nav_result.startswith("ERROR"):
|
|
return {"status": "error", "message": nav_result}
|
|
|
|
step1_result = self.step1_search_claim()
|
|
if step1_result.startswith("ERROR"):
|
|
return {"status": "error", "message": step1_result}
|
|
|
|
step2_result = self.step2_extract_paid_amount()
|
|
if isinstance(step2_result, str) and step2_result.startswith("ERROR"):
|
|
return {"status": "error", "message": step2_result}
|
|
|
|
return step2_result
|
|
except Exception as e:
|
|
return {"status": "error", "message": str(e)}
|
|
finally:
|
|
self.driver.quit()
|