87d7ce9ed9
- Extract patient first/last name from Patient Information DOM section (scoped to avoid duplicate Subscriber Information column values) - Switch to latest tab at start of step2 (Eligibility Identifier opens in new tab) - DOB: double-click + ActionChains.send_keys (no pyperclip, avoids Chrome crash) - BCBS MA button changed to variant="default" to match nearby buttons - Backend processor uses extracted names from selenium result Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
509 lines
22 KiB
Python
509 lines
22 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 = 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
|
||
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}")
|
||
|
||
# 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"
|
||
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 DOM — scope to "Patient Information" column only
|
||
# to avoid picking up the duplicate values in the Subscriber Information column.
|
||
first_name = self.first_name
|
||
last_name = self.last_name
|
||
try:
|
||
# Find the "Patient Information" header, then search within its parent container
|
||
patient_section = self.driver.find_element(By.XPATH,
|
||
"//*[normalize-space(text())='Patient Information']/ancestor::*[3]"
|
||
)
|
||
section_text = patient_section.text
|
||
print(f"[BCBS MA step2] Patient section text:\n{section_text[:300]}")
|
||
|
||
lines = section_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].split("Member ID:")[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
|
||
|
||
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}"}
|