Files
DentalManagementMH06/apps/SeleniumService/selenium_DDMA_claimSubmitWorker.py
Gitead 6cfca0d015 feat: select DDMA provider by NPI settings instead of always first
Pass the user's primary NPI provider name through the eligibility and
claim routes so the Selenium workers click the matching option in the
DDMA member-search provider dropdown (data-testid=member-search_provider_select-btn)
rather than always falling back to the first entry.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-11 15:32:01 -04:00

1120 lines
55 KiB
Python

from selenium.common.exceptions import TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time
import os
import json
import base64
from datetime import date
from ddma_browser_manager import get_browser_manager
MEMBERS_URL = "https://providers.deltadentalma.com/members"
# Absolute path to the Backend app root (two levels up from this file's SeleniumService dir)
_SERVICE_DIR = os.path.dirname(os.path.abspath(__file__))
_BACKEND_CWD = os.path.normpath(os.path.join(_SERVICE_DIR, "..", "Backend"))
class AutomationDDMAClaimSubmit:
def __init__(self, data):
self.headless = False
self.driver = None
raw = data if isinstance(data, dict) else {}
claim = raw.get("claim", {}) or {}
patient_name = (claim.get("patientName") or "").strip()
parts = patient_name.split()
first_name = parts[0] if parts else ""
last_name = " ".join(parts[1:]) if len(parts) > 1 else ""
self.massddma_username = claim.get("massddmaUsername", "") or raw.get("massddmaUsername", "")
self.massddma_password = claim.get("massddmaPassword", "") or raw.get("massddmaPassword", "")
self.memberId = claim.get("memberId", "")
self.dateOfBirth = claim.get("dateOfBirth", "")
self.serviceDate = claim.get("serviceDate", "")
self.firstName = claim.get("firstName", "") or first_name
self.lastName = claim.get("lastName", "") or last_name
self.serviceLines = claim.get("serviceLines", []) or []
self.claimFiles = claim.get("claimFiles", []) or []
self.providerName = claim.get("providerName", "") or raw.get("providerName", "")
self.download_dir = get_browser_manager().download_dir
print(f"[DDMA Claim] Init — member={self.memberId}, "
f"patient={self.firstName} {self.lastName}, "
f"lines={len(self.serviceLines)}")
def config_driver(self):
self.driver = get_browser_manager().get_driver(self.headless)
# ------------------------------------------------------------------ #
# Login — identical to DDMA eligibility worker #
# ------------------------------------------------------------------ #
def _force_logout(self):
try:
print("[DDMA Claim login] Forcing logout due to credential change...")
browser_manager = get_browser_manager()
try:
self.driver.get("https://providers.deltadentalma.com/")
time.sleep(2)
for selector in [
"//button[contains(text(), 'Log out') or contains(text(), 'Logout') or contains(text(), 'Sign out')]",
"//a[contains(text(), 'Log out') or contains(text(), 'Logout') or contains(text(), 'Sign out')]",
]:
try:
btn = WebDriverWait(self.driver, 3).until(
EC.element_to_be_clickable((By.XPATH, selector))
)
btn.click()
time.sleep(2)
break
except TimeoutException:
continue
except Exception as e:
print(f"[DDMA Claim login] Could not click logout: {e}")
try:
self.driver.delete_all_cookies()
except Exception:
pass
browser_manager.clear_credentials_hash()
except Exception as e:
print(f"[DDMA Claim login] Error during forced logout: {e}")
def login(self, url):
wait = WebDriverWait(self.driver, 30)
browser_manager = get_browser_manager()
try:
if self.massddma_username and browser_manager.credentials_changed(self.massddma_username):
self._force_logout()
self.driver.get(url)
time.sleep(2)
# Check if already on a logged-in page
try:
current_url = self.driver.current_url
print(f"[DDMA Claim login] Current URL: {current_url}")
logged_in_patterns = ["member", "dashboard", "eligibility", "search", "patients"]
is_logged_in_url = any(p in current_url.lower() for p in logged_in_patterns)
if is_logged_in_url and "onboarding" not in current_url.lower():
# Navigate to members page to confirm
if "member" not in current_url.lower():
try:
WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]'))
)
print("[DDMA Claim login] Already logged in")
return "ALREADY_LOGGED_IN"
except TimeoutException:
self.driver.get(MEMBERS_URL)
time.sleep(2)
try:
WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]'))
)
print("[DDMA Claim login] Already logged in — member search found")
return "ALREADY_LOGGED_IN"
except TimeoutException:
print("[DDMA Claim login] Could not find member search, will try login")
except Exception as e:
print(f"[DDMA Claim login] Error checking current state: {e}")
self.driver.get(url)
time.sleep(2)
# Check if session redirected us straight to member search
try:
current_url = self.driver.current_url
if "onboarding" not in current_url.lower():
member_search = WebDriverWait(self.driver, 3).until(
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]'))
)
if member_search:
print("[DDMA Claim login] Session valid — skipping login")
return "ALREADY_LOGGED_IN"
except TimeoutException:
print("[DDMA Claim login] Proceeding with login form")
# Dismiss any "Authentication flow continued in another tab" modal
modal_dismissed = False
try:
ok_button = WebDriverWait(self.driver, 3).until(
EC.element_to_be_clickable((By.XPATH,
"//button[normalize-space(text())='Ok' or normalize-space(text())='OK']"
))
)
ok_button.click()
print("[DDMA Claim login] Dismissed authentication modal")
modal_dismissed = True
time.sleep(2)
all_windows = self.driver.window_handles
if len(all_windows) > 1:
original_window = self.driver.current_window_handle
for window in all_windows:
if window != original_window:
self.driver.switch_to.window(window)
break
try:
otp_candidate = WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.XPATH,
"//input[contains(@aria-label,'Verification code') or "
"contains(@placeholder,'Enter your verification code')]"
))
)
if otp_candidate:
print("[DDMA Claim login] OTP required (popup)")
return "OTP_REQUIRED"
except TimeoutException:
self.driver.switch_to.window(original_window)
except TimeoutException:
pass
if modal_dismissed:
time.sleep(2)
try:
WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="Search by member ID"]'))
)
print("[DDMA Claim login] Already authenticated after modal dismiss")
return "ALREADY_LOGGED_IN"
except TimeoutException:
pass
# Fill login form
try:
email_field = WebDriverWait(self.driver, 10).until(
EC.element_to_be_clickable((By.XPATH, "//input[@name='username' and @type='text']"))
)
except TimeoutException:
return "ERROR: Login form not found"
email_field.clear()
email_field.send_keys(self.massddma_username)
password_field = wait.until(
EC.presence_of_element_located((By.XPATH, "//input[@name='password' and @type='password']"))
)
password_field.clear()
password_field.send_keys(self.massddma_password)
try:
remember_me = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//label[.//span[contains(text(),'Remember me')]]")
))
remember_me.click()
except Exception:
print("[DDMA Claim login] Remember me checkbox not found (continuing)")
login_button = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//button[@type='submit' and @aria-label='Sign in']")
))
login_button.click()
if self.massddma_username:
browser_manager.save_credentials_hash(self.massddma_username)
# OTP detection
try:
otp_candidate = WebDriverWait(self.driver, 30).until(
EC.presence_of_element_located((By.XPATH,
"//input[contains(@aria-lable,'Verification code') or "
"contains(@placeholder,'Enter your verification code')]"
))
)
if otp_candidate:
print("[DDMA Claim login] OTP required")
return "OTP_REQUIRED"
except TimeoutException:
try:
current_url = self.driver.current_url.lower()
if "member" in current_url or "dashboard" in current_url:
print("[DDMA Claim login] Login succeeded without OTP")
return "SUCCESS"
except Exception:
pass
if "onboarding" in self.driver.current_url.lower() or "login" in self.driver.current_url.lower():
return "ERROR: LOGIN FAILED: Still on login/onboarding page"
print("[DDMA Claim login] Assuming login succeeded")
return "SUCCESS"
except Exception as e:
print(f"[DDMA Claim login] Exception: {e}")
return f"ERROR:LOGIN FAILED: {e}"
def _select_provider_dropdown(self):
"""
Click the provider dropdown on the member search page and select the option
matching self.providerName (case-insensitive). Falls back to the first option.
The button data-testid is 'member-search_provider_select-btn'.
"""
try:
short_wait = WebDriverWait(self.driver, 5)
try:
provider_btn = short_wait.until(
EC.element_to_be_clickable(
(By.XPATH, '//button[@data-testid="member-search_provider_select-btn"]')
)
)
except TimeoutException:
print("[DDMA Claim step1] No provider dropdown found — skipping")
return
provider_btn.click()
time.sleep(0.5)
try:
WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, "//*[@role='option']"))
)
except TimeoutException:
print("[DDMA Claim step1] Provider listbox did not open")
return
options = self.driver.find_elements(By.XPATH, "//*[@role='option']")
print(f"[DDMA Claim step1] Provider options: {[o.text.strip() for o in options]}")
target = (self.providerName or "").strip().lower()
selected = False
if target:
for opt in options:
if target in opt.text.lower():
opt.click()
print(f"[DDMA Claim step1] Selected provider: '{opt.text.strip()}'")
selected = True
break
if not selected and options:
options[0].click()
print(f"[DDMA Claim step1] No match for '{self.providerName}', selected first: '{options[0].text.strip()}'")
time.sleep(0.3)
except Exception as e:
print(f"[DDMA Claim step1] Provider selection error (non-fatal): {e}")
# ------------------------------------------------------------------ #
# Step 1 — Navigate directly to /members then search patient #
# (same as eligibility — bypasses onboarding date/location screen) #
# ------------------------------------------------------------------ #
def step1_search_patient(self):
"""Search for the patient — identical to DDMA eligibility step1.
Does NOT navigate: login already left the browser on the search page."""
wait = WebDriverWait(self.driver, 30)
def replace_with_sendkeys(el, value):
el.click()
el.send_keys(Keys.CONTROL, "a")
el.send_keys(Keys.BACKSPACE)
el.send_keys(value)
try:
print(f"[DDMA Claim step1] Current URL: {self.driver.current_url}")
print(f"[DDMA Claim step1] Waiting for member search input...")
# Select provider from dropdown based on NPI settings
self._select_provider_dropdown()
# Fill Member ID
if self.memberId:
try:
member_id_input = wait.until(EC.presence_of_element_located(
(By.XPATH, '//input[@placeholder="Search by member ID"]')
))
member_id_input.clear()
member_id_input.send_keys(self.memberId)
print(f"[DDMA Claim step1] Entered Member ID: {self.memberId}")
time.sleep(0.2)
except Exception as e:
print(f"[DDMA Claim step1] Warning: Could not fill Member ID: {e}")
# Fill DOB
if self.dateOfBirth:
try:
dob_parts = self.dateOfBirth.split("-")
year = dob_parts[0]
month = dob_parts[1].zfill(2)
day = dob_parts[2].zfill(2)
dob_container = wait.until(EC.presence_of_element_located(
(By.XPATH, "//div[@data-testid='member-search_date-of-birth']")
))
month_elem = dob_container.find_element(By.XPATH, ".//span[@data-type='month' and @contenteditable='true']")
day_elem = dob_container.find_element(By.XPATH, ".//span[@data-type='day' and @contenteditable='true']")
year_elem = dob_container.find_element(By.XPATH, ".//span[@data-type='year' and @contenteditable='true']")
replace_with_sendkeys(month_elem, month)
time.sleep(0.05)
replace_with_sendkeys(day_elem, day)
time.sleep(0.05)
replace_with_sendkeys(year_elem, year)
print(f"[DDMA Claim step1] Filled DOB: {month}/{day}/{year}")
except Exception as e:
print(f"[DDMA Claim step1] Warning: Could not fill DOB: {e}")
time.sleep(0.3)
# Click Search
search_btn = wait.until(EC.element_to_be_clickable(
(By.XPATH, '//button[@data-testid="member-search_search-button"]')
))
search_btn.click()
print("[DDMA Claim step1] Clicked Search button")
# Wait for the member search result page to load
WebDriverWait(self.driver, 15).until(
EC.any_of(
EC.presence_of_element_located((By.XPATH, "//tbody//tr")),
EC.presence_of_element_located((By.XPATH, '//div[@data-testid="member-search-result-no-results"]')),
)
)
time.sleep(4)
try:
no_results = self.driver.find_element(By.XPATH, '//div[@data-testid="member-search-result-no-results"]')
if no_results:
return "ERROR: No patient found with given search criteria"
except Exception:
pass
print("[DDMA Claim step1] Search completed")
return "SUCCESS"
except Exception as e:
print(f"[DDMA Claim step1] Exception: {e}")
return f"ERROR: step1 failed: {e}"
# ------------------------------------------------------------------ #
# Step 2 — Click patient name → Member Information page #
# ------------------------------------------------------------------ #
def step2_open_member_page(self):
"""Navigate to member detail page — same approach as DDMA eligibility step2."""
try:
# Wait for the results table to appear, then let it stabilize
try:
WebDriverWait(self.driver, 20).until(
EC.presence_of_element_located((By.XPATH, "//tbody//tr"))
)
time.sleep(3) # wait for the list to fully stabilize
except TimeoutException:
print("[DDMA Claim step2] Warning: Results table not found within timeout")
# Wait until the patient name link in the first row is clickable
detail_url = None
PATIENT_LINK_XPATHS = [
"(//table//tbody//tr)[1]//td[1]//a",
"(//tbody//tr)[1]//a[contains(@href,'member-details')]",
"(//tbody//tr)[1]//a[contains(@href,'member')]",
"//a[contains(@href,'member-details')]",
]
for selector in PATIENT_LINK_XPATHS:
try:
link_el = WebDriverWait(self.driver, 20).until(
EC.element_to_be_clickable((By.XPATH, selector))
)
href = link_el.get_attribute("href")
if href and "member-details" in href:
detail_url = href
print(f"[DDMA Claim step2] Patient link is clickable: {href}")
break
except Exception:
continue
if not detail_url:
return "ERROR: step2 failed: could not find member-details link"
self.driver.get(detail_url)
print(f"[DDMA Claim step2] Navigating to: {detail_url}")
try:
WebDriverWait(self.driver, 15).until(
lambda d: "member-details" in d.current_url
)
print(f"[DDMA Claim step2] Member Information page loaded: {self.driver.current_url}")
except TimeoutException:
print(f"[DDMA Claim step2] Warning — URL: {self.driver.current_url}")
try:
# Wait until Create claim button is visible AND enabled (not just in the DOM)
WebDriverWait(self.driver, 15).until(
EC.element_to_be_clickable((By.XPATH, "//button[@aria-label='Create claim']"))
)
print("[DDMA Claim step2] 'Create claim' button is clickable")
time.sleep(3) # let React finish any remaining re-renders
except TimeoutException:
print("[DDMA Claim step2] Warning: 'Create claim' button not clickable in time")
time.sleep(2)
return "SUCCESS"
except Exception as e:
print(f"[DDMA Claim step2] Exception: {e}")
return f"ERROR: step2 failed: {e}"
# ------------------------------------------------------------------ #
# Step 3 — Click "Create claim" button #
# ------------------------------------------------------------------ #
def step3_click_create_claim(self):
"""Click the 'Create claim' button on the Member Information page."""
try:
print(f"[DDMA Claim step3] Current URL: {self.driver.current_url}")
handles_before = set(self.driver.window_handles)
url_before = self.driver.current_url
self.driver.execute_script("window.scrollTo(0, 0);")
time.sleep(0.5)
# Re-find the button fresh each attempt to avoid stale element references.
# The page may re-render after step2, so we do not cache the element.
CLAIM_BTN_XPATHS = [
"//button[@aria-label='Create claim']",
"//button[contains(normalize-space(text()),'Create claim')]",
"//button[.//span[contains(text(),'Create claim')]]",
]
btn = None
for xpath in CLAIM_BTN_XPATHS:
try:
btn = WebDriverWait(self.driver, 8).until(
EC.element_to_be_clickable((By.XPATH, xpath))
)
if btn:
print(f"[DDMA Claim step3] Found 'Create claim' via: {xpath}")
break
except Exception:
continue
if not btn:
all_btns = self.driver.find_elements(By.XPATH, "//button")
print(f"[DDMA Claim step3] Buttons on page: {[b.get_attribute('aria-label') or b.text for b in all_btns]}")
return "ERROR: step3 failed: 'Create claim' button not found"
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", btn)
time.sleep(0.5)
# Try direct Selenium click first, fall back to JS events
clicked = False
for attempt, method in enumerate(["selenium", "js_events", "js_click"]):
try:
# Re-find fresh on each attempt to avoid stale reference
for xpath in CLAIM_BTN_XPATHS:
try:
btn = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.XPATH, xpath))
)
if btn:
break
except Exception:
continue
if method == "selenium":
btn.click()
elif method == "js_events":
self.driver.execute_script("""
var el = arguments[0];
el.dispatchEvent(new PointerEvent('pointerover', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new PointerEvent('pointerdown', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new PointerEvent('pointerup', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new MouseEvent('click', {bubbles:true, cancelable:true, composed:true}));
""", btn)
else:
self.driver.execute_script("arguments[0].click();", btn)
print(f"[DDMA Claim step3] Clicked via {method} (attempt {attempt+1})")
time.sleep(2)
# Check if navigation happened (URL changed or new tab opened)
handles_after = set(self.driver.window_handles)
new_handles = handles_after - handles_before
if new_handles:
self.driver.switch_to.window(new_handles.pop())
print(f"[DDMA Claim step3] Switched to new tab: {self.driver.current_url}")
clicked = True
break
if self.driver.current_url != url_before:
print(f"[DDMA Claim step3] URL changed to: {self.driver.current_url}")
clicked = True
break
print(f"[DDMA Claim step3] URL unchanged after {method}, retrying...")
except Exception as e:
print(f"[DDMA Claim step3] {method} click failed: {e}, retrying...")
if not clicked:
page_text = self.driver.execute_script("return document.body.innerText;")[:300]
print(f"[DDMA Claim step3] No navigation after all click attempts — page: {page_text}")
return "ERROR: step3 failed: button clicked but no navigation occurred"
# Wait for claim form to load
try:
WebDriverWait(self.driver, 20).until(
EC.any_of(
EC.presence_of_element_located((By.XPATH, "//input[contains(@id,'procedureCode')]")),
EC.presence_of_element_located((By.XPATH, "//span[@data-type='month' and @contenteditable='true']")),
EC.presence_of_element_located((By.XPATH,
"//*[contains(text(),'date of service') or contains(text(),'Date of service') "
"or contains(text(),'Procedure code')]"
)),
)
)
print(f"[DDMA Claim step3] Claim form loaded: {self.driver.current_url}")
except TimeoutException:
page_text = self.driver.execute_script("return document.body.innerText;")[:400]
print(f"[DDMA Claim step3] Claim form not detected — page: {page_text}")
time.sleep(1)
return "SUCCESS"
except Exception as e:
print(f"[DDMA Claim step3] Exception: {e}")
return f"ERROR: step3 failed: {e}"
# ------------------------------------------------------------------ #
# Step 4 — Fill service date and procedure code #
# ------------------------------------------------------------------ #
def _parse_service_date(self):
s = str(self.serviceDate or "").strip()
if not s:
return None, None, None
if "-" in s:
parts = s.split("-")
if len(parts) == 3 and len(parts[0]) == 4:
return parts[1].zfill(2), parts[2].zfill(2), parts[0]
if len(parts) == 3 and len(parts[2]) == 4:
return parts[0].zfill(2), parts[1].zfill(2), parts[2]
if "/" in s:
parts = s.split("/")
if len(parts) == 3:
return parts[0].zfill(2), parts[1].zfill(2), parts[2]
return None, None, None
def _fill_spinbutton(self, label_fragment, value):
for sel in [
f"//span[@contenteditable='true' and contains(@aria-label,'{label_fragment}')]",
f"//span[@data-type='{label_fragment}' and @contenteditable='true']",
]:
try:
elem = WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, sel))
)
elem.click()
elem.send_keys(Keys.CONTROL, "a")
elem.send_keys(Keys.BACKSPACE)
elem.send_keys(value)
time.sleep(0.1)
print(f"[DDMA Claim step4] Filled spinbutton '{label_fragment}' = {value!r}")
return True
except Exception:
continue
print(f"[DDMA Claim step4] Warning: spinbutton '{label_fragment}' not found")
return False
def _fill_combobox(self, inp, value, label="field"):
"""Type value into a combobox and select the first matching dropdown option."""
try:
inp.click()
inp.send_keys(Keys.CONTROL + "a")
inp.send_keys(Keys.DELETE)
inp.send_keys(str(value))
time.sleep(0.5)
# Try to click matching option in listbox
listbox_id = inp.get_attribute("aria-controls") or ""
try:
if listbox_id:
option = WebDriverWait(self.driver, 4).until(
EC.element_to_be_clickable((By.XPATH,
f"//*[@id='{listbox_id}']//*[@role='option'][1]"
))
)
else:
option = WebDriverWait(self.driver, 4).until(
EC.element_to_be_clickable((By.XPATH,
f"//*[@role='listbox']//*[@role='option' and contains(normalize-space(.),'{value}')]"
f" | //*[@role='listbox']//*[@role='option'][1]"
))
)
option.click()
print(f"[DDMA Claim step4] {label}: selected '{value}'")
except TimeoutException:
# No dropdown — press Enter or Tab to confirm free text
inp.send_keys(Keys.TAB)
print(f"[DDMA Claim step4] {label}: typed '{value}' (no dropdown)")
except Exception as e:
print(f"[DDMA Claim step4] Warning: could not fill {label}: {e}")
def _fill_text_input(self, inp, value, label="field"):
"""Clear and type value into a plain text input."""
try:
inp.click()
inp.send_keys(Keys.CONTROL + "a")
inp.send_keys(Keys.DELETE)
inp.send_keys(str(value))
time.sleep(0.1)
print(f"[DDMA Claim step4] {label}: typed '{value}'")
except Exception as e:
print(f"[DDMA Claim step4] Warning: could not fill {label}: {e}")
def step4_fill_claim_form(self):
"""Fill service date then all procedure line fields."""
try:
month, day, year = self._parse_service_date()
# ── Service date (once, at the top of the form) ──────────────────
if month and day and year:
print(f"[DDMA Claim step4] Filling service date: {month}/{day}/{year}")
try:
dos_container = WebDriverWait(self.driver, 8).until(
EC.presence_of_element_located((By.XPATH,
"//*[@data-testid and contains(@data-testid,'date-of-service')] | "
"//*[contains(@aria-label,'Select date of service')]/ancestor::div[1]"
))
)
month_el = dos_container.find_element(By.XPATH, ".//span[@data-type='month' and @contenteditable='true']")
day_el = dos_container.find_element(By.XPATH, ".//span[@data-type='day' and @contenteditable='true']")
year_el = dos_container.find_element(By.XPATH, ".//span[@data-type='year' and @contenteditable='true']")
for elem, val in [(month_el, month), (day_el, day), (year_el, year)]:
elem.click()
elem.send_keys(Keys.CONTROL, "a")
elem.send_keys(Keys.BACKSPACE)
elem.send_keys(val)
time.sleep(0.05)
print("[DDMA Claim step4] Service date filled")
except Exception:
self._fill_spinbutton("month", month)
self._fill_spinbutton("day", day)
self._fill_spinbutton("year", year)
else:
print(f"[DDMA Claim step4] No valid service date: {self.serviceDate!r}")
time.sleep(0.3)
active_lines = [ln for ln in self.serviceLines if str(ln.get("procedureCode") or "").strip()]
print(f"[DDMA Claim step4] {len(active_lines)} service line(s)")
for idx, line in enumerate(active_lines):
code = str(line.get("procedureCode") or "").strip().upper()
tooth = str(line.get("toothNumber") or line.get("tooth") or "").strip()
arch = str(line.get("arch") or "").strip()
quad = str(line.get("quad") or line.get("quadrant") or "").strip()
surface = str(line.get("toothSurface") or line.get("surface") or "").strip().upper()
billed = str(line.get("totalBilled") or line.get("billedAmount") or line.get("fee") or "").strip()
billed = billed.replace("$", "").strip()
print(f"[DDMA Claim step4] Line {idx}: code={code} tooth={tooth!r} arch={arch!r} "
f"quad={quad!r} surface={surface!r} billed={billed!r}")
# ── Click "Add a procedure" for lines after the first ─────────
if idx > 0:
try:
add_btn = WebDriverWait(self.driver, 10).until(
EC.element_to_be_clickable((By.XPATH,
"//button[.//span[contains(text(),'Add a procedure')]] | "
"//button[contains(normalize-space(text()),'Add a procedure')] | "
"//*[contains(text(),'Add a procedure') and @role='button']"
))
)
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", add_btn)
add_btn.click()
print(f"[DDMA Claim step4] Clicked 'Add a procedure' for line {idx}")
time.sleep(1)
except Exception as e:
print(f"[DDMA Claim step4] Could not click 'Add a procedure': {e}")
# ── Procedure code ────────────────────────────────────────────
if code:
try:
proc_inputs = self.driver.find_elements(By.XPATH,
"//input[contains(@id,'procedureCode') and contains(@id,'-input')]"
)
proc_inp = proc_inputs[idx] if idx < len(proc_inputs) else proc_inputs[-1]
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", proc_inp)
self._fill_combobox(proc_inp, code, f"procedureCode[{idx}]")
# Wait for the tooth input on the second row to be present AND interactable
if tooth:
try:
WebDriverWait(self.driver, 8).until(
lambda d: (
len(d.find_elements(By.XPATH, "//input[@aria-label='Tooth']")) > idx and
d.find_elements(By.XPATH, "//input[@aria-label='Tooth']")[idx].is_enabled() and
d.find_elements(By.XPATH, "//input[@aria-label='Tooth']")[idx].is_displayed()
)
)
except Exception:
time.sleep(1) # fallback: just wait a second
except Exception as e:
print(f"[DDMA Claim step4] Could not fill procedure code: {e}")
# ── Tooth ─────────────────────────────────────────────────────
if tooth:
try:
tooth_inputs = self.driver.find_elements(By.XPATH, "//input[@aria-label='Tooth']")
if idx < len(tooth_inputs):
ti = tooth_inputs[idx]
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", ti)
time.sleep(0.3)
listbox_id = ti.get_attribute("aria-controls") or ""
# Click the "^" chevron toggle button to open the dropdown.
# It lives in the same container as the tooth input —
# try progressively wider ancestor scopes.
toggle_clicked = False
for ancestor_steps in range(1, 6):
ancestor_axis = "/".join([".."] * ancestor_steps)
toggle_xpath = (
f"//input[@aria-label='Tooth'][{idx+1}]"
f"/{ancestor_axis}//button"
)
toggles = self.driver.find_elements(By.XPATH, toggle_xpath)
if toggles:
try:
self.driver.execute_script("arguments[0].click();", toggles[0])
print(f"[DDMA Claim step4] tooth[{idx}]: clicked toggle btn (ancestor depth {ancestor_steps})")
toggle_clicked = True
break
except Exception:
continue
if not toggle_clicked:
# Fallback: JS focus the input to open the combobox
self.driver.execute_script("arguments[0].focus();", ti)
print(f"[DDMA Claim step4] tooth[{idx}]: no toggle btn found, used JS focus")
time.sleep(0.5)
# Select the matching tooth number from the 1-32 listbox
try:
if listbox_id:
option = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.XPATH,
f"//*[@id='{listbox_id}']//*[@role='option' and normalize-space(.)='{tooth}']"
))
)
else:
option = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.XPATH,
f"//*[@role='listbox']//*[@role='option' and normalize-space(.)='{tooth}']"
))
)
option.click()
print(f"[DDMA Claim step4] tooth[{idx}]: selected '{tooth}' from dropdown")
except TimeoutException:
print(f"[DDMA Claim step4] tooth[{idx}]: option '{tooth}' not found in dropdown")
time.sleep(0.3)
except Exception as e:
print(f"[DDMA Claim step4] Could not fill tooth: {e}")
# ── Arch ──────────────────────────────────────────────────────
if arch:
try:
arch_inputs = self.driver.find_elements(By.XPATH, "//input[@aria-label='Arch']")
if idx < len(arch_inputs):
self._fill_combobox(arch_inputs[idx], arch, f"arch[{idx}]")
time.sleep(0.3)
except Exception as e:
print(f"[DDMA Claim step4] Could not fill arch: {e}")
# ── Quad ──────────────────────────────────────────────────────
if quad:
try:
quad_inputs = self.driver.find_elements(By.XPATH, "//input[@aria-label='Quad']")
if idx < len(quad_inputs):
self._fill_combobox(quad_inputs[idx], quad, f"quad[{idx}]")
time.sleep(0.3)
except Exception as e:
print(f"[DDMA Claim step4] Could not fill quad: {e}")
# ── Surface (type directly then Tab to billed amount) ─────────
if surface:
try:
surface_inputs = self.driver.find_elements(By.XPATH, "//input[@aria-label='Surface']")
if idx < len(surface_inputs):
surf_inp = surface_inputs[idx]
surf_inp.click()
surf_inp.send_keys(Keys.CONTROL + "a")
surf_inp.send_keys(Keys.DELETE)
surf_inp.send_keys(surface)
print(f"[DDMA Claim step4] surface[{idx}]: typed '{surface}'")
time.sleep(0.2)
except Exception as e:
print(f"[DDMA Claim step4] Could not fill surface: {e}")
# ── Billed amount ─────────────────────────────────────────────
if billed:
try:
billed_inputs = self.driver.find_elements(By.XPATH,
"//input[@aria-label='Enter billed amount']"
)
if idx < len(billed_inputs):
self._fill_text_input(billed_inputs[idx], billed, f"billedAmount[{idx}]")
time.sleep(0.2)
except Exception as e:
print(f"[DDMA Claim step4] Could not fill billed amount: {e}")
print("[DDMA Claim step4] Done")
return "SUCCESS"
except Exception as e:
print(f"[DDMA Claim step4] Exception: {e}")
return f"ERROR: step4 failed: {e}"
# ------------------------------------------------------------------ #
# Step 5 — Attach files #
# ------------------------------------------------------------------ #
def step5_attach_files(self):
"""For each claimFile with a filePath, click 'Add a file' and upload it."""
if not self.claimFiles:
print("[DDMA Claim step5] No files to attach")
return "SUCCESS"
attached = 0
for cf in self.claimFiles:
relative_path = cf.get("filePath") or ""
if not relative_path:
print(f"[DDMA Claim step5] Skipping file with no filePath: {cf}")
continue
# Build absolute path — filePath is like /uploads/patients/Name/file.pdf
abs_path = os.path.normpath(os.path.join(_BACKEND_CWD, relative_path.lstrip("/")))
if not os.path.isfile(abs_path):
print(f"[DDMA Claim step5] File not found on disk: {abs_path}")
continue
print(f"[DDMA Claim step5] Attaching: {abs_path}")
try:
# Click "Add a file" button/link
add_file_btn = WebDriverWait(self.driver, 10).until(
EC.element_to_be_clickable((By.XPATH,
"//button[.//span[contains(text(),'Add a file')]] | "
"//button[contains(normalize-space(text()),'Add a file')] | "
"//*[contains(text(),'Add a file') and (@role='button' or self::label)]"
))
)
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", add_file_btn)
ActionChains(self.driver).move_to_element(add_file_btn).click().perform()
time.sleep(1)
# Find the file input that appeared (may be hidden)
file_input = WebDriverWait(self.driver, 8).until(
EC.presence_of_element_located((By.XPATH, "//input[@type='file']"))
)
# Make hidden inputs interactable
self.driver.execute_script("arguments[0].style.display='block';", file_input)
file_input.send_keys(abs_path)
time.sleep(1.5)
print(f"[DDMA Claim step5] Attached: {os.path.basename(abs_path)}")
attached += 1
except Exception as e:
print(f"[DDMA Claim step5] Could not attach {abs_path}: {e}")
print(f"[DDMA Claim step5] Attached {attached}/{len(self.claimFiles)} file(s)")
return "SUCCESS"
# ------------------------------------------------------------------ #
# Step 6 — Click Next step #
# ------------------------------------------------------------------ #
def step6_click_next(self):
"""Click the 'Next step' button (React Aria — dispatches pointer events directly)."""
try:
print(f"[DDMA Claim step6] Current URL: {self.driver.current_url}")
btn = WebDriverWait(self.driver, 15).until(
EC.element_to_be_clickable((By.XPATH,
"//button[@data-testid='next-step-btn'] | "
"//button[@aria-label='Next step']"
))
)
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", btn)
time.sleep(0.5)
self.driver.execute_script("""
var el = arguments[0];
el.dispatchEvent(new PointerEvent('pointerover', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new PointerEvent('pointerdown', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new PointerEvent('pointerup', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new MouseEvent('click', {bubbles:true, cancelable:true, composed:true}));
""", btn)
print(f"[DDMA Claim step6] Clicked 'Next step'")
time.sleep(2)
print(f"[DDMA Claim step6] URL after Next: {self.driver.current_url}")
return "SUCCESS"
except Exception as e:
print(f"[DDMA Claim step6] Exception: {e}")
return f"ERROR: step6 failed: {e}"
# ------------------------------------------------------------------ #
# Step 7 — Claims summary: check acknowledgement + submit #
# ------------------------------------------------------------------ #
def step7_submit_claim(self):
"""On the claims summary page, tick the acknowledgement checkbox then click Submit claim."""
try:
print(f"[DDMA Claim step7] Current URL: {self.driver.current_url}")
# Wait for the acknowledgement checkbox to appear
checkbox = WebDriverWait(self.driver, 15).until(
EC.presence_of_element_located((By.XPATH,
"//input[@type='checkbox'] | "
"//*[@role='checkbox'] | "
"//label[contains(.,'submitting this claim')]//input | "
"//*[contains(@aria-label,'submitting this claim')]"
))
)
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", checkbox)
time.sleep(0.3)
self.driver.execute_script("""
var el = arguments[0];
el.dispatchEvent(new PointerEvent('pointerover', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new PointerEvent('pointerdown', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new PointerEvent('pointerup', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new MouseEvent('click', {bubbles:true, cancelable:true, composed:true}));
""", checkbox)
print("[DDMA Claim step7] Checked acknowledgement checkbox")
time.sleep(0.5)
# Click Submit claim button
submit_btn = None
for xpath in [
"//button[.//span[contains(text(),'Submit claim')]]",
"//button[contains(normalize-space(text()),'Submit claim')]",
"//button[@aria-label='Submit claim']",
]:
try:
submit_btn = WebDriverWait(self.driver, 8).until(
EC.element_to_be_clickable((By.XPATH, xpath))
)
if submit_btn:
break
except Exception:
continue
if not submit_btn:
return "ERROR: step7 failed: Submit claim button not found"
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", submit_btn)
time.sleep(0.3)
self.driver.execute_script("""
var el = arguments[0];
el.dispatchEvent(new PointerEvent('pointerover', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new PointerEvent('pointerdown', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new PointerEvent('pointerup', {bubbles:true, cancelable:true, composed:true}));
el.dispatchEvent(new MouseEvent('click', {bubbles:true, cancelable:true, composed:true}));
""", submit_btn)
print("[DDMA Claim step7] Clicked 'Submit claim'")
time.sleep(2)
print(f"[DDMA Claim step7] URL after submit: {self.driver.current_url}")
return "SUCCESS"
except Exception as e:
print(f"[DDMA Claim step7] Exception: {e}")
return f"ERROR: step7 failed: {e}"
def step8_save_confirmation_pdf(self):
"""Wait for the Thank-you page, extract the claim number, save page as PDF."""
import re
try:
# Wait for confirmation page
WebDriverWait(self.driver, 30).until(
lambda d: "thank" in d.page_source.lower() or "submitted claim" in d.page_source.lower()
)
time.sleep(2)
print(f"[DDMA Claim step8] Confirmation page URL: {self.driver.current_url}")
# Extract claim number from page text
claim_number = None
try:
body_text = self.driver.find_element(By.TAG_NAME, "body").text
match = re.search(r'submitted claim\s+(\d{10,})', body_text, re.IGNORECASE)
if match:
claim_number = match.group(1)
print(f"[DDMA Claim step8] Extracted claim number: {claim_number}")
else:
# Fallback: any long digit sequence
match = re.search(r'\b(\d{12,})\b', body_text)
if match:
claim_number = match.group(1)
print(f"[DDMA Claim step8] Extracted claim number (fallback): {claim_number}")
except Exception as e:
print(f"[DDMA Claim step8] Could not extract claim number: {e}")
# Save page as PDF via Chrome DevTools Protocol
# Use the shared 'downloads/' dir that agent.py serves as static files
shared_downloads = os.path.join(_SERVICE_DIR, "downloads")
os.makedirs(shared_downloads, exist_ok=True)
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
safe_claim = ("_" + claim_number[:20]) if claim_number else ""
timestamp = time.strftime("%Y%m%d_%H%M%S")
pdf_filename = f"ddma_claim_confirmation_{safe_member}{safe_claim}_{timestamp}.pdf"
pdf_path = os.path.join(shared_downloads, pdf_filename)
try:
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {
"printBackground": True,
"paperWidth": 8.5,
"paperHeight": 11,
"marginTop": 0.4,
"marginBottom": 0.4,
"marginLeft": 0.4,
"marginRight": 0.4,
})
pdf_bytes = base64.b64decode(pdf_data["data"])
with open(pdf_path, "wb") as f:
f.write(pdf_bytes)
print(f"[DDMA Claim step8] PDF saved: {pdf_path}")
except Exception as e:
print(f"[DDMA Claim step8] PDF capture failed: {e}")
return f"ERROR: step8 PDF failed: {e}"
return {
"status": "success",
"pdf_path": pdf_path,
"claimNumber": claim_number,
}
except Exception as e:
print(f"[DDMA Claim step8] Exception: {e}")
return f"ERROR: step8 failed: {e}"
# ------------------------------------------------------------------ #
# Fee schedule helpers #
# ------------------------------------------------------------------ #
def _load_ddma_fee_schedule(self):
base = os.path.dirname(os.path.abspath(__file__))
json_path = os.path.join(base, "..", "Frontend", "src", "assets", "data", "procedureCodesDDMA.json")
try:
with open(json_path) as f:
rows = json.load(f)
fee_map = {}
for row in rows:
code = str(row.get("Procedure Code", "")).strip().upper()
if code:
fee_map[code] = row
print(f"[DDMA Claim] Loaded {len(fee_map)} fee codes")
return fee_map
except Exception as e:
print(f"[DDMA Claim] Could not load fee schedule: {e}")
return {}
def _get_patient_age(self):
if not self.dateOfBirth:
return None
try:
parts = self.dateOfBirth.split("-")
dob = date(int(parts[0]), int(parts[1]), int(parts[2]))
today = date.today()
return today.year - dob.year - ((today.month, today.day) < (dob.month, dob.day))
except Exception:
return None
def _get_fee(self, code, fee_map, age):
row = fee_map.get(str(code).strip().upper(), {})
if not row:
return ""
if "Price" in row:
val = str(row["Price"])
elif age is not None and age <= 21:
val = str(row.get("PriceLTEQ21") or row.get("PriceGT21") or "")
else:
val = str(row.get("PriceGT21") or row.get("PriceLTEQ21") or "")
return "" if val in ("NC", "IC", "None", "") else val