feat: add Check MH Payment automation and MH Paid column
- 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>
This commit is contained in:
Gitead
@@ -8,6 +8,7 @@ from selenium_claimSubmitWorker import AutomationMassHealthClaimsLogin
|
||||
from selenium_eligibilityCheckWorker import AutomationMassHealthEligibilityCheck
|
||||
from selenium_claimStatusCheckWorker import AutomationMassHealthClaimStatusCheck
|
||||
from selenium_preAuthWorker import AutomationMassHealthPreAuth
|
||||
from selenium_MHPaymentCheckWorker import AutomationMassHealthPaymentCheck
|
||||
import os
|
||||
import time
|
||||
import helpers_ddma_eligibility as hddma
|
||||
@@ -188,6 +189,34 @@ async def start_workflow(request: Request):
|
||||
async with lock:
|
||||
active_jobs -= 1
|
||||
|
||||
# Endpoint: 5 — Check MassHealth payment for a given claim number
|
||||
@app.post("/mh-payment-check")
|
||||
async def mh_payment_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 = AutomationMassHealthPaymentCheck(data)
|
||||
result = bot.main_workflow("https://provider.masshealth-dental.org/mh_provider_login")
|
||||
|
||||
if result.get("status") != "success":
|
||||
return {"status": "error", "message": result.get("message")}
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
finally:
|
||||
async with lock:
|
||||
active_jobs -= 1
|
||||
|
||||
|
||||
# Endpoint: 4 — Start the automation of cheking claim pre auth
|
||||
@app.post("/claim-pre-auth")
|
||||
async def start_workflow(request: Request):
|
||||
|
||||
218
apps/SeleniumService/selenium_MHPaymentCheckWorker.py
Normal file
218
apps/SeleniumService/selenium_MHPaymentCheckWorker.py
Normal file
@@ -0,0 +1,218 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user