GiteadandClaude Opus 5.5
a5beeb7ba9
Reuses the MetLife eligibility login (persistent profile, email MFA, OTP from app or browser), then Submit Claims -> ClaimConnect -> DentalXChange Single Claim: billing/rendering provider, member ID, last name, payer Metlife 65978, patient picked by DOB/name, signature on file Yes, one claim item per CDT code (date, code, tooth, surfaces, fee; Add Claim Item past 3), Finish, Submit Claim, then Back to Claim to read the Claim ID and save the page as PDF. Waits out Wicket ajax/veil before each click. MetLife now has its own fee schedule (seeded from DDMA) instead of falling back to MassHealth prices. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
317 lines
13 KiB
Python
317 lines
13 KiB
Python
import os
|
|
import time
|
|
import asyncio
|
|
from typing import Dict, Any
|
|
from selenium.common.exceptions import WebDriverException
|
|
|
|
from selenium_MetLife_claimSubmitWorker import AutomationMetLifeClaimSubmit
|
|
from selenium_MetLife_eligibilityCheckWorker import METLIFE_HOME_URL
|
|
|
|
sessions: Dict[str, Dict[str, Any]] = {}
|
|
|
|
SESSION_OTP_TIMEOUT = int(os.getenv("SESSION_OTP_TIMEOUT", "120"))
|
|
|
|
|
|
def make_session_entry() -> str:
|
|
import uuid
|
|
sid = str(uuid.uuid4())
|
|
sessions[sid] = {
|
|
"status": "created",
|
|
"created_at": time.time(),
|
|
"last_activity": time.time(),
|
|
"bot": None,
|
|
"driver": None,
|
|
"otp_event": asyncio.Event(),
|
|
"otp_value": None,
|
|
"result": None,
|
|
"message": None,
|
|
}
|
|
return sid
|
|
|
|
|
|
async def cleanup_session(sid: str, message: str | None = None):
|
|
"""
|
|
Set final state and remove the session entry.
|
|
Does NOT quit the browser — the persistent MetLife 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_claim] 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_claim_run(sid: str, data: dict, url: str):
|
|
"""
|
|
Run the MetLife claim workflow — login/OTP mirrors helpers_metlife_eligibility,
|
|
then continues into claim-specific steps.
|
|
|
|
OTP handling uses two complementary strategies:
|
|
1. Accept OTP submitted from the app (via /submit-otp → otp_value)
|
|
2. Poll the browser 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 = AutomationMetLifeClaimSubmit(data)
|
|
bot.config_driver()
|
|
|
|
s["bot"] = bot
|
|
s["driver"] = bot.driver
|
|
s["last_activity"] = time.time()
|
|
|
|
try:
|
|
bot.driver.maximize_window()
|
|
except Exception:
|
|
pass
|
|
|
|
# --- 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 (persistent session) ────────────────────────────
|
|
if login_result == "ALREADY_LOGGED_IN":
|
|
print("[MetLife Claim] Session persisted — skipping OTP")
|
|
s["status"] = "running"
|
|
s["message"] = "Session persisted"
|
|
|
|
# ── 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 Claim 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 Claim OTP poll {poll+1}] OTP received from app, typing it in...")
|
|
s["otp_value"] = None
|
|
if bot.submit_otp_code(otp_value):
|
|
print("[MetLife Claim OTP] OTP typed and submitted via app")
|
|
await asyncio.sleep(3)
|
|
else:
|
|
print("[MetLife Claim 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 Claim OTP] Search form found — login successful!")
|
|
login_success = True
|
|
break
|
|
|
|
print(f"[MetLife Claim OTP poll {poll+1}/{SESSION_OTP_TIMEOUT}] URL: {driver.current_url[:70]}...")
|
|
except Exception as poll_err:
|
|
print(f"[MetLife Claim OTP poll {poll+1}] Error: {poll_err}")
|
|
|
|
if not login_success:
|
|
# Final check — navigate directly to the portal home page
|
|
try:
|
|
print("[MetLife Claim 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 Claim OTP] Proceeding to claim steps...")
|
|
|
|
# ── Login succeeded without OTP ───────────────────────────────────────
|
|
elif login_result == "SUCCESS":
|
|
print("[MetLife Claim] Login succeeded without OTP")
|
|
s["status"] = "running"
|
|
s["message"] = "Login succeeded"
|
|
|
|
# ── 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: Open Submit Claims ---
|
|
step1_result = bot.step1_open_submit_claims()
|
|
print(f"[MetLife Claim] step1 result: {step1_result}")
|
|
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: Open ClaimConnect ---
|
|
step2_result = bot.step2_open_claimconnect()
|
|
print(f"[MetLife Claim] step2 result: {step2_result}")
|
|
if isinstance(step2_result, str) and step2_result.startswith("ERROR"):
|
|
s["status"] = "error"
|
|
s["message"] = step2_result
|
|
await cleanup_session(sid)
|
|
return {"status": "error", "message": step2_result}
|
|
|
|
# --- Step 3: Open Single Claim on DentalXChange ---
|
|
step3_result = bot.step3_open_single_claim()
|
|
print(f"[MetLife Claim] step3 result: {step3_result}")
|
|
if isinstance(step3_result, str) and step3_result.startswith("ERROR"):
|
|
s["status"] = "error"
|
|
s["message"] = step3_result
|
|
await cleanup_session(sid)
|
|
return {"status": "error", "message": step3_result}
|
|
|
|
# --- Step 4: Start New Claim (provider, patient, payer → Next) ---
|
|
step4_result = bot.step4_start_new_claim()
|
|
print(f"[MetLife Claim] step4 result: {step4_result}")
|
|
if isinstance(step4_result, str) and step4_result.startswith("ERROR"):
|
|
s["status"] = "error"
|
|
s["message"] = step4_result
|
|
await cleanup_session(sid)
|
|
return {"status": "error", "message": step4_result}
|
|
|
|
# --- Step 5: Select Patient → Create Claim ---
|
|
step5_result = bot.step5_select_patient()
|
|
print(f"[MetLife Claim] step5 result: {step5_result}")
|
|
if isinstance(step5_result, str) and step5_result.startswith("ERROR"):
|
|
s["status"] = "error"
|
|
s["message"] = step5_result
|
|
await cleanup_session(sid)
|
|
return {"status": "error", "message": step5_result}
|
|
|
|
# --- Step 6: Claim Detail — Signature on file → Next ---
|
|
step6_result = bot.step6_claim_detail()
|
|
print(f"[MetLife Claim] step6 result: {step6_result}")
|
|
if isinstance(step6_result, str) and step6_result.startswith("ERROR"):
|
|
s["status"] = "error"
|
|
s["message"] = step6_result
|
|
await cleanup_session(sid)
|
|
return {"status": "error", "message": step6_result}
|
|
|
|
# --- Step 7: Claim Item Detail — procedure lines → Finish ---
|
|
step7_result = bot.step7_fill_claim_items()
|
|
print(f"[MetLife Claim] step7 result: {step7_result}")
|
|
if isinstance(step7_result, str) and step7_result.startswith("ERROR"):
|
|
s["status"] = "error"
|
|
s["message"] = step7_result
|
|
await cleanup_session(sid)
|
|
return {"status": "error", "message": step7_result}
|
|
|
|
# --- Step 8: Review page → Submit Claim ---
|
|
step8_result = bot.step8_submit_claim()
|
|
print(f"[MetLife Claim] step8 result: {step8_result}")
|
|
if isinstance(step8_result, str) and step8_result.startswith("ERROR"):
|
|
s["status"] = "error"
|
|
s["message"] = step8_result
|
|
await cleanup_session(sid)
|
|
return {"status": "error", "message": step8_result}
|
|
|
|
# --- Step 9: Back to Claim → Claim ID + PDF ---
|
|
# The claim is already submitted, so a failure here is non-fatal.
|
|
step9_result = bot.step9_save_claim()
|
|
print(f"[MetLife Claim] step9 result: {step9_result}")
|
|
if not isinstance(step9_result, dict):
|
|
step9_result = {}
|
|
|
|
claim_number = step9_result.get("claimNumber")
|
|
pdf_path = step9_result.get("pdf_path")
|
|
pdf_url = None
|
|
if pdf_path:
|
|
port = os.getenv("PORT", "5002")
|
|
url_host = os.getenv("HOST", "localhost")
|
|
pdf_url = f"http://{url_host}:{port}/downloads/{os.path.basename(pdf_path)}"
|
|
print(f"[MetLife Claim] pdf_url: {pdf_url}")
|
|
|
|
result = {
|
|
"status": "success",
|
|
"message": "MetLife claim submitted via DentalXChange",
|
|
"claimNumber": claim_number,
|
|
"pdf_url": pdf_url,
|
|
}
|
|
s["status"] = "completed"
|
|
s["result"] = result
|
|
s["message"] = "completed"
|
|
|
|
asyncio.create_task(_remove_session_later(sid, 60))
|
|
return result
|
|
|
|
except Exception as e:
|
|
if s:
|
|
s["status"] = "error"
|
|
s["message"] = f"worker exception: {e}"
|
|
await cleanup_session(sid)
|
|
return {"status": "error", "message": f"worker exception: {e}"}
|
|
|
|
|
|
def submit_otp(sid: str, otp: str) -> Dict[str, Any]:
|
|
s = sessions.get(sid)
|
|
if not s:
|
|
return {"status": "error", "message": "session not found"}
|
|
if s.get("status") != "waiting_for_otp":
|
|
return {"status": "error", "message": f"session not waiting for otp (state={s.get('status')})"}
|
|
s["otp_value"] = otp
|
|
s["last_activity"] = time.time()
|
|
try:
|
|
s["otp_event"].set()
|
|
except Exception:
|
|
pass
|
|
return {"status": "ok", "message": "otp accepted"}
|
|
|
|
|
|
def get_session_status(sid: str) -> Dict[str, Any]:
|
|
s = sessions.get(sid)
|
|
if not s:
|
|
return {"status": "not_found"}
|
|
return {
|
|
"session_id": sid,
|
|
"status": s.get("status"),
|
|
"message": s.get("message"),
|
|
"created_at": s.get("created_at"),
|
|
"last_activity": s.get("last_activity"),
|
|
"result": s.get("result") if s.get("status") in ("completed", "error") else None,
|
|
}
|