Gitead/DentalManagement09
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add United Ucard claim worker and fix batch-column site-key routing

Browse Source 
Adds a "United Ucard Claim" button (claims/preauth insurance form) that
submits through a full Selenium pipeline mirroring the Tufts SCO claim
worker: frontend handler -> backend route/queue processor -> Python
agent, reusing the persistent login/OTP session from the existing
United Ucard eligibility check. Implements the UHC Dental claim wizard
end-to-end (Provider Information, Patient Information search/select,
Claim Information procedures + questions, Upload Documents, Review and
Sign, Submit); the final confirmation-page PDF/claim-number capture is
still a TODO stub pending that page's DOM.

Also wires United Ucard into the AI chatbot's eligibility flow
(insuranceAliases.json, siteKeyToAutoCheck, clarification options), and
fixes a bug where the AI-column batch claim feature silently defaulted
United Ucard patients to MassHealth because batchColumnDeriveSiteKey
had no matching branch and no job-dispatch case for UNITED_UCARD.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Gitead
2026-09-14 23:27:39 -04:00
co-authored by Claude Sonnet 5
parent f30016e674
commit e570a0aa13
15 changed files with 1695 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/SeleniumService/helpers_uniteducard_claim.py
+322
View File
@@ -0,0 +1,322 @@
import os
import time
import asyncio
from typing import Dict, Any
from selenium.webdriver.common.by import By
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_UnitedUcard_claimSubmitWorker import AutomationUnitedUcardClaimSubmit
sessions: Dict[str, Dict[str, Any]] = {}
SESSION_OTP_TIMEOUT = int(os.getenv("SESSION_OTP_TIMEOUT", "120"))
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()
# NOTE: Do NOT quit driver - the persistent browser (uniteducard_browser_manager)
# is shared with the eligibility check worker; keep it alive for the next run.
finally:
sessions.pop(sid, None)
print(f"[helpers_uniteducard_claim] cleaned session {sid}")
async def _remove_session_later(sid: str, delay: int = 20):
await asyncio.sleep(delay)
await cleanup_session(sid)
def _minimize_browser(bot):
try:
if bot and bot.driver:
try:
bot.driver.get("about:blank")
except Exception:
pass
try:
bot.driver.minimize_window()
print("[UnitedUcard Claim] Browser minimized after error")
return
except Exception:
pass
try:
bot.driver.set_window_position(-10000, -10000)
print("[UnitedUcard Claim] Browser moved off-screen after error")
except Exception:
pass
except Exception as e:
print(f"[UnitedUcard Claim] Could not hide browser: {e}")
async def start_uniteducard_claim_run(sid: str, data: dict, url: str):
"""
Run the United Ucard (UHC Dental) claim workflow.
Login/OTP handling mirrors helpers_uniteducard_eligibility.py — same
persistent browser/profile, so a device already trusted by a prior
eligibility check (or claim) skips OTP here too.
Claim-specific steps call selenium_UnitedUcard_claimSubmitWorker (currently
TODO stubs — see that file).
"""
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 = AutomationUnitedUcardClaimSubmit(data)
bot.config_driver()
s["bot"] = bot
s["driver"] = bot.driver
s["last_activity"] = time.time()
try:
bot.driver.maximize_window()
bot.driver.get(url)
await asyncio.sleep(1)
except Exception as e:
s["status"] = "error"
s["message"] = f"Navigation failed: {e}"
await cleanup_session(sid)
return {"status": "error", "message": s["message"]}
# --- Login ---
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"])
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"])
return {"status": "error", "message": s["message"]}
# --- Already logged in ---
if isinstance(login_result, str) and login_result == "ALREADY_LOGGED_IN":
print("[UnitedUcard Claim] Session persisted — skipping OTP")
s["status"] = "running"
s["message"] = "Session persisted"
# --- OTP required ---
elif isinstance(login_result, str) and login_result == "OTP_REQUIRED":
s["status"] = "waiting_for_otp"
s["message"] = "OTP required for login — please enter OTP"
s["last_activity"] = time.time()
driver = s["driver"]
max_polls = SESSION_OTP_TIMEOUT
login_success = False
print(f"[UnitedUcard 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"[UnitedUcard Claim OTP poll {poll+1}] OTP received from app, typing...")
try:
otp_input = driver.find_element(By.XPATH,
"//input[@id='otpBox' or contains(@id,'otp') or contains(@name,'otp') or "
"contains(@name,'code') or @type='tel' or contains(@aria-label,'Verification') or "
"contains(@placeholder,'code')]"
)
otp_input.clear()
otp_input.send_keys(otp_value)
try:
try:
verify_btn = driver.find_element(By.ID, "continuebtn")
except Exception:
verify_btn = driver.find_element(By.XPATH,
"//button[@type='submit' or contains(text(),'Verify') or "
"contains(text(),'Submit') or contains(text(),'Continue')]"
)
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", verify_btn)
verify_btn.click()
print("[UnitedUcard Claim OTP] Clicked verify button")
except Exception:
otp_input.send_keys("\n")
print("[UnitedUcard Claim OTP] Pressed Enter as fallback")
print("[UnitedUcard Claim OTP] OTP typed and submitted via app")
s["otp_value"] = None
await asyncio.sleep(3)
except Exception as type_err:
print(f"[UnitedUcard Claim OTP] Failed to type OTP from app: {type_err}")
current_url = driver.current_url.lower()
print(f"[UnitedUcard Claim OTP Poll {poll+1}/{max_polls}] URL: {current_url[:60]}...")
if "secure.uhcdental.com" in current_url:
print("[UnitedUcard Claim OTP] Reached UHC Dental dashboard - login successful!")
login_success = True
break
try:
driver.find_element(By.XPATH,
"//input[@id='otpBox' or contains(@id,'otp') or contains(@name,'otp') or "
"contains(@name,'code') or @type='tel' or contains(@aria-label,'Verification') or "
"contains(@placeholder,'code') or contains(@placeholder,'Code')]"
)
print(f"[UnitedUcard Claim OTP Poll {poll+1}] OTP input still visible - waiting...")
except Exception:
print(f"[UnitedUcard Claim OTP Poll {poll+1}] OTP input not found, continuing to poll...")
except Exception as poll_err:
print(f"[UnitedUcard Claim OTP Poll {poll+1}] Error: {poll_err}")
if not login_success:
s["status"] = "error"
s["message"] = "OTP timeout - login not completed"
await cleanup_session(sid)
return {"status": "error", "message": "OTP not completed in time"}
s["status"] = "running"
s["message"] = "Login successful after OTP"
print("[UnitedUcard Claim OTP] Proceeding to claim steps...")
elif isinstance(login_result, str) and login_result.startswith("ERROR"):
s["status"] = "error"
s["message"] = login_result
await cleanup_session(sid)
return {"status": "error", "message": login_result}
elif isinstance(login_result, str) and login_result == "SUCCESS":
print("[UnitedUcard Claim] Login succeeded without OTP")
s["status"] = "running"
s["message"] = "Login succeeded"
# --- Claim steps (Dental Claims wizard) ---
for step_name, step_fn in [
("step1_open_claims_page", bot.step1_open_claims_page),
("step2_confirm_provider_information", bot.step2_confirm_provider_information),
("step3_fill_patient_information", bot.step3_fill_patient_information),
("step4_fill_claim_information", bot.step4_fill_claim_information),
("step5_upload_documents", bot.step5_upload_documents),
("step6_review_and_sign", bot.step6_review_and_sign),
]:
result = step_fn()
print(f"[UnitedUcard Claim] {step_name} result: {result}")
if isinstance(result, str) and result.startswith("ERROR"):
s["status"] = "error"
s["message"] = result
_minimize_browser(bot)
asyncio.create_task(_remove_session_later(sid, 30))
return {"status": "error", "message": result}
# --- PDF + claim number ---
step7_result = bot.step7_save_confirmation_pdf()
print(f"[UnitedUcard Claim] step7 result: {step7_result}")
if isinstance(step7_result, str) and step7_result.startswith("ERROR"):
print(f"[UnitedUcard Claim] step7 warning (non-fatal): {step7_result}")
step7_result = {}
pdf_path = step7_result.get("pdf_path") if isinstance(step7_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"[UnitedUcard Claim] pdf_url: {pdf_url}")
claim_number = step7_result.get("claimNumber") if isinstance(step7_result, dict) else None
result = {
"status": "success",
"message": "United Ucard claim submitted successfully",
"claimNumber": claim_number,
"pdf_url": pdf_url,
}
s["status"] = "completed"
s["result"] = result
s["message"] = "completed"
try:
bot.driver.get("about:blank")
try:
bot.driver.minimize_window()
except Exception:
try:
bot.driver.set_window_position(-10000, -10000)
except Exception:
pass
print("[UnitedUcard Claim] Browser minimized - session preserved for next run")
except Exception as close_err:
print(f"[UnitedUcard Claim] Could not minimize browser (non-fatal): {close_err}")
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}"
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,
}