feat: BCBS MA persistent session, Expand All fix, address/gender extraction

- Persistent Chrome profile (bcbs_ma_browser_manager) reused across patients;
  login probes /myportal/ to skip login + OTP while the session is active
- Expand All: JS-click the expandCollapseViews span, verify sections opened,
  and unclip fixed-height result tables so the PDF includes every row
- Extract patient address and gender from the results page and save to patient
- Uppercase member IDs (XXA...), matching older lowercase records
- Add /clear-bcbs-ma-session endpoint

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-29 18:27:46 -04:00
co-authored by Claude Opus 5.5
parent 011869f641
commit 13be6e97d5
6 changed files with 474 additions and 89 deletions
@@ -7,54 +7,30 @@ 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
from bcbs_ma_browser_manager import get_browser_manager
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
DOWNLOAD_DIR = get_browser_manager().download_dir
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).
Uses the persistent BCBS MA browser (bcbs_ma_browser_manager) — one Chrome
with a saved profile, reused across patients. While the portal login is
still active, login() returns SUCCESS and no OTP is needed.
Flow: login(url) → OTP_REQUIRED → [caller polls for OTP] → submit_otp_step(otp)
Flow: login(url) → OTP_REQUIRED | SUCCESS → [caller polls for OTP] → submit_otp_step(otp)
→ step1() [search member] → step2() [extract + PDF]
"""
# Public portal — always shows the login form, even when already logged in
LOGIN_URL = "https://provider.bluecrossma.com/ProviderHome/portal/"
# Logged-in portal — only loads with an active session; otherwise BCBS
# redirects away from /myportal/
MYPORTAL_URL = "https://provider.bluecrossma.com/ProviderHome/myportal/"
def __init__(self, data: dict):
raw = data.get("data", data)
@@ -70,21 +46,44 @@ class AutomationBCBSMAEligibilityCheck:
# ── Driver lifecycle ──────────────────────────────────────────────────────
def config_driver(self):
self.driver = _fresh_driver()
# Persistent browser from manager (keeps the BCBS MA login between patients)
headless = os.getenv("SELENIUM_HEADLESS", "false").lower() == "true"
self.driver = get_browser_manager().get_driver(headless)
self._close_extra_tabs()
def close_driver(self):
def _close_extra_tabs(self):
"""Close ConnectCenter tabs left over from the previous patient, keep the first tab."""
try:
if self.driver:
self.driver.quit()
except Exception:
pass
self.driver = None
handles = self.driver.window_handles
for handle in handles[1:]:
self.driver.switch_to.window(handle)
self.driver.close()
self.driver.switch_to.window(handles[0])
if len(handles) > 1:
print(f"[BCBS MA] Closed {len(handles) - 1} leftover tab(s)")
except Exception as e:
print(f"[BCBS MA] Could not close leftover tabs: {e}")
def _force_logout(self):
"""Clear all BCBS MA / ConnectCenter cookies when the saved username changes."""
try:
print("[BCBS MA login] Forcing logout due to credential change...")
self.driver.execute_cdp_cmd("Network.clearBrowserCookies", {})
get_browser_manager().clear_credentials_hash()
print("[BCBS MA login] Cookies cleared")
except Exception as e:
print(f"[BCBS MA login] Error during forced logout: {e}")
# ── Helpers ───────────────────────────────────────────────────────────────
def _wait(self, timeout=15):
return WebDriverWait(self.driver, timeout)
def _is_logged_in_page(self) -> bool:
"""On the logged-in portal (/myportal/) with no login form showing."""
return ("/myportal/" in self.driver.current_url.lower()
and not self.driver.find_elements(By.ID, "txtUsername0"))
def _dob_mmddyyyy(self) -> str:
"""Convert YYYY-MM-DD → MM/DD/YYYY."""
try:
@@ -97,18 +96,38 @@ class AutomationBCBSMAEligibilityCheck:
def login(self, url: str = "") -> str:
"""
Navigate to BCBS MA Provider Central, fill credentials, click Log in.
Check for an active session on the logged-in portal (/myportal/); if
there is one, return SUCCESS straight away. Otherwise open the public
portal, fill credentials and click Log in.
Returns:
"OTP_REQUIRED" – redirected to MFA challenge page (/mga/sps/authsvc)
"SUCCESS" – landed on dashboard without OTP
"SUCCESS" – on the logged-in portal (existing session, or login without OTP)
"ERROR:..." – something went wrong
"""
target = url or self.LOGIN_URL
try:
print(f"[BCBS MA login] Navigating to {target}")
if self.username and get_browser_manager().credentials_changed(self.username):
self._force_logout()
# 1. Existing session? The public portal shows the login form even when
# logged in, so probe /myportal/, which only loads with a live session.
print(f"[BCBS MA login] Checking for active session at {self.MYPORTAL_URL}")
self.driver.get(self.MYPORTAL_URL)
self._wait(20).until(lambda d: d.execute_script("return document.readyState") == "complete")
time.sleep(2) # let any session-expired redirect finish
print(f"[BCBS MA login] Landed on {self.driver.current_url[:80]}")
if self._is_logged_in_page():
print("[BCBS MA login] Session still active — skipping login and OTP")
return "SUCCESS"
if "authsvc" in self.driver.current_url.lower():
print("[BCBS MA login] Landed on OTP page")
return "OTP_REQUIRED"
# 2. Not logged in — log in from the public portal
print(f"[BCBS MA login] Not logged in — 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"))
)
@@ -149,17 +168,10 @@ class AutomationBCBSMAEligibilityCheck:
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
# eTools shows on the public page too — only /myportal/ proves login
if self._is_logged_in_page():
print("[BCBS MA] Logged in without OTP")
return "SUCCESS"
print("[BCBS MA] Could not determine post-login state")
return "OTP_REQUIRED" # default assumption for BCBS MA
@@ -211,7 +223,7 @@ class AutomationBCBSMAEligibilityCheck:
def step1(self) -> str:
"""
After OTP login:
After login, on the Provider Central dashboard:
1. Click eTools menu
2. Click ConnectCenter in the dropdown
3. Click Go Now on the ConnectCenter launch page
@@ -311,6 +323,90 @@ class AutomationBCBSMAEligibilityCheck:
# ── Step 2: Fill Eligibility Identifier form and get results ─────────────
def _expand_all_sections(self):
"""
Click "Expand All" on the results page and confirm the sections opened.
The label may be nested inside an h6/a/span, so match the innermost
element whose visible text is "Expand All", then fall back to a JS
click on it and its clickable ancestors until the page height grows.
"""
label_xpath = (
"//*[normalize-space(.)='Expand All' and "
"not(*[normalize-space(.)='Expand All'])]"
)
try:
label = self._wait(15).until(
EC.presence_of_element_located((By.XPATH, label_xpath))
)
except TimeoutException:
print("[BCBS MA step2] Expand All not found — PDF will have collapsed sections")
return
# Each benefit section (Copay, Coinsurance, ...) is an ng-grid table that
# only becomes visible once expanded — count them to verify the click
def visible_grids():
return self.driver.execute_script(
"return [...document.querySelectorAll('.ngViewport')]"
".filter(v => v.offsetParent && v.clientHeight > 0).length;"
)
# Let the benefit sections finish rendering before expanding them
time.sleep(2)
before = visible_grids()
# Markup: <span ng-click="expandCollapseViews(false)"><h6>Expand All</h6></span>.
# JS-click the span (an overlapping View Options panel intercepts native
# clicks); fall back to the label itself. Only re-click while nothing has
# expanded, so a click can never collapse sections that already opened.
for attempt in range(1, 4):
try:
targets = self.driver.find_elements(By.XPATH,
label_xpath + "/ancestor::*[starts-with(@ng-click,'expandCollapseViews')][1]"
) or [label]
self.driver.execute_script("arguments[0].click();", targets[0])
except Exception as e:
print(f"[BCBS MA step2] Expand All click attempt {attempt} failed: {e}")
for _ in range(10):
time.sleep(0.5)
if visible_grids() > before:
time.sleep(1.5) # let the rows inside each section render
print(f"[BCBS MA step2] Expand All worked (attempt {attempt}, "
f"visible sections {before} → {visible_grids()})")
self._unclip_grids()
return
print(f"[BCBS MA step2] Expand All attempt {attempt} did not expand sections "
f"(visible sections still {visible_grids()})")
print("[BCBS MA step2] Expand All failed — PDF will have collapsed sections")
def _unclip_grids(self):
"""
Each expanded section's table is a fixed-height (368px) scroll box, so a
PDF only shows its first rows. All rows are already in the DOM — grow each
box (and its ng-grid wrapper) to full height so the PDF includes them.
"""
try:
count = self.driver.execute_script("""
let n = 0;
document.querySelectorAll('.ngViewport').forEach(v => {
if (!v.offsetParent || v.scrollHeight <= v.clientHeight + 2) return;
const grow = v.scrollHeight - v.clientHeight;
v.style.height = v.scrollHeight + 'px';
v.style.overflow = 'visible';
const grid = v.closest('.ngGrid');
if (grid) {
grid.style.height = (grid.offsetHeight + grow) + 'px';
grid.style.overflow = 'visible';
}
n++;
});
return n;
""")
print(f"[BCBS MA step2] Unclipped {count} scrolling table(s) for the PDF")
except Exception as e:
print(f"[BCBS MA step2] Could not unclip tables: {e}")
def step2(self) -> dict:
"""
On the Eligibility Identifier page:
@@ -430,17 +526,7 @@ class AutomationBCBSMAEligibilityCheck:
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}")
self._expand_all_sections()
# Extract eligibility status and patient name from live page text
page_text = self.driver.find_element(By.TAG_NAME, "body").text
@@ -458,6 +544,7 @@ class AutomationBCBSMAEligibilityCheck:
import re
first_name = self.first_name
last_name = self.last_name
street = city = state = zip_code = gender = ""
try:
# Slice out just the Patient Information section
patient_match = re.search(
@@ -490,6 +577,27 @@ class AutomationBCBSMAEligibilityCheck:
capturing_last = False
if first_name and last_name and not capturing_last:
break
# Address — "Street: 1041 MADISON PLACE" / "City State Zip: SOUTHBOROUGH MA 01772"
street_match = re.search(r"Street:[ \t]*([^\n]*)", patient_text)
if street_match:
street = street_match.group(1).strip()
csz_match = re.search(
r"City State Zip:\s*(.+?)\s+([A-Z]{2})\s+(\d{5}(?:-?\d{4})?)",
patient_text
)
if csz_match:
city = csz_match.group(1).strip()
state = csz_match.group(2)
zip_code = csz_match.group(3)
print(f"[BCBS MA step2] Extracted address — Street: '{street}', "
f"City: '{city}', State: '{state}', Zip: '{zip_code}'")
# Gender — "Gender: M"
gender_match = re.search(r"Gender:[ \t]*([^\n]*)", patient_text)
if gender_match:
gender = gender_match.group(1).strip()
print(f"[BCBS MA step2] Extracted gender: '{gender}'")
else:
print("[BCBS MA step2] Patient section not found in page text")
@@ -533,6 +641,11 @@ class AutomationBCBSMAEligibilityCheck:
"firstName": first_name,
"lastName": last_name,
"memberId": self.member_id,
"address": street,
"city": city,
"state": state,
"zipCode": zip_code,
"gender": gender,
"insurerName": "BCBS MA",
"pdfBase64": pdf_base64,
"pdf_path": pdf_path,