feat: MassHealth PDF import auto-pays full balance + patient name fix
- PDF import now marks payments as PAID when MassHealth patient's mhPaidAmount >= totalBilled (no patient balance) - Newly created patients from MH vouchers get insuranceProvider = 'MassHealth' - Existing patients with blank insuranceProvider get it filled on import - Fix: update patient name from PDF if existing record has empty name - Various frontend/selenium/route updates Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -10,7 +10,8 @@ from selenium_MH_eligibilityHistoryCheckWorker import AutomationMassHealthEligib
|
||||
from selenium_CMSP_eligibilityHistoryRemainingCheckWorker import AutomationCMSPEligibilityHistoryRemainingCheck
|
||||
from selenium_claimStatusCheckWorker import AutomationMassHealthClaimStatusCheck
|
||||
from selenium_preAuthWorker import AutomationMassHealthPreAuth
|
||||
from selenium_MHPaymentCheckWorker import AutomationMassHealthPaymentCheck
|
||||
from selenium_MHSinglePaymentCheckWorker import AutomationMassHealthSinglePaymentCheck
|
||||
from selenium_MHBatchPaymentCheckWorker import AutomationMassHealthBatchPaymentCheck
|
||||
import os
|
||||
import time
|
||||
import helpers_ddma_eligibility as hddma
|
||||
@@ -290,7 +291,35 @@ async def mh_payment_check(request: Request):
|
||||
waiting_jobs -= 1
|
||||
active_jobs += 1
|
||||
try:
|
||||
bot = AutomationMassHealthPaymentCheck(data)
|
||||
bot = AutomationMassHealthSinglePaymentCheck(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: 6 — Check MassHealth payments in batch by date range
|
||||
@app.post("/mh-batch-payment-check")
|
||||
async def mh_batch_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 = AutomationMassHealthBatchPaymentCheck(data)
|
||||
result = bot.main_workflow("https://provider.masshealth-dental.org/mh_provider_login")
|
||||
|
||||
if result.get("status") != "success":
|
||||
|
||||
@@ -1,218 +0,0 @@
|
||||
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