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 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}"}