57c18b404f
Use appear-then-disappear pattern: briefly wait for spinner to show up (so we don't pass through before it starts), then wait for it to clear. Prevents element click interception on Find Provider button in step2. Also restores element_to_be_clickable for New Eligibility Request (step1). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
543 lines
23 KiB
Python
543 lines
23 KiB
Python
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_XPATH = (
|
||
"//a[contains(@ng-click,'newEligibility')] | "
|
||
"//a[contains(normalize-space(text()),'New Eligibility Request')] | "
|
||
"//a[contains(normalize-space(text()),'New Eligibility')]"
|
||
)
|
||
try:
|
||
new_elig = self._wait(8).until(
|
||
EC.element_to_be_clickable((By.XPATH, NEW_ELIG_XPATH))
|
||
)
|
||
except TimeoutException:
|
||
# Dropdown may have closed — re-open Verification and try again
|
||
print("[BCBS MA step1] New Eligibility Request not found, re-clicking Verification...")
|
||
verification.click()
|
||
time.sleep(2)
|
||
new_elig = self._wait(8).until(
|
||
EC.element_to_be_clickable((By.XPATH, NEW_ELIG_XPATH))
|
||
)
|
||
try:
|
||
new_elig.click()
|
||
except Exception:
|
||
self.driver.execute_script("arguments[0].click();", new_elig)
|
||
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
|
||
from selenium.webdriver.common.action_chains import ActionChains
|
||
from selenium.webdriver.common.keys import Keys
|
||
|
||
try:
|
||
# New Eligibility Request opens in a new tab — switch to the latest one
|
||
self._wait(10).until(lambda d: len(d.window_handles) >= 2)
|
||
self.driver.switch_to.window(self.driver.window_handles[-1])
|
||
print(f"[BCBS MA step2] Switched to tab ({len(self.driver.window_handles)} open): {self.driver.current_url[:70]}")
|
||
time.sleep(2)
|
||
|
||
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}")
|
||
|
||
# Wait for spinner cycle to complete, then click Find Provider
|
||
try:
|
||
self._wait(2).until(
|
||
EC.visibility_of_element_located((By.CSS_SELECTOR, "div.ccProgressBar"))
|
||
)
|
||
except TimeoutException:
|
||
pass # spinner never appeared — page loaded instantly
|
||
self._wait(10).until(
|
||
EC.invisibility_of_element_located((By.CSS_SELECTOR, "div.ccProgressBar"))
|
||
)
|
||
find_provider = self._wait(10).until(
|
||
EC.element_to_be_clickable((By.XPATH,
|
||
"//button[@ng-click='searchProviders()']"
|
||
))
|
||
)
|
||
try:
|
||
find_provider.click()
|
||
except Exception:
|
||
self.driver.execute_script("arguments[0].click();", find_provider)
|
||
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"
|
||
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 and patient name from live page text
|
||
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"
|
||
|
||
# Extract first/last name from Patient Information column only.
|
||
# "Relationship:" is unique to the Patient column (not in Subscriber column).
|
||
# Use it as anchor: grab text from "Relationship:" up to "Member ID:" (Subscriber starts there).
|
||
import re
|
||
first_name = self.first_name
|
||
last_name = self.last_name
|
||
try:
|
||
# Slice out just the Patient Information section
|
||
patient_match = re.search(
|
||
r"Relationship:(.+?)(?=Member ID:|Subscriber Information|Plan Name:)",
|
||
page_text,
|
||
re.DOTALL
|
||
)
|
||
if patient_match:
|
||
patient_text = patient_match.group(1)
|
||
print(f"[BCBS MA step2] Patient section:\n{patient_text[:200]}")
|
||
|
||
lines = patient_text.split("\n")
|
||
capturing_last = False
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if "First Name:" in stripped and not first_name:
|
||
val = stripped.split("First Name:", 1)[1].strip()
|
||
val = val.split("Middle Name:")[0].split("Last Name:")[0].strip()
|
||
if val:
|
||
first_name = val
|
||
elif "Last Name:" in stripped and not last_name:
|
||
val = stripped.split("Last Name:", 1)[1].strip()
|
||
val = val.split("SSN:")[0].split("Date of Birth:")[0].strip()
|
||
if val:
|
||
last_name = val
|
||
capturing_last = True
|
||
elif capturing_last:
|
||
if stripped and ":" not in stripped and stripped == stripped.upper():
|
||
last_name += " " + stripped
|
||
capturing_last = False
|
||
if first_name and last_name and not capturing_last:
|
||
break
|
||
else:
|
||
print("[BCBS MA step2] Patient section not found in page text")
|
||
|
||
print(f"[BCBS MA step2] Extracted — First: '{first_name}', Last: '{last_name}'")
|
||
except Exception as e:
|
||
print(f"[BCBS MA step2] Name extraction failed: {e}")
|
||
|
||
# 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}")
|
||
|
||
|
||
patient_name = f"{first_name} {last_name}".strip()
|
||
print(f"[BCBS MA step2] Eligibility: {eligibility}, Patient: {patient_name}")
|
||
|
||
return {
|
||
"status": "success",
|
||
"eligibility": eligibility,
|
||
"patientName": patient_name,
|
||
"firstName": first_name,
|
||
"lastName": last_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}"}
|