feat: United/DentalHub + Tufts SCO pre-auth, cloud storage search & file thumbnails
- Add United/DentalHub pre-authorization Selenium worker, helpers, backend route/processor/client, and frontend OTP modal + wired button - Add Tufts SCO pre-authorization Selenium worker, helpers, backend route/processor/client, and frontend OTP modal + wired button - Fix btnSubmitAuthorization selector for UnitedDH preauth step2 - Fix Tufts SCO preauth step3 to target <span>pre-authorization</span> button - Cloud storage search: default to "both" mode so patient folder names match on first search - Cloud storage folder panel: auto-scroll to panel when opened from search results - Cloud storage files: show inline image thumbnails and PDF badge in file grid Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
ff
345
apps/SeleniumService/helpers_tuftssco_preauth.py
Normal file
345
apps/SeleniumService/helpers_tuftssco_preauth.py
Normal file
@@ -0,0 +1,345 @@
|
||||
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_TuftsSCO_preAuthWorker import AutomationTuftsSCOPreAuth
|
||||
|
||||
sessions: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
SESSION_OTP_TIMEOUT = int(os.getenv("SESSION_OTP_TIMEOUT", "120"))
|
||||
|
||||
MEMBERS_URL = "https://providers.dentaquest.com/members"
|
||||
|
||||
|
||||
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_tuftssco_preauth] 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("[TuftsSCO PreAuth] Browser minimized after error")
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
bot.driver.set_window_position(-10000, -10000)
|
||||
print("[TuftsSCO PreAuth] Browser moved off-screen after error")
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"[TuftsSCO PreAuth] Could not hide browser: {e}")
|
||||
|
||||
|
||||
async def start_tuftssco_preauth_run(sid: str, data: dict, url: str):
|
||||
"""
|
||||
Run the Tufts SCO (DentaQuest) pre-authorization workflow.
|
||||
Login/OTP handling mirrors helpers_tuftssco_claim.py exactly.
|
||||
Pre-auth steps call selenium_TuftsSCO_preAuthWorker.
|
||||
"""
|
||||
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 = AutomationTuftsSCOPreAuth(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("[TuftsSCO PreAuth] 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 in browser"
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
driver = s["driver"]
|
||||
max_polls = SESSION_OTP_TIMEOUT
|
||||
login_success = False
|
||||
|
||||
print(f"[TuftsSCO PreAuth OTP] Waiting for user to enter OTP (polling browser for {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"[TuftsSCO PreAuth OTP] OTP received from app: {otp_value}")
|
||||
try:
|
||||
otp_input = driver.find_element(By.XPATH,
|
||||
"//input[contains(@name,'otp') or contains(@name,'code') or @type='tel' or "
|
||||
"contains(@aria-label,'Verification') or contains(@aria-label,'verification') or "
|
||||
"contains(@placeholder,'code') or contains(@placeholder,'Code')]"
|
||||
)
|
||||
otp_input.clear()
|
||||
otp_input.send_keys(otp_value)
|
||||
try:
|
||||
verify_btn = driver.find_element(By.XPATH,
|
||||
"//button[@type='submit'] | "
|
||||
"//button[contains(text(),'Verify') or contains(text(),'Submit') or contains(text(),'Confirm')]"
|
||||
)
|
||||
verify_btn.click()
|
||||
print("[TuftsSCO PreAuth OTP] Clicked verify button")
|
||||
except:
|
||||
otp_input.send_keys("\n")
|
||||
print("[TuftsSCO PreAuth OTP] Pressed Enter as fallback")
|
||||
print("[TuftsSCO PreAuth OTP] OTP typed and submitted via app")
|
||||
s["otp_value"] = None
|
||||
await asyncio.sleep(3)
|
||||
except Exception as type_err:
|
||||
print(f"[TuftsSCO PreAuth OTP] Failed to type OTP from app: {type_err}")
|
||||
|
||||
current_url = driver.current_url.lower()
|
||||
print(f"[TuftsSCO PreAuth OTP Poll {poll+1}/{max_polls}] URL: {current_url[:60]}...")
|
||||
|
||||
if "member" in current_url or "dashboard" in current_url or "eligibility" in current_url or "home" in current_url:
|
||||
try:
|
||||
WebDriverWait(driver, 5).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
'//input[@placeholder="Search by member ID"] | //input[contains(@placeholder,"Search")] | //*[contains(@class,"dashboard")]'
|
||||
))
|
||||
)
|
||||
print("[TuftsSCO PreAuth OTP] Dashboard/search element found - login successful!")
|
||||
login_success = True
|
||||
break
|
||||
except TimeoutException:
|
||||
print("[TuftsSCO PreAuth OTP] On member page but search input not found, continuing to poll...")
|
||||
|
||||
try:
|
||||
driver.find_element(By.XPATH,
|
||||
"//input[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"[TuftsSCO PreAuth OTP Poll {poll+1}] OTP input still visible - waiting...")
|
||||
except:
|
||||
if "login" in current_url or "auth" in current_url:
|
||||
print("[TuftsSCO PreAuth OTP] OTP input gone, trying to navigate to members page...")
|
||||
try:
|
||||
driver.get(MEMBERS_URL)
|
||||
await asyncio.sleep(2)
|
||||
except:
|
||||
pass
|
||||
|
||||
except Exception as poll_err:
|
||||
print(f"[TuftsSCO PreAuth OTP Poll {poll+1}] Error: {poll_err}")
|
||||
|
||||
if not login_success:
|
||||
try:
|
||||
print("[TuftsSCO PreAuth OTP] Final attempt - navigating to members page...")
|
||||
driver.get(MEMBERS_URL)
|
||||
await asyncio.sleep(3)
|
||||
WebDriverWait(driver, 10).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
'//input[@placeholder="Search by member ID"] | //input[contains(@placeholder,"Search")] | //*[contains(@class,"dashboard")]'
|
||||
))
|
||||
)
|
||||
print("[TuftsSCO PreAuth OTP] Members page element found - login successful!")
|
||||
login_success = True
|
||||
except TimeoutException:
|
||||
s["status"] = "error"
|
||||
s["message"] = "OTP timeout - login not completed"
|
||||
await cleanup_session(sid)
|
||||
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)
|
||||
return {"status": "error", "message": s["message"]}
|
||||
|
||||
if login_success:
|
||||
s["status"] = "running"
|
||||
s["message"] = "Login successful after OTP"
|
||||
print("[TuftsSCO PreAuth OTP] Proceeding to pre-auth steps...")
|
||||
|
||||
# ── Login succeeded without OTP ───────────────────────────────────────
|
||||
elif isinstance(login_result, str) and login_result == "SUCCESS":
|
||||
print("[TuftsSCO PreAuth] Login succeeded without OTP")
|
||||
s["status"] = "running"
|
||||
s["message"] = "Login succeeded"
|
||||
|
||||
# ── Login error ───────────────────────────────────────────────────────
|
||||
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}
|
||||
|
||||
# Ensure we're on the members search page before starting
|
||||
try:
|
||||
current_url = bot.driver.current_url
|
||||
if "member" not in current_url.lower():
|
||||
bot.driver.get(MEMBERS_URL)
|
||||
await asyncio.sleep(3)
|
||||
print(f"[TuftsSCO PreAuth] Navigated to members: {bot.driver.current_url}")
|
||||
except Exception as e:
|
||||
print(f"[TuftsSCO PreAuth] Warning: could not ensure members URL: {e}")
|
||||
|
||||
# --- Pre-auth steps ---
|
||||
for step_name, step_fn in [
|
||||
("step1_search_patient", bot.step1_search_patient),
|
||||
("step2_open_member_page", bot.step2_open_member_page),
|
||||
("step3_click_create_preauth", bot.step3_click_create_preauth),
|
||||
("step4_fill_preauth_form", bot.step4_fill_preauth_form),
|
||||
("step5_attach_files", bot.step5_attach_files),
|
||||
("step6_click_next", bot.step6_click_next),
|
||||
("step7_submit_preauth", bot.step7_submit_preauth),
|
||||
]:
|
||||
result = step_fn()
|
||||
print(f"[TuftsSCO PreAuth] {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}
|
||||
|
||||
# --- Step 8: PDF + pre-auth number ---
|
||||
step8_result = bot.step8_save_confirmation_pdf()
|
||||
print(f"[TuftsSCO PreAuth] step8 result: {step8_result}")
|
||||
if isinstance(step8_result, str) and step8_result.startswith("ERROR"):
|
||||
print(f"[TuftsSCO PreAuth] step8 warning (non-fatal): {step8_result}")
|
||||
step8_result = {}
|
||||
|
||||
pdf_path = step8_result.get("pdf_path") if isinstance(step8_result, dict) else None
|
||||
preauth_number = step8_result.get("preAuthNumber") if isinstance(step8_result, dict) else None
|
||||
|
||||
result = {
|
||||
"status": "success",
|
||||
"message": "Tufts SCO pre-authorization submitted successfully",
|
||||
"preAuthNumber": preauth_number,
|
||||
"pdf_path": pdf_path,
|
||||
}
|
||||
s["status"] = "completed"
|
||||
s["result"] = result
|
||||
s["message"] = "completed"
|
||||
|
||||
# Preserve session: navigate away and minimize (do NOT quit the driver)
|
||||
try:
|
||||
if bot and bot.driver:
|
||||
bot.driver.get("about:blank")
|
||||
await asyncio.sleep(0.5)
|
||||
bot.driver.minimize_window()
|
||||
print("[TuftsSCO PreAuth] Session preserved (about:blank + minimize)")
|
||||
except Exception as close_err:
|
||||
print(f"[TuftsSCO PreAuth] Could not preserve session (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 {
|
||||
"status": s.get("status"),
|
||||
"message": s.get("message"),
|
||||
"result": s.get("result"),
|
||||
}
|
||||
Reference in New Issue
Block a user