Files
DentalManagement09/apps/SeleniumService/selenium_DDMA_claimSubmitWorker.py
T
GiteadandClaude Sonnet 5 7b8a7af5d7 fix: update DDMA claim worker for new provider portal
DDMA redesigned their claims-submission flow, so the worker now navigates
via the dashboard's "Submit Claims" quick link, the "Member Search" and
"Claim Detail" sidebar sections, checkbox-based member selection, indexed
treatment-row fields, a multi-file upload input, and the new confirmation
page layout.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-18 14:36:14 -04:00

853 lines
41 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 re
import json
import base64
import calendar
from datetime import date
from ddma_browser_manager import get_browser_manager
MEMBERS_URL = "https://www.deltadentalma.com/private/provider/dashboard"
# 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://www.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", "private/provider"]
is_logged_in_url = any(p in current_url.lower() for p in logged_in_patterns)
if is_logged_in_url and "sign-in" 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, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
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, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
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 "sign-in" not in current_url.lower():
member_search = WebDriverWait(self.driver, 3).until(
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
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[@data-test-id='MfaCode']"
))
)
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, "//a[contains(@href, 'check-patient-benefits-search')]"))
)
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[@data-test-id='Username']"))
)
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[@data-test-id='Password']"))
)
password_field.clear()
password_field.send_keys(self.massddma_password)
login_button = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//button[@data-test-id='SigninSubmitButton']")
))
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[@data-test-id='MfaCode']"
))
)
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 "sign-in" in self.driver.current_url.lower():
return "ERROR: LOGIN FAILED: Still on login 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}"
# ------------------------------------------------------------------ #
# Step 1 — Navigate to claims-submission page, search by Subscriber #
# ID + Birthdate (same search form as the eligibility flow) #
# ------------------------------------------------------------------ #
def step1_search_patient(self):
"""From the dashboard, click 'Submit Claims' and search for the patient."""
wait = WebDriverWait(self.driver, 30)
try:
claims_url = "https://www.deltadentalma.com/private/provider/claims-submission"
if "dashboard" not in self.driver.current_url.lower():
print(f"[DDMA Claim step1] Not on dashboard — navigating to {MEMBERS_URL}")
self.driver.get(MEMBERS_URL)
time.sleep(2)
try:
submit_claims_link = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//a[@href='/private/provider/claims-submission']")
))
submit_claims_link.click()
print("[DDMA Claim step1] Clicked 'Submit Claims' quick link")
except Exception as e:
print(f"[DDMA Claim step1] Could not click 'Submit Claims' link ({e}) — falling back to direct navigation")
self.driver.get(claims_url)
time.sleep(2)
if "claims-submission" not in self.driver.current_url.lower():
print(f"[DDMA Claim step1] Warning: Ended up at {self.driver.current_url} instead of claims-submission page")
# Claims-submission page opens on the "Service Office" step — click the
# "Member Search" sidebar item to get to the birthdate/subscriber search form.
try:
member_search_link = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//span[contains(@class,'mainLabel') and normalize-space(text())='Member Search']")
))
member_search_link.click()
print("[DDMA Claim step1] Clicked 'Member Search' sidebar link")
time.sleep(1)
except Exception as e:
print(f"[DDMA Claim step1] Warning: Could not click 'Member Search' sidebar link: {e}")
# Fill Member ID
if self.memberId:
try:
member_id_input = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//input[@data-test-id='memberId']")
))
member_id_input.click()
member_id_input.send_keys(Keys.CONTROL, "a")
member_id_input.send_keys(Keys.BACKSPACE)
member_id_input.send_keys(self.memberId)
print(f"[DDMA Claim step1] Filled Member ID: {self.memberId}")
except Exception as e:
print(f"[DDMA Claim step1] Warning: Could not fill Member ID: {e}")
# Fill Birthdate (masked mm/dd/yyyy input — send digits only, mask inserts slashes)
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)
birthdate_input = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//input[@data-test-id='memberBirthDate']")
))
birthdate_input.click()
birthdate_input.send_keys(Keys.CONTROL, "a")
birthdate_input.send_keys(Keys.BACKSPACE)
birthdate_input.send_keys(f"{month}{day}{year}")
print(f"[DDMA Claim step1] Filled Birthdate: {month}/{day}/{year}")
except Exception as e:
print(f"[DDMA Claim step1] Warning: Could not fill Birthdate: {e}")
time.sleep(0.3)
# Click Search
search_btn = wait.until(EC.element_to_be_clickable(
(By.XPATH, "//button[@data-test-id='memberSearchButton']")
))
search_btn.click()
print("[DDMA Claim step1] Clicked Search button")
# Wait for the Members results table to render
try:
WebDriverWait(self.driver, 15).until(
EC.presence_of_element_located((By.XPATH, "//table//tbody//tr"))
)
time.sleep(2)
except TimeoutException:
print("[DDMA Claim step1] Warning: No results table appeared within timeout")
return "ERROR: No patient found with given search criteria"
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 — Select the matching family member row → Member detail page #
# ------------------------------------------------------------------ #
def step2_open_member_page(self):
"""Pick the family member row matching this patient and check its selection box.
The results table (columns: checkbox, Member, Member Id, Role, Group Name,
Date Of Birth, Termination Date) stays on the same page — no navigation happens.
"""
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, "//table//tbody//tr"))
)
time.sleep(2)
except TimeoutException:
print("[DDMA Claim step2] Warning: Results table not found within timeout")
rows = self.driver.find_elements(By.XPATH, "//table//tbody//tr")
print(f"[DDMA Claim step2] Found {len(rows)} family member row(s)")
target_row = None
if self.dateOfBirth and rows:
try:
dob_parts = self.dateOfBirth.split("-")
dob_year = dob_parts[0]
dob_month = dob_parts[1].zfill(2)
dob_day = dob_parts[2].zfill(2)
expected_dob = f"{dob_month}/{dob_day}/{dob_year}"
for row in rows:
cells = row.find_elements(By.XPATH, "./td")
# cells: [0]=checkbox, [1]=Member, [2]=Member Id, [3]=Role,
# [4]=Group Name, [5]=Date Of Birth, [6]=Termination Date
if len(cells) >= 6 and cells[5].text.strip() == expected_dob:
target_row = row
break
except Exception as e:
print(f"[DDMA Claim step2] DOB matching error (non-fatal): {e}")
if target_row is None and self.lastName and rows:
for row in rows:
if self.lastName.strip().lower() in row.text.lower():
target_row = row
break
if target_row is None and rows:
target_row = rows[0]
print("[DDMA Claim step2] No DOB/name match — defaulting to first family member row")
if target_row is None:
return "ERROR: step2 failed: no family member rows found"
try:
checkbox = target_row.find_element(By.XPATH, ".//i[contains(@class,'fa-check-square')]")
checkbox.click()
print("[DDMA Claim step2] Checked the matching family member row")
except Exception as e:
print(f"[DDMA Claim step2] Warning: Could not click member checkbox: {e}")
return f"ERROR: step2 failed: could not click member checkbox: {e}"
time.sleep(1)
try:
claim_detail_link = WebDriverWait(self.driver, 10).until(
EC.element_to_be_clickable(
(By.XPATH, "//span[contains(@class,'mainLabel') and normalize-space(text())='Claim Detail']")
)
)
claim_detail_link.click()
print("[DDMA Claim step2] Clicked 'Claim Detail' sidebar link")
time.sleep(2)
except Exception as e:
print(f"[DDMA Claim step2] Warning: Could not click 'Claim Detail' sidebar link: {e}")
return f"ERROR: step2 failed: could not click 'Claim Detail' link: {e}"
return "SUCCESS"
except Exception as e:
print(f"[DDMA Claim step2] Exception: {e}")
return f"ERROR: step2 failed: {e}"
# ------------------------------------------------------------------ #
# Step 3 — No-op: step2 already navigates into the Claim Detail form #
# on the new site (no separate "Create claim" button exists there). #
# ------------------------------------------------------------------ #
def step3_click_create_claim(self):
"""Confirm the Claim Detail / Treatment Form is present (navigation already happened in step2)."""
try:
WebDriverWait(self.driver, 15).until(
EC.presence_of_element_located((By.XPATH,
"//*[contains(text(),'Treatment Form')] | //input[@data-test-id='claimRowServiceDate0']"
))
)
print("[DDMA Claim step3] Treatment Form is present — no-op (step2 already navigated here)")
return "SUCCESS"
except TimeoutException:
page_text = self.driver.execute_script("return document.body.innerText;")[:400]
print(f"[DDMA Claim step3] Treatment Form not detected — page: {page_text}")
return "ERROR: step3 failed: Treatment Form not found"
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 each treatment row: service date, procedure code, tooth number, surface, amount.
The new Claim Detail Treatment Form has one row per procedure, each with
indexed fields (ServiceDate{i}, Surface{i}, Amount{i} by data-test-id;
procedure code and tooth number have no id, so they're matched positionally).
"""
try:
month, day, year = self._parse_service_date()
service_date_str = f"{month}/{day}/{year}" if (month and day and year) else None
if not service_date_str:
print(f"[DDMA Claim step4] No valid service date: {self.serviceDate!r}")
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()
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} surface={surface!r} billed={billed!r}")
# ── Click "+ Add another treatment" for lines after the first ──
if idx > 0:
try:
add_link = WebDriverWait(self.driver, 10).until(
EC.element_to_be_clickable((By.XPATH, "//a[@data-test-id='addNewRowButton']"))
)
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", add_link)
add_link.click()
print(f"[DDMA Claim step4] Clicked 'Add another treatment' for line {idx}")
time.sleep(1)
except Exception as e:
print(f"[DDMA Claim step4] Could not click 'Add another treatment': {e}")
# ── Service date ────────────────────────────────────────────
if service_date_str:
try:
date_inp = WebDriverWait(self.driver, 8).until(
EC.element_to_be_clickable((By.XPATH, f"//input[@data-test-id='claimRowServiceDate{idx}']"))
)
date_inp.click()
date_inp.send_keys(Keys.CONTROL, "a")
date_inp.send_keys(Keys.BACKSPACE)
date_inp.send_keys(f"{month}{day}{year}")
print(f"[DDMA Claim step4] serviceDate[{idx}]: filled {service_date_str}")
except Exception as e:
print(f"[DDMA Claim step4] Could not fill service date[{idx}]: {e}")
time.sleep(0.2)
# ── Procedure code (no id/data-test-id — matched positionally) ─
if code:
try:
proc_inputs = self.driver.find_elements(By.XPATH, "//input[@aria-label='e.g. D1234']")
proc_inp = proc_inputs[idx] if idx < len(proc_inputs) else (proc_inputs[-1] if proc_inputs else None)
if proc_inp:
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", proc_inp)
self._fill_combobox(proc_inp, code, f"procedureCode[{idx}]")
else:
print(f"[DDMA Claim step4] procedureCode[{idx}]: input not found")
except Exception as e:
print(f"[DDMA Claim step4] Could not fill procedure code[{idx}]: {e}")
time.sleep(0.2)
# ── Tooth Number (custom scrollable dropdown, no native <select>) ─
if tooth:
try:
tooth_triggers = self.driver.find_elements(By.XPATH, "//span[contains(@class,'filtrable')]")
if idx < len(tooth_triggers):
trigger = tooth_triggers[idx]
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", trigger)
trigger.click()
time.sleep(0.4)
option = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.XPATH,
f"//*[@role='option' and normalize-space(.)='{tooth}'] | "
f"//li[normalize-space(.)='{tooth}'] | "
f"//*[contains(@class,'dropdown')]//*[normalize-space(.)='{tooth}']"
))
)
option.click()
print(f"[DDMA Claim step4] tooth[{idx}]: selected '{tooth}'")
else:
print(f"[DDMA Claim step4] tooth[{idx}]: dropdown trigger not found")
except Exception as e:
print(f"[DDMA Claim step4] Could not fill tooth[{idx}]: {e}")
time.sleep(0.2)
# ── Surface ──────────────────────────────────────────────────
if surface:
try:
surf_inp = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.XPATH, f"//input[@data-test-id='claimRowSurface{idx}']"))
)
self._fill_text_input(surf_inp, surface, f"surface[{idx}]")
except Exception as e:
print(f"[DDMA Claim step4] Could not fill surface[{idx}]: {e}")
time.sleep(0.2)
# ── Amount ───────────────────────────────────────────────────
if billed:
try:
amount_inp = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.XPATH, f"//input[@data-test-id='claimRowAmount{idx}']"))
)
self._fill_text_input(amount_inp, billed, f"amount[{idx}]")
except Exception as e:
print(f"[DDMA Claim step4] Could not fill amount[{idx}]: {e}")
time.sleep(0.2)
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):
"""Upload all claimFiles at once via the multi-file drag-and-drop input."""
if not self.claimFiles:
print("[DDMA Claim step5] No files to attach")
return "SUCCESS"
abs_paths = []
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
abs_paths.append(abs_path)
if not abs_paths:
print("[DDMA Claim step5] No valid files found on disk to attach")
return "SUCCESS"
try:
file_input = WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.XPATH, "//input[@data-test-id='fileUploadInput']"))
)
# The input may be visually hidden behind a drag-and-drop overlay.
self.driver.execute_script("arguments[0].style.display='block';", file_input)
# <input multiple> accepts several paths joined by newlines in one send_keys call.
file_input.send_keys("\n".join(abs_paths))
time.sleep(1.5)
print(f"[DDMA Claim step5] Attached {len(abs_paths)}/{len(self.claimFiles)} file(s): "
f"{[os.path.basename(p) for p in abs_paths]}")
except Exception as e:
print(f"[DDMA Claim step5] Could not attach files: {e}")
return f"ERROR: step5 failed: {e}"
return "SUCCESS"
# ------------------------------------------------------------------ #
# Step 6 — Click Next step #
# ------------------------------------------------------------------ #
def step6_click_next(self):
"""Click the sidebar 'Submit Claim' navigation button to advance from the Claim Detail form."""
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-test-id='submitClaimNavigationButton']"))
)
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 — No-op: step6's "Submit Claim" click already finalizes the #
# claim on the new site (no separate acknowledgement/summary page). #
# ------------------------------------------------------------------ #
def step7_submit_claim(self):
"""Confirm the confirmation page is present (submission already happened in step6)."""
try:
WebDriverWait(self.driver, 20).until(
lambda d: "claim submitted successfully" in d.page_source.lower()
)
print("[DDMA Claim step7] Confirmation page detected — no-op (step6 already submitted)")
return "SUCCESS"
except TimeoutException:
print(f"[DDMA Claim step7] Confirmation page not detected — URL: {self.driver.current_url}")
return "ERROR: step7 failed: confirmation page not detected"
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 confirmation page, extract the claim number, save page as PDF."""
import re
try:
# Wait for confirmation page
WebDriverWait(self.driver, 30).until(
lambda d: "claim submitted successfully" 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 (row: "Claim Number" label + value, e.g. "202626032473000")
claim_number = None
try:
body_text = self.driver.find_element(By.TAG_NAME, "body").text
match = re.search(r'claim number\s*\n?\s*(\d{6,})', 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