from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from webdriver_manager.chrome import ChromeDriverManager
import time
import os
import re
import base64
from datetime import date, datetime
# Case-insensitive XPath text match helper
_LOWER = "translate({0}, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz')"
def _ci_contains(expr, needle):
return f"contains({_LOWER.format(expr)}, '{needle.lower()}')"
class AutomationAltusEligibilityCheck:
"""
Altus Dental provider portal eligibility check.
Plain username/password login (no OTP) — fresh browser per run, like MH.
"""
def __init__(self, data):
self.headless = False
self.driver = None
self.extracted_data = {}
self.data = data.get("data")
self.altus_username = self.data.get("altusUsername", "")
self.altus_password = self.data.get("altusPassword", "")
self.dateOfBirth = self.data.get("dateOfBirth", "")
self.memberId = self.data.get("memberId", "")
self.firstName = self.data.get("firstName", "") or ""
self.lastName = self.data.get("lastName", "") or ""
# Convert dateOfBirth from YYYY-MM-DD to MM/DD/YYYY
if self.dateOfBirth and "-" in self.dateOfBirth:
parts = self.dateOfBirth.split("-")
if len(parts) == 3:
year, month, day = parts
self.dateOfBirth = f"{month.zfill(2)}/{day.zfill(2)}/{year}"
self.download_dir = os.path.abspath("downloads")
os.makedirs(self.download_dir, exist_ok=True)
# ── driver ──────────────────────────────────────────────────────────────────
def config_driver(self):
options = webdriver.ChromeOptions()
if self.headless:
options.add_argument("--headless")
prefs = {
"download.default_directory": self.download_dir,
"plugins.always_open_pdf_externally": False,
"download.prompt_for_download": False,
"download.directory_upgrade": True,
}
options.add_experimental_option("prefs", prefs)
s = Service(ChromeDriverManager().install())
self.driver = webdriver.Chrome(service=s, options=options)
# ── debug ───────────────────────────────────────────────────────────────────
def _dump_debug(self, tag):
"""Save page HTML + screenshot so selectors can be tuned after a failure."""
try:
base = os.path.join(self.download_dir, f"altus_debug_{tag}")
with open(base + ".html", "w", encoding="utf-8") as f:
f.write(self.driver.page_source)
self.driver.save_screenshot(base + ".png")
print(f"[debug] saved {base}.html / .png URL={self.driver.current_url}")
except Exception as e:
print(f"[debug] dump failed: {e}")
def _visible(self, xpath):
for el in self.driver.find_elements(By.XPATH, xpath):
try:
if el.is_displayed():
return el
except Exception:
continue
return None
def _type(self, el, value):
el.click()
el.send_keys(Keys.CONTROL, "a")
el.send_keys(Keys.DELETE)
el.send_keys(value)
# ── login ───────────────────────────────────────────────────────────────────
def login(self):
wait = WebDriverWait(self.driver, 30)
if not self.altus_username or not self.altus_password:
return "ERROR: Altus username/password missing — update them in Settings (ALTUS)."
try:
username_field = wait.until(EC.element_to_be_clickable((By.ID, "Username")))
password_field = wait.until(EC.element_to_be_clickable((By.ID, "Password")))
# Fields use their value as a placeholder ("Username"/"Password"), swapped
# by onfocus/onblur, and jQuery validate rejects empty or placeholder
# values. Type into each field, then verify; fall back to setting the
# value via JS if the onfocus/onblur handlers interfered.
self._type(username_field, self.altus_username)
self._type(password_field, self.altus_password)
for el_id, expected in (("Username", self.altus_username), ("Password", self.altus_password)):
actual = self.driver.execute_script(f"return document.getElementById('{el_id}').value;")
if actual != expected:
print(f"[login] {el_id} value mismatch after typing — setting via JS")
self.driver.execute_script(
"var el = document.getElementById(arguments[0]); el.value = arguments[1];"
"el.dispatchEvent(new Event('input', {bubbles: true}));"
"el.dispatchEvent(new Event('change', {bubbles: true}));",
el_id, expected,
)
# Click "Log In" ()
try:
login_link = self.driver.find_element(
By.XPATH, "//form[@id='LoginForm']//a[contains(@class,'smallButton') and normalize-space()='Log In']"
)
self.driver.execute_script("arguments[0].click();", login_link)
except Exception:
self.driver.execute_script("SubmitForm();")
time.sleep(1)
if "/provider/login" in self.driver.current_url.lower():
invalid = self.driver.find_elements(By.CSS_SELECTOR, "#LoginForm input.error[aria-invalid='true']")
if invalid:
ids = [el.get_attribute("id") for el in invalid]
self._dump_debug("login_validation")
return f"ERROR: Altus login form rejected {', '.join(ids)} (empty or placeholder value)."
# Wait to leave the login page, or for a visible validation error
deadline = time.time() + 30
while time.time() < deadline:
time.sleep(0.5)
url = self.driver.current_url.lower()
if "/provider/login" not in url and "/authorization/login" not in url:
break
err = self._visible("//div[contains(@class,'formValidationError')]")
if err and err.text.strip():
print(f"[login] portal error: {err.text.strip()}")
return f"ERROR: Altus login failed — {err.text.strip()}"
else:
self._dump_debug("login")
return "ERROR: Altus login timed out — check credentials or portal availability."
print(f"[login] logged in, URL: {self.driver.current_url}")
self._accept_agreement_if_shown()
return "Success"
except Exception as e:
print(f"[login] Error: {e}")
self._dump_debug("login")
return "ERROR:LOGIN FAILED"
def _accept_agreement_if_shown(self):
"""The login form has a ShowOnlineAccessAgreement flag — accept it if it appears."""
time.sleep(2)
btn = self._visible(
"//*[self::a or self::button or self::input[@type='submit' or @type='button']]"
f"[{_ci_contains('normalize-space(.)', 'accept')} or {_ci_contains('@value', 'accept')}"
f" or {_ci_contains('normalize-space(.)', 'i agree')} or {_ci_contains('@value', 'agree')}]"
)
if btn:
print("[login] accepting online access agreement")
self.driver.execute_script("arguments[0].click();", btn)
time.sleep(2)
# ── step 1 — open eligibility search, search member ─────────────────────────
def _fill_verified(self, el, value, digits_fallback=False):
"""Type into a field and confirm it took; retry digits-only (masked inputs) then JS."""
self._type(el, value)
if (el.get_attribute("value") or "") == value:
return
if digits_fallback:
digits = "".join(c for c in value if c.isdigit())
self._type(el, digits)
if (el.get_attribute("value") or "") in (value, digits):
return
print(f"[step1] {el.get_attribute('id')} value mismatch — setting via JS")
self.driver.execute_script(
"arguments[0].value = arguments[1];"
"arguments[0].dispatchEvent(new Event('change', {bubbles: true}));",
el, value,
)
def step1(self):
wait = WebDriverWait(self.driver, 30)
substep = "init"
try:
# Left menu on Dentist Home: "Benefits & Claims Information" → /Provider/SearchBenefits
substep = "benefits_claims_link"
try:
link = WebDriverWait(self.driver, 15).until(
EC.element_to_be_clickable((By.CSS_SELECTOR, "a[href*='/Provider/SearchBenefits']"))
)
print(f"[step1] clicking '{link.text.strip()}'")
self.driver.execute_script("arguments[0].click();", link)
except Exception:
print("[step1] Benefits & Claims link not found — navigating directly")
self.driver.get("https://www.altusdental.com/Provider/SearchBenefits")
wait.until(lambda d: "/provider/searchbenefits" in d.current_url.lower())
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
time.sleep(2)
print(f"[step1] URL: {self.driver.current_url}")
# Patient Lookup: Subscriber ID + Patient First Name + Patient D.O.B.
# Without a Subscriber ID, "Click Here" switches the first box to Subscriber Last Name.
if self.memberId:
substep = "subscriber_id_input"
subscriber_input = wait.until(EC.element_to_be_clickable((By.ID, "SubscriberId")))
self._fill_verified(subscriber_input, self.memberId)
else:
substep = "alternate_search"
alt_btn = wait.until(EC.element_to_be_clickable((By.ID, "alternateSearch")))
alt_btn.click() # onclick="return ShowAlternateSearch();"
substep = "last_name_input"
last_input = wait.until(EC.element_to_be_clickable((By.ID, "LastName")))
self._fill_verified(last_input, self.lastName)
substep = "first_name_input"
self._fill_verified(self.driver.find_element(By.ID, "FirstName"), self.firstName)
substep = "dob_input"
dob_input = self.driver.find_element(By.ID, "DateOfBirth")
self._fill_verified(dob_input, self.dateOfBirth, digits_fallback=True)
substep = "submit_button"
submit_btn = self.driver.find_element(By.ID, "submitPatientLookup")
# Native click so the button's onclick="return ProcessPatientSearch();" runs
submit_btn.click()
# ProcessPatientSearch() may reject input with a JS alert
try:
alert = WebDriverWait(self.driver, 2).until(EC.alert_is_present())
alert_text = alert.text
alert.accept()
print(f"[step1] portal alert: {alert_text}")
return f"ERROR: Altus search rejected — {alert_text}"
except Exception:
pass
# Results: patient list table (holds up to 20 patients from this session's searches)
substep = "wait_results"
row = WebDriverWait(self.driver, 30).until(lambda d: self._find_result_row() or self._not_found())
if row is True:
self.extracted_data = {"eligibility": None, "notFound": True}
return "Success"
self.extracted_data = self._extract_row_data(row)
print(f"[step1] row data: {self.extracted_data}")
# View Details →
substep = "view_details"
details_link = row.find_element(By.XPATH, ".//a[contains(@href,'ViewPatientBenefits')]")
list_url = self.driver.current_url
handles_before = set(self.driver.window_handles)
self.driver.execute_script("arguments[0].click();", details_link)
try:
WebDriverWait(self.driver, 30).until(
lambda d: len(d.window_handles) > len(handles_before)
or d.current_url != list_url
or not self._is_element_attached(details_link)
)
except Exception:
# Details may render in place without a navigation — print whatever is shown
print("[step1] no navigation detected after View Details — continuing")
self._dump_debug("view_details")
new_handles = [h for h in self.driver.window_handles if h not in handles_before]
if new_handles:
self.driver.switch_to.window(new_handles[0])
WebDriverWait(self.driver, 30).until(
lambda d: d.execute_script("return document.readyState") == "complete"
)
time.sleep(2)
print(f"[step1] benefit details URL: {self.driver.current_url}")
# The results row only has First Name — read the last name from the details page
last_name = self._extract_last_name()
if last_name:
self.extracted_data["lastName"] = last_name
# Fill in plan name from the details page if it shows one
if not self.extracted_data.get("group"):
self.extracted_data["group"] = self._value_after_label(["Group Number", "Group #", "Group"])
plan_name = self._value_after_label(["Plan Name", "Group Name"])
if plan_name:
self.extracted_data["insurance"] = f"Altus Dental {plan_name}"
return "Success"
except Exception as e:
print(f"[step1] FAILED at substep='{substep}': {e}")
print(f"[step1] URL at failure: {self.driver.current_url}")
self._dump_debug(f"step1_{substep}")
return f"ERROR:STEP1:{substep}"
# ── helpers ─────────────────────────────────────────────────────────────────
_NOT_FOUND = ["no records found", "no member found", "no patients found", "not found",
"no results", "unable to locate", "does not match", "could not be found"]
def _page_text(self):
try:
return (self.driver.execute_script("return document.body.innerText;") or "").lower()
except Exception:
return ""
def _not_found(self):
text = self._page_text()
return any(p in text for p in self._NOT_FOUND)
def _is_element_attached(self, el):
try:
el.is_enabled()
return True
except Exception:
return False
def _normalize_id(self, s):
# Subscriber IDs are shown zero-padded (0001249957) — compare without leading zeros
return "".join(c for c in str(s) if c.isalnum()).lower().lstrip("0")
_DATE_RE = re.compile(r"^\d{2}/\d{2}/\d{4}$")
def _row_cells(self, row):
return [(td.text or "").strip() for td in row.find_elements(By.TAG_NAME, "td")]
def _column_value(self, row, header):
"""Cell under the results-table column titled `header` (e.g. "Group")."""
try:
table = row.find_element(By.XPATH, "./ancestor::table[1]")
headers = table.find_elements(By.XPATH, ".//th")
if not headers:
headers = table.find_elements(By.XPATH, ".//tr[1]/td")
names = [(h.text or "").strip().lower() for h in headers]
if header.lower() not in names:
return ""
cells = self._row_cells(row)
if len(cells) != len(names):
return "" # column layout doesn't line up — let the caller fall back
return cells[names.index(header.lower())]
except Exception:
return ""
def _find_result_row(self):
"""Row whose Subscriber ID (if searched), First Name and D.O.B. match this patient."""
rows = self.driver.find_elements(
By.XPATH, "//tr[.//a[contains(@href,'ViewPatientBenefits')]]"
)
first = self.firstName.strip().lower()
for row in rows:
try:
cells = self._row_cells(row)
except Exception:
continue
lowered = [c.lower() for c in cells]
if self.dateOfBirth not in cells or first not in lowered:
continue
if self.memberId:
href = row.find_element(By.XPATH, ".//a[contains(@href,'ViewPatientBenefits')]").get_attribute("href") or ""
ids = [self._normalize_id(c) for c in cells] + [self._normalize_id(x) for x in re.findall(r"'(\w+)'", href)]
if self._normalize_id(self.memberId) not in ids:
continue
return row
return None
def _parse_date(self, s):
try:
return datetime.strptime(s.strip(), "%m/%d/%Y").date()
except Exception:
return None
def _extract_row_data(self, row):
"""Row columns: Subscriber ID | First Name | Group | D.O.B. | Coverage | View Details."""
cells = self._row_cells(row)
coverage = next((c for c in cells if " - " in c), "")
subscriber_id = next((c for c in cells if re.fullmatch(r"\d{6,}", c)), "")
# Group number (saved to the patient's Group Number — used on ADA claim box 16)
group = self._column_value(row, "Group") or next(
(c for c in cells if re.fullmatch(r"\d+-\d+", c)), ""
)
first = next((c for c in cells if c.lower() == self.firstName.strip().lower()), self.firstName)
# Coverage "MM/DD/YYYY - Present" or "MM/DD/YYYY - MM/DD/YYYY": active if today is inside it.
eligibility = None
if coverage:
start_s, end_s = [p.strip() for p in coverage.split(" - ", 1)]
start = self._parse_date(start_s)
end = None if end_s.lower() == "present" else self._parse_date(end_s)
today = date.today()
if start:
eligibility = "Y" if start <= today and (end is None or today <= end) else "N"
# The (i) icon marks inactive coverage — trust it over the date range
try:
row_html = (row.get_attribute("innerHTML") or "").lower()
if "inactive" in row_html:
eligibility = "N"
except Exception:
pass
print(f"[extraction] subscriber='{subscriber_id}' first='{first}' group='{group}' coverage='{coverage}' → {eligibility}")
return {
"eligibility": eligibility,
"memberId": subscriber_id,
# First name: always the one typed in the app (the row shows it upper-cased)
"firstName": self.firstName.strip(),
"coverage": coverage,
"group": group,
"insurance": "Altus Dental",
}
_NAME_LABELS = ["Patient Name", "Member Name", "Patient", "Member", "Name", "Subscriber Name", "Subscriber"]
def _split_last_name(self, full_name):
"""Last name from 'LAST, FIRST M' or 'FIRST M LAST' — only if it contains our first name."""
first = self.firstName.strip().lower()
full_name = re.sub(r"\s+", " ", full_name or "").strip()
if not first or first not in full_name.lower():
return ""
if "," in full_name:
last, rest = [p.strip() for p in full_name.split(",", 1)]
return last if first in rest.lower() else ""
tokens = full_name.split(" ")
lowered = [t.lower() for t in tokens]
if first not in lowered:
return ""
i = lowered.index(first)
after = [t for t in tokens[i + 1:] if len(t.strip(".")) > 1] # skip middle initials
if after:
return " ".join(after)
before = [t for t in tokens[:i] if len(t.strip(".")) > 1]
return " ".join(before)
def _extract_last_name(self):
"""Find the patient's full name on the benefit details page and return the last name."""
first = self.firstName.strip()
if not first:
return ""
candidates = []
# 1) Value next to an exact name label (exact match so headers like
# "Patient First Name" are never mistaken for a value)
for label in self._NAME_LABELS:
for lab in (label, label + ":"):
xp = (
f"//*[self::td or self::th or self::dt or self::label or self::span or self::strong or self::b]"
f"[normalize-space(.)='{lab}']"
"/following-sibling::*[1]"
)
for el in self.driver.find_elements(By.XPATH, xp):
try:
if el.is_displayed() and el.text.strip():
candidates.append(el.text.strip())
except Exception:
continue
# 2) Any line of page text containing the first name next to another word
text = self.driver.execute_script("return document.body.innerText;") or ""
pat = re.compile(
rf"([A-Za-z'\-]+,[ ]*{re.escape(first)}(?:[ ]+[A-Za-z]\.?)?\b"
rf"|\b{re.escape(first)}[ ]+(?:[A-Za-z]\.?[ ]+)?[A-Za-z'\-]{{2,}}(?:[ ]+[A-Za-z'\-]{{2,}})?)",
re.IGNORECASE,
)
candidates += [m.group(0) for m in pat.finditer(text)]
for cand in candidates:
last = self._split_last_name(cand)
# Reject label words that can sit next to a name in page text
if last and last.lower() not in {"dob", "d.o.b.", "id", "subscriber", "group", "patient", "member"}:
last = last.title() if last.isupper() else last
print(f"[extraction] last name '{last}' from '{cand}'")
return last
print("[extraction] last name not found on details page")
return ""
def _value_after_label(self, labels):
"""Read the value shown next to a label (table cell, dt/dd, or 'Label: value')."""
for label in labels:
xps = [
f"//td[{_ci_contains('normalize-space(.)', label)}]/following-sibling::td[1]",
f"//th[{_ci_contains('normalize-space(.)', label)}]/following-sibling::td[1]",
f"//dt[{_ci_contains('normalize-space(.)', label)}]/following-sibling::dd[1]",
f"//label[{_ci_contains('normalize-space(.)', label)}]/following-sibling::*[1]",
f"//span[{_ci_contains('normalize-space(.)', label)}]/following-sibling::span[1]",
]
for xp in xps:
el = self._visible(xp)
if el and el.text.strip():
return el.text.strip()
return ""
# ── step 2 — print results page as PDF via CDP ──────────────────────────────
def step2_eligibility_pdf(self):
try:
WebDriverWait(self.driver, 30).until(
lambda d: d.execute_script("return document.readyState") == "complete"
)
time.sleep(1)
key = self.memberId or f"{self.lastName}_{self.firstName}"
safe_member = "".join(c for c in str(key) if c.isalnum() or c in "-_.")
pdf_filename = f"eligibility_altus_{safe_member}.pdf"
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
pdf_path = os.path.join(self.download_dir, pdf_filename)
with open(pdf_path, "wb") as f:
f.write(base64.b64decode(pdf_data["data"]))
print("Eligibility PDF saved at:", pdf_path)
return pdf_path
except Exception as e:
print(f"[step2_eligibility_pdf] failed: {e}")
return None
# ── main workflow ────────────────────────────────────────────────────────────
def main_workflow(self, url):
if not (self.firstName and self.dateOfBirth and (self.memberId or self.lastName)):
return {"status": "error", "message": "Altus requires First Name, Date of Birth, and Member ID or Last Name."}
try:
self.config_driver()
self.driver.maximize_window()
self.driver.get(url)
time.sleep(3)
login_result = self.login()
if login_result.startswith("ERROR"):
return {"status": "error", "message": login_result}
step1_result = self.step1()
if step1_result.startswith("ERROR"):
return {"status": "error", "message": step1_result}
if self.extracted_data.get("notFound"):
return {"status": "error", "message": "Altus: member not found — check Member ID / Last Name, First Name and DOB."}
pdf_path = self.step2_eligibility_pdf()
result = {
"status": "success",
"pdf_path": pdf_path,
"file_type": "pdf",
"message": "Altus eligibility PDF captured",
}
if self.extracted_data:
result.update(self.extracted_data)
return result
except Exception as e:
return {"status": "error", "message": str(e)}
finally:
if self.driver:
self.driver.quit()