feat: add United Ucard eligibility check with OTP + persistent session
Adds a full login/OTP/search/capture flow for United Ucard (OneHealthcareID -> UHC Dental provider portal), mirroring the United SCO/DDMA pattern: persistent Chrome profile, OTP polling, member eligibility search, and PDF capture saved to the patient's Eligibility folder. Also fixes a validation bug in the browser-safe patient schema where email/address/city/zipCode/gender/phone were optional but not nullable, rejecting patient records with null values on those fields during edits (e.g. "Expected string, received null"). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -17,6 +17,7 @@ import time
|
||||
import helpers_ddma_eligibility as hddma
|
||||
import helpers_deltains_eligibility as hdeltains
|
||||
import helpers_unitedsco_eligibility as hunitedsco
|
||||
import helpers_uniteducard_eligibility as huniteducard
|
||||
import helpers_dentaquest_eligibility as hdentaquest
|
||||
import helpers_cca_eligibility as hcca
|
||||
import helpers_cca_claim as hcca_claim
|
||||
@@ -32,6 +33,7 @@ import helpers_bcbs_ma_eligibility as hbcbs_ma
|
||||
from ddma_browser_manager import clear_ddma_session_on_startup
|
||||
from deltains_browser_manager import clear_deltains_session_on_startup
|
||||
from unitedsco_browser_manager import clear_unitedsco_session_on_startup
|
||||
from uniteducard_browser_manager import clear_uniteducard_session_on_startup
|
||||
from dentaquest_browser_manager import clear_dentaquest_session_on_startup
|
||||
from cca_browser_manager import clear_cca_session_on_startup
|
||||
|
||||
@@ -45,6 +47,7 @@ print("=" * 50)
|
||||
clear_ddma_session_on_startup()
|
||||
clear_deltains_session_on_startup()
|
||||
clear_unitedsco_session_on_startup()
|
||||
clear_uniteducard_session_on_startup()
|
||||
clear_dentaquest_session_on_startup()
|
||||
clear_cca_session_on_startup()
|
||||
print("=" * 50)
|
||||
@@ -497,6 +500,47 @@ async def unitedsco_eligibility(request: Request):
|
||||
return {"status": "started", "session_id": sid}
|
||||
|
||||
|
||||
async def _uniteducard_worker_wrapper(sid: str, data: dict, url: str):
|
||||
"""Background worker for United Ucard — acquires semaphore, updates counters."""
|
||||
global active_jobs, waiting_jobs
|
||||
async with semaphore:
|
||||
async with lock:
|
||||
waiting_jobs -= 1
|
||||
active_jobs += 1
|
||||
try:
|
||||
await huniteducard.start_uniteducard_run(sid, data, url)
|
||||
finally:
|
||||
async with lock:
|
||||
active_jobs -= 1
|
||||
|
||||
|
||||
@app.post("/uniteducard-eligibility")
|
||||
async def uniteducard_eligibility(request: Request):
|
||||
"""
|
||||
Starts a United Ucard (OneHealthcareID) eligibility session in the background.
|
||||
Body: { "data": { ... } }
|
||||
Returns: { status: "started", session_id: "<uuid>" }
|
||||
"""
|
||||
global waiting_jobs
|
||||
|
||||
body = await request.json()
|
||||
data = body.get("data", {})
|
||||
|
||||
sid = huniteducard.make_session_entry()
|
||||
huniteducard.sessions[sid]["type"] = "uniteducard_eligibility"
|
||||
huniteducard.sessions[sid]["last_activity"] = time.time()
|
||||
|
||||
async with lock:
|
||||
waiting_jobs += 1
|
||||
|
||||
asyncio.create_task(_uniteducard_worker_wrapper(
|
||||
sid, data,
|
||||
url="https://www.uhcdental.com/content/provider/dental.html?state=abc"
|
||||
))
|
||||
|
||||
return {"status": "started", "session_id": sid}
|
||||
|
||||
|
||||
async def _dentaquest_worker_wrapper(sid: str, data: dict, url: str):
|
||||
"""Background worker for DentaQuest (Tufts SCO) — acquires semaphore, updates counters."""
|
||||
global active_jobs, waiting_jobs
|
||||
@@ -932,6 +976,8 @@ async def submit_otp(request: Request):
|
||||
res = hdeltains.submit_otp(sid, otp)
|
||||
elif sid in hunitedsco.sessions:
|
||||
res = hunitedsco.submit_otp(sid, otp)
|
||||
elif sid in huniteducard.sessions:
|
||||
res = huniteducard.submit_otp(sid, otp)
|
||||
elif sid in hdentaquest.sessions:
|
||||
res = hdentaquest.submit_otp(sid, otp)
|
||||
elif sid in hddma_claim.sessions:
|
||||
@@ -963,6 +1009,8 @@ async def session_status(sid: str):
|
||||
s = hdeltains.get_session_status(sid)
|
||||
elif sid in hunitedsco.sessions:
|
||||
s = hunitedsco.get_session_status(sid)
|
||||
elif sid in huniteducard.sessions:
|
||||
s = huniteducard.get_session_status(sid)
|
||||
elif sid in hdentaquest.sessions:
|
||||
s = hdentaquest.get_session_status(sid)
|
||||
elif sid in hcca.sessions:
|
||||
@@ -1022,6 +1070,16 @@ async def clear_unitedsco_session_endpoint():
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
|
||||
@app.post("/clear-uniteducard-session")
|
||||
async def clear_uniteducard_session_endpoint():
|
||||
"""Clears the United Ucard browser session. Call when credentials are deleted or changed."""
|
||||
try:
|
||||
clear_uniteducard_session_on_startup()
|
||||
return {"status": "success", "message": "United Ucard session cleared"}
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
|
||||
@app.post("/clear-cca-session")
|
||||
async def clear_cca_session_endpoint():
|
||||
"""Clears the CCA browser session. Call when credentials are deleted or changed."""
|
||||
|
||||
@@ -0,0 +1,323 @@
|
||||
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_eligibilityCheckWorker import AutomationUnitedUcardEligibilityCheck
|
||||
|
||||
# In-memory session store
|
||||
sessions: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
SESSION_OTP_TIMEOUT = int(os.getenv("SESSION_OTP_TIMEOUT", "120")) # seconds
|
||||
|
||||
|
||||
def make_session_entry() -> str:
|
||||
"""Create a new session entry and return its ID."""
|
||||
import uuid
|
||||
sid = str(uuid.uuid4())
|
||||
sessions[sid] = {
|
||||
"status": "created", # created -> running -> waiting_for_otp -> otp_submitted -> completed / error
|
||||
"created_at": time.time(),
|
||||
"last_activity": time.time(),
|
||||
"bot": None, # worker instance
|
||||
"driver": None, # selenium webdriver
|
||||
"otp_event": asyncio.Event(),
|
||||
"otp_value": None,
|
||||
"result": None,
|
||||
"message": None,
|
||||
"type": None,
|
||||
}
|
||||
return sid
|
||||
|
||||
|
||||
async def cleanup_session(sid: str, message: str | None = None):
|
||||
"""
|
||||
Close driver (if any), wake OTP waiter, set final state, and remove session entry.
|
||||
Idempotent: safe to call multiple times.
|
||||
"""
|
||||
s = sessions.get(sid)
|
||||
if not s:
|
||||
return
|
||||
try:
|
||||
try:
|
||||
if s.get("status") not in ("completed", "error", "not_found"):
|
||||
s["status"] = "error"
|
||||
if message:
|
||||
s["message"] = message
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
ev = s.get("otp_event")
|
||||
if ev and not ev.is_set():
|
||||
ev.set()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# NOTE: Do NOT quit driver - keep browser alive for next patient
|
||||
# Browser manager handles the persistent browser instance
|
||||
|
||||
finally:
|
||||
sessions.pop(sid, None)
|
||||
|
||||
|
||||
async def _remove_session_later(sid: str, delay: int = 20):
|
||||
await asyncio.sleep(delay)
|
||||
await cleanup_session(sid)
|
||||
|
||||
|
||||
def _minimize_browser(bot):
|
||||
"""Hide the browser window so it doesn't stay in the user's way."""
|
||||
try:
|
||||
if bot and bot.driver:
|
||||
try:
|
||||
bot.driver.get("about:blank")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
bot.driver.minimize_window()
|
||||
print("[UnitedUcard] Browser minimized after error")
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
bot.driver.set_window_position(-10000, -10000)
|
||||
print("[UnitedUcard] Browser moved off-screen after error")
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard] Could not hide browser: {e}")
|
||||
|
||||
|
||||
async def start_uniteducard_run(sid: str, data: dict, url: str):
|
||||
"""
|
||||
Run the United Ucard workflow for a session (WITHOUT managing semaphore/counters).
|
||||
Called by agent.py inside a wrapper that handles queue/counters.
|
||||
"""
|
||||
s = sessions.get(sid)
|
||||
if not s:
|
||||
return {"status": "error", "message": "session not found"}
|
||||
|
||||
s["status"] = "running"
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
try:
|
||||
bot = AutomationUnitedUcardEligibilityCheck({"data": data})
|
||||
bot.config_driver()
|
||||
|
||||
s["bot"] = bot
|
||||
s["driver"] = bot.driver
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
# Navigate to login URL
|
||||
try:
|
||||
if not url:
|
||||
raise ValueError("URL not provided for United Ucard run")
|
||||
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 - session persisted from profile, skip to step1
|
||||
if isinstance(login_result, str) and login_result == "ALREADY_LOGGED_IN":
|
||||
s["status"] = "running"
|
||||
s["message"] = "Session persisted"
|
||||
print("[start_uniteducard_run] Session persisted - skipping OTP")
|
||||
|
||||
# OTP required path - POLL THE BROWSER to detect when user enters OTP
|
||||
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"[UnitedUcard 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:
|
||||
# Check if OTP was submitted via API (from app)
|
||||
otp_value = s.get("otp_value")
|
||||
if otp_value:
|
||||
print(f"[UnitedUcard OTP] OTP received from app: {otp_value}")
|
||||
try:
|
||||
# Confirmed against live site (2026-09-13): #/rba/options/text page
|
||||
# <input id="otpBox" data-testid="otpBox" maxlength="10" type="text">
|
||||
# <button id="continuebtn" data-cy="data-continuebtn-field">Continue</button>
|
||||
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 the confirmed id first — the broader text/type
|
||||
# fallback below can match the WRONG button if more
|
||||
# than one element on the page qualifies (find_element
|
||||
# returns the first DOM match, not necessarily the
|
||||
# right one), which was landing on a blank page.
|
||||
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 OTP] Clicked verify button")
|
||||
except Exception:
|
||||
otp_input.send_keys("\n") # Press Enter as fallback
|
||||
print("[UnitedUcard OTP] Pressed Enter as fallback")
|
||||
print("[UnitedUcard OTP] OTP typed and submitted via app")
|
||||
s["otp_value"] = None # Clear so we don't submit again
|
||||
await asyncio.sleep(3) # Wait for verification
|
||||
except Exception as type_err:
|
||||
print(f"[UnitedUcard OTP] Failed to type OTP from app: {type_err}")
|
||||
|
||||
# Check current URL - success lands on the UHC Dental provider dashboard
|
||||
# (secure.uhcdental.com/.../postloginhomescreen.html)
|
||||
current_url = driver.current_url.lower()
|
||||
print(f"[UnitedUcard OTP Poll {poll+1}/{max_polls}] URL: {current_url[:60]}...")
|
||||
|
||||
if "secure.uhcdental.com" in current_url:
|
||||
print("[UnitedUcard OTP] Reached UHC Dental dashboard - login successful!")
|
||||
login_success = True
|
||||
break
|
||||
|
||||
# Also check if OTP input is still visible
|
||||
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 OTP Poll {poll+1}] OTP input still visible - waiting...")
|
||||
except Exception:
|
||||
print(f"[UnitedUcard OTP Poll {poll+1}] OTP input not found, continuing to poll...")
|
||||
|
||||
except Exception as poll_err:
|
||||
print(f"[UnitedUcard 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"}
|
||||
|
||||
if login_success:
|
||||
s["status"] = "running"
|
||||
s["message"] = "Login successful after OTP"
|
||||
print("[UnitedUcard OTP] Proceeding to step1...")
|
||||
|
||||
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}
|
||||
|
||||
# Login succeeded without OTP (SUCCESS)
|
||||
elif isinstance(login_result, str) and login_result == "SUCCESS":
|
||||
print("[start_uniteducard_run] Login succeeded without OTP")
|
||||
s["status"] = "running"
|
||||
s["message"] = "Login succeeded"
|
||||
|
||||
# Step 1
|
||||
step1_result = bot.step1()
|
||||
if isinstance(step1_result, str) and step1_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = step1_result
|
||||
s["result"] = {"status": "error", "message": step1_result}
|
||||
_minimize_browser(bot)
|
||||
asyncio.create_task(_remove_session_later(sid, 30))
|
||||
return {"status": "error", "message": step1_result}
|
||||
|
||||
# Step 2 (PDF)
|
||||
step2_result = bot.step2()
|
||||
if isinstance(step2_result, dict) and step2_result.get("status") == "success":
|
||||
s["status"] = "completed"
|
||||
s["result"] = step2_result
|
||||
s["message"] = "completed"
|
||||
asyncio.create_task(_remove_session_later(sid, 30))
|
||||
return step2_result
|
||||
else:
|
||||
s["status"] = "error"
|
||||
if isinstance(step2_result, dict):
|
||||
s["message"] = step2_result.get("message", "unknown error")
|
||||
else:
|
||||
s["message"] = str(step2_result)
|
||||
s["result"] = {"status": "error", "message": s["message"]}
|
||||
_minimize_browser(bot)
|
||||
asyncio.create_task(_remove_session_later(sid, 30))
|
||||
return {"status": "error", "message": s["message"]}
|
||||
|
||||
except Exception as e:
|
||||
s["status"] = "error"
|
||||
s["message"] = f"worker exception: {e}"
|
||||
try:
|
||||
if bot and bot.driver:
|
||||
bot.driver.minimize_window()
|
||||
except Exception:
|
||||
pass
|
||||
s["result"] = {"status": "error", "message": s["message"]}
|
||||
asyncio.create_task(_remove_session_later(sid, 30))
|
||||
return {"status": "error", "message": s["message"]}
|
||||
|
||||
|
||||
def submit_otp(sid: str, otp: str) -> Dict[str, Any]:
|
||||
"""Set OTP for a session and wake waiting runner."""
|
||||
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,632 @@
|
||||
"""
|
||||
United Ucard (OneHealthcareID) eligibility check worker.
|
||||
|
||||
Scaffolded from selenium_UnitedSCO_eligibilityCheckWorker.py — same shape
|
||||
(config_driver / login / step1 / step2 / main_workflow) so it plugs into the
|
||||
same helpers_*_eligibility.py + agent.py session/OTP pattern.
|
||||
|
||||
Login flow (confirmed against the live site, 2026-09-13):
|
||||
0. https://www.uhcdental.com/content/provider/dental.html?state=abc (public site)
|
||||
span.header-signin-btn ("Sign In") -> a.sign-in-button-link ("Provider Login")
|
||||
which OIDC-redirects to identity.onehealthcareid.com (or straight back to the
|
||||
dashboard if there's already a valid SSO session)
|
||||
1. https://identity.onehealthcareid.com/oneapp/index.html#/login
|
||||
input#username -> button#btnLogin ("Continue")
|
||||
2. #/signin/password
|
||||
input#login-pwd -> button#btnLogin ("Continue")
|
||||
3. #/rba/options ("Verify Your Identity")
|
||||
button#textMsg ("Via Text Message")
|
||||
4. #/rba/options/text ("Access Code")
|
||||
input#otpBox -> button#continuebtn ("Continue")
|
||||
5. Lands on the UHC Dental provider dashboard:
|
||||
https://secure.uhcdental.com/.../postloginhomescreen.html?...
|
||||
|
||||
step1()/step2() (the actual eligibility search + PDF capture, once logged in)
|
||||
are left as TODO stubs — fill these in once we know the exact DOM for the
|
||||
member search + eligibility result pages on the dashboard.
|
||||
"""
|
||||
from selenium.common.exceptions import WebDriverException, TimeoutException
|
||||
from selenium.webdriver.common.by import By
|
||||
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 uniteducard_browser_manager import get_browser_manager
|
||||
|
||||
|
||||
class AutomationUnitedUcardEligibilityCheck:
|
||||
def __init__(self, data):
|
||||
self.headless = False
|
||||
self.driver = None
|
||||
|
||||
self.data = data.get("data", {}) if isinstance(data, dict) else {}
|
||||
|
||||
# Flatten values for convenience
|
||||
self.memberId = self.data.get("memberId", "")
|
||||
self.dateOfBirth = self.data.get("dateOfBirth", "")
|
||||
self.firstName = self.data.get("firstName", "")
|
||||
self.lastName = self.data.get("lastName", "")
|
||||
self.uniteducard_username = self.data.get("uniteducardUsername", "")
|
||||
self.uniteducard_password = self.data.get("uniteducardPassword", "")
|
||||
self.payment_group_id = self.data.get("paymentGroupId", "")
|
||||
|
||||
# Use browser manager's download dir
|
||||
self.download_dir = get_browser_manager().download_dir
|
||||
os.makedirs(self.download_dir, exist_ok=True)
|
||||
|
||||
def config_driver(self):
|
||||
# Use persistent browser from manager (keeps device trust tokens)
|
||||
self.driver = get_browser_manager().get_driver(self.headless)
|
||||
|
||||
def _force_logout(self):
|
||||
"""Force logout by clearing cookies for the One Healthcare ID domain."""
|
||||
try:
|
||||
print("[UnitedUcard login] Forcing logout due to credential change...")
|
||||
browser_manager = get_browser_manager()
|
||||
|
||||
try:
|
||||
self.driver.delete_all_cookies()
|
||||
print("[UnitedUcard login] Cleared all cookies")
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard login] Error clearing cookies: {e}")
|
||||
|
||||
browser_manager.clear_credentials_hash()
|
||||
print("[UnitedUcard login] Logout complete")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard login] Error during forced logout: {e}")
|
||||
return False
|
||||
|
||||
def login(self, url):
|
||||
"""
|
||||
Log in via https://www.uhcdental.com/content/provider/dental.html (Sign In ->
|
||||
Provider Login), which OIDC-redirects through identity.onehealthcareid.com.
|
||||
|
||||
Returns one of:
|
||||
"ALREADY_LOGGED_IN" / "SUCCESS" - logged in, caller should proceed to step1
|
||||
"OTP_REQUIRED" - caller should poll for OTP entry
|
||||
"ERROR: <msg>" - login failed
|
||||
"""
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
browser_manager = get_browser_manager()
|
||||
|
||||
try:
|
||||
# Check if credentials have changed - if so, force logout first
|
||||
if self.uniteducard_username and browser_manager.credentials_changed(self.uniteducard_username):
|
||||
self._force_logout()
|
||||
self.driver.get(url)
|
||||
time.sleep(2)
|
||||
|
||||
# Navigate to the public UHC Dental provider portal (NOT directly to
|
||||
# identity.onehealthcareid.com) — Sign In -> Provider Login triggers
|
||||
# the OIDC redirect to the identity login page.
|
||||
self.driver.get(url)
|
||||
time.sleep(3)
|
||||
|
||||
current_url = self.driver.current_url
|
||||
print(f"[UnitedUcard login] Current URL: {current_url}")
|
||||
|
||||
# If we've already landed on the UHC Dental provider dashboard
|
||||
# (secure.uhcdental.com/.../postloginhomescreen.html), a prior
|
||||
# session is still valid.
|
||||
if "secure.uhcdental.com" in current_url.lower():
|
||||
print("[UnitedUcard login] Already logged in - on UHC Dental dashboard")
|
||||
return "ALREADY_LOGGED_IN"
|
||||
|
||||
# ── Step 0: public site -> Sign In -> Provider Login ────────────
|
||||
# Confirmed against live site (2026-09-13):
|
||||
# <span class="header-signin-btn">Sign In</span> (opens dropdown)
|
||||
# <a class="sign-in-button-link" href="https://identity.onehealthcareid.com/
|
||||
# oidc/authorize?client_id=DBP92444&response_type=code&scope=openid+profile+email&
|
||||
# redirect_uri=https://secure.uhcdental.com/content/dental-benefits-provider/
|
||||
# en/secure/postloginhomescreen.html">Provider Login</a>
|
||||
if "onehealthcareid.com" not in current_url.lower():
|
||||
try:
|
||||
signin_btn = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//span[contains(@class,'header-signin-btn') or contains(text(),'Sign In')]"))
|
||||
)
|
||||
signin_btn.click()
|
||||
print("[UnitedUcard login] Clicked 'Sign In' to open dropdown")
|
||||
time.sleep(1)
|
||||
except TimeoutException:
|
||||
return "ERROR: 'Sign In' button not found on UHC Dental provider site"
|
||||
|
||||
# Match by exact visible text "Provider Login" — the dropdown
|
||||
# also has an "Administrator Login" link that shares the same
|
||||
# class AND the same oidc/authorize href pattern (just a
|
||||
# different client_id), so text is the only reliable
|
||||
# differentiator. It's also always the first item in the
|
||||
# dropdown (Provider Login above Administrator Login).
|
||||
clicked_provider_login = False
|
||||
try:
|
||||
provider_login_link = WebDriverWait(self.driver, 10).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//a[normalize-space(text())='Provider Login']"))
|
||||
)
|
||||
self.driver.execute_script(
|
||||
"arguments[0].scrollIntoView({block:'center'});", provider_login_link
|
||||
)
|
||||
time.sleep(0.3)
|
||||
try:
|
||||
provider_login_link.click()
|
||||
print("[UnitedUcard login] Clicked 'Provider Login' (direct)")
|
||||
clicked_provider_login = True
|
||||
except Exception as direct_click_err:
|
||||
print(f"[UnitedUcard login] Direct click failed ({direct_click_err}); trying JS click")
|
||||
self.driver.execute_script("arguments[0].click();", provider_login_link)
|
||||
print("[UnitedUcard login] Clicked 'Provider Login' (JS)")
|
||||
clicked_provider_login = True
|
||||
time.sleep(3)
|
||||
except TimeoutException:
|
||||
pass
|
||||
|
||||
if not clicked_provider_login:
|
||||
# Fallback: the sign-in dropdown's first link is always
|
||||
# "Provider Login" (Administrator Login is second) —
|
||||
# click by position instead of matching text/href.
|
||||
try:
|
||||
dropdown_link = WebDriverWait(self.driver, 5).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"(//a[contains(@class,'sign-in-button-link')])[1]"))
|
||||
)
|
||||
self.driver.execute_script("arguments[0].click();", dropdown_link)
|
||||
print("[UnitedUcard login] Clicked first dropdown link (Provider Login, by position)")
|
||||
clicked_provider_login = True
|
||||
time.sleep(3)
|
||||
except Exception as e:
|
||||
return f"ERROR: 'Provider Login' link not found in Sign In dropdown: {e}"
|
||||
|
||||
current_url = self.driver.current_url
|
||||
print(f"[UnitedUcard login] After Provider Login click URL: {current_url}")
|
||||
|
||||
# SSO may skip straight back to the dashboard if already authenticated
|
||||
if "secure.uhcdental.com" in current_url.lower():
|
||||
print("[UnitedUcard login] Already logged in - redirected straight to dashboard")
|
||||
return "ALREADY_LOGGED_IN"
|
||||
|
||||
# Check for OTP input first (in case device left mid-OTP)
|
||||
try:
|
||||
WebDriverWait(self.driver, 3).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//input[@type='tel' or contains(@placeholder,'code') or contains(@placeholder,'Code') or "
|
||||
"contains(@aria-label,'Verification') or contains(@id,'otp') or contains(@name,'otp')]"))
|
||||
)
|
||||
print("[UnitedUcard login] OTP input found on landing")
|
||||
return "OTP_REQUIRED"
|
||||
except TimeoutException:
|
||||
pass
|
||||
|
||||
# ── Step 1: "One Healthcare ID or Email Address" ────────────────
|
||||
# Confirmed against live site (2026-09-13):
|
||||
# <input id="username" data-testid="username" ...>
|
||||
# <button id="btnLogin" data-cy="data-btnLogin-field">Continue</button>
|
||||
try:
|
||||
username_field = wait.until(
|
||||
EC.element_to_be_clickable((By.ID, "username"))
|
||||
)
|
||||
username_field.clear()
|
||||
username_field.send_keys(self.uniteducard_username)
|
||||
print(f"[UnitedUcard login] Entered username: {self.uniteducard_username}")
|
||||
except TimeoutException:
|
||||
return "ERROR: Username field not found on login page"
|
||||
|
||||
try:
|
||||
continue_btn = wait.until(
|
||||
EC.element_to_be_clickable((By.ID, "btnLogin"))
|
||||
)
|
||||
continue_btn.click()
|
||||
print("[UnitedUcard login] Clicked Continue (btnLogin) after username")
|
||||
time.sleep(2)
|
||||
except TimeoutException:
|
||||
return "ERROR: Continue button (btnLogin) not found after entering username"
|
||||
|
||||
# ── Step 2: "Enter Your Password" page (#/signin/password) ─────
|
||||
# Confirmed against live site (2026-09-13):
|
||||
# <input id="login-pwd" data-testid="login-pwd" type="password" ...>
|
||||
# <button id="btnLogin" data-cy="data-btnLogin-field">Continue</button>
|
||||
# The email field on this page is pre-filled/read-only from step 1.
|
||||
try:
|
||||
password_field = wait.until(
|
||||
EC.presence_of_element_located((By.ID, "login-pwd"))
|
||||
)
|
||||
password_field.clear()
|
||||
password_field.send_keys(self.uniteducard_password)
|
||||
print("[UnitedUcard login] Entered password")
|
||||
except TimeoutException:
|
||||
return "ERROR: Password field not found on login page"
|
||||
|
||||
try:
|
||||
signin_button = wait.until(
|
||||
EC.element_to_be_clickable((By.ID, "btnLogin"))
|
||||
)
|
||||
signin_button.click()
|
||||
print("[UnitedUcard login] Clicked Continue (btnLogin) on password page")
|
||||
except TimeoutException:
|
||||
return "ERROR: Continue button (btnLogin) not found on password page"
|
||||
|
||||
if self.uniteducard_username:
|
||||
browser_manager.save_credentials_hash(self.uniteducard_username)
|
||||
|
||||
time.sleep(5) # Wait for login to process
|
||||
|
||||
current_url_after_login = self.driver.current_url.lower()
|
||||
print(f"[UnitedUcard login] After login URL: {current_url_after_login}")
|
||||
|
||||
if "secure.uhcdental.com" in current_url_after_login:
|
||||
print("[UnitedUcard login] Login successful - redirected to UHC Dental dashboard")
|
||||
return "SUCCESS"
|
||||
|
||||
# ── Step 3: "Verify Your Identity" MFA method page (#/rba/options) ──
|
||||
# Confirmed against live site (2026-09-13):
|
||||
# <button id="textMsg" data-cy="data-textMsg-field">Via Text Message</button>
|
||||
if "rba/options" in current_url_after_login:
|
||||
try:
|
||||
text_msg_btn = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.ID, "textMsg"))
|
||||
)
|
||||
text_msg_btn.click()
|
||||
print("[UnitedUcard login] Clicked 'Via Text Message' on MFA options page")
|
||||
time.sleep(3)
|
||||
except TimeoutException:
|
||||
print("[UnitedUcard login] MFA options page detected but 'Via Text Message' button not found")
|
||||
|
||||
current_url_after_login = self.driver.current_url.lower()
|
||||
print(f"[UnitedUcard login] After MFA method selection URL: {current_url_after_login}")
|
||||
|
||||
# Check for OTP input after submitting credentials
|
||||
try:
|
||||
WebDriverWait(self.driver, 15).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//input[@type='tel' or contains(@placeholder,'code') or contains(@placeholder,'Code') or "
|
||||
"contains(@aria-label,'Verification') or contains(@aria-label,'verification') or "
|
||||
"contains(@name,'otp') or contains(@name,'code') or contains(@id,'otp') or contains(@id,'code')]"
|
||||
))
|
||||
)
|
||||
print("[UnitedUcard login] OTP input detected -> OTP_REQUIRED")
|
||||
return "OTP_REQUIRED"
|
||||
except TimeoutException:
|
||||
print("[UnitedUcard login] No OTP input detected")
|
||||
|
||||
# Re-check after waiting for OTP check
|
||||
current_url_after_login = self.driver.current_url.lower()
|
||||
if "secure.uhcdental.com" in current_url_after_login:
|
||||
print("[UnitedUcard login] Login successful - redirected to UHC Dental dashboard")
|
||||
return "SUCCESS"
|
||||
|
||||
# Check for error messages on the page
|
||||
try:
|
||||
error_elem = self.driver.find_element(By.XPATH,
|
||||
"//*[contains(@class,'error') or contains(@class,'alert')]")
|
||||
error_text = error_elem.text
|
||||
if error_text:
|
||||
print(f"[UnitedUcard login] Error on page: {error_text}")
|
||||
return f"ERROR: {error_text}"
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return "ERROR: Login did not complete - still on identity domain"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard login] Exception: {e}")
|
||||
return f"ERROR:LOGIN FAILED: {e}"
|
||||
|
||||
def _format_dob(self, dob_str):
|
||||
"""Convert DOB from YYYY-MM-DD to MM/DD/YYYY format"""
|
||||
if dob_str and "-" in dob_str:
|
||||
dob_parts = dob_str.split("-")
|
||||
if len(dob_parts) == 3:
|
||||
# YYYY-MM-DD -> MM/DD/YYYY
|
||||
return f"{dob_parts[1]}/{dob_parts[2]}/{dob_parts[0]}"
|
||||
return dob_str
|
||||
|
||||
def step1(self):
|
||||
"""
|
||||
Fill in the "Eligibility Search" form — it's the default tab already
|
||||
showing on the dashboard (postloginhomescreen.html) right after login.
|
||||
|
||||
Confirmed against live site (2026-09-13):
|
||||
<input id="serviceDate"> - defaults blank; fill with today's date, MM/DD/YYYY
|
||||
<input id="eligMemDob"> - Member Date of Birth, MM/DD/YYYY
|
||||
<input id="eligSubsId"> - Subscriber ID
|
||||
<input type="button" value="search" ng-click="submit()">
|
||||
"Search By: Subscriber ID" and "Search For: Individual" are already
|
||||
the default radio selections, so no extra clicks needed there.
|
||||
"""
|
||||
from datetime import date
|
||||
|
||||
def _fill_field(field_id, value, label):
|
||||
"""Wait for field to be clickable, scroll into view, then fill it.
|
||||
Falls back to a JS value-set (+dispatch input/change events, which
|
||||
Angular's ng-model listens for) if send_keys hits an
|
||||
ElementNotInteractable error — the dashboard's Angular app can
|
||||
take a beat to finish rendering/enabling these fields."""
|
||||
field = WebDriverWait(self.driver, 20).until(
|
||||
EC.presence_of_element_located((By.ID, field_id))
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", field)
|
||||
time.sleep(0.3)
|
||||
try:
|
||||
WebDriverWait(self.driver, 10).until(EC.element_to_be_clickable((By.ID, field_id)))
|
||||
field.clear()
|
||||
field.send_keys(value)
|
||||
print(f"[UnitedUcard step1] Entered {label}: {value}")
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard step1] send_keys failed for {label} ({e}); falling back to JS set")
|
||||
self.driver.execute_script("""
|
||||
var el = arguments[0];
|
||||
var val = arguments[1];
|
||||
var nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
|
||||
nativeSetter.call(el, val);
|
||||
el.dispatchEvent(new Event('input', { bubbles: true }));
|
||||
el.dispatchEvent(new Event('change', { bubbles: true }));
|
||||
el.dispatchEvent(new Event('keyup', { bubbles: true }));
|
||||
""", field, value)
|
||||
print(f"[UnitedUcard step1] Entered {label} via JS: {value}")
|
||||
return field
|
||||
|
||||
try:
|
||||
print(f"[UnitedUcard step1] Starting eligibility search for memberId={self.memberId}, DOB={self.dateOfBirth}")
|
||||
|
||||
# Give the dashboard's Angular app a moment to finish rendering
|
||||
# after the redirect from login — filling fields too early was
|
||||
# throwing "element not interactable".
|
||||
WebDriverWait(self.driver, 20).until(
|
||||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||||
)
|
||||
time.sleep(2)
|
||||
|
||||
# Service Date - default to today.
|
||||
# NOTE: serviceDate/eligMemDob both use onkeydown="dateFunction(...)",
|
||||
# a JS input mask that auto-inserts "/" as you type raw digits
|
||||
# (same as manually typing it). Sending an already-formatted
|
||||
# "MM/DD/YYYY" string double-processes the slashes through the
|
||||
# mask and corrupts the stored value — send digits only instead,
|
||||
# matching what a human typing the field actually does.
|
||||
try:
|
||||
today_digits = date.today().strftime("%m%d%Y")
|
||||
_fill_field("serviceDate", today_digits, "Service Date")
|
||||
except TimeoutException:
|
||||
return "ERROR: Service Date field (serviceDate) not found on Eligibility Search page"
|
||||
except Exception as e:
|
||||
return f"ERROR: Could not enter Service Date: {e}"
|
||||
|
||||
# Member Date of Birth
|
||||
try:
|
||||
dob_digits = re.sub(r"\D", "", self._format_dob(self.dateOfBirth))
|
||||
_fill_field("eligMemDob", dob_digits, "Member DOB")
|
||||
except Exception as e:
|
||||
return f"ERROR: Could not enter Member Date of Birth: {e}"
|
||||
|
||||
# Subscriber ID
|
||||
try:
|
||||
_fill_field("eligSubsId", self.memberId, "Subscriber ID")
|
||||
except Exception as e:
|
||||
return f"ERROR: Could not enter Subscriber ID: {e}"
|
||||
|
||||
time.sleep(0.5)
|
||||
|
||||
# Click SEARCH button
|
||||
try:
|
||||
search_btn = WebDriverWait(self.driver, 10).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//input[@type='button' and @value='search']"))
|
||||
)
|
||||
self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", search_btn)
|
||||
time.sleep(0.3)
|
||||
try:
|
||||
search_btn.click()
|
||||
except Exception:
|
||||
self.driver.execute_script("arguments[0].click();", search_btn)
|
||||
print("[UnitedUcard step1] Clicked SEARCH button")
|
||||
except Exception as e:
|
||||
return f"ERROR: Could not click SEARCH button: {e}"
|
||||
|
||||
time.sleep(3)
|
||||
|
||||
# TODO: confirm what the results look like (same page update? new
|
||||
# tab? modal?) once we see it, then extend step2() accordingly.
|
||||
print("[UnitedUcard step1] Eligibility search submitted")
|
||||
return "Success"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard step1] Exception: {e}")
|
||||
return f"ERROR:STEP1 - {e}"
|
||||
|
||||
def step2(self):
|
||||
"""
|
||||
Extract patient/eligibility info from the "Eligibility Summary" page
|
||||
(secure.uhcdental.com/.../eligibility-summary.html) that step1()'s
|
||||
SEARCH click lands on, then capture it as a PDF.
|
||||
|
||||
Confirmed against live site (2026-09-13) via screenshot/PDF export:
|
||||
"Patient" panel -> patient full name (e.g. "WINGHUN KWONG")
|
||||
"Insurance Information" -> status badges: "In Network",
|
||||
"Eligible" (or "Not Eligible"/"Ineligible"), "No Essential
|
||||
Health Benefits", and a MASKED Subscriber ID (e.g. "******444")
|
||||
— we deliberately do NOT use that masked ID to overwrite the
|
||||
patient record.
|
||||
Exact DOM ids/classes for the badges haven't been inspected yet, so
|
||||
this uses text-based extraction (same approach as UnitedSCO/DDMA)
|
||||
as a robust starting point.
|
||||
"""
|
||||
try:
|
||||
print("[UnitedUcard step2] Starting eligibility capture")
|
||||
|
||||
try:
|
||||
WebDriverWait(self.driver, 20).until(
|
||||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(2)
|
||||
|
||||
current_url = self.driver.current_url
|
||||
print(f"[UnitedUcard step2] Current URL: {current_url}")
|
||||
|
||||
page_text = ""
|
||||
try:
|
||||
page_text = self.driver.find_element(By.TAG_NAME, "body").text
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── Eligibility status ───────────────────────────────────────
|
||||
eligibilityText = "unknown"
|
||||
lower_text = page_text.lower()
|
||||
if "not eligible" in lower_text or "ineligible" in lower_text:
|
||||
eligibilityText = "inactive"
|
||||
elif "eligible" in lower_text:
|
||||
eligibilityText = "active"
|
||||
print(f"[UnitedUcard step2] Eligibility status: {eligibilityText}")
|
||||
|
||||
# ── Patient name ─────────────────────────────────────────────
|
||||
patientName = f"{self.firstName} {self.lastName}".strip()
|
||||
name_extracted = False
|
||||
|
||||
# Strategy 1: text right after the "Patient" heading, e.g.
|
||||
# "Patient\nWINGHUN KWONG\nSUBSCRIBER/INSURED"
|
||||
try:
|
||||
name_match = re.search(
|
||||
r'Patient\s*\n\s*([A-Z][A-Za-z\-\']+(?:\s+[A-Z][A-Za-z\-\']+)+)\s*\n',
|
||||
page_text
|
||||
)
|
||||
if name_match:
|
||||
candidate = name_match.group(1).strip()
|
||||
if candidate and len(candidate) < 60 and "eligible" not in candidate.lower():
|
||||
patientName = candidate.title()
|
||||
name_extracted = True
|
||||
print(f"[UnitedUcard step2] Extracted patient name from text: {patientName}")
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard step2] Name regex failed: {e}")
|
||||
|
||||
# Strategy 2: DOM heuristic fallback — heading/strong element
|
||||
# following a "Patient" label
|
||||
if not name_extracted:
|
||||
try:
|
||||
elems = self.driver.find_elements(By.XPATH,
|
||||
"//*[contains(text(),'Patient')]/following::*[self::h1 or self::h2 or self::h3 or self::strong][1]"
|
||||
)
|
||||
for elem in elems:
|
||||
txt = elem.text.strip()
|
||||
if txt and len(txt.split()) >= 2 and txt[0].isupper() and len(txt) < 60:
|
||||
patientName = txt.title()
|
||||
name_extracted = True
|
||||
print(f"[UnitedUcard step2] Extracted patient name from DOM: {patientName}")
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not name_extracted:
|
||||
print("[UnitedUcard step2] WARNING: Could not extract patient name from page; using form values")
|
||||
|
||||
# ── Capture PDF ──────────────────────────────────────────────
|
||||
pdf_path = self._capture_pdf(self.memberId)
|
||||
if not pdf_path:
|
||||
return {"status": "error", "message": "STEP2 FAILED: Could not generate PDF"}
|
||||
|
||||
print(f"[UnitedUcard step2] PDF saved: {pdf_path}")
|
||||
|
||||
self._hide_browser()
|
||||
|
||||
print("[UnitedUcard step2] Eligibility capture complete")
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"eligibility": eligibilityText,
|
||||
"ss_path": pdf_path,
|
||||
"pdf_path": pdf_path,
|
||||
"patientName": patientName,
|
||||
# NOTE: Subscriber ID shown on this page is masked (e.g.
|
||||
# "******444"), so we return the ID we searched with rather
|
||||
# than a scraped-and-truncated value.
|
||||
"memberId": self.memberId,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard step2] Exception: {e}")
|
||||
return {"status": "error", "message": f"STEP2 FAILED: {str(e)}"}
|
||||
|
||||
def _capture_pdf(self, member_id):
|
||||
"""Capture the current page as PDF using Chrome DevTools Protocol."""
|
||||
try:
|
||||
pdf_options = {
|
||||
"landscape": False,
|
||||
"displayHeaderFooter": False,
|
||||
"printBackground": True,
|
||||
"preferCSSPageSize": True,
|
||||
"paperWidth": 8.5,
|
||||
"paperHeight": 11,
|
||||
"marginTop": 0.4,
|
||||
"marginBottom": 0.4,
|
||||
"marginLeft": 0.4,
|
||||
"marginRight": 0.4,
|
||||
"scale": 0.9,
|
||||
}
|
||||
|
||||
file_identifier = member_id if member_id else f"{self.firstName}_{self.lastName}"
|
||||
|
||||
result = self.driver.execute_cdp_cmd("Page.printToPDF", pdf_options)
|
||||
pdf_data = base64.b64decode(result.get('data', ''))
|
||||
pdf_path = os.path.join(self.download_dir, f"uniteducard_eligibility_{file_identifier}_{int(time.time())}.pdf")
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_data)
|
||||
return pdf_path
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard _capture_pdf] Error: {e}")
|
||||
return None
|
||||
|
||||
def _hide_browser(self):
|
||||
"""Hide the browser window after task completion using multiple strategies."""
|
||||
try:
|
||||
try:
|
||||
self.driver.get("about:blank")
|
||||
time.sleep(0.5)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
self.driver.minimize_window()
|
||||
print("[UnitedUcard step2] Browser window minimized")
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
self.driver.set_window_position(-10000, -10000)
|
||||
print("[UnitedUcard step2] Browser window moved off-screen")
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard step2] Could not hide browser: {e}")
|
||||
|
||||
def main_workflow(self, url):
|
||||
"""Main workflow that runs all steps."""
|
||||
try:
|
||||
self.config_driver()
|
||||
|
||||
login_result = self.login(url)
|
||||
print(f"[main_workflow] Login result: {login_result}")
|
||||
|
||||
if login_result == "OTP_REQUIRED":
|
||||
return {"status": "otp_required", "message": "OTP required after login"}
|
||||
|
||||
if isinstance(login_result, str) and login_result.startswith("ERROR"):
|
||||
return {"status": "error", "message": login_result}
|
||||
|
||||
step1_result = self.step1()
|
||||
print(f"[main_workflow] Step1 result: {step1_result}")
|
||||
|
||||
if isinstance(step1_result, str) and step1_result.startswith("ERROR"):
|
||||
return {"status": "error", "message": step1_result}
|
||||
|
||||
step2_result = self.step2()
|
||||
print(f"[main_workflow] Step2 result: {step2_result}")
|
||||
|
||||
return step2_result
|
||||
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
@@ -0,0 +1,243 @@
|
||||
"""
|
||||
Minimal browser manager for United Ucard (OneHealthcareID) - only handles persistent
|
||||
profile and keeping browser alive. Clears session cookies on startup (after PC restart)
|
||||
to force fresh login. Tracks credentials to detect changes mid-session.
|
||||
|
||||
Mirrors unitedsco_browser_manager.py — see that file for the pattern this is based on.
|
||||
"""
|
||||
import os
|
||||
import hashlib
|
||||
import threading
|
||||
import subprocess
|
||||
import time
|
||||
from selenium import webdriver
|
||||
from selenium.webdriver.chrome.service import Service
|
||||
from webdriver_manager.chrome import ChromeDriverManager
|
||||
|
||||
# Ensure DISPLAY is set for Chrome to work (needed when running from SSH/background)
|
||||
if not os.environ.get("DISPLAY"):
|
||||
os.environ["DISPLAY"] = ":0"
|
||||
|
||||
|
||||
class UnitedUcardBrowserManager:
|
||||
"""
|
||||
Singleton that manages a persistent Chrome browser instance for United Ucard.
|
||||
- Uses --user-data-dir for persistent profile (device trust tokens)
|
||||
- Clears session cookies on startup (after PC restart)
|
||||
- Tracks credentials to detect changes mid-session
|
||||
"""
|
||||
_instance = None
|
||||
_lock = threading.Lock()
|
||||
|
||||
def __new__(cls):
|
||||
with cls._lock:
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
cls._instance._driver = None
|
||||
cls._instance.profile_dir = os.path.abspath("chrome_profile_uniteducard")
|
||||
cls._instance.download_dir = os.path.abspath("seleniumDownloads")
|
||||
cls._instance._credentials_file = os.path.join(cls._instance.profile_dir, ".last_credentials")
|
||||
cls._instance._needs_session_clear = False # Flag to clear session on next driver creation
|
||||
os.makedirs(cls._instance.profile_dir, exist_ok=True)
|
||||
os.makedirs(cls._instance.download_dir, exist_ok=True)
|
||||
return cls._instance
|
||||
|
||||
def clear_session_on_startup(self):
|
||||
"""
|
||||
On startup, only clear the Cookies file so the login session is reset
|
||||
but device trust tokens (Local Storage, IndexedDB) are preserved.
|
||||
"""
|
||||
print("[UnitedUcard BrowserManager] Clearing cookies on startup (preserving device trust)...")
|
||||
|
||||
try:
|
||||
# Clear credentials tracking so the next login re-saves the hash
|
||||
if os.path.exists(self._credentials_file):
|
||||
os.remove(self._credentials_file)
|
||||
print("[UnitedUcard BrowserManager] Cleared credentials tracking file")
|
||||
|
||||
# Only remove cookie files — leave everything else intact
|
||||
cookie_files = ["Cookies", "Cookies-journal"]
|
||||
for filename in cookie_files:
|
||||
for base in [os.path.join(self.profile_dir, "Default"), self.profile_dir]:
|
||||
filepath = os.path.join(base, filename)
|
||||
if os.path.exists(filepath):
|
||||
try:
|
||||
os.remove(filepath)
|
||||
print(f"[UnitedUcard BrowserManager] Removed {filepath}")
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard BrowserManager] Could not remove {filepath}: {e}")
|
||||
|
||||
self._needs_session_clear = False
|
||||
print("[UnitedUcard BrowserManager] Cookies cleared — device trust tokens preserved")
|
||||
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard BrowserManager] Error clearing session: {e}")
|
||||
|
||||
def _hash_credentials(self, username: str) -> str:
|
||||
"""Create a hash of the username to track credential changes."""
|
||||
return hashlib.sha256(username.encode()).hexdigest()[:16]
|
||||
|
||||
def get_last_credentials_hash(self) -> str | None:
|
||||
"""Get the hash of the last-used credentials."""
|
||||
try:
|
||||
if os.path.exists(self._credentials_file):
|
||||
with open(self._credentials_file, 'r') as f:
|
||||
return f.read().strip()
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
def save_credentials_hash(self, username: str):
|
||||
"""Save the hash of the current credentials."""
|
||||
try:
|
||||
cred_hash = self._hash_credentials(username)
|
||||
with open(self._credentials_file, 'w') as f:
|
||||
f.write(cred_hash)
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard BrowserManager] Failed to save credentials hash: {e}")
|
||||
|
||||
def credentials_changed(self, username: str) -> bool:
|
||||
"""Check if the credentials have changed since last login."""
|
||||
last_hash = self.get_last_credentials_hash()
|
||||
if last_hash is None:
|
||||
return False # No previous credentials, not a change
|
||||
current_hash = self._hash_credentials(username)
|
||||
changed = last_hash != current_hash
|
||||
if changed:
|
||||
print(f"[UnitedUcard BrowserManager] Credentials changed - logout required")
|
||||
return changed
|
||||
|
||||
def clear_credentials_hash(self):
|
||||
"""Clear the saved credentials hash (used after logout)."""
|
||||
try:
|
||||
if os.path.exists(self._credentials_file):
|
||||
os.remove(self._credentials_file)
|
||||
except Exception as e:
|
||||
print(f"[UnitedUcard BrowserManager] Failed to clear credentials hash: {e}")
|
||||
|
||||
def _kill_existing_chrome_for_profile(self):
|
||||
"""Kill any existing Chrome processes using this profile."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["pgrep", "-f", f"user-data-dir={self.profile_dir}"],
|
||||
capture_output=True, text=True
|
||||
)
|
||||
if result.stdout.strip():
|
||||
pids = result.stdout.strip().split('\n')
|
||||
for pid in pids:
|
||||
try:
|
||||
subprocess.run(["kill", "-9", pid], check=False)
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(1)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Remove SingletonLock if exists
|
||||
lock_file = os.path.join(self.profile_dir, "SingletonLock")
|
||||
try:
|
||||
if os.path.islink(lock_file) or os.path.exists(lock_file):
|
||||
os.remove(lock_file)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def get_driver(self, headless=False):
|
||||
"""Get or create the persistent browser instance."""
|
||||
with self._lock:
|
||||
if self._driver is None:
|
||||
print("[UnitedUcard BrowserManager] Driver is None, creating new driver")
|
||||
self._kill_existing_chrome_for_profile()
|
||||
self._create_driver(headless)
|
||||
elif not self._is_alive():
|
||||
print("[UnitedUcard BrowserManager] Driver not alive, recreating")
|
||||
self._kill_existing_chrome_for_profile()
|
||||
self._create_driver(headless)
|
||||
else:
|
||||
print("[UnitedUcard BrowserManager] Reusing existing driver")
|
||||
return self._driver
|
||||
|
||||
def _is_alive(self):
|
||||
"""Check if browser is still responsive."""
|
||||
try:
|
||||
if self._driver is None:
|
||||
return False
|
||||
_ = self._driver.current_url
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _create_driver(self, headless=False):
|
||||
"""Create browser with persistent profile."""
|
||||
if self._driver:
|
||||
try:
|
||||
self._driver.quit()
|
||||
except Exception:
|
||||
pass
|
||||
self._driver = None
|
||||
time.sleep(1)
|
||||
|
||||
options = webdriver.ChromeOptions()
|
||||
if headless:
|
||||
options.add_argument("--headless")
|
||||
|
||||
# Persistent profile - THIS IS THE KEY for device trust
|
||||
options.add_argument(f"--user-data-dir={self.profile_dir}")
|
||||
options.add_argument("--no-sandbox")
|
||||
options.add_argument("--disable-dev-shm-usage")
|
||||
|
||||
# Anti-detection options (prevent bot detection)
|
||||
options.add_argument("--disable-blink-features=AutomationControlled")
|
||||
options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
||||
options.add_experimental_option("useAutomationExtension", False)
|
||||
options.add_argument("--disable-infobars")
|
||||
|
||||
prefs = {
|
||||
"download.default_directory": self.download_dir,
|
||||
"plugins.always_open_pdf_externally": True,
|
||||
"download.prompt_for_download": False,
|
||||
"download.directory_upgrade": True,
|
||||
# Disable password save dialog that blocks page interactions
|
||||
"credentials_enable_service": False,
|
||||
"profile.password_manager_enabled": False,
|
||||
"profile.password_manager_leak_detection": False,
|
||||
}
|
||||
options.add_experimental_option("prefs", prefs)
|
||||
|
||||
service = Service(ChromeDriverManager().install())
|
||||
self._driver = webdriver.Chrome(service=service, options=options)
|
||||
self._driver.maximize_window()
|
||||
|
||||
# Remove webdriver property to avoid detection
|
||||
try:
|
||||
self._driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
self._needs_session_clear = False
|
||||
|
||||
def quit_driver(self):
|
||||
"""Quit browser (only call on shutdown)."""
|
||||
with self._lock:
|
||||
if self._driver:
|
||||
try:
|
||||
self._driver.quit()
|
||||
except Exception:
|
||||
pass
|
||||
self._driver = None
|
||||
self._kill_existing_chrome_for_profile()
|
||||
|
||||
|
||||
# Singleton accessor
|
||||
_manager = None
|
||||
|
||||
def get_browser_manager():
|
||||
global _manager
|
||||
if _manager is None:
|
||||
_manager = UnitedUcardBrowserManager()
|
||||
return _manager
|
||||
|
||||
|
||||
def clear_uniteducard_session_on_startup():
|
||||
"""Called by agent.py on startup to clear session."""
|
||||
manager = get_browser_manager()
|
||||
manager.clear_session_on_startup()
|
||||
Reference in New Issue
Block a user