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) # Check "Remember this device" so future logins skip OTP. # Confirmed against live site (2026-09-15): # try: remember_checkbox = driver.find_element(By.CSS_SELECTOR, "[data-cy='data-checkbox-rememberDevice']") if remember_checkbox.get_attribute("aria-checked") != "true": driver.execute_script( "arguments[0].scrollIntoView({block:'center'});", remember_checkbox) remember_checkbox.click() print("[UnitedUcard Claim OTP] Checked 'Remember this device'") except Exception as remember_err: print(f"[UnitedUcard Claim OTP] Could not check 'Remember this device': {remember_err}") 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, }