feat: add Deltains claim submission flow
Adds a full "Deltains Claim" button next to Delta MA Claim, mirroring the DDMA claim flow across frontend, backend queue/routes, and a new Selenium worker. Reuses the existing Deltains eligibility login/OTP code and was iteratively wired against the live portal (patient roster search, per-service-line fields, tooth number/surface selects, review and submit). Also fixes chatbot/insurance-alias carrier recognition so "deltains" resolves correctly instead of falling back to MassHealth.
This commit is contained in:
@@ -23,6 +23,7 @@ import helpers_cca_eligibility as hcca
|
||||
import helpers_cca_claim as hcca_claim
|
||||
import helpers_cca_preauth as hcca_preauth
|
||||
import helpers_ddma_claim as hddma_claim
|
||||
import helpers_deltains_claim as hdeltains_claim
|
||||
import helpers_uniteddh_claim as huniteddh_claim
|
||||
import helpers_uniteddh_preauth as huniteddh_preauth
|
||||
import helpers_tuftssco_claim as htuftssco_claim
|
||||
@@ -749,6 +750,48 @@ async def ddma_claim(request: Request):
|
||||
return {"status": "started", "session_id": sid}
|
||||
|
||||
|
||||
async def _deltains_claim_worker_wrapper(sid: str, data: dict, url: str):
|
||||
"""Background worker for Deltains claim submission."""
|
||||
global active_jobs, waiting_jobs
|
||||
async with semaphore:
|
||||
async with lock:
|
||||
waiting_jobs -= 1
|
||||
active_jobs += 1
|
||||
try:
|
||||
await hdeltains_claim.start_deltains_claim_run(sid, data, url)
|
||||
finally:
|
||||
async with lock:
|
||||
active_jobs -= 1
|
||||
|
||||
|
||||
@app.post("/deltains-claim")
|
||||
async def deltains_claim(request: Request):
|
||||
"""
|
||||
Starts a Deltains claim submission session in the background.
|
||||
Logs in, searches patient, opens the member's eligibility/benefits page,
|
||||
finds the claim-submission entry point, fills service lines, and submits.
|
||||
Body: { "claim": { "deltainsUsername": "...", "deltainsPassword": "...", ... } }
|
||||
Returns: { status: "started", session_id: "<uuid>" }
|
||||
"""
|
||||
global waiting_jobs
|
||||
|
||||
body = await request.json()
|
||||
|
||||
sid = hdeltains_claim.make_session_entry()
|
||||
hdeltains_claim.sessions[sid]["type"] = "deltains_claim"
|
||||
hdeltains_claim.sessions[sid]["last_activity"] = time.time()
|
||||
|
||||
async with lock:
|
||||
waiting_jobs += 1
|
||||
|
||||
asyncio.create_task(_deltains_claim_worker_wrapper(
|
||||
sid, body,
|
||||
url="https://www.deltadentalins.com/ciam/login?TARGET=%2Fprovider-tools%2Fv2"
|
||||
))
|
||||
|
||||
return {"status": "started", "session_id": sid}
|
||||
|
||||
|
||||
async def _uniteddh_claim_worker_wrapper(sid: str, data: dict, url: str):
|
||||
"""Background worker for United/DentalHub claim submission."""
|
||||
global active_jobs, waiting_jobs
|
||||
@@ -1025,6 +1068,8 @@ async def submit_otp(request: Request):
|
||||
res = hdentaquest.submit_otp(sid, otp)
|
||||
elif sid in hddma_claim.sessions:
|
||||
res = hddma_claim.submit_otp(sid, otp)
|
||||
elif sid in hdeltains_claim.sessions:
|
||||
res = hdeltains_claim.submit_otp(sid, otp)
|
||||
elif sid in huniteddh_claim.sessions:
|
||||
res = huniteddh_claim.submit_otp(sid, otp)
|
||||
elif sid in huniteddh_preauth.sessions:
|
||||
@@ -1066,6 +1111,8 @@ async def session_status(sid: str):
|
||||
s = hcca_preauth.get_session_status(sid)
|
||||
elif sid in hddma_claim.sessions:
|
||||
s = hddma_claim.get_session_status(sid)
|
||||
elif sid in hdeltains_claim.sessions:
|
||||
s = hdeltains_claim.get_session_status(sid)
|
||||
elif sid in huniteddh_claim.sessions:
|
||||
s = huniteddh_claim.get_session_status(sid)
|
||||
elif sid in huniteddh_preauth.sessions:
|
||||
|
||||
@@ -0,0 +1,349 @@
|
||||
import os
|
||||
import time
|
||||
import asyncio
|
||||
from typing import Dict, Any
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.common.keys import Keys
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
from selenium.common.exceptions import WebDriverException, TimeoutException
|
||||
|
||||
from selenium_DeltaIns_claimSubmitWorker import AutomationDeltaInsClaimSubmit
|
||||
from deltains_browser_manager import get_browser_manager
|
||||
|
||||
sessions: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
SESSION_OTP_TIMEOUT = int(os.getenv("SESSION_OTP_TIMEOUT", "240"))
|
||||
|
||||
|
||||
def make_session_entry() -> str:
|
||||
import uuid
|
||||
sid = str(uuid.uuid4())
|
||||
sessions[sid] = {
|
||||
"status": "created",
|
||||
"created_at": time.time(),
|
||||
"last_activity": time.time(),
|
||||
"bot": None,
|
||||
"driver": None,
|
||||
"otp_event": asyncio.Event(),
|
||||
"otp_value": None,
|
||||
"result": None,
|
||||
"message": None,
|
||||
}
|
||||
return sid
|
||||
|
||||
|
||||
async def cleanup_session(sid: str, message: str | None = None):
|
||||
s = sessions.get(sid)
|
||||
if not s:
|
||||
return
|
||||
try:
|
||||
if s.get("status") not in ("completed", "error", "not_found"):
|
||||
s["status"] = "error"
|
||||
if message:
|
||||
s["message"] = message
|
||||
ev = s.get("otp_event")
|
||||
if ev and not ev.is_set():
|
||||
ev.set()
|
||||
finally:
|
||||
sessions.pop(sid, None)
|
||||
print(f"[helpers_deltains_claim] cleaned session {sid}")
|
||||
|
||||
|
||||
async def _remove_session_later(sid: str, delay: int = 30):
|
||||
await asyncio.sleep(delay)
|
||||
await cleanup_session(sid)
|
||||
|
||||
|
||||
def _close_browser(bot):
|
||||
"""Save cookies and close the browser after task completion."""
|
||||
try:
|
||||
bm = get_browser_manager()
|
||||
try:
|
||||
bm.save_cookies()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
bm.quit_driver()
|
||||
print("[Deltains Claim] Browser closed")
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim] Could not close browser: {e}")
|
||||
|
||||
|
||||
async def start_deltains_claim_run(sid: str, data: dict, url: str):
|
||||
"""
|
||||
Run the Deltains claim workflow — reuses the Okta login/OTP handling
|
||||
proven in helpers_deltains_eligibility.py, then continues into
|
||||
claim-specific steps mirroring helpers_ddma_claim.py's structure.
|
||||
"""
|
||||
s = sessions.get(sid)
|
||||
if not s:
|
||||
return {"status": "error", "message": "session not found"}
|
||||
|
||||
s["status"] = "running"
|
||||
s["last_activity"] = time.time()
|
||||
bot = None
|
||||
|
||||
try:
|
||||
bot = AutomationDeltaInsClaimSubmit(data)
|
||||
bot.config_driver()
|
||||
|
||||
s["bot"] = bot
|
||||
s["driver"] = bot.driver
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
try:
|
||||
bot.driver.maximize_window()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
login_result = bot.login(url)
|
||||
except WebDriverException as wde:
|
||||
s["status"] = "error"
|
||||
s["message"] = f"Selenium driver error during login: {wde}"
|
||||
await cleanup_session(sid, s["message"])
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": s["message"]}
|
||||
except Exception as e:
|
||||
s["status"] = "error"
|
||||
s["message"] = f"Unexpected error during login: {e}"
|
||||
await cleanup_session(sid, s["message"])
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": s["message"]}
|
||||
|
||||
# ── Already logged in (persistent session) ────────────────────────────
|
||||
if isinstance(login_result, str) and login_result == "ALREADY_LOGGED_IN":
|
||||
print("[Deltains Claim] Session persisted — skipping OTP")
|
||||
s["status"] = "running"
|
||||
s["message"] = "Session persisted"
|
||||
get_browser_manager().save_cookies()
|
||||
|
||||
# ── OTP required ──────────────────────────────────────────────────────
|
||||
elif isinstance(login_result, str) and login_result == "OTP_REQUIRED":
|
||||
s["status"] = "waiting_for_otp"
|
||||
s["message"] = "OTP required — please enter the code sent to your email"
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
driver = s["driver"]
|
||||
max_polls = SESSION_OTP_TIMEOUT
|
||||
login_success = False
|
||||
|
||||
print(f"[Deltains Claim OTP] Polling for OTP completion (up to {SESSION_OTP_TIMEOUT}s)...")
|
||||
|
||||
for poll in range(max_polls):
|
||||
await asyncio.sleep(1)
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
try:
|
||||
otp_value = s.get("otp_value")
|
||||
if otp_value:
|
||||
print(f"[Deltains Claim OTP poll {poll+1}] OTP received from app, typing it in...")
|
||||
try:
|
||||
otp_input = driver.find_element(By.XPATH,
|
||||
"//input[@name='credentials.passcode' and @type='text'] | "
|
||||
"//input[contains(@name,'passcode')]")
|
||||
otp_input.clear()
|
||||
otp_input.send_keys(otp_value)
|
||||
try:
|
||||
verify_btn = driver.find_element(By.XPATH,
|
||||
"//input[@type='submit'] | "
|
||||
"//button[@type='submit']")
|
||||
verify_btn.click()
|
||||
except Exception:
|
||||
otp_input.send_keys(Keys.RETURN)
|
||||
print("[Deltains Claim OTP] OTP typed and submitted via app")
|
||||
s["otp_value"] = None
|
||||
await asyncio.sleep(8)
|
||||
except Exception as type_err:
|
||||
print(f"[Deltains Claim OTP] Failed to type OTP from app: {type_err}")
|
||||
|
||||
current_url = driver.current_url.lower()
|
||||
if poll % 10 == 0:
|
||||
print(f"[Deltains Claim OTP poll {poll+1}/{max_polls}] URL: {current_url[:80]}")
|
||||
|
||||
if "provider-tools" in current_url and "login" not in current_url and "ciam" not in current_url:
|
||||
print("[Deltains Claim OTP] Login successful!")
|
||||
login_success = True
|
||||
break
|
||||
|
||||
except Exception as poll_err:
|
||||
if poll % 10 == 0:
|
||||
print(f"[Deltains Claim OTP poll {poll+1}] Error: {poll_err}")
|
||||
|
||||
if not login_success:
|
||||
try:
|
||||
current_url = driver.current_url.lower()
|
||||
if "provider-tools" in current_url and "login" not in current_url and "ciam" not in current_url:
|
||||
login_success = True
|
||||
else:
|
||||
s["status"] = "error"
|
||||
s["message"] = "OTP timeout — login not completed"
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": "OTP not completed in time"}
|
||||
except Exception as final_err:
|
||||
s["status"] = "error"
|
||||
s["message"] = f"OTP verification failed: {final_err}"
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": s["message"]}
|
||||
|
||||
if login_success:
|
||||
s["status"] = "running"
|
||||
s["message"] = "Login successful after OTP"
|
||||
print("[Deltains Claim OTP] Proceeding to claim steps...")
|
||||
get_browser_manager().save_cookies()
|
||||
|
||||
# ── Login succeeded without OTP ───────────────────────────────────────
|
||||
elif isinstance(login_result, str) and login_result == "SUCCESS":
|
||||
print("[Deltains Claim] Login succeeded without OTP")
|
||||
s["status"] = "running"
|
||||
s["message"] = "Login succeeded"
|
||||
get_browser_manager().save_cookies()
|
||||
|
||||
# ── Login error ───────────────────────────────────────────────────────
|
||||
elif isinstance(login_result, str) and login_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = login_result
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": login_result}
|
||||
|
||||
# --- Step 1: Search patient ---
|
||||
step1_result = bot.step1_search_patient()
|
||||
print(f"[Deltains Claim] step1 result: {step1_result}")
|
||||
if isinstance(step1_result, str) and step1_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = step1_result
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": step1_result}
|
||||
|
||||
# --- Step 2: Select transaction type (search lands directly on the ---
|
||||
# --- "Submit a claim" page, so no separate member-page/create-claim ---
|
||||
# --- click is needed here, unlike the DDMA flow) ---
|
||||
step2_result = bot.step2_select_transaction_type()
|
||||
print(f"[Deltains Claim] step2 result: {step2_result}")
|
||||
if isinstance(step2_result, str) and step2_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = step2_result
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": step2_result}
|
||||
|
||||
# --- Step 4: Fill service date + procedure lines ---
|
||||
step4_result = bot.step4_fill_claim_form()
|
||||
print(f"[Deltains Claim] step4 result: {step4_result}")
|
||||
if isinstance(step4_result, str) and step4_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = step4_result
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": step4_result}
|
||||
|
||||
# --- Step 5: Attach files ---
|
||||
step5_result = bot.step5_attach_files()
|
||||
print(f"[Deltains Claim] step5 result: {step5_result}")
|
||||
if isinstance(step5_result, str) and step5_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = step5_result
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": step5_result}
|
||||
|
||||
# --- Step 6: Click Next step ---
|
||||
step6_result = bot.step6_click_next()
|
||||
print(f"[Deltains Claim] step6 result: {step6_result}")
|
||||
if isinstance(step6_result, str) and step6_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = step6_result
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": step6_result}
|
||||
|
||||
# --- Step 7: Acknowledge + Submit claim ---
|
||||
step7_result = bot.step7_submit_claim()
|
||||
print(f"[Deltains Claim] step7 result: {step7_result}")
|
||||
if isinstance(step7_result, str) and step7_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = step7_result
|
||||
await cleanup_session(sid)
|
||||
_close_browser(bot)
|
||||
return {"status": "error", "message": step7_result}
|
||||
|
||||
# --- Step 8: Extract claim number + save confirmation PDF ---
|
||||
step8_result = bot.step8_save_confirmation_pdf()
|
||||
print(f"[Deltains Claim] step8 result: {step8_result}")
|
||||
if isinstance(step8_result, str) and step8_result.startswith("ERROR"):
|
||||
# Non-fatal: claim was submitted; log but don't abort
|
||||
print(f"[Deltains Claim] step8 warning (non-fatal): {step8_result}")
|
||||
step8_result = {}
|
||||
|
||||
pdf_path = step8_result.get("pdf_path") if isinstance(step8_result, dict) else None
|
||||
pdf_url = None
|
||||
if pdf_path:
|
||||
import os as _os
|
||||
filename = _os.path.basename(pdf_path)
|
||||
port = _os.getenv("PORT", "5002")
|
||||
url_host = _os.getenv("HOST", "localhost")
|
||||
pdf_url = f"http://{url_host}:{port}/downloads/{filename}"
|
||||
print(f"[Deltains Claim] pdf_url: {pdf_url}")
|
||||
|
||||
claim_number = step8_result.get("claimNumber") if isinstance(step8_result, dict) else None
|
||||
|
||||
result = {
|
||||
"status": "success",
|
||||
"message": "Deltains claim submitted successfully",
|
||||
"claimNumber": claim_number,
|
||||
"pdf_url": pdf_url,
|
||||
}
|
||||
s["status"] = "completed"
|
||||
s["result"] = result
|
||||
s["message"] = "completed"
|
||||
|
||||
# Leave the browser window open on the "My claims" page (persistent session)
|
||||
print("[Deltains Claim] Claim completed - browser left open on My claims page")
|
||||
|
||||
asyncio.create_task(_remove_session_later(sid, 60))
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
if s:
|
||||
s["status"] = "error"
|
||||
s["message"] = f"worker exception: {e}"
|
||||
if bot:
|
||||
_close_browser(bot)
|
||||
await cleanup_session(sid)
|
||||
return {"status": "error", "message": f"worker exception: {e}"}
|
||||
|
||||
|
||||
def submit_otp(sid: str, otp: str) -> Dict[str, Any]:
|
||||
s = sessions.get(sid)
|
||||
if not s:
|
||||
return {"status": "error", "message": "session not found"}
|
||||
if s.get("status") != "waiting_for_otp":
|
||||
return {"status": "error", "message": f"session not waiting for otp (state={s.get('status')})"}
|
||||
s["otp_value"] = otp
|
||||
s["last_activity"] = time.time()
|
||||
try:
|
||||
s["otp_event"].set()
|
||||
except Exception:
|
||||
pass
|
||||
return {"status": "ok", "message": "otp accepted"}
|
||||
|
||||
|
||||
def get_session_status(sid: str) -> Dict[str, Any]:
|
||||
s = sessions.get(sid)
|
||||
if not s:
|
||||
return {"status": "not_found"}
|
||||
return {
|
||||
"session_id": sid,
|
||||
"status": s.get("status"),
|
||||
"message": s.get("message"),
|
||||
"created_at": s.get("created_at"),
|
||||
"last_activity": s.get("last_activity"),
|
||||
"result": s.get("result") if s.get("status") in ("completed", "error") else None,
|
||||
}
|
||||
@@ -0,0 +1,821 @@
|
||||
from selenium.common.exceptions import TimeoutException
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.common.keys import Keys
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
import time
|
||||
import os
|
||||
import re
|
||||
import base64
|
||||
|
||||
from deltains_browser_manager import get_browser_manager
|
||||
|
||||
LOGIN_URL = "https://www.deltadentalins.com/ciam/login?TARGET=%2Fprovider-tools%2Fv2"
|
||||
PROVIDER_TOOLS_URL = "https://www.deltadentalins.com/provider-tools/v2"
|
||||
|
||||
# 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 AutomationDeltaInsClaimSubmit:
|
||||
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.deltains_username = claim.get("deltainsUsername", "") or raw.get("deltainsUsername", "")
|
||||
self.deltains_password = claim.get("deltainsPassword", "") or raw.get("deltainsPassword", "")
|
||||
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
|
||||
os.makedirs(self.download_dir, exist_ok=True)
|
||||
|
||||
print(f"[Deltains 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 — copied from AutomationDeltaInsEligibilityCheck.login() #
|
||||
# (Okta-based, already proven working for the eligibility flow) #
|
||||
# ------------------------------------------------------------------ #
|
||||
def _dismiss_cookie_banner(self):
|
||||
try:
|
||||
accept_btn = WebDriverWait(self.driver, 5).until(
|
||||
EC.element_to_be_clickable((By.ID, "onetrust-accept-btn-handler"))
|
||||
)
|
||||
accept_btn.click()
|
||||
print("[Deltains Claim login] Dismissed cookie consent banner")
|
||||
time.sleep(1)
|
||||
except TimeoutException:
|
||||
print("[Deltains Claim login] No cookie consent banner found")
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim login] Error dismissing cookie banner: {e}")
|
||||
|
||||
def _force_logout(self):
|
||||
try:
|
||||
print("[Deltains Claim login] Forcing logout due to credential change...")
|
||||
browser_manager = get_browser_manager()
|
||||
try:
|
||||
self.driver.delete_all_cookies()
|
||||
print("[Deltains Claim login] Cleared all cookies")
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim login] Error clearing cookies: {e}")
|
||||
browser_manager.clear_credentials_hash()
|
||||
print("[Deltains Claim login] Logout complete")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim login] Error during forced logout: {e}")
|
||||
return False
|
||||
|
||||
def login(self, url):
|
||||
"""
|
||||
Multi-step login flow for DeltaIns (Okta-based) — identical to the
|
||||
eligibility worker's login():
|
||||
1. Enter username (name='identifier') -> click Next
|
||||
2. Enter password (type='password') -> click Submit
|
||||
3. Handle MFA: click 'Send me an email' -> wait for OTP
|
||||
Returns: ALREADY_LOGGED_IN, SUCCESS, OTP_REQUIRED, or ERROR:...
|
||||
"""
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
browser_manager = get_browser_manager()
|
||||
|
||||
try:
|
||||
if self.deltains_username and browser_manager.credentials_changed(self.deltains_username):
|
||||
self._force_logout()
|
||||
self.driver.get(url)
|
||||
time.sleep(3)
|
||||
|
||||
try:
|
||||
current_url = self.driver.current_url
|
||||
print(f"[Deltains Claim login] Current URL: {current_url}")
|
||||
if "provider-tools" in current_url and "login" not in current_url.lower() and "ciam" not in current_url.lower():
|
||||
print("[Deltains Claim login] Already on provider tools page - logged in")
|
||||
return "ALREADY_LOGGED_IN"
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim login] Error checking current state: {e}")
|
||||
|
||||
print("[Deltains Claim login] Trying provider-tools URL to check session...")
|
||||
self.driver.get(PROVIDER_TOOLS_URL)
|
||||
time.sleep(5)
|
||||
|
||||
current_url = self.driver.current_url
|
||||
print(f"[Deltains Claim login] After provider-tools nav URL: {current_url}")
|
||||
|
||||
if "provider-tools" in current_url and "login" not in current_url.lower() and "ciam" not in current_url.lower():
|
||||
print("[Deltains Claim login] Session still valid - already logged in")
|
||||
return "ALREADY_LOGGED_IN"
|
||||
|
||||
print("[Deltains Claim login] Session not valid, navigating to login page...")
|
||||
self.driver.get(url)
|
||||
time.sleep(3)
|
||||
|
||||
current_url = self.driver.current_url
|
||||
print(f"[Deltains Claim login] After login nav URL: {current_url}")
|
||||
|
||||
if "provider-tools" in current_url and "login" not in current_url.lower() and "ciam" not in current_url.lower():
|
||||
print("[Deltains Claim login] Already logged in - on provider tools")
|
||||
return "ALREADY_LOGGED_IN"
|
||||
|
||||
self._dismiss_cookie_banner()
|
||||
|
||||
print("[Deltains Claim login] Looking for username field...")
|
||||
username_entered = False
|
||||
for sel in [
|
||||
(By.NAME, "identifier"),
|
||||
(By.ID, "okta-signin-username"),
|
||||
(By.XPATH, "//input[@type='text' and @autocomplete='username']"),
|
||||
(By.XPATH, "//input[@type='text']"),
|
||||
]:
|
||||
try:
|
||||
field = WebDriverWait(self.driver, 8).until(EC.presence_of_element_located(sel))
|
||||
if field.is_displayed():
|
||||
field.clear()
|
||||
field.send_keys(self.deltains_username)
|
||||
username_entered = True
|
||||
print(f"[Deltains Claim login] Username entered via {sel}")
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if not username_entered:
|
||||
return "ERROR: Could not find username field"
|
||||
|
||||
time.sleep(1)
|
||||
for sel in [
|
||||
(By.XPATH, "//input[@type='submit' and @value='Next']"),
|
||||
(By.XPATH, "//input[@type='submit']"),
|
||||
(By.XPATH, "//button[@type='submit']"),
|
||||
]:
|
||||
try:
|
||||
btn = self.driver.find_element(*sel)
|
||||
if btn.is_displayed():
|
||||
btn.click()
|
||||
print(f"[Deltains Claim login] Clicked Next via {sel}")
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
time.sleep(4)
|
||||
|
||||
current_url = self.driver.current_url
|
||||
if "provider-tools" in current_url and "login" not in current_url.lower() and "ciam" not in current_url.lower():
|
||||
return "ALREADY_LOGGED_IN"
|
||||
|
||||
print("[Deltains Claim login] Looking for password field...")
|
||||
pw_entered = False
|
||||
for sel in [
|
||||
(By.XPATH, "//input[@type='password']"),
|
||||
(By.ID, "okta-signin-password"),
|
||||
(By.NAME, "password"),
|
||||
]:
|
||||
try:
|
||||
field = WebDriverWait(self.driver, 10).until(EC.presence_of_element_located(sel))
|
||||
if field.is_displayed():
|
||||
field.clear()
|
||||
field.send_keys(self.deltains_password)
|
||||
pw_entered = True
|
||||
print(f"[Deltains Claim login] Password entered via {sel}")
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if not pw_entered:
|
||||
current_url = self.driver.current_url
|
||||
if "provider-tools" in current_url and "login" not in current_url.lower():
|
||||
return "ALREADY_LOGGED_IN"
|
||||
return "ERROR: Password field not found"
|
||||
|
||||
time.sleep(1)
|
||||
for sel in [
|
||||
(By.ID, "okta-signin-submit"),
|
||||
(By.XPATH, "//input[@type='submit']"),
|
||||
(By.XPATH, "//button[@type='submit']"),
|
||||
]:
|
||||
try:
|
||||
btn = self.driver.find_element(*sel)
|
||||
if btn.is_displayed():
|
||||
btn.click()
|
||||
print(f"[Deltains Claim login] Clicked Sign In via {sel}")
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if self.deltains_username:
|
||||
browser_manager.save_credentials_hash(self.deltains_username)
|
||||
|
||||
time.sleep(6)
|
||||
|
||||
current_url = self.driver.current_url
|
||||
print(f"[Deltains Claim login] After password submit URL: {current_url}")
|
||||
|
||||
if "provider-tools" in current_url and "login" not in current_url.lower() and "ciam" not in current_url.lower():
|
||||
print("[Deltains Claim login] Login successful - on provider tools")
|
||||
return "SUCCESS"
|
||||
|
||||
print("[Deltains Claim login] Handling MFA...")
|
||||
|
||||
try:
|
||||
body_text = self.driver.find_element(By.TAG_NAME, "body").text
|
||||
if "security method" in body_text.lower() or "select from the following" in body_text.lower():
|
||||
print("[Deltains Claim login] MFA method selection page detected")
|
||||
email_select = None
|
||||
for sel in [
|
||||
(By.CSS_SELECTOR, "div[data-se='okta_email'] a.select-factor"),
|
||||
(By.XPATH, "//div[@data-se='okta_email']//a[contains(@class,'select-factor')]"),
|
||||
(By.XPATH, "//a[contains(@aria-label,'Select Email')]"),
|
||||
(By.XPATH, "//div[@data-se='okta_email']//a[@data-se='button']"),
|
||||
(By.CSS_SELECTOR, "a.select-factor.link-button"),
|
||||
]:
|
||||
try:
|
||||
btn = self.driver.find_element(*sel)
|
||||
if btn.is_displayed():
|
||||
email_select = btn
|
||||
print(f"[Deltains Claim login] Found Email Select via {sel}")
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if email_select:
|
||||
email_select.click()
|
||||
print("[Deltains Claim login] Clicked 'Select' for Email MFA")
|
||||
time.sleep(5)
|
||||
else:
|
||||
print("[Deltains Claim login] Could not find Email Select button")
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim login] Error checking MFA method selection: {e}")
|
||||
|
||||
try:
|
||||
send_btn = WebDriverWait(self.driver, 8).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//input[@type='submit' and @value='Send me an email'] | "
|
||||
"//input[@value='Send me an email'] | "
|
||||
"//button[contains(text(),'Send me an email')]"))
|
||||
)
|
||||
send_btn.click()
|
||||
print("[Deltains Claim login] Clicked 'Send me an email'")
|
||||
time.sleep(5)
|
||||
except TimeoutException:
|
||||
print("[Deltains Claim login] No 'Send me an email' button, checking for OTP input...")
|
||||
|
||||
try:
|
||||
otp_input = WebDriverWait(self.driver, 10).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//input[@name='credentials.passcode' and @type='text'] | "
|
||||
"//input[contains(@name,'passcode')]"))
|
||||
)
|
||||
print("[Deltains Claim login] OTP input found -> OTP_REQUIRED")
|
||||
return "OTP_REQUIRED"
|
||||
except TimeoutException:
|
||||
pass
|
||||
|
||||
current_url = self.driver.current_url
|
||||
if "provider-tools" in current_url and "login" not in current_url.lower() and "ciam" not in current_url.lower():
|
||||
return "SUCCESS"
|
||||
|
||||
try:
|
||||
error_elem = self.driver.find_element(By.XPATH,
|
||||
"//*[contains(@class,'error') or contains(@class,'alert-error')]")
|
||||
error_text = error_elem.text.strip()[:200]
|
||||
if error_text:
|
||||
return f"ERROR: {error_text}"
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
print("[Deltains Claim login] Could not determine login state - returning OTP_REQUIRED as fallback")
|
||||
return "OTP_REQUIRED"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim login] Exception: {e}")
|
||||
return f"ERROR:LOGIN FAILED: {e}"
|
||||
|
||||
def _format_dob(self, dob_str):
|
||||
if dob_str and "-" in dob_str:
|
||||
dob_parts = dob_str.split("-")
|
||||
if len(dob_parts) == 3:
|
||||
return f"{dob_parts[1]}/{dob_parts[2]}/{dob_parts[0]}"
|
||||
return dob_str
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Step 1 — Search patient (same search form as the eligibility flow) #
|
||||
# ------------------------------------------------------------------ #
|
||||
def step1_search_patient(self):
|
||||
"""
|
||||
Click the 'Submit claim' dashboard tile, which opens a 'Patient search'
|
||||
modal, and search the patient roster by full name.
|
||||
"""
|
||||
try:
|
||||
print(f"[Deltains Claim step1] Starting — memberId={self.memberId}, "
|
||||
f"patient={self.firstName} {self.lastName}")
|
||||
|
||||
print("[Deltains Claim step1] Clicking 'Submit claim' dashboard tile...")
|
||||
try:
|
||||
submit_claim_tile = WebDriverWait(self.driver, 15).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//button[@data-testid='moduleTile'][.//a[contains(text(),'Submit claim')]] | "
|
||||
"//a[@data-testid='moduleTileLabelLink'][contains(text(),'Submit claim')] | "
|
||||
"//a[contains(text(),'Submit claim')]"))
|
||||
)
|
||||
submit_claim_tile.click()
|
||||
time.sleep(5)
|
||||
print("[Deltains Claim step1] Clicked 'Submit claim' tile")
|
||||
except TimeoutException:
|
||||
return "ERROR: 'Submit claim' dashboard tile not found"
|
||||
|
||||
print("[Deltains Claim step1] Waiting for 'Patient search' modal...")
|
||||
try:
|
||||
WebDriverWait(self.driver, 15).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//h1[contains(text(),'Patient search')] | "
|
||||
"//h2[contains(text(),'Patient search')]"))
|
||||
)
|
||||
print("[Deltains Claim step1] 'Patient search' modal loaded")
|
||||
except TimeoutException:
|
||||
print("[Deltains Claim step1] 'Patient search' modal heading not found, continuing anyway...")
|
||||
|
||||
full_name = f"{self.firstName} {self.lastName}".strip()
|
||||
print(f"[Deltains Claim step1] Searching patient roster by name: '{full_name}'...")
|
||||
try:
|
||||
name_field = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.ID, "patientRosterAutocomplete"))
|
||||
)
|
||||
name_field.click()
|
||||
name_field.send_keys(Keys.CONTROL + "a")
|
||||
name_field.send_keys(Keys.DELETE)
|
||||
time.sleep(0.3)
|
||||
name_field.send_keys(full_name)
|
||||
time.sleep(1.5)
|
||||
print(f"[Deltains Claim step1] Name entered: '{name_field.get_attribute('value')}'")
|
||||
except TimeoutException:
|
||||
return "ERROR: Patient roster search field not found"
|
||||
|
||||
print("[Deltains Claim step1] Clicking search icon button...")
|
||||
try:
|
||||
search_btn = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//button[.//*[contains(@class,'dd-icon-Search')]] | "
|
||||
"//button[@data-testid='autocompleteSearchButton']"))
|
||||
)
|
||||
search_btn.click()
|
||||
time.sleep(5)
|
||||
print("[Deltains Claim step1] Clicked search icon button")
|
||||
except TimeoutException:
|
||||
return "ERROR: Search icon button not found"
|
||||
|
||||
print("[Deltains Claim step1] Checking for results...")
|
||||
try:
|
||||
patient_card = WebDriverWait(self.driver, 15).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//div[contains(@class,'patient-card-root')] | "
|
||||
"//div[@data-testid='patientCard'] | "
|
||||
"//div[starts-with(@data-testid,'patientCard')]"))
|
||||
)
|
||||
print("[Deltains Claim step1] Patient card found!")
|
||||
self._patient_card = patient_card
|
||||
except TimeoutException:
|
||||
try:
|
||||
body_text = self.driver.find_element(By.TAG_NAME, "body").text
|
||||
if "no results" in body_text.lower() or "not found" in body_text.lower() or "no patient" in body_text.lower():
|
||||
return "ERROR: No patient found with the provided name"
|
||||
alerts = self.driver.find_elements(By.XPATH, "//*[@role='alert']")
|
||||
for alert in alerts:
|
||||
if alert.is_displayed():
|
||||
return f"ERROR: {alert.text.strip()[:200]}"
|
||||
except Exception:
|
||||
pass
|
||||
return "ERROR: No patient results found within timeout"
|
||||
|
||||
print("[Deltains Claim step1] Search completed")
|
||||
return "SUCCESS"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step1] Exception: {e}")
|
||||
return f"ERROR: step1 failed: {e}"
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Step 2 — Select transaction type on the "Submit a claim" page #
|
||||
# (search lands directly on this page — no separate member-page or #
|
||||
# "Create claim" click is needed, unlike the DDMA flow) #
|
||||
# ------------------------------------------------------------------ #
|
||||
def step2_select_transaction_type(self):
|
||||
"""
|
||||
Confirm the "Submit a claim" page loaded, then click the
|
||||
"Statement of completed services" radio button under
|
||||
"Type of transaction".
|
||||
"""
|
||||
try:
|
||||
print(f"[Deltains Claim step2] Current URL: {self.driver.current_url}")
|
||||
try:
|
||||
WebDriverWait(self.driver, 20).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//h1[contains(text(),'Submit a claim')] | "
|
||||
"//*[contains(text(),'Type of transaction')]"))
|
||||
)
|
||||
print("[Deltains Claim step2] 'Submit a claim' page loaded")
|
||||
except TimeoutException:
|
||||
page_text = self.driver.execute_script("return document.body.innerText;")[:400]
|
||||
print(f"[Deltains Claim step2] Claim form not detected — page: {page_text}")
|
||||
return "ERROR: step2 failed: 'Submit a claim' page did not load"
|
||||
|
||||
print("[Deltains Claim step2] Clicking 'Statement of completed services' radio...")
|
||||
try:
|
||||
radio = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//input[@type='radio'][following-sibling::*[contains(text(),'Statement of completed services')]] | "
|
||||
"//label[contains(.,'Statement of completed services')]//input[@type='radio'] | "
|
||||
"//input[@type='radio'][ancestor::*[contains(.,'Statement of completed services')]][1]"))
|
||||
)
|
||||
radio.click()
|
||||
time.sleep(1)
|
||||
print("[Deltains Claim step2] Selected 'Statement of completed services'")
|
||||
except TimeoutException:
|
||||
return "ERROR: step2 failed: 'Statement of completed services' radio not found"
|
||||
|
||||
return "SUCCESS"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step2] Exception: {e}")
|
||||
return f"ERROR: step2 failed: {e}"
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Step 4 — Fill service date and procedure lines #
|
||||
# ------------------------------------------------------------------ #
|
||||
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)
|
||||
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"[Deltains Claim step4] {label}: selected '{value}'")
|
||||
except TimeoutException:
|
||||
inp.send_keys(Keys.TAB)
|
||||
print(f"[Deltains Claim step4] {label}: typed '{value}' (no dropdown)")
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step4] Warning: could not fill {label}: {e}")
|
||||
|
||||
def _fill_by_id(self, element_id, value, label="field"):
|
||||
"""Fill a plain text/tel input located by its exact id attribute."""
|
||||
try:
|
||||
inp = WebDriverWait(self.driver, 8).until(
|
||||
EC.element_to_be_clickable((By.ID, element_id))
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", inp)
|
||||
inp.click()
|
||||
inp.send_keys(Keys.CONTROL + "a")
|
||||
inp.send_keys(Keys.DELETE)
|
||||
inp.send_keys(str(value))
|
||||
time.sleep(0.2)
|
||||
print(f"[Deltains Claim step4] {label}: typed '{value}'")
|
||||
return inp
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step4] Warning: could not fill {label} (#{element_id}): {e}")
|
||||
return None
|
||||
|
||||
def step4_fill_claim_form(self):
|
||||
"""
|
||||
Fill each service line: date of service, procedure code, and fee
|
||||
via their per-line ids (serviceLines.{idx}.dateOfService,
|
||||
serviceLines.{idx}.procedure, serviceLines.{idx}.fee — confirmed
|
||||
against the live Deltains claim page). Tooth/arch/quad/surface
|
||||
selectors below remain unverified.
|
||||
"""
|
||||
try:
|
||||
active_lines = [ln for ln in self.serviceLines if str(ln.get("procedureCode") or "").strip()]
|
||||
print(f"[Deltains 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"[Deltains Claim step4] Line {idx}: code={code} tooth={tooth!r} arch={arch!r} "
|
||||
f"quad={quad!r} surface={surface!r} billed={billed!r}")
|
||||
|
||||
if idx > 0:
|
||||
try:
|
||||
add_btn = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//button[contains(normalize-space(text()),'Add service')]"
|
||||
))
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", add_btn)
|
||||
add_btn.click()
|
||||
print(f"[Deltains Claim step4] Clicked 'Add service' for line {idx}")
|
||||
time.sleep(1)
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step4] Could not click 'Add service': {e}")
|
||||
|
||||
formatted_dos = self._format_dob(self.serviceDate) if self.serviceDate else ""
|
||||
if formatted_dos:
|
||||
self._fill_by_id(f"serviceLines.{idx}.dateOfService", formatted_dos, f"dateOfService[{idx}]")
|
||||
|
||||
if code:
|
||||
try:
|
||||
proc_inp = WebDriverWait(self.driver, 8).until(
|
||||
EC.element_to_be_clickable((By.ID, f"serviceLines.{idx}.procedure"))
|
||||
)
|
||||
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"[Deltains Claim step4] Could not fill procedure code: {e}")
|
||||
|
||||
# Tooth number — MUI single-select (only rendered for codes
|
||||
# that require it, e.g. fillings like D2392)
|
||||
if tooth:
|
||||
try:
|
||||
tooth_select = WebDriverWait(self.driver, 8).until(
|
||||
EC.element_to_be_clickable((By.ID, f"mui-component-select-serviceLines.{idx}.toothNumber"))
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", tooth_select)
|
||||
tooth_select.click()
|
||||
time.sleep(0.5)
|
||||
padded = tooth.zfill(2)
|
||||
option = WebDriverWait(self.driver, 5).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
f"//li[@role='option'][normalize-space(.)='{padded}'] | "
|
||||
f"//li[@role='option'][normalize-space(.)='{tooth}']"))
|
||||
)
|
||||
option.click()
|
||||
time.sleep(0.3)
|
||||
print(f"[Deltains Claim step4] tooth[{idx}]: selected '{tooth}'")
|
||||
except TimeoutException:
|
||||
print(f"[Deltains Claim step4] tooth[{idx}]: select not present for this procedure code")
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step4] Could not fill tooth number: {e}")
|
||||
|
||||
# Tooth surfaces — MUI multi-select; each letter (B, D, L, M, O...)
|
||||
# is a checkbox option that stays open until clicked elsewhere.
|
||||
if surface:
|
||||
try:
|
||||
surface_select = WebDriverWait(self.driver, 8).until(
|
||||
EC.element_to_be_clickable((By.ID, f"serviceLines.{idx}.toothSurfaces"))
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", surface_select)
|
||||
surface_select.click()
|
||||
time.sleep(0.5)
|
||||
surface_letters = [c for c in surface if c.isalpha()]
|
||||
for letter in surface_letters:
|
||||
try:
|
||||
opt = WebDriverWait(self.driver, 5).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
f"//li[@role='option'][normalize-space(.)='{letter}']"))
|
||||
)
|
||||
opt.click()
|
||||
time.sleep(0.2)
|
||||
print(f"[Deltains Claim step4] surface[{idx}]: checked '{letter}'")
|
||||
except TimeoutException:
|
||||
print(f"[Deltains Claim step4] surface[{idx}]: option '{letter}' not found")
|
||||
self.driver.find_element(By.TAG_NAME, "body").send_keys(Keys.ESCAPE)
|
||||
time.sleep(0.3)
|
||||
except TimeoutException:
|
||||
print(f"[Deltains Claim step4] surface[{idx}]: select not present for this procedure code")
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step4] Could not fill tooth surfaces: {e}")
|
||||
|
||||
if billed:
|
||||
self._fill_by_id(f"serviceLines.{idx}.fee", billed, f"fee[{idx}]")
|
||||
time.sleep(0.2)
|
||||
|
||||
print("[Deltains Claim step4] Done")
|
||||
return "SUCCESS"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step4] Exception: {e}")
|
||||
return f"ERROR: step4 failed: {e}"
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Step 5 — Attach files #
|
||||
# ------------------------------------------------------------------ #
|
||||
def step5_attach_files(self):
|
||||
"""
|
||||
Expand the 'Attachments' accordion section under 'Additional
|
||||
information' (max 5 files, PDF/JPEG only), then for each claimFile
|
||||
with a filePath, send it directly to the hidden file input behind
|
||||
the 'Choose a file' label.
|
||||
"""
|
||||
if not self.claimFiles:
|
||||
print("[Deltains Claim step5] No files to attach")
|
||||
return "SUCCESS"
|
||||
|
||||
try:
|
||||
section = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//*[contains(text(),'Attachments') and contains(text(),'x-rays')]"))
|
||||
)
|
||||
if section.get_attribute("aria-expanded") == "false":
|
||||
section.click()
|
||||
time.sleep(1)
|
||||
print("[Deltains Claim step5] Expanded 'Attachments' section")
|
||||
except TimeoutException:
|
||||
print("[Deltains Claim step5] 'Attachments' section header not found, "
|
||||
"assuming already expanded")
|
||||
|
||||
attached = 0
|
||||
for cf in self.claimFiles:
|
||||
relative_path = cf.get("filePath") or ""
|
||||
if not relative_path:
|
||||
print(f"[Deltains Claim step5] Skipping file with no filePath: {cf}")
|
||||
continue
|
||||
|
||||
abs_path = os.path.normpath(os.path.join(_BACKEND_CWD, relative_path.lstrip("/")))
|
||||
if not os.path.isfile(abs_path):
|
||||
print(f"[Deltains Claim step5] File not found on disk: {abs_path}")
|
||||
continue
|
||||
|
||||
print(f"[Deltains Claim step5] Attaching: {abs_path}")
|
||||
try:
|
||||
file_input = WebDriverWait(self.driver, 8).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//label[contains(@class,'dd-file-uploader-btn')]//input[@type='file'] | "
|
||||
"//input[@type='file']"))
|
||||
)
|
||||
self.driver.execute_script(
|
||||
"arguments[0].style.display='block'; arguments[0].style.visibility='visible';",
|
||||
file_input)
|
||||
file_input.send_keys(abs_path)
|
||||
time.sleep(1.5)
|
||||
print(f"[Deltains Claim step5] Attached: {os.path.basename(abs_path)}")
|
||||
attached += 1
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step5] Could not attach {abs_path}: {e}")
|
||||
|
||||
print(f"[Deltains Claim step5] Attached {attached}/{len(self.claimFiles)} file(s)")
|
||||
return "SUCCESS"
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Step 6 — Click Review #
|
||||
# ------------------------------------------------------------------ #
|
||||
def step6_click_next(self):
|
||||
"""Click the 'Review' button (data-testid='ctaReviewButton') to
|
||||
advance from the claim-entry form to the review screen."""
|
||||
try:
|
||||
print(f"[Deltains Claim step6] Current URL: {self.driver.current_url}")
|
||||
btn = WebDriverWait(self.driver, 15).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//button[@data-testid='ctaReviewButton']"
|
||||
))
|
||||
)
|
||||
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("[Deltains Claim step6] Clicked 'Review'")
|
||||
time.sleep(2)
|
||||
print(f"[Deltains Claim step6] URL after Review: {self.driver.current_url}")
|
||||
return "SUCCESS"
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step6] Exception: {e}")
|
||||
return f"ERROR: step6 failed: {e}"
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Step 7 — Claims summary: check acknowledgement + submit #
|
||||
# ------------------------------------------------------------------ #
|
||||
def step7_submit_claim(self):
|
||||
"""Click the 'Submit' button (data-testid='ctaSubmitButton') on the
|
||||
review page to finalize the claim."""
|
||||
try:
|
||||
print(f"[Deltains Claim step7] Current URL: {self.driver.current_url}")
|
||||
|
||||
submit_btn = WebDriverWait(self.driver, 15).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//button[@data-testid='ctaSubmitButton']"
|
||||
))
|
||||
)
|
||||
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("[Deltains Claim step7] Clicked 'Submit'")
|
||||
time.sleep(2)
|
||||
print(f"[Deltains Claim step7] URL after submit: {self.driver.current_url}")
|
||||
return "SUCCESS"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step7] Exception: {e}")
|
||||
return f"ERROR: step7 failed: {e}"
|
||||
|
||||
def step8_save_confirmation_pdf(self):
|
||||
"""
|
||||
Wait for the 'Claim submitted successfully' confirmation modal,
|
||||
extract the claim identification number, save the page as PDF
|
||||
via CDP.
|
||||
"""
|
||||
try:
|
||||
WebDriverWait(self.driver, 30).until(
|
||||
lambda d: "claim submitted successfully" in d.page_source.lower()
|
||||
)
|
||||
time.sleep(2)
|
||||
print(f"[Deltains Claim step8] Confirmation page URL: {self.driver.current_url}")
|
||||
|
||||
claim_number = None
|
||||
try:
|
||||
body_text = self.driver.find_element(By.TAG_NAME, "body").text
|
||||
match = re.search(r'claim identification number\s*:\s*(\d+)', body_text, re.IGNORECASE)
|
||||
if match:
|
||||
claim_number = match.group(1)
|
||||
print(f"[Deltains Claim step8] Extracted claim number: {claim_number}")
|
||||
else:
|
||||
match = re.search(r'\b(\d{10,})\b', body_text)
|
||||
if match:
|
||||
claim_number = match.group(1)
|
||||
print(f"[Deltains Claim step8] Extracted claim number (fallback): {claim_number}")
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step8] Could not extract claim number: {e}")
|
||||
|
||||
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"deltains_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"[Deltains Claim step8] PDF saved: {pdf_path}")
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step8] PDF capture failed: {e}")
|
||||
return f"ERROR: step8 PDF failed: {e}"
|
||||
|
||||
print("[Deltains Claim step8] Clicking 'My claims'...")
|
||||
try:
|
||||
my_claims_btn = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//button[contains(normalize-space(text()),'My claims')]"))
|
||||
)
|
||||
my_claims_btn.click()
|
||||
time.sleep(3)
|
||||
print(f"[Deltains Claim step8] Navigated to My claims: {self.driver.current_url}")
|
||||
except TimeoutException:
|
||||
print("[Deltains Claim step8] 'My claims' button not found — leaving confirmation modal open")
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"pdf_path": pdf_path,
|
||||
"claimNumber": claim_number,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[Deltains Claim step8] Exception: {e}")
|
||||
return f"ERROR: step8 failed: {e}"
|
||||
Reference in New Issue
Block a user