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.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}" # ------------------------------------------------------------------ # # 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...") # 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: try: WebDriverWait(self.driver, 10).until( EC.presence_of_element_located((By.XPATH, "//tbody//tr")) ) time.sleep(2) except TimeoutException: print("[DDMA Claim step2] Warning: Results table not found within timeout") # Find member-details URL from first row — identical to eligibility step2 detail_url = None for selector in [ "(//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')]", ]: try: link_el = WebDriverWait(self.driver, 5).until( EC.presence_of_element_located((By.XPATH, selector)) ) href = link_el.get_attribute("href") if href and "member-details" in href: detail_url = href print(f"[DDMA Claim step2] Found detail URL: {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: WebDriverWait(self.driver, 15).until( EC.presence_of_element_located((By.XPATH, "//button[@aria-label='Create claim']")) ) print("[DDMA Claim step2] 'Create claim' button found") except TimeoutException: print("[DDMA Claim step2] Warning: 'Create claim' button not found") 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) self.driver.execute_script("window.scrollTo(0, 0);") time.sleep(0.5) # Log all buttons on page for debugging 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]}") btn = WebDriverWait(self.driver, 10).until( EC.element_to_be_clickable((By.XPATH, "//button[@aria-label='Create claim' and @data-react-aria-pressable='true']" )) ) print(f"[DDMA Claim step3] Found 'Create claim' button, displayed={btn.is_displayed()}, enabled={btn.is_enabled()}") self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", btn) time.sleep(0.5) # Try all click methods in sequence until one causes navigation 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("[DDMA Claim step3] Dispatched pointer+click events on 'Create claim'") time.sleep(2) # Switch to new tab if one 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") print(f"[DDMA Claim step3] Post-click URL: {self.driver.current_url}") # Wait for claim form — just log, don't fail 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") 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}]") time.sleep(0.5) 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): self._fill_combobox(tooth_inputs[idx], tooth, f"tooth[{idx}]") 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 (free-text combobox — type directly) ────────────── 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) # Dismiss any listbox with Escape so it doesn't block next field time.sleep(0.3) surf_inp.send_keys(Keys.ESCAPE) 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 = WebDriverWait(self.driver, 10).until( EC.element_to_be_clickable((By.XPATH, "//button[.//span[contains(text(),'Submit claim')]] | " "//button[contains(normalize-space(text()),'Submit claim')] | " "//button[@aria-label='Submit claim']" )) ) 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