feat: MetLife Dental eligibility check with persistent session
Mirrors the DDMA flow: persistent Chrome profile + credential-change logout, PingOne login with email MFA selection and OTP from app or browser, Eligibility & Plan Details search (masked Subscriber ID/SSN), family member picked by DOB and re-verified on the plan details page, PDF saved to the patient's Eligibility folder and status set ACTIVE/INACTIVE. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -31,6 +31,7 @@ import helpers_tuftssco_claim as htuftssco_claim
|
||||
import helpers_tuftssco_preauth as htuftssco_preauth
|
||||
import helpers_uniteducard_claim as huniteducard_claim
|
||||
import helpers_bcbs_ma_eligibility as hbcbs_ma
|
||||
import helpers_metlife_eligibility as hmetlife
|
||||
|
||||
# Import startup session-clear functions
|
||||
from ddma_browser_manager import clear_ddma_session_on_startup
|
||||
@@ -40,6 +41,7 @@ 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
|
||||
from metlife_browser_manager import clear_metlife_session_on_startup
|
||||
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
@@ -55,6 +57,7 @@ clear_unitedsco_session_on_startup()
|
||||
clear_uniteducard_session_on_startup()
|
||||
clear_dentaquest_session_on_startup()
|
||||
clear_cca_session_on_startup()
|
||||
clear_metlife_session_on_startup()
|
||||
print("=" * 50)
|
||||
print("SESSION CLEAR COMPLETE")
|
||||
print("=" * 50)
|
||||
@@ -423,6 +426,46 @@ async def ddma_eligibility(request: Request):
|
||||
return {"status": "started", "session_id": sid}
|
||||
|
||||
|
||||
# Endpoint: MetLife Dental eligibility (background, OTP, persistent session)
|
||||
|
||||
async def _metlife_worker_wrapper(sid: str, data: dict, url: str):
|
||||
"""Background worker for MetLife eligibility — acquires semaphore, updates counters."""
|
||||
global active_jobs, waiting_jobs
|
||||
async with semaphore:
|
||||
async with lock:
|
||||
waiting_jobs -= 1
|
||||
active_jobs += 1
|
||||
try:
|
||||
await hmetlife.start_metlife_run(sid, data, url)
|
||||
finally:
|
||||
async with lock:
|
||||
active_jobs -= 1
|
||||
|
||||
|
||||
@app.post("/metlife-eligibility")
|
||||
async def metlife_eligibility(request: Request):
|
||||
"""
|
||||
Starts a MetLife 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 = hmetlife.make_session_entry()
|
||||
hmetlife.sessions[sid]["type"] = "metlife_eligibility"
|
||||
hmetlife.sessions[sid]["last_activity"] = time.time()
|
||||
|
||||
async with lock:
|
||||
waiting_jobs += 1
|
||||
|
||||
asyncio.create_task(_metlife_worker_wrapper(sid, data, url="https://dentalprovider.metlife.com/home"))
|
||||
|
||||
return {"status": "started", "session_id": sid}
|
||||
|
||||
|
||||
async def _deltaindiv_worker_wrapper(sid: str, data: dict, url: str):
|
||||
"""Background worker for Delta Indiv eligibility — acquires semaphore, updates counters."""
|
||||
global active_jobs, waiting_jobs
|
||||
@@ -1126,6 +1169,8 @@ async def submit_otp(request: Request):
|
||||
res = huniteducard_claim.submit_otp(sid, otp)
|
||||
elif sid in hbcbs_ma.sessions:
|
||||
res = hbcbs_ma.submit_otp(sid, otp)
|
||||
elif sid in hmetlife.sessions:
|
||||
res = hmetlife.submit_otp(sid, otp)
|
||||
else:
|
||||
raise HTTPException(status_code=404, detail="session not found")
|
||||
|
||||
@@ -1171,6 +1216,8 @@ async def session_status(sid: str):
|
||||
s = huniteducard_claim.get_session_status(sid)
|
||||
elif sid in hbcbs_ma.sessions:
|
||||
s = hbcbs_ma.get_session_status(sid)
|
||||
elif sid in hmetlife.sessions:
|
||||
s = hmetlife.get_session_status(sid)
|
||||
else:
|
||||
s = {"status": "not_found"}
|
||||
if s.get("status") == "not_found":
|
||||
@@ -1240,6 +1287,16 @@ async def clear_cca_session_endpoint():
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
|
||||
@app.post("/clear-metlife-session")
|
||||
async def clear_metlife_session_endpoint():
|
||||
"""Clears the MetLife browser session. Call when credentials are deleted or changed."""
|
||||
try:
|
||||
clear_metlife_session_on_startup()
|
||||
return {"status": "success", "message": "MetLife session cleared"}
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
|
||||
# ✅ Health Check Endpoint
|
||||
@app.get("/")
|
||||
async def health_check():
|
||||
|
||||
@@ -0,0 +1,252 @@
|
||||
import os
|
||||
import time
|
||||
import asyncio
|
||||
from typing import Dict, Any
|
||||
from selenium.common.exceptions import WebDriverException
|
||||
|
||||
from selenium_MetLife_eligibilityCheckWorker import AutomationMetLifeEligibilityCheck, METLIFE_HOME_URL
|
||||
|
||||
# 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 → completed / error
|
||||
"created_at": time.time(),
|
||||
"last_activity": time.time(),
|
||||
"bot": None, # AutomationMetLifeEligibilityCheck instance
|
||||
"driver": None, # selenium webdriver
|
||||
"otp_event": asyncio.Event(),
|
||||
"otp_value": None, # OTP submitted from the app
|
||||
"result": None,
|
||||
"message": None,
|
||||
"type": None,
|
||||
}
|
||||
return sid
|
||||
|
||||
|
||||
async def cleanup_session(sid: str, message: str | None = None):
|
||||
"""
|
||||
Wake any OTP waiter, set final state, and remove the session entry.
|
||||
Safe to call multiple times (idempotent).
|
||||
NOTE: Does NOT quit the browser driver — the persistent browser stays alive.
|
||||
"""
|
||||
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_metlife] cleaned session {sid}")
|
||||
|
||||
|
||||
async def _remove_session_later(sid: str, delay: int = 20):
|
||||
await asyncio.sleep(delay)
|
||||
await cleanup_session(sid)
|
||||
|
||||
|
||||
async def start_metlife_run(sid: str, data: dict, url: str):
|
||||
"""
|
||||
Run the full MetLife eligibility workflow for one session.
|
||||
Called by agent.py inside a wrapper that manages the semaphore/counters.
|
||||
|
||||
OTP handling uses two complementary strategies:
|
||||
1. Accept OTP submitted from the app (via /submit-otp endpoint → otp_value field)
|
||||
2. Poll the browser directly to detect when the user completes MFA themselves
|
||||
"""
|
||||
s = sessions.get(sid)
|
||||
if not s:
|
||||
return {"status": "error", "message": "session not found"}
|
||||
|
||||
s["status"] = "running"
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
try:
|
||||
bot = AutomationMetLifeEligibilityCheck({"data": data})
|
||||
bot.config_driver()
|
||||
|
||||
s["bot"] = bot
|
||||
s["driver"] = bot.driver
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
try:
|
||||
if not url:
|
||||
raise ValueError("URL not provided for MetLife run")
|
||||
bot.driver.maximize_window()
|
||||
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"]}
|
||||
|
||||
# ── Path: already logged in (persistent session) ──────────────────────
|
||||
if login_result == "ALREADY_LOGGED_IN":
|
||||
print("[start_metlife_run] Session persisted - skipping OTP")
|
||||
s["status"] = "running"
|
||||
s["message"] = "Session persisted"
|
||||
|
||||
# ── Path: OTP required ────────────────────────────────────────────────
|
||||
elif 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"]
|
||||
login_success = False
|
||||
|
||||
print(f"[MetLife OTP] Polling for OTP completion (up to {SESSION_OTP_TIMEOUT}s)...")
|
||||
|
||||
for poll in range(SESSION_OTP_TIMEOUT):
|
||||
await asyncio.sleep(1)
|
||||
s["last_activity"] = time.time()
|
||||
|
||||
try:
|
||||
# a) App submitted OTP via /submit-otp endpoint
|
||||
otp_value = s.get("otp_value")
|
||||
if otp_value:
|
||||
print(f"[MetLife OTP poll {poll+1}] OTP received from app, typing it in...")
|
||||
s["otp_value"] = None # Clear so we don't re-submit
|
||||
if bot.submit_otp_code(otp_value):
|
||||
print("[MetLife OTP] OTP typed and submitted via app")
|
||||
await asyncio.sleep(3)
|
||||
else:
|
||||
print("[MetLife OTP] OTP input not found — enter it in the browser window")
|
||||
|
||||
# b) Back on the portal with the search form → login completed
|
||||
if bot.is_logged_in(timeout=2):
|
||||
print("[MetLife OTP] Search form found — login successful!")
|
||||
login_success = True
|
||||
break
|
||||
|
||||
print(f"[MetLife OTP poll {poll+1}/{SESSION_OTP_TIMEOUT}] URL: {driver.current_url[:70]}...")
|
||||
except Exception as poll_err:
|
||||
print(f"[MetLife OTP poll {poll+1}] Error: {poll_err}")
|
||||
|
||||
if not login_success:
|
||||
# Final check — navigate directly to the portal home page
|
||||
try:
|
||||
print("[MetLife OTP] Final attempt - navigating to home page...")
|
||||
driver.get(METLIFE_HOME_URL)
|
||||
await asyncio.sleep(3)
|
||||
login_success = bot.is_logged_in(timeout=10)
|
||||
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 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("[MetLife OTP] Proceeding to step1...")
|
||||
|
||||
# ── Path: login succeeded without OTP ─────────────────────────────────
|
||||
elif login_result == "SUCCESS":
|
||||
print("[start_metlife_run] Login succeeded without OTP")
|
||||
s["status"] = "running"
|
||||
s["message"] = "Login succeeded"
|
||||
|
||||
# ── Path: login error ─────────────────────────────────────────────────
|
||||
else:
|
||||
msg = login_result if isinstance(login_result, str) else "ERROR: unknown login result"
|
||||
s["status"] = "error"
|
||||
s["message"] = msg
|
||||
await cleanup_session(sid)
|
||||
return {"status": "error", "message": msg}
|
||||
|
||||
# ── Step 1: search ────────────────────────────────────────────────────
|
||||
step1_result = bot.step1()
|
||||
if isinstance(step1_result, str) and step1_result.startswith("ERROR"):
|
||||
s["status"] = "error"
|
||||
s["message"] = step1_result
|
||||
await cleanup_session(sid)
|
||||
return {"status": "error", "message": step1_result}
|
||||
|
||||
# ── Step 2: patient selection + 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
|
||||
|
||||
s["status"] = "error"
|
||||
if isinstance(step2_result, dict):
|
||||
s["message"] = step2_result.get("message", "unknown error")
|
||||
else:
|
||||
s["message"] = str(step2_result)
|
||||
await cleanup_session(sid)
|
||||
return {"status": "error", "message": s["message"]}
|
||||
|
||||
except Exception as e:
|
||||
s["status"] = "error"
|
||||
s["message"] = f"worker exception: {e}"
|
||||
await cleanup_session(sid)
|
||||
return {"status": "error", "message": s["message"]}
|
||||
|
||||
|
||||
def submit_otp(sid: str, otp: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Called when the app sends an OTP via POST /submit-otp.
|
||||
Sets otp_value on the session so the polling loop picks it up.
|
||||
"""
|
||||
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") == "completed" else None,
|
||||
}
|
||||
@@ -0,0 +1,253 @@
|
||||
"""
|
||||
Browser manager for MetLife Dental (MetDental provider portal) - persistent profile with session management.
|
||||
- Uses --user-data-dir for persistent profile (device trust tokens)
|
||||
- Clears session cookies on startup (after PC restart) to force fresh login
|
||||
- Tracks credentials to detect changes mid-session (triggers logout)
|
||||
- Anti-detection options to avoid bot detection
|
||||
"""
|
||||
import os
|
||||
import glob
|
||||
import shutil
|
||||
import hashlib
|
||||
import threading
|
||||
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 MetLifeBrowserManager:
|
||||
"""
|
||||
Singleton that manages a persistent Chrome browser instance.
|
||||
- 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_metlife")
|
||||
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
|
||||
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):
|
||||
"""
|
||||
Clear only login cookies on startup to force credential re-entry after restart.
|
||||
NEVER clears Local Storage or IndexedDB — those hold the MetLife (PingOne) device trust token
|
||||
that allows the portal to skip OTP for recognised devices.
|
||||
"""
|
||||
print("[MetLife BrowserManager] Clearing login cookies on startup (preserving device trust)...")
|
||||
|
||||
try:
|
||||
# Reset credentials tracking so the next login re-saves the hash
|
||||
if os.path.exists(self._credentials_file):
|
||||
os.remove(self._credentials_file)
|
||||
print("[MetLife BrowserManager] Cleared credentials tracking file")
|
||||
|
||||
# Only remove cookie / login-data files — these expire the session so the
|
||||
# user must re-enter credentials, but the device trust token is untouched.
|
||||
session_files = [
|
||||
"Cookies",
|
||||
"Cookies-journal",
|
||||
"Login Data",
|
||||
"Login Data-journal",
|
||||
]
|
||||
for filename in session_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"[MetLife BrowserManager] Removed {filename}")
|
||||
except Exception as e:
|
||||
print(f"[MetLife BrowserManager] Could not remove {filename}: {e}")
|
||||
|
||||
# Local Storage, IndexedDB, and Session Storage are intentionally
|
||||
# NOT cleared — they contain the MetLife (PingOne) device trust token that prevents
|
||||
# OTP from being required on every login.
|
||||
|
||||
self._needs_session_clear = True
|
||||
print("[MetLife BrowserManager] Startup clear done — device trust preserved, OTP not required")
|
||||
|
||||
except Exception as e:
|
||||
print(f"[MetLife BrowserManager] Error clearing session: {e}")
|
||||
|
||||
# ── Credential hash tracking ──────────────────────────────────────────────
|
||||
|
||||
def _hash_credentials(self, username: str) -> str:
|
||||
return hashlib.sha256(username.encode()).hexdigest()[:16]
|
||||
|
||||
def get_last_credentials_hash(self) -> str | None:
|
||||
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):
|
||||
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"[MetLife BrowserManager] Failed to save credentials hash: {e}")
|
||||
|
||||
def credentials_changed(self, username: str) -> bool:
|
||||
last_hash = self.get_last_credentials_hash()
|
||||
if last_hash is None:
|
||||
return False # No previous credentials stored — not a change
|
||||
current_hash = self._hash_credentials(username)
|
||||
changed = last_hash != current_hash
|
||||
if changed:
|
||||
print("[MetLife BrowserManager] Credentials changed — logout required")
|
||||
return changed
|
||||
|
||||
def clear_credentials_hash(self):
|
||||
try:
|
||||
if os.path.exists(self._credentials_file):
|
||||
os.remove(self._credentials_file)
|
||||
except Exception as e:
|
||||
print(f"[MetLife BrowserManager] Failed to clear credentials hash: {e}")
|
||||
|
||||
# ── Chrome process management ─────────────────────────────────────────────
|
||||
|
||||
def _kill_existing_chrome_for_profile(self):
|
||||
"""Kill any existing Chrome processes using this profile and remove lock files."""
|
||||
import subprocess
|
||||
import time as time_module
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["pgrep", "-f", f"user-data-dir={self.profile_dir}"],
|
||||
capture_output=True, text=True
|
||||
)
|
||||
if result.stdout.strip():
|
||||
for pid in result.stdout.strip().split('\n'):
|
||||
try:
|
||||
subprocess.run(["kill", "-9", pid], check=False)
|
||||
except Exception:
|
||||
pass
|
||||
time_module.sleep(1)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
for lock_file in ["SingletonLock", "SingletonSocket", "SingletonCookie"]:
|
||||
lock_path = os.path.join(self.profile_dir, lock_file)
|
||||
try:
|
||||
if os.path.islink(lock_path) or os.path.exists(lock_path):
|
||||
os.remove(lock_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── Driver lifecycle ──────────────────────────────────────────────────────
|
||||
|
||||
def get_driver(self, headless=False):
|
||||
"""Get or create the persistent browser instance."""
|
||||
with self._lock:
|
||||
if self._driver is None:
|
||||
print("[MetLife BrowserManager] Driver is None, creating new driver")
|
||||
self._kill_existing_chrome_for_profile()
|
||||
self._create_driver(headless)
|
||||
elif not self._is_alive():
|
||||
print("[MetLife BrowserManager] Driver not alive, recreating")
|
||||
self._kill_existing_chrome_for_profile()
|
||||
self._create_driver(headless)
|
||||
else:
|
||||
print("[MetLife BrowserManager] Reusing existing driver")
|
||||
return self._driver
|
||||
|
||||
def _is_alive(self):
|
||||
try:
|
||||
if self._driver is None:
|
||||
return False
|
||||
url = self._driver.current_url
|
||||
print(f"[MetLife BrowserManager] Driver alive, current URL: {url[:50]}...")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"[MetLife BrowserManager] Driver not alive: {e}")
|
||||
return False
|
||||
|
||||
def _create_driver(self, headless=False):
|
||||
"""Create browser with persistent profile and anti-detection options."""
|
||||
if self._driver:
|
||||
try:
|
||||
self._driver.quit()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
options = webdriver.ChromeOptions()
|
||||
if headless:
|
||||
options.add_argument("--headless")
|
||||
|
||||
# Persistent profile — keeps device trust tokens between runs
|
||||
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.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,
|
||||
}
|
||||
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 — NOT between patients)."""
|
||||
with self._lock:
|
||||
if self._driver:
|
||||
try:
|
||||
self._driver.quit()
|
||||
except Exception:
|
||||
pass
|
||||
self._driver = None
|
||||
|
||||
|
||||
# ── Singleton accessor ────────────────────────────────────────────────────────
|
||||
|
||||
_manager = None
|
||||
|
||||
def get_browser_manager() -> MetLifeBrowserManager:
|
||||
global _manager
|
||||
if _manager is None:
|
||||
_manager = MetLifeBrowserManager()
|
||||
return _manager
|
||||
|
||||
|
||||
def clear_metlife_session_on_startup():
|
||||
"""Called by agent.py on startup to clear MetLife session (after PC restart)."""
|
||||
manager = get_browser_manager()
|
||||
manager.clear_session_on_startup()
|
||||
@@ -0,0 +1,834 @@
|
||||
from selenium.common.exceptions import WebDriverException, 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
|
||||
from datetime import datetime
|
||||
import time
|
||||
import os
|
||||
import re
|
||||
import base64
|
||||
|
||||
from metlife_browser_manager import get_browser_manager
|
||||
|
||||
# MetDental provider portal (React SPA, PingOne DaVinci login)
|
||||
METLIFE_BASE_URL = "https://dentalprovider.metlife.com"
|
||||
METLIFE_HOME_URL = f"{METLIFE_BASE_URL}/home"
|
||||
METLIFE_PRESIGNIN_URL = f"{METLIFE_BASE_URL}/presignin"
|
||||
|
||||
# Home page search form: "Subscriber ID or SSN" (max 11 alphanumeric chars)
|
||||
# The portal's design-system components may put the id on a wrapper instead of the
|
||||
# <input>/<button> itself, so accept both.
|
||||
# Actual input: <input id="subscriber-id-input-input"> (label for="subscriber-id-input-input")
|
||||
SUBSCRIBER_INPUT_XPATH = (
|
||||
"//input[@id='subscriber-id-input-input']"
|
||||
" | //input[@id='subscriber-id-input'] | //*[@id='subscriber-id-input']//input"
|
||||
" | //input[contains(@aria-label,'Subscriber ID') or contains(@placeholder,'Subscriber ID')]"
|
||||
)
|
||||
SEARCH_SUBMIT_XPATH = (
|
||||
"//button[@id='submit-button'] | //*[@id='submit-button']//button"
|
||||
" | //form[.//input[starts-with(@id,'subscriber-id-input')]]//button[@type='submit']"
|
||||
)
|
||||
# Home page card link: <a id="benefit-link">Eligibility & Plan Details</a> (rendered several times)
|
||||
BENEFIT_LINK_XPATH = "//a[@id='benefit-link']"
|
||||
# Extra inputs the portal asks for when the ID alone is ambiguous (reason codes 200/205/225)
|
||||
# Same "<id>-input" naming as the subscriber box
|
||||
EXTRA_LASTNAME_XPATH = "//input[@id='input-lastname-input' or @id='input-lastname'] | //*[@id='input-lastname']//input"
|
||||
EXTRA_ZIPCODE_XPATH = "//input[@id='input-zipcode-input' or @id='input-zipcode'] | //*[@id='input-zipcode']//input"
|
||||
# One tile per covered person on /eligibility
|
||||
RESULT_TILE_CSS = ".eligibility-result__tile-content"
|
||||
# "Sign in" button on /presignin (goes to the PingOne login page)
|
||||
PRESIGNIN_BUTTON_XPATH = (
|
||||
"//button[@id='mldc-react-button-secondary-white-signin']"
|
||||
" | //*[@id='hero-section__text--login']/following::button[1]"
|
||||
)
|
||||
|
||||
# Finds a visible element of a given kind in the current document.
|
||||
# The PingOne login widget markup isn't published, so match on attributes/text
|
||||
# rather than fixed ids.
|
||||
FIND_ELEMENT_JS = r"""
|
||||
const kind = arguments[0];
|
||||
const vis = el => {
|
||||
if (!el) return false;
|
||||
const st = getComputedStyle(el);
|
||||
if (st.visibility === 'hidden' || st.display === 'none') return false;
|
||||
return !!(el.offsetWidth || el.offsetHeight || el.getClientRects().length);
|
||||
};
|
||||
const attrs = el => [el.id, el.name, el.getAttribute('autocomplete'), el.getAttribute('placeholder'),
|
||||
el.getAttribute('aria-label'), el.getAttribute('data-id'), el.getAttribute('data-testid')]
|
||||
.filter(Boolean).join(' ').toLowerCase();
|
||||
const textInputs = Array.from(document.querySelectorAll('input')).filter(i =>
|
||||
vis(i) && !['hidden', 'checkbox', 'radio', 'submit', 'button'].includes((i.type || '').toLowerCase()));
|
||||
|
||||
if (kind === 'password') {
|
||||
return textInputs.find(i => i.type === 'password') || null;
|
||||
}
|
||||
if (kind === 'otp') {
|
||||
return textInputs.find(i =>
|
||||
i.getAttribute('autocomplete') === 'one-time-code' ||
|
||||
(i.type !== 'password' && /otp|passcode|verification|one.?time|mfa|security.?code|\bcode\b/.test(attrs(i)))
|
||||
) || null;
|
||||
}
|
||||
if (kind === 'username') {
|
||||
return textInputs.find(i => i.type === 'email' || (i.type !== 'password' && /user|login|email/.test(attrs(i))))
|
||||
|| textInputs.find(i => (i.type || 'text') === 'text')
|
||||
|| null;
|
||||
}
|
||||
if (kind === 'submit') {
|
||||
const btns = Array.from(document.querySelectorAll('button, input[type=submit], [role=button]')).filter(vis);
|
||||
const label = b => ((b.innerText || b.value || b.getAttribute('aria-label') || '') + '').trim().toLowerCase();
|
||||
const rx = /^(sign ?in|log ?in|next|continue|submit|verify|confirm)\b/;
|
||||
return btns.find(b => rx.test(label(b)))
|
||||
|| btns.find(b => (b.type || '').toLowerCase() === 'submit')
|
||||
|| null;
|
||||
}
|
||||
return null;
|
||||
"""
|
||||
|
||||
|
||||
class AutomationMetLifeEligibilityCheck:
|
||||
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.subscriberLastName = self.data.get("subscriberLastName", "") or self.lastName
|
||||
self.zipCode = self.data.get("zipCode", "")
|
||||
self.metlife_username = self.data.get("metlifeUsername", "")
|
||||
self.metlife_password = self.data.get("metlifePassword", "")
|
||||
|
||||
# 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)
|
||||
|
||||
# ── Element helpers ───────────────────────────────────────────────────────
|
||||
|
||||
def _find(self, kind):
|
||||
"""
|
||||
Find a visible element of `kind` in the main document or any iframe.
|
||||
On success the driver is left switched into the frame holding it.
|
||||
"""
|
||||
self.driver.switch_to.default_content()
|
||||
try:
|
||||
el = self.driver.execute_script(FIND_ELEMENT_JS, kind)
|
||||
if el:
|
||||
return el
|
||||
except Exception:
|
||||
pass
|
||||
for frame in self.driver.find_elements(By.TAG_NAME, "iframe"):
|
||||
try:
|
||||
self.driver.switch_to.default_content()
|
||||
self.driver.switch_to.frame(frame)
|
||||
el = self.driver.execute_script(FIND_ELEMENT_JS, kind)
|
||||
if el:
|
||||
return el
|
||||
except Exception:
|
||||
continue
|
||||
self.driver.switch_to.default_content()
|
||||
return None
|
||||
|
||||
def _type_into(self, el, value):
|
||||
try:
|
||||
el.click()
|
||||
except Exception:
|
||||
pass
|
||||
el.send_keys(Keys.CONTROL, "a")
|
||||
el.send_keys(Keys.BACKSPACE)
|
||||
el.send_keys(value)
|
||||
|
||||
def _click_submit(self, fallback_el=None):
|
||||
btn = self._find("submit")
|
||||
if btn:
|
||||
try:
|
||||
btn.click()
|
||||
except Exception:
|
||||
self.driver.execute_script("arguments[0].click();", btn)
|
||||
return True
|
||||
if fallback_el is not None:
|
||||
fallback_el.send_keys(Keys.ENTER)
|
||||
return True
|
||||
return False
|
||||
|
||||
def _select_email_mfa_method(self):
|
||||
"""
|
||||
On the "Select the authentication method" screen (federate.sso.metlife.com),
|
||||
click the email tile (<div class="tile-button__content">KQ****@GMAIL.COM</div>)
|
||||
so the code gets sent. Returns True if a tile was clicked.
|
||||
"""
|
||||
js = r"""
|
||||
const tiles = Array.from(document.querySelectorAll('.tile-button__content'))
|
||||
.filter(el => el.offsetWidth || el.offsetHeight || el.getClientRects().length);
|
||||
const email = tiles.find(el => (el.innerText || '').includes('@'));
|
||||
if (!email) return null;
|
||||
return email.closest('button, a, [role=button], .tile-button') || email;
|
||||
"""
|
||||
self.driver.switch_to.default_content()
|
||||
frames = [None] + self.driver.find_elements(By.TAG_NAME, "iframe")
|
||||
for frame in frames:
|
||||
try:
|
||||
self.driver.switch_to.default_content()
|
||||
if frame is not None:
|
||||
self.driver.switch_to.frame(frame)
|
||||
tile = self.driver.execute_script(js)
|
||||
if not tile:
|
||||
continue
|
||||
try:
|
||||
tile.click()
|
||||
except Exception:
|
||||
self.driver.execute_script("arguments[0].click();", tile)
|
||||
print("[MetLife login] Selected email authentication method")
|
||||
self.driver.switch_to.default_content()
|
||||
return True
|
||||
except Exception:
|
||||
continue
|
||||
self.driver.switch_to.default_content()
|
||||
return False
|
||||
|
||||
def _on_portal(self):
|
||||
try:
|
||||
return METLIFE_BASE_URL in self.driver.current_url and "/presignin" not in self.driver.current_url
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _search_form_present(self, timeout=5):
|
||||
"""True once the logged-in portal UI (search box or Eligibility link) has rendered."""
|
||||
self.driver.switch_to.default_content()
|
||||
try:
|
||||
WebDriverWait(self.driver, timeout).until(
|
||||
lambda d: d.find_elements(By.XPATH, SUBSCRIBER_INPUT_XPATH)
|
||||
or d.find_elements(By.XPATH, BENEFIT_LINK_XPATH)
|
||||
)
|
||||
return True
|
||||
except TimeoutException:
|
||||
return False
|
||||
|
||||
def is_logged_in(self, timeout=5):
|
||||
return self._on_portal() and self._search_form_present(timeout)
|
||||
|
||||
# ── Login ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _force_logout(self):
|
||||
"""Force logout by clearing cookies when credentials change."""
|
||||
try:
|
||||
print("[MetLife login] Forcing logout due to credential change...")
|
||||
browser_manager = get_browser_manager()
|
||||
|
||||
try:
|
||||
self.driver.get(METLIFE_HOME_URL)
|
||||
time.sleep(2)
|
||||
logout_selectors = [
|
||||
"//button[contains(translate(normalize-space(.),'LOGUT','logut'),'logout') or contains(translate(normalize-space(.),'LOGUT','logut'),'log out')]",
|
||||
"//a[contains(translate(normalize-space(.),'LOGUT','logut'),'logout') or contains(translate(normalize-space(.),'LOGUT','logut'),'log out')]",
|
||||
]
|
||||
for selector in logout_selectors:
|
||||
try:
|
||||
logout_btn = WebDriverWait(self.driver, 3).until(
|
||||
EC.element_to_be_clickable((By.XPATH, selector))
|
||||
)
|
||||
logout_btn.click()
|
||||
print("[MetLife login] Clicked logout button")
|
||||
time.sleep(2)
|
||||
break
|
||||
except TimeoutException:
|
||||
continue
|
||||
except Exception as e:
|
||||
print(f"[MetLife login] Could not click logout button: {e}")
|
||||
|
||||
# Login spans the portal and the PingOne domain, so clear cookies for all domains
|
||||
try:
|
||||
self.driver.execute_cdp_cmd("Network.clearBrowserCookies", {})
|
||||
print("[MetLife login] Cleared all cookies")
|
||||
except Exception as e:
|
||||
print(f"[MetLife login] Error clearing cookies: {e}")
|
||||
|
||||
browser_manager.clear_credentials_hash()
|
||||
print("[MetLife login] Logout complete")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"[MetLife login] Error during forced logout: {e}")
|
||||
return False
|
||||
|
||||
def _open_login_page(self):
|
||||
"""From /presignin, click "Sign in" to reach the PingOne login page."""
|
||||
if "/presignin" not in self.driver.current_url:
|
||||
return
|
||||
try:
|
||||
btn = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable((By.XPATH, PRESIGNIN_BUTTON_XPATH))
|
||||
)
|
||||
btn.click()
|
||||
print("[MetLife login] Clicked 'Sign in' on presignin page")
|
||||
except TimeoutException:
|
||||
print("[MetLife login] 'Sign in' button not found on presignin page")
|
||||
return
|
||||
try:
|
||||
WebDriverWait(self.driver, 20).until(lambda d: "/presignin" not in d.current_url)
|
||||
except TimeoutException:
|
||||
pass
|
||||
time.sleep(2)
|
||||
|
||||
def login(self, url):
|
||||
browser_manager = get_browser_manager()
|
||||
|
||||
try:
|
||||
# Check if credentials changed — force logout first
|
||||
if self.metlife_username and browser_manager.credentials_changed(self.metlife_username):
|
||||
self._force_logout()
|
||||
|
||||
# Persistent session: home page shows the subscriber search form when logged in
|
||||
self.driver.get(url)
|
||||
time.sleep(3)
|
||||
print(f"[MetLife login] URL after navigation: {self.driver.current_url}")
|
||||
if self.is_logged_in(timeout=8):
|
||||
print("[MetLife login] Session valid - ALREADY_LOGGED_IN")
|
||||
return "ALREADY_LOGGED_IN"
|
||||
|
||||
self._open_login_page()
|
||||
|
||||
# The SSO page may bounce straight back to the portal if the IdP session is alive
|
||||
if self.is_logged_in(timeout=5):
|
||||
print("[MetLife login] IdP session valid - ALREADY_LOGGED_IN")
|
||||
return "ALREADY_LOGGED_IN"
|
||||
|
||||
# Login may already be at the MFA step (e.g. a previous attempt was interrupted)
|
||||
if self._select_email_mfa_method():
|
||||
self._wait_for_otp_input()
|
||||
print("[MetLife login] Email code requested -> OTP_REQUIRED")
|
||||
return "OTP_REQUIRED"
|
||||
if self._find("otp") and not self._find("password"):
|
||||
self.driver.switch_to.default_content()
|
||||
print("[MetLife login] OTP input already showing -> OTP_REQUIRED")
|
||||
return "OTP_REQUIRED"
|
||||
|
||||
# Username (PingOne may ask for username and password on separate screens)
|
||||
username_el = None
|
||||
for _ in range(20):
|
||||
username_el = self._find("username")
|
||||
if username_el:
|
||||
break
|
||||
time.sleep(1)
|
||||
if not username_el:
|
||||
self.driver.switch_to.default_content()
|
||||
print(f"[MetLife login] Login form not found. URL: {self.driver.current_url}")
|
||||
return "ERROR: Login form not found"
|
||||
|
||||
self._type_into(username_el, self.metlife_username)
|
||||
print("[MetLife login] Filled username")
|
||||
|
||||
password_el = self._find("password")
|
||||
if not password_el:
|
||||
# Two-step flow: submit username, then wait for password screen
|
||||
username_el = self._find("username")
|
||||
self._click_submit(username_el)
|
||||
for _ in range(15):
|
||||
time.sleep(1)
|
||||
password_el = self._find("password")
|
||||
if password_el:
|
||||
break
|
||||
if not password_el:
|
||||
self.driver.switch_to.default_content()
|
||||
return "ERROR:LOGIN FAILED: Password field not found"
|
||||
|
||||
self._type_into(password_el, self.metlife_password)
|
||||
print("[MetLife login] Filled password")
|
||||
self._click_submit(password_el)
|
||||
self.driver.switch_to.default_content()
|
||||
|
||||
# Save credentials hash after login attempt
|
||||
if self.metlife_username:
|
||||
browser_manager.save_credentials_hash(self.metlife_username)
|
||||
|
||||
# Wait up to 30s for: portal search form (success), OTP / MFA screen, or an error
|
||||
for _ in range(30):
|
||||
time.sleep(1)
|
||||
if self.is_logged_in(timeout=1):
|
||||
print("[MetLife login] Login successful without OTP")
|
||||
return "SUCCESS"
|
||||
if self._on_portal():
|
||||
continue # SPA still loading after redirect
|
||||
# "Select the authentication method" screen — pick the email tile
|
||||
if self._select_email_mfa_method():
|
||||
self._wait_for_otp_input()
|
||||
print("[MetLife login] Email code requested -> OTP_REQUIRED")
|
||||
return "OTP_REQUIRED"
|
||||
if self._find("otp"):
|
||||
self.driver.switch_to.default_content()
|
||||
print("[MetLife login] OTP input detected -> OTP_REQUIRED")
|
||||
return "OTP_REQUIRED"
|
||||
self.driver.switch_to.default_content()
|
||||
try:
|
||||
page_text = self.driver.execute_script("return document.body.innerText;").lower()
|
||||
except Exception:
|
||||
page_text = ""
|
||||
if re.search(r"incorrect|invalid (user|password|credentials)|not recognized|locked", page_text):
|
||||
snippet = re.search(r"[^\n]*(incorrect|invalid|not recognized|locked)[^\n]*", page_text)
|
||||
msg = snippet.group(0).strip() if snippet else "invalid credentials"
|
||||
print(f"[MetLife login] Login failed: {msg}")
|
||||
return f"ERROR:LOGIN FAILED: {msg}"
|
||||
# MFA method picker ("send code by email / text") — the user picks it in the
|
||||
# browser window; treat it as the OTP step so the app shows the OTP modal.
|
||||
if re.search(r"verification|verify your identity|one-time|passcode|send code|authenticat", page_text):
|
||||
print("[MetLife login] MFA screen detected -> OTP_REQUIRED")
|
||||
return "OTP_REQUIRED"
|
||||
|
||||
if self.is_logged_in(timeout=5):
|
||||
return "SUCCESS"
|
||||
if self._find("password"):
|
||||
self.driver.switch_to.default_content()
|
||||
return "ERROR:LOGIN FAILED: Still on login page"
|
||||
self.driver.switch_to.default_content()
|
||||
# Unknown intermediate screen — let the user finish it in the browser
|
||||
print(f"[MetLife login] Unrecognized post-login screen ({self.driver.current_url}) -> OTP_REQUIRED")
|
||||
return "OTP_REQUIRED"
|
||||
|
||||
except Exception as e:
|
||||
try:
|
||||
self.driver.switch_to.default_content()
|
||||
except Exception:
|
||||
pass
|
||||
print("[MetLife login] Exception during login:", e)
|
||||
return f"ERROR:LOGIN FAILED: {e}"
|
||||
|
||||
def _wait_for_otp_input(self, timeout=20):
|
||||
"""After choosing the email method, wait for the code input to appear."""
|
||||
for _ in range(timeout):
|
||||
time.sleep(1)
|
||||
if self._find("otp"):
|
||||
self.driver.switch_to.default_content()
|
||||
print("[MetLife login] OTP input is showing")
|
||||
return True
|
||||
self.driver.switch_to.default_content()
|
||||
print("[MetLife login] OTP input did not appear yet — user can finish in the browser")
|
||||
return False
|
||||
|
||||
def submit_otp_code(self, otp):
|
||||
"""Type an OTP submitted from the app into the PingOne OTP input."""
|
||||
otp_el = self._find("otp")
|
||||
if not otp_el:
|
||||
self.driver.switch_to.default_content()
|
||||
return False
|
||||
self._type_into(otp_el, otp)
|
||||
self._click_submit(otp_el)
|
||||
self.driver.switch_to.default_content()
|
||||
return True
|
||||
|
||||
# ── Step 1: search ────────────────────────────────────────────────────────
|
||||
|
||||
def _wait_for_search_outcome(self, timeout=30):
|
||||
"""Return 'tiles', 'extra_inputs', 'no_results' or 'timeout'."""
|
||||
end = time.time() + timeout
|
||||
while time.time() < end:
|
||||
if self.driver.find_elements(By.CSS_SELECTOR, RESULT_TILE_CSS):
|
||||
return "tiles"
|
||||
if self.driver.find_elements(By.XPATH, EXTRA_LASTNAME_XPATH) or \
|
||||
self.driver.find_elements(By.XPATH, EXTRA_ZIPCODE_XPATH):
|
||||
return "extra_inputs"
|
||||
try:
|
||||
text = self.driver.execute_script("return document.body.innerText;")
|
||||
except Exception:
|
||||
text = ""
|
||||
if "Your search for" in text or "making sure that the information you've entered is correct" in text:
|
||||
return "no_results"
|
||||
time.sleep(0.5)
|
||||
return "timeout"
|
||||
|
||||
def _fill_input(self, el, value):
|
||||
"""
|
||||
Type into a portal input that Selenium may consider not interactable
|
||||
(masked SSN box under a floating label). Focus it via JS, type with key
|
||||
actions, and fall back to React's native value setter. Returns True if
|
||||
the input ends up holding `value`.
|
||||
"""
|
||||
from selenium.webdriver.common.action_chains import ActionChains
|
||||
|
||||
try:
|
||||
self.driver.execute_script(
|
||||
"arguments[0].scrollIntoView({block:'center'}); arguments[0].focus();", el
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 1) Normal typing
|
||||
try:
|
||||
self._type_into(el, value)
|
||||
except Exception as e:
|
||||
print(f"[MetLife] direct typing failed ({type(e).__name__}); using key actions")
|
||||
# 2) Click the label / focus, then type into whatever has focus
|
||||
try:
|
||||
el_id = el.get_attribute("id")
|
||||
labels = self.driver.find_elements(By.XPATH, f"//label[@for='{el_id}']") if el_id else []
|
||||
if labels:
|
||||
self.driver.execute_script("arguments[0].click();", labels[0])
|
||||
self.driver.execute_script("arguments[0].focus();", el)
|
||||
ActionChains(self.driver).key_down(Keys.CONTROL).send_keys("a").key_up(Keys.CONTROL) \
|
||||
.send_keys(Keys.BACKSPACE).send_keys(value).perform()
|
||||
except Exception as e2:
|
||||
print(f"[MetLife] key actions failed: {e2}")
|
||||
|
||||
current = self.driver.execute_script("return arguments[0].value;", el) or ""
|
||||
if self._value_matches(current, value):
|
||||
return True
|
||||
if current:
|
||||
# Something was typed but reads back differently (masking variants) —
|
||||
# don't overwrite it; the search result will tell if it was wrong.
|
||||
print(f"[MetLife] input reads back as '{current}' (masked?) — keeping it")
|
||||
return True
|
||||
|
||||
# 3) Nothing landed — React-controlled input: set via the native setter and fire input/change
|
||||
print("[MetLife] input is empty after typing, setting it via native setter")
|
||||
self.driver.execute_script(
|
||||
"""
|
||||
const el = arguments[0], v = arguments[1];
|
||||
const setter = Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, 'value').set;
|
||||
setter.call(el, v);
|
||||
el.dispatchEvent(new Event('input', { bubbles: true }));
|
||||
el.dispatchEvent(new Event('change', { bubbles: true }));
|
||||
""",
|
||||
el, value,
|
||||
)
|
||||
current = self.driver.execute_script("return arguments[0].value;", el) or ""
|
||||
return self._value_matches(current, value)
|
||||
|
||||
@staticmethod
|
||||
def _value_matches(current, value):
|
||||
"""
|
||||
The Subscriber ID / SSN box masks what was typed (dots until the eye
|
||||
button is clicked), so a same-length run of mask characters counts as a match.
|
||||
"""
|
||||
if current == value:
|
||||
return True
|
||||
return len(current) == len(value) and all(c in "•●*·∙◦xX" for c in current)
|
||||
|
||||
def _click_search(self):
|
||||
btn = WebDriverWait(self.driver, 15).until(
|
||||
EC.presence_of_element_located((By.XPATH, SEARCH_SUBMIT_XPATH))
|
||||
)
|
||||
try:
|
||||
btn.click()
|
||||
except Exception:
|
||||
self.driver.execute_script("arguments[0].click();", btn)
|
||||
print("[MetLife step1] Clicked Search button")
|
||||
|
||||
def _open_eligibility_page(self):
|
||||
"""From the home page, click "Eligibility & Plan Details" to reach /eligibility."""
|
||||
if "/eligibility" in self.driver.current_url:
|
||||
return
|
||||
if "/home" not in self.driver.current_url:
|
||||
self.driver.get(METLIFE_HOME_URL)
|
||||
try:
|
||||
WebDriverWait(self.driver, 20).until(
|
||||
EC.presence_of_element_located((By.XPATH, BENEFIT_LINK_XPATH))
|
||||
)
|
||||
links = [l for l in self.driver.find_elements(By.XPATH, BENEFIT_LINK_XPATH) if l.is_displayed()]
|
||||
link = links[0] if links else self.driver.find_element(By.XPATH, BENEFIT_LINK_XPATH)
|
||||
try:
|
||||
link.click()
|
||||
except Exception:
|
||||
# href="javascript:;" — the React onClick does the navigation
|
||||
self.driver.execute_script("arguments[0].click();", link)
|
||||
print("[MetLife step1] Clicked 'Eligibility & Plan Details'")
|
||||
WebDriverWait(self.driver, 15).until(lambda d: "/eligibility" in d.current_url)
|
||||
print(f"[MetLife step1] On eligibility page: {self.driver.current_url}")
|
||||
time.sleep(1.5)
|
||||
except TimeoutException:
|
||||
# Fall back to the search box on the home page
|
||||
print(f"[MetLife step1] Could not open eligibility page; searching from {self.driver.current_url}")
|
||||
|
||||
def step1(self):
|
||||
"""Open Eligibility & Plan Details and search by Subscriber ID / SSN."""
|
||||
wait = WebDriverWait(self.driver, 30)
|
||||
|
||||
try:
|
||||
# Portal accepts letters and digits only (SSNs are entered without dashes)
|
||||
search_id = re.sub(r"[^A-Za-z0-9]", "", self.memberId or "")
|
||||
if not search_id:
|
||||
return "ERROR: Member ID (Subscriber ID / SSN) is required for MetLife"
|
||||
print(f"[MetLife step1] Searching SubscriberID: {search_id}")
|
||||
|
||||
self._open_eligibility_page()
|
||||
# Wait for presence only: the masked input sits under its floating label, so
|
||||
# Selenium may never consider it "clickable" even though it's usable.
|
||||
try:
|
||||
subscriber_input = WebDriverWait(self.driver, 10).until(
|
||||
EC.presence_of_element_located((By.XPATH, SUBSCRIBER_INPUT_XPATH))
|
||||
)
|
||||
except TimeoutException:
|
||||
print("[MetLife step1] No search box on eligibility page — using the home page search")
|
||||
self.driver.get(METLIFE_HOME_URL)
|
||||
try:
|
||||
subscriber_input = wait.until(EC.presence_of_element_located((By.XPATH, SUBSCRIBER_INPUT_XPATH)))
|
||||
except TimeoutException:
|
||||
inputs = self.driver.execute_script(
|
||||
"return Array.from(document.querySelectorAll('input')).map(i => "
|
||||
"[i.id, i.name, i.type, i.offsetWidth > 0, i.getAttribute('aria-label')].join(' | '));"
|
||||
)
|
||||
print(f"[MetLife step1] Search box not found. Inputs on page: {inputs}")
|
||||
return "ERROR: MetLife search box not found"
|
||||
time.sleep(0.5)
|
||||
if not self._fill_input(subscriber_input, search_id):
|
||||
return "ERROR: Could not type the Subscriber ID into the MetLife search box"
|
||||
time.sleep(0.3)
|
||||
|
||||
self._click_search()
|
||||
|
||||
outcome = self._wait_for_search_outcome()
|
||||
|
||||
# Ambiguous ID — portal asks for the subscriber's last name and/or zip code
|
||||
if outcome == "extra_inputs":
|
||||
print("[MetLife step1] Portal requested extra subscriber info")
|
||||
filled_any = False
|
||||
lastname_inputs = self.driver.find_elements(By.XPATH, EXTRA_LASTNAME_XPATH)
|
||||
if lastname_inputs and self.subscriberLastName:
|
||||
filled_any = self._fill_input(lastname_inputs[0], self.subscriberLastName) or filled_any
|
||||
zip_inputs = self.driver.find_elements(By.XPATH, EXTRA_ZIPCODE_XPATH)
|
||||
if zip_inputs and self.zipCode:
|
||||
filled_any = self._fill_input(zip_inputs[0], self.zipCode) or filled_any
|
||||
if not filled_any:
|
||||
return "ERROR: MetLife requires the subscriber's last name or zip code for this ID"
|
||||
self._click_search()
|
||||
outcome = self._wait_for_search_outcome()
|
||||
|
||||
if outcome == "tiles":
|
||||
time.sleep(1.5) # let all family tiles render
|
||||
print("[MetLife step1] Search completed successfully")
|
||||
return "Success"
|
||||
if outcome == "no_results":
|
||||
return "ERROR: INVALID SEARCH CRITERIA"
|
||||
if outcome == "extra_inputs":
|
||||
return "ERROR: MetLife could not match the subscriber with the provided last name / zip code"
|
||||
print(f"[MetLife step1] No results within timeout. URL: {self.driver.current_url}")
|
||||
return "ERROR: INVALID SEARCH CRITERIA"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[MetLife step1] Exception: {e}")
|
||||
return f"ERROR:STEP1 - {e}"
|
||||
|
||||
# ── Step 2: pick patient, open plan details, PDF ──────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _parse_tile(text):
|
||||
"""Extract name, DOB, status and coverage dates from a tile's innerText."""
|
||||
lines = [l.strip() for l in text.split("\n") if l.strip()]
|
||||
info = {"name": lines[0] if lines else "", "dob": "", "status": "", "start": "", "end": ""}
|
||||
m = re.search(r"DOB:\s*(\d{2}/\d{2}/\d{4})", text)
|
||||
if m:
|
||||
info["dob"] = m.group(1)
|
||||
m = re.search(r"Status:\s*([^|\n]+)", text)
|
||||
if m:
|
||||
info["status"] = m.group(1).strip()
|
||||
m = re.search(r"Start Date:\s*([^|\n]+)", text)
|
||||
if m:
|
||||
info["start"] = m.group(1).strip()
|
||||
m = re.search(r"End Date:\s*([^|\n]+)", text)
|
||||
if m:
|
||||
info["end"] = m.group(1).strip()
|
||||
return info
|
||||
|
||||
@staticmethod
|
||||
def _eligibility_from(status, end_date):
|
||||
s = (status or "").lower()
|
||||
if any(w in s for w in ("inactive", "termin", "termed", "cancel", "expired", "not eligible")):
|
||||
return "inactive"
|
||||
try:
|
||||
if end_date and datetime.strptime(end_date, "%m/%d/%Y").date() < datetime.now().date():
|
||||
return "inactive"
|
||||
except ValueError:
|
||||
pass
|
||||
if "active" in s or "eligible" in s:
|
||||
return "active"
|
||||
return s or "unknown"
|
||||
|
||||
STATUS_WORDS = ("active", "inactive", "cancelled", "canceled", "terminated", "termed", "expired")
|
||||
|
||||
def _read_detail_page(self):
|
||||
"""
|
||||
Read the plan details page: patient card ("Subscriber | 68 | DOB: 03/19/1958 | Female"),
|
||||
status badge (e.g. "Cancelled"), End Date, and the
|
||||
"coverage is no longer in effect" notice.
|
||||
"""
|
||||
out = {"dob": "", "status": "", "end": "", "no_longer_in_effect": False}
|
||||
try:
|
||||
text = self.driver.execute_script("return document.body.innerText;") or ""
|
||||
except Exception:
|
||||
return out
|
||||
m = re.search(r"DOB:\s*(\d{2}/\d{2}/\d{4})", text)
|
||||
if m:
|
||||
out["dob"] = m.group(1)
|
||||
# The badge is its own line containing only the status word
|
||||
for line in text.split("\n"):
|
||||
if line.strip().lower() in self.STATUS_WORDS:
|
||||
out["status"] = line.strip()
|
||||
break
|
||||
m = re.search(r"End Date\s*:?\s*\n?\s*(\d{2}/\d{2}/\d{4})", text)
|
||||
if m:
|
||||
out["end"] = m.group(1)
|
||||
out["no_longer_in_effect"] = "no longer in effect" in text.lower()
|
||||
return out
|
||||
|
||||
def step2(self):
|
||||
try:
|
||||
eligibilityText = "unknown"
|
||||
foundMemberId = self.memberId or ""
|
||||
patientName = f"{self.firstName} {self.lastName}".strip() if self.firstName and self.lastName else ""
|
||||
known_patient_name = bool(patientName)
|
||||
|
||||
tiles = self.driver.find_elements(By.CSS_SELECTOR, RESULT_TILE_CSS)
|
||||
print(f"[MetLife step2] Found {len(tiles)} covered person tile(s)")
|
||||
parsed = [(t, self._parse_tile(t.text)) for t in tiles]
|
||||
|
||||
target = None
|
||||
if self.dateOfBirth:
|
||||
try:
|
||||
y, mo, d = self.dateOfBirth.split("-")[:3]
|
||||
expected_dob = f"{mo.zfill(2)}/{d.zfill(2)}/{y}"
|
||||
matches = [p for p in parsed if p[1]["dob"] == expected_dob]
|
||||
# Same DOB (e.g. twins) — narrow by first name
|
||||
if len(matches) > 1 and self.firstName:
|
||||
by_name = [p for p in matches if self.firstName.lower() in p[1]["name"].lower()]
|
||||
matches = by_name or matches
|
||||
if matches:
|
||||
target = matches[0]
|
||||
except Exception as e:
|
||||
print(f"[MetLife step2] DOB matching error (non-fatal): {e}")
|
||||
if target is None and self.firstName:
|
||||
target = next((p for p in parsed if self.firstName.lower() in p[1]["name"].lower()), None)
|
||||
if target is None and parsed:
|
||||
target = parsed[0]
|
||||
print("[MetLife step2] No DOB/name match — defaulting to first tile")
|
||||
if target is None:
|
||||
return {"status": "error", "message": "No covered persons found for this subscriber"}
|
||||
|
||||
tile, info = target
|
||||
print(f"[MetLife step2] Selected tile: {info}")
|
||||
if not known_patient_name and info["name"]:
|
||||
patientName = re.sub(r"\s+", " ", info["name"]).strip()
|
||||
eligibilityText = self._eligibility_from(info["status"], info["end"])
|
||||
print(f"[MetLife step2] Eligibility: {eligibilityText} (status='{info['status']}', end='{info['end']}')")
|
||||
|
||||
# Selecting a tile navigates to /eligibility/plandetails
|
||||
try:
|
||||
tile.click()
|
||||
except Exception:
|
||||
self.driver.execute_script("arguments[0].click();", tile)
|
||||
try:
|
||||
WebDriverWait(self.driver, 20).until(lambda d: "/eligibility/plandetails" in d.current_url)
|
||||
print(f"[MetLife step2] On plan details page: {self.driver.current_url}")
|
||||
except TimeoutException:
|
||||
print(f"[MetLife step2] Warning: plan details page did not open; printing {self.driver.current_url}")
|
||||
|
||||
try:
|
||||
WebDriverWait(self.driver, 30).until(
|
||||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
# Plan details sections load asynchronously after navigation
|
||||
time.sleep(5)
|
||||
|
||||
# Double-check the detail page: DOB must match the input, and its status
|
||||
# badge / coverage notice is the final word on eligibility.
|
||||
details = self._read_detail_page()
|
||||
print(f"[MetLife step2] Detail page: {details}")
|
||||
if self.dateOfBirth and details["dob"]:
|
||||
y, mo, d = self.dateOfBirth.split("-")[:3]
|
||||
expected_dob = f"{mo.zfill(2)}/{d.zfill(2)}/{y}"
|
||||
if details["dob"] != expected_dob:
|
||||
return {
|
||||
"status": "error",
|
||||
"message": f"DOB mismatch on MetLife detail page: portal shows {details['dob']}, expected {expected_dob}",
|
||||
}
|
||||
print(f"[MetLife step2] DOB verified: {expected_dob}")
|
||||
if details["no_longer_in_effect"]:
|
||||
eligibilityText = "inactive"
|
||||
elif details["status"]:
|
||||
eligibilityText = self._eligibility_from(details["status"], details["end"] or info["end"])
|
||||
print(f"[MetLife step2] Final eligibility: {eligibilityText}")
|
||||
|
||||
print("[MetLife step2] Generating PDF of plan details page...")
|
||||
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,
|
||||
}
|
||||
result = self.driver.execute_cdp_cmd("Page.printToPDF", pdf_options)
|
||||
pdf_data = base64.b64decode(result.get("data", ""))
|
||||
pdf_id = re.sub(r"[^A-Za-z0-9]", "", foundMemberId) or "unknown"
|
||||
pdf_path = os.path.join(self.download_dir, f"eligibility_metlife_{pdf_id}.pdf")
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(pdf_data)
|
||||
print(f"[MetLife step2] PDF saved at: {pdf_path}")
|
||||
print(f"[MetLife step2] Final — PatientName: '{patientName}', MemberID: '{foundMemberId}'")
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"eligibility": eligibilityText,
|
||||
"ss_path": pdf_path, # kept for parity with other workers
|
||||
"pdf_path": pdf_path,
|
||||
"patientName": patientName,
|
||||
"memberId": foundMemberId,
|
||||
"dateOfBirth": self.dateOfBirth,
|
||||
"coverageStart": info["start"],
|
||||
"coverageEnd": info["end"],
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print("ERROR in MetLife step2:", e)
|
||||
# Cleanup download dir on error
|
||||
try:
|
||||
dl = os.path.abspath(self.download_dir)
|
||||
if os.path.isdir(dl):
|
||||
for name in os.listdir(dl):
|
||||
item = os.path.join(dl, name)
|
||||
try:
|
||||
if os.path.isfile(item) or os.path.islink(item):
|
||||
os.remove(item)
|
||||
except Exception as rm_err:
|
||||
print(f"[cleanup] failed to remove {item}: {rm_err}")
|
||||
except Exception as cleanup_exc:
|
||||
print(f"[cleanup] unexpected error: {cleanup_exc}")
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
def main_workflow(self, url):
|
||||
try:
|
||||
self.config_driver()
|
||||
self.driver.maximize_window()
|
||||
time.sleep(3)
|
||||
|
||||
login_result = self.login(url)
|
||||
if login_result.startswith("ERROR"):
|
||||
return {"status": "error", "message": login_result}
|
||||
if login_result == "OTP_REQUIRED":
|
||||
return {"status": "otp_required", "message": "OTP required after login"}
|
||||
|
||||
step1_result = self.step1()
|
||||
if step1_result.startswith("ERROR"):
|
||||
return {"status": "error", "message": step1_result}
|
||||
|
||||
step2_result = self.step2()
|
||||
if step2_result.get("status") == "error":
|
||||
return {"status": "error", "message": step2_result.get("message")}
|
||||
|
||||
return step2_result
|
||||
except WebDriverException as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": str(e)}
|
||||
# NOTE: Do NOT quit driver — keep browser alive for next patient
|
||||
Reference in New Issue
Block a user