feat: add BCBS MA eligibility check with OTP flow
- New Selenium worker (fresh Chrome per run, no persistent session) login → OTP modal → eTools → ConnectCenter → Verification → New Eligibility Request → fill form (NPI, member ID, DOB) → Expand All → CDP PDF back to app - Backend route fetches BCBS_MA credentials + provider NPI from settings - Frontend OTP modal with 6-digit code entry - BCBS MA added to insurance credentials dropdown in settings Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
ff
462
apps/SeleniumService/selenium_BCBS_MA_eligibilityCheckWorker.py
Normal file
462
apps/SeleniumService/selenium_BCBS_MA_eligibilityCheckWorker.py
Normal file
@@ -0,0 +1,462 @@
|
||||
import os
|
||||
import time
|
||||
import base64
|
||||
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 TimeoutException, WebDriverException
|
||||
from selenium import webdriver
|
||||
from selenium.webdriver.chrome.options import Options
|
||||
from selenium.webdriver.chrome.service import Service
|
||||
|
||||
|
||||
DOWNLOAD_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "downloads", "bcbs_ma"))
|
||||
|
||||
|
||||
def _fresh_driver() -> webdriver.Chrome:
|
||||
"""Create a disposable Chrome instance for a single BCBS MA session."""
|
||||
os.makedirs(DOWNLOAD_DIR, exist_ok=True)
|
||||
options = Options()
|
||||
options.add_argument("--no-sandbox")
|
||||
options.add_argument("--disable-dev-shm-usage")
|
||||
options.add_argument("--disable-blink-features=AutomationControlled")
|
||||
options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
||||
options.add_experimental_option("useAutomationExtension", False)
|
||||
options.add_experimental_option("prefs", {
|
||||
"download.default_directory": DOWNLOAD_DIR,
|
||||
"download.prompt_for_download": False,
|
||||
"plugins.always_open_pdf_externally": True,
|
||||
})
|
||||
headless = os.getenv("SELENIUM_HEADLESS", "false").lower() == "true"
|
||||
if headless:
|
||||
options.add_argument("--headless=new")
|
||||
|
||||
try:
|
||||
from webdriver_manager.chrome import ChromeDriverManager
|
||||
service = Service(ChromeDriverManager().install())
|
||||
except Exception:
|
||||
service = Service()
|
||||
|
||||
driver = webdriver.Chrome(service=service, options=options)
|
||||
driver.maximize_window()
|
||||
return driver
|
||||
|
||||
|
||||
class AutomationBCBSMAEligibilityCheck:
|
||||
"""
|
||||
BCBS MA Provider Central eligibility check.
|
||||
|
||||
No persistent session — fresh Chrome every run because BCBS MA
|
||||
always requires OTP on new login (the OTP prefix changes each time).
|
||||
|
||||
Flow: login(url) → OTP_REQUIRED → [caller polls for OTP] → submit_otp_step(otp)
|
||||
→ step1() [search member] → step2() [extract + PDF]
|
||||
"""
|
||||
|
||||
LOGIN_URL = "https://provider.bluecrossma.com/ProviderHome/portal/"
|
||||
|
||||
def __init__(self, data: dict):
|
||||
raw = data.get("data", data)
|
||||
self.member_id = raw.get("memberId", "")
|
||||
self.dob = raw.get("dateOfBirth", "") # YYYY-MM-DD from frontend
|
||||
self.first_name = raw.get("firstName", "")
|
||||
self.last_name = raw.get("lastName", "")
|
||||
self.username = raw.get("bcbsMaUsername", "")
|
||||
self.password = raw.get("bcbsMaPassword", "")
|
||||
self.provider_npi = raw.get("providerNpi", "")
|
||||
self.driver: webdriver.Chrome | None = None
|
||||
|
||||
# ── Driver lifecycle ──────────────────────────────────────────────────────
|
||||
|
||||
def config_driver(self):
|
||||
self.driver = _fresh_driver()
|
||||
|
||||
def close_driver(self):
|
||||
try:
|
||||
if self.driver:
|
||||
self.driver.quit()
|
||||
except Exception:
|
||||
pass
|
||||
self.driver = None
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────
|
||||
|
||||
def _wait(self, timeout=15):
|
||||
return WebDriverWait(self.driver, timeout)
|
||||
|
||||
def _dob_mmddyyyy(self) -> str:
|
||||
"""Convert YYYY-MM-DD → MM/DD/YYYY."""
|
||||
try:
|
||||
parts = self.dob.split("-")
|
||||
return f"{parts[1]}/{parts[2]}/{parts[0]}"
|
||||
except Exception:
|
||||
return self.dob
|
||||
|
||||
# ── Step: Login ───────────────────────────────────────────────────────────
|
||||
|
||||
def login(self, url: str = "") -> str:
|
||||
"""
|
||||
Navigate to BCBS MA Provider Central, fill credentials, click Log in.
|
||||
Returns:
|
||||
"OTP_REQUIRED" – redirected to MFA challenge page (/mga/sps/authsvc)
|
||||
"SUCCESS" – landed on dashboard without OTP
|
||||
"ERROR:..." – something went wrong
|
||||
"""
|
||||
target = url or self.LOGIN_URL
|
||||
try:
|
||||
print(f"[BCBS MA login] Navigating to {target}")
|
||||
self.driver.get(target)
|
||||
|
||||
# Wait for username field: id="txtUsername0"
|
||||
username_input = self._wait(20).until(
|
||||
EC.presence_of_element_located((By.ID, "txtUsername0"))
|
||||
)
|
||||
username_input.clear()
|
||||
username_input.send_keys(self.username)
|
||||
print("[BCBS MA login] Username filled")
|
||||
|
||||
# Password field: id="txtPassword"
|
||||
password_input = self._wait(10).until(
|
||||
EC.presence_of_element_located((By.ID, "txtPassword"))
|
||||
)
|
||||
password_input.clear()
|
||||
password_input.send_keys(self.password)
|
||||
print("[BCBS MA login] Password filled")
|
||||
|
||||
# Log in button: id="ns_Z7_09ME1282N8N3B0QGV9ND6N20G2_loginSubmit"
|
||||
login_btn = self._wait(10).until(
|
||||
EC.element_to_be_clickable((By.ID, "ns_Z7_09ME1282N8N3B0QGV9ND6N20G2_loginSubmit"))
|
||||
)
|
||||
login_btn.click()
|
||||
print("[BCBS MA login] Log in clicked, waiting for response...")
|
||||
time.sleep(3)
|
||||
|
||||
return self._detect_post_login_state()
|
||||
|
||||
except Exception as e:
|
||||
print(f"[BCBS MA login] Error: {e}")
|
||||
return f"ERROR: login failed: {e}"
|
||||
|
||||
def _detect_post_login_state(self) -> str:
|
||||
"""Check current URL/DOM to decide what happened after login."""
|
||||
for _ in range(6):
|
||||
time.sleep(1)
|
||||
url = self.driver.current_url.lower()
|
||||
print(f"[BCBS MA] post-login URL: {url[:80]}")
|
||||
|
||||
if "authsvc" in url or "/mga/sps/" in url:
|
||||
print("[BCBS MA] OTP page detected")
|
||||
return "OTP_REQUIRED"
|
||||
|
||||
if "providerhome" in url or "portal" in url:
|
||||
# Check if we're on a real portal page (not login form)
|
||||
try:
|
||||
self.driver.find_element(By.XPATH,
|
||||
"//*[contains(@href,'eTools') or contains(text(),'eTools') or "
|
||||
"contains(@href,'eligibility') or contains(text(),'Eligibility')]"
|
||||
)
|
||||
print("[BCBS MA] Dashboard detected — logged in without OTP")
|
||||
return "SUCCESS"
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
print("[BCBS MA] Could not determine post-login state")
|
||||
return "OTP_REQUIRED" # default assumption for BCBS MA
|
||||
|
||||
# ── Step: Submit OTP ──────────────────────────────────────────────────────
|
||||
|
||||
def submit_otp_step(self, otp: str) -> str:
|
||||
"""
|
||||
Enter the 6-digit OTP into the BCBS MA verification page and click Submit.
|
||||
OTP input: id="otppswd", name="otp.user.otp"
|
||||
Submit btn: id="submitButton" (starts disabled, enables after OTP typed)
|
||||
Returns "SUCCESS" or "ERROR:..."
|
||||
"""
|
||||
try:
|
||||
print(f"[BCBS MA OTP] Submitting OTP: {otp}")
|
||||
|
||||
# OTP input field: id="otppswd"
|
||||
otp_input = self._wait(15).until(
|
||||
EC.presence_of_element_located((By.ID, "otppswd"))
|
||||
)
|
||||
otp_input.clear()
|
||||
otp_input.send_keys(otp)
|
||||
print("[BCBS MA OTP] OTP entered")
|
||||
|
||||
# Submit button starts disabled — wait for it to become clickable
|
||||
submit_btn = self._wait(10).until(
|
||||
EC.element_to_be_clickable((By.ID, "submitButton"))
|
||||
)
|
||||
submit_btn.click()
|
||||
print("[BCBS MA OTP] Submit clicked, waiting for dashboard...")
|
||||
time.sleep(4)
|
||||
|
||||
# Wait for dashboard
|
||||
for _ in range(15):
|
||||
time.sleep(1)
|
||||
url = self.driver.current_url.lower()
|
||||
print(f"[BCBS MA OTP] URL: {url[:80]}")
|
||||
if "authsvc" not in url and "/mga/sps/" not in url:
|
||||
print("[BCBS MA OTP] Left OTP page — login successful")
|
||||
return "SUCCESS"
|
||||
|
||||
return "ERROR: OTP page still visible after submission"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[BCBS MA OTP] Error: {e}")
|
||||
return f"ERROR: OTP submission failed: {e}"
|
||||
|
||||
# ── Step 1: Navigate to ConnectCenter → New Eligibility Request ──────────
|
||||
|
||||
def step1(self) -> str:
|
||||
"""
|
||||
After OTP login:
|
||||
1. Click eTools menu
|
||||
2. Click ConnectCenter in the dropdown
|
||||
3. Click Go Now on the ConnectCenter launch page
|
||||
4. Click Continue in the popup (opens ConnectCenter in new tab)
|
||||
5. Switch to new tab, click Verification
|
||||
6. Click New Eligibility Request in dropdown
|
||||
Returns "SUCCESS" (on Eligibility Identifier page) or "ERROR:..."
|
||||
"""
|
||||
try:
|
||||
from selenium.webdriver.common.keys import Keys
|
||||
|
||||
# 1. Click eTools
|
||||
print("[BCBS MA step1] Clicking eTools...")
|
||||
etools = self._wait(15).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//span[text()='eTools'] | //a[.//span[text()='eTools']]"
|
||||
))
|
||||
)
|
||||
etools.click()
|
||||
print("[BCBS MA step1] eTools clicked, waiting for dropdown...")
|
||||
time.sleep(2)
|
||||
|
||||
# 2. Click ConnectCenter link in the dropdown
|
||||
print("[BCBS MA step1] Clicking ConnectCenter...")
|
||||
connect_center = self._wait(15).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//a[contains(@href,'connectcenter')]"
|
||||
))
|
||||
)
|
||||
connect_center.click()
|
||||
print("[BCBS MA step1] ConnectCenter clicked, waiting for page to load...")
|
||||
time.sleep(3)
|
||||
|
||||
# 3. Click Go Now on the ConnectCenter page
|
||||
print("[BCBS MA step1] Clicking Go Now...")
|
||||
go_now = self._wait(20).until(
|
||||
EC.element_to_be_clickable((By.ID, "GoNow-2c338de9-6a2f-4d71-b5fb-a6e4ae14be80"))
|
||||
)
|
||||
go_now.click()
|
||||
time.sleep(2)
|
||||
print("[BCBS MA step1] Go Now clicked — popup opened, pressing Enter to continue...")
|
||||
|
||||
# 4. Press Enter to auto-confirm the popup
|
||||
from selenium.webdriver.common.action_chains import ActionChains
|
||||
ActionChains(self.driver).send_keys(Keys.ENTER).perform()
|
||||
time.sleep(4)
|
||||
print("[BCBS MA step1] Enter pressed — continuing to ConnectCenter")
|
||||
|
||||
# 5. Switch to the new tab that ConnectCenter opened
|
||||
self._wait(10).until(lambda d: len(d.window_handles) > 1)
|
||||
self.driver.switch_to.window(self.driver.window_handles[-1])
|
||||
print(f"[BCBS MA step1] Switched to new tab: {self.driver.current_url[:60]}")
|
||||
time.sleep(3)
|
||||
|
||||
# 6. Click Verification menu
|
||||
print("[BCBS MA step1] Clicking Verification...")
|
||||
verification = self._wait(15).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//a[text()='Verification' or normalize-space(text())='Verification']"
|
||||
))
|
||||
)
|
||||
verification.click()
|
||||
time.sleep(2)
|
||||
print("[BCBS MA step1] Verification clicked")
|
||||
|
||||
# 7. Click New Eligibility Request in dropdown
|
||||
print("[BCBS MA step1] Clicking New Eligibility Request...")
|
||||
new_elig = self._wait(10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//a[@ng-click='newEligibility();']"
|
||||
))
|
||||
)
|
||||
new_elig.click()
|
||||
time.sleep(3)
|
||||
print("[BCBS MA step1] New Eligibility Request clicked — on Eligibility Identifier page")
|
||||
|
||||
return "SUCCESS"
|
||||
|
||||
except Exception as e:
|
||||
print(f"[BCBS MA step1] Error: {e}")
|
||||
return f"ERROR: step1 failed: {e}"
|
||||
|
||||
# ── Step 2: Fill Eligibility Identifier form and get results ─────────────
|
||||
|
||||
def step2(self) -> dict:
|
||||
"""
|
||||
On the Eligibility Identifier page:
|
||||
1. Enter provider NPI → click Find Provider
|
||||
2. Select payer search option: Member ID, Subscriber Date Of Birth (value=2)
|
||||
3. Select service type: Dental Care [35] (value=33)
|
||||
4. Select place of service: OFFICE [11] (value=30)
|
||||
5. Enter member ID
|
||||
6. Enter date of birth (MM/DD/YYYY)
|
||||
7. Click Submit → wait for results page → PDF
|
||||
"""
|
||||
from selenium.webdriver.support.ui import Select
|
||||
|
||||
try:
|
||||
print("[BCBS MA step2] Filling Eligibility Identifier form...")
|
||||
|
||||
# 1. Enter provider NPI
|
||||
provider_id_input = self._wait(15).until(
|
||||
EC.presence_of_element_located((By.ID, "providerID"))
|
||||
)
|
||||
provider_id_input.clear()
|
||||
provider_id_input.send_keys(self.provider_npi)
|
||||
print(f"[BCBS MA step2] Provider NPI entered: {self.provider_npi}")
|
||||
|
||||
# Click Find Provider
|
||||
find_provider = self._wait(10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//button[@ng-click='searchProviders()']"
|
||||
))
|
||||
)
|
||||
find_provider.click()
|
||||
print("[BCBS MA step2] Find Provider clicked, waiting...")
|
||||
time.sleep(3)
|
||||
|
||||
# 2. Payer search options → value="2": Member ID, Subscriber Date Of Birth
|
||||
payer_select = self._wait(10).until(
|
||||
EC.presence_of_element_located((By.ID, "payerSearchOptions"))
|
||||
)
|
||||
Select(payer_select).select_by_value("2")
|
||||
print("[BCBS MA step2] Payer search option selected: Member ID + DOB")
|
||||
time.sleep(1)
|
||||
|
||||
# 3. Service type → value="33": Dental Care [35]
|
||||
service_select = self._wait(10).until(
|
||||
EC.presence_of_element_located((By.ID, "serviceType"))
|
||||
)
|
||||
Select(service_select).select_by_value("33")
|
||||
print("[BCBS MA step2] Service type selected: Dental Care [35]")
|
||||
|
||||
# 4. Place of service → value="30": OFFICE [11]
|
||||
pos_select = self._wait(10).until(
|
||||
EC.presence_of_element_located((By.ID, "placeOfService"))
|
||||
)
|
||||
Select(pos_select).select_by_value("30")
|
||||
print("[BCBS MA step2] Place of service selected: OFFICE [11]")
|
||||
|
||||
# 5. Member ID — field name is subscriberMedicaidID
|
||||
member_input = self._wait(10).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//input[@name='subscriberMedicaidID' or @id='subscriberMedicaidID']"
|
||||
))
|
||||
)
|
||||
member_input.clear()
|
||||
member_input.send_keys(self.member_id)
|
||||
print(f"[BCBS MA step2] Member ID entered: {self.member_id}")
|
||||
|
||||
# 6. Date of birth — id="subscriberDateOfBirth", placeholder="mm/dd/yyyy"
|
||||
from selenium.webdriver.common.action_chains import ActionChains
|
||||
from selenium.webdriver.common.keys import Keys
|
||||
|
||||
dob_formatted = self._dob_mmddyyyy()
|
||||
dob_input = self._wait(10).until(
|
||||
EC.presence_of_element_located((By.ID, "subscriberDateOfBirth"))
|
||||
)
|
||||
# Double-click to focus, then type directly via ActionChains
|
||||
ActionChains(self.driver).double_click(dob_input).perform()
|
||||
time.sleep(0.3)
|
||||
ActionChains(self.driver).send_keys(dob_formatted).perform()
|
||||
print(f"[BCBS MA step2] DOB typed: {dob_formatted}")
|
||||
time.sleep(1)
|
||||
|
||||
# 7. Click Submit
|
||||
submit_btn = self._wait(10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//button[@ng-click='submit()' and @type='submit']"
|
||||
))
|
||||
)
|
||||
submit_btn.click()
|
||||
print("[BCBS MA step2] Submit clicked, waiting for results...")
|
||||
time.sleep(5)
|
||||
|
||||
# Wait for results page to load
|
||||
self._wait(20).until(
|
||||
EC.presence_of_element_located((By.XPATH,
|
||||
"//*[contains(text(),'Eligible') or contains(text(),'Not Eligible') or "
|
||||
"contains(text(),'Active') or contains(text(),'Inactive') or "
|
||||
"contains(text(),'Coverage') or contains(text(),'Benefit')]"
|
||||
))
|
||||
)
|
||||
print("[BCBS MA step2] Results page loaded")
|
||||
|
||||
# Click Expand All to expand all eligibility sections before PDFing
|
||||
try:
|
||||
expand_all = self._wait(10).until(
|
||||
EC.element_to_be_clickable((By.XPATH,
|
||||
"//h6[normalize-space(text())='Expand All']"
|
||||
))
|
||||
)
|
||||
expand_all.click()
|
||||
print("[BCBS MA step2] Expand All clicked, waiting for sections to expand...")
|
||||
time.sleep(3)
|
||||
except Exception as e:
|
||||
print(f"[BCBS MA step2] Expand All not found or failed: {e}")
|
||||
|
||||
# Extract eligibility status
|
||||
page_text = self.driver.find_element(By.TAG_NAME, "body").text
|
||||
text_lower = page_text.lower()
|
||||
if "not eligible" in text_lower or "inactive" in text_lower or "terminated" in text_lower:
|
||||
eligibility = "Not Eligible"
|
||||
elif "eligible" in text_lower or "active" in text_lower or "covered" in text_lower:
|
||||
eligibility = "Eligible"
|
||||
else:
|
||||
eligibility = "Unknown"
|
||||
|
||||
patient_name = f"{self.first_name} {self.last_name}".strip()
|
||||
print(f"[BCBS MA step2] Eligibility: {eligibility}")
|
||||
|
||||
# PDF the results page via CDP
|
||||
pdf_base64 = ""
|
||||
pdf_path = None
|
||||
try:
|
||||
result = self.driver.execute_cdp_cmd("Page.printToPDF", {
|
||||
"printBackground": True,
|
||||
"paperWidth": 8.5,
|
||||
"paperHeight": 11,
|
||||
"marginTop": 0.5,
|
||||
"marginBottom": 0.5,
|
||||
"marginLeft": 0.5,
|
||||
"marginRight": 0.5,
|
||||
})
|
||||
pdf_data = result.get("data", "")
|
||||
if pdf_data:
|
||||
os.makedirs(DOWNLOAD_DIR, exist_ok=True)
|
||||
filename = f"bcbs_ma_eligibility_{self.member_id}_{int(time.time())}.pdf"
|
||||
pdf_path = os.path.join(DOWNLOAD_DIR, filename)
|
||||
with open(pdf_path, "wb") as f:
|
||||
f.write(base64.b64decode(pdf_data))
|
||||
pdf_base64 = pdf_data
|
||||
print(f"[BCBS MA step2] PDF saved: {pdf_path}")
|
||||
except Exception as e:
|
||||
print(f"[BCBS MA step2] PDF generation failed: {e}")
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"eligibility": eligibility,
|
||||
"patientName": patient_name,
|
||||
"memberId": self.member_id,
|
||||
"insurerName": "BCBS MA",
|
||||
"pdfBase64": pdf_base64,
|
||||
"pdf_path": pdf_path,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[BCBS MA step2] Error: {e}")
|
||||
return {"status": "error", "message": f"step2 failed: {e}"}
|
||||
Reference in New Issue
Block a user