feat: MassHealth member detail demographics, wait for name/plan before PDF
- Extract gender, address (street/city/state/zip) and home phone from the MassHealth Member Detail page and save to the patient; phone is only filled when the patient has none - Shared extract_eligibility_row polls up to 30s until the matching results row has both name and plan before the PDF is printed (AngularJS fills cell text after the rows appear); used by all three MH eligibility workers - Fall back to the Member Detail heading for the name when the results table never filled it in Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,80 @@
|
||||
import time
|
||||
|
||||
from selenium.webdriver.common.by import By
|
||||
|
||||
# Cell texts of every row in the Eligible / Ineligible results tables.
|
||||
# innerText (not .text) so cells rendered off-screen still return their text.
|
||||
_ROWS_JS = """
|
||||
const out = [];
|
||||
for (const label of ['Eligible', 'Ineligible']) {
|
||||
const rows = document.evaluate(
|
||||
`//h4[normalize-space(text())='${label}']/following::table[1]/tbody/tr`,
|
||||
document, null, XPathResult.ORDERED_NODE_SNAPSHOT_TYPE, null);
|
||||
for (let i = 0; i < rows.snapshotLength; i++) {
|
||||
const cells = Array.from(rows.snapshotItem(i).querySelectorAll('td'))
|
||||
.map(td => (td.innerText || td.textContent || '').trim());
|
||||
out.push([label, cells]);
|
||||
}
|
||||
}
|
||||
return out;
|
||||
"""
|
||||
|
||||
|
||||
def _normalize_id(s):
|
||||
return "".join(c for c in str(s) if c.isalnum()).lower()
|
||||
|
||||
|
||||
def extract_eligibility_row(driver, member_id, timeout=30):
|
||||
"""Wait for the MassHealth results row matching member_id and return
|
||||
{eligibility, firstName, lastName, insurance}.
|
||||
|
||||
AngularJS adds the rows first and fills the cell text afterwards — the
|
||||
member ID can appear before the name and plan do. Keep polling until the
|
||||
matching row has both a name and a plan, so callers never print the PDF
|
||||
before the data is captured. After the timeout, returns whatever was
|
||||
found ({"eligibility": None} if no row matched).
|
||||
"""
|
||||
norm_self = _normalize_id(member_id)
|
||||
best = {"eligibility": None}
|
||||
deadline = time.time() + timeout
|
||||
attempt = 0
|
||||
|
||||
while True:
|
||||
attempt += 1
|
||||
try:
|
||||
rows = driver.execute_script(_ROWS_JS) or []
|
||||
except Exception as e:
|
||||
print(f"[extraction] reading rows failed: {e}")
|
||||
rows = []
|
||||
|
||||
for label, cells in rows:
|
||||
if len(cells) < 3:
|
||||
continue
|
||||
norm_cell = _normalize_id(cells[2])
|
||||
if not (norm_self and norm_cell and (norm_self in norm_cell or norm_cell in norm_self)):
|
||||
continue
|
||||
|
||||
# name in cells[4], plan in cells[6] (fallback to last cell)
|
||||
full_name = cells[4] if len(cells) > 4 else ""
|
||||
plan_name = cells[6] if len(cells) > 6 else (cells[-1] if len(cells) > 4 else "")
|
||||
name_parts = full_name.split()
|
||||
best = {
|
||||
"eligibility": "Y" if label == "Eligible" else "N",
|
||||
"firstName": name_parts[0] if name_parts else "",
|
||||
"lastName": " ".join(name_parts[1:]),
|
||||
"insurance": plan_name,
|
||||
}
|
||||
if full_name and plan_name:
|
||||
print(f"[extraction] MATCHED {label} → name='{full_name}' plan='{plan_name}' (attempt {attempt})")
|
||||
return best
|
||||
break
|
||||
|
||||
if time.time() >= deadline:
|
||||
break
|
||||
if attempt % 5 == 1:
|
||||
state = "name/plan still empty" if best["eligibility"] else "no matching row yet"
|
||||
print(f"[extraction] {state} for memberId='{member_id}' (attempt {attempt}), waiting...")
|
||||
time.sleep(1)
|
||||
|
||||
print(f"[extraction] gave up after {timeout}s for memberId='{member_id}' → {best}")
|
||||
return best
|
||||
@@ -0,0 +1,74 @@
|
||||
import re
|
||||
|
||||
from selenium.webdriver.common.by import By
|
||||
|
||||
# Labels shown in the MassHealth Member Detail block, e.g.
|
||||
# Date of Birth: 4/21/1958
|
||||
# Gender: Female
|
||||
# Address: 169 CYPRESS ST, NEWTON, MA 02459
|
||||
# Home Phone: 6036789376
|
||||
# Registered Not Registered in Member Community
|
||||
_LABELS = ("Date of Birth", "Gender", "Address", "Home Phone", "Phone", "Registered")
|
||||
|
||||
|
||||
def _label_value(text, label):
|
||||
"""Value next to "Label:" — same line or the next line (innerText of a
|
||||
two-column layout can put the value on its own line). Empty when the
|
||||
portal leaves the field blank."""
|
||||
m = re.search(rf"(?:^|\n)[ \t]*{re.escape(label)}:[ \t]*\n?[ \t]*([^\n]*)", text)
|
||||
if not m:
|
||||
return ""
|
||||
value = m.group(1).strip()
|
||||
# Blank field — we grabbed the next row's label instead of a value
|
||||
if value.endswith(":") or any(value.startswith(f"{l}:") or value == l for l in _LABELS):
|
||||
return ""
|
||||
return value
|
||||
|
||||
|
||||
def extract_member_demographics(driver):
|
||||
"""Read gender, address and phone from the MassHealth Member Detail page.
|
||||
|
||||
Returns only the fields that were found:
|
||||
gender — "Female" / "Male" as shown by the portal
|
||||
address, city, state, zipCode — split from "STREET, CITY, ST 12345"
|
||||
phone — 10 digits
|
||||
memberName — the heading above "Date of Birth:" (e.g. "OLHA SVYSHCH")
|
||||
"""
|
||||
result = {}
|
||||
try:
|
||||
text = driver.find_element(By.TAG_NAME, "body").get_attribute("innerText") or ""
|
||||
|
||||
# Name is the last non-empty line before the "Date of Birth:" row
|
||||
dob = re.search(r"(?:^|\n)[ \t]*Date of Birth:", text)
|
||||
if dob:
|
||||
lines = [l.strip() for l in text[:dob.start()].split("\n") if l.strip()]
|
||||
if lines and ":" not in lines[-1]:
|
||||
result["memberName"] = lines[-1]
|
||||
|
||||
gender = _label_value(text, "Gender")
|
||||
if gender:
|
||||
result["gender"] = gender
|
||||
|
||||
address = _label_value(text, "Address")
|
||||
if address:
|
||||
# "169 CYPRESS ST, NEWTON, MA 02459" — street may itself contain commas (APT 2)
|
||||
m = re.match(r"^(.*),\s*([^,]+?),?\s+([A-Z]{2})\s+(\d{5})(?:-?\d{4})?$", address)
|
||||
if m:
|
||||
result["address"] = m.group(1).strip()
|
||||
result["city"] = m.group(2).strip()
|
||||
result["state"] = m.group(3)
|
||||
result["zipCode"] = m.group(4)
|
||||
else:
|
||||
result["address"] = address
|
||||
|
||||
phone = _label_value(text, "Home Phone") or _label_value(text, "Phone")
|
||||
digits = re.sub(r"\D", "", phone)
|
||||
if len(digits) == 11 and digits.startswith("1"):
|
||||
digits = digits[1:]
|
||||
if len(digits) == 10:
|
||||
result["phone"] = digits
|
||||
|
||||
print(f"[extract_member_demographics] {result}")
|
||||
except Exception as e:
|
||||
print(f"[extract_member_demographics] failed: {e}")
|
||||
return result
|
||||
@@ -9,6 +9,9 @@ import time
|
||||
import os
|
||||
import base64
|
||||
|
||||
from mh_eligibility_results import extract_eligibility_row
|
||||
from mh_member_details import extract_member_demographics
|
||||
|
||||
|
||||
class AutomationCMSPEligibilityHistoryRemainingCheck:
|
||||
def __init__(self, data):
|
||||
@@ -161,62 +164,9 @@ class AutomationCMSPEligibilityHistoryRemainingCheck:
|
||||
|
||||
# ── helpers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _cell_text(self, cell):
|
||||
text = cell.text.strip()
|
||||
if not text:
|
||||
try:
|
||||
text = (self.driver.execute_script("return arguments[0].innerText;", cell) or "").strip()
|
||||
except Exception:
|
||||
pass
|
||||
return text
|
||||
|
||||
def _normalize_id(self, s):
|
||||
return "".join(c for c in str(s) if c.isalnum()).lower()
|
||||
|
||||
def _extract_data_from_page(self):
|
||||
wait = WebDriverWait(self.driver, 15)
|
||||
try:
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']/following::table[1]/tbody/tr")
|
||||
)
|
||||
)
|
||||
# AngularJS populates cell text after elements appear in the DOM.
|
||||
# Retry a few times until a row with non-empty cell[2] is found.
|
||||
for attempt in range(4):
|
||||
for status_label, elig_flag in [("Eligible", "Y"), ("Ineligible", "N")]:
|
||||
rows = self.driver.find_elements(
|
||||
By.XPATH,
|
||||
f"//h4[text()='{status_label}']/following::table[1]/tbody/tr"
|
||||
)
|
||||
for row in rows:
|
||||
cells = row.find_elements(By.TAG_NAME, "td")
|
||||
if len(cells) < 3:
|
||||
continue
|
||||
norm_cell = self._normalize_id(self._cell_text(cells[2]))
|
||||
norm_self = self._normalize_id(self.memberId)
|
||||
if norm_self and norm_cell and (norm_self in norm_cell or norm_cell in norm_self):
|
||||
full_name = self._cell_text(cells[4]) if len(cells) > 4 else ""
|
||||
plan_name = (
|
||||
self._cell_text(cells[6]) if len(cells) > 6
|
||||
else (self._cell_text(cells[-1]) if len(cells) > 4 else "")
|
||||
)
|
||||
name_parts = full_name.split()
|
||||
return {
|
||||
"eligibility": elig_flag,
|
||||
"firstName": name_parts[0] if name_parts else "",
|
||||
"lastName": " ".join(name_parts[1:]) if len(name_parts) > 1 else "",
|
||||
"insurance": plan_name,
|
||||
}
|
||||
if attempt < 3:
|
||||
print(f"[extraction] cells not yet populated (attempt {attempt + 1}), retrying...")
|
||||
time.sleep(1.5)
|
||||
|
||||
print(f"[extraction] No matching row for memberId='{self.memberId}'")
|
||||
return {"eligibility": None}
|
||||
except Exception as e:
|
||||
print("Extraction error:", e)
|
||||
return {"eligibility": None}
|
||||
# Waits until the matching row's name and plan are filled in
|
||||
return extract_eligibility_row(self.driver, self.memberId)
|
||||
|
||||
# ── step 2 — print eligibility PDF, return to results tab ───────────────────
|
||||
|
||||
@@ -336,6 +286,16 @@ class AutomationCMSPEligibilityHistoryRemainingCheck:
|
||||
|
||||
self.extract_third_party_insurance()
|
||||
|
||||
# Gender, address and phone from the Member Detail block
|
||||
details = extract_member_demographics(self.driver)
|
||||
member_name = details.pop("memberName", "")
|
||||
# Results table never filled the name in — use the Member Detail heading
|
||||
if member_name and not self.extracted_data.get("firstName"):
|
||||
parts = member_name.split()
|
||||
details["firstName"] = parts[0]
|
||||
details["lastName"] = " ".join(parts[1:])
|
||||
self.extracted_data.update(details)
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"mh_member_details_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
|
||||
@@ -9,6 +9,9 @@ import time
|
||||
import os
|
||||
import base64
|
||||
|
||||
from mh_eligibility_results import extract_eligibility_row
|
||||
from mh_member_details import extract_member_demographics
|
||||
|
||||
|
||||
class AutomationMassHealthEligibilityHistoryCheck:
|
||||
def __init__(self, data):
|
||||
@@ -191,67 +194,9 @@ class AutomationMassHealthEligibilityHistoryCheck:
|
||||
|
||||
# ── helpers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _cell_text(self, cell):
|
||||
text = cell.text.strip()
|
||||
if not text:
|
||||
try:
|
||||
text = (self.driver.execute_script("return arguments[0].innerText;", cell) or "").strip()
|
||||
except Exception:
|
||||
pass
|
||||
return text
|
||||
|
||||
def _normalize_id(self, s):
|
||||
return "".join(c for c in str(s) if c.isalnum()).lower()
|
||||
|
||||
def _extract_data_from_page(self):
|
||||
wait = WebDriverWait(self.driver, 15)
|
||||
extracted = {}
|
||||
try:
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']/following::table[1]/tbody/tr")
|
||||
)
|
||||
)
|
||||
|
||||
# AngularJS populates cell text after elements appear in the DOM.
|
||||
# Retry a few times until a row with non-empty cell[2] is found.
|
||||
for attempt in range(4):
|
||||
for status_label, elig_flag in [("Eligible", "Y"), ("Ineligible", "N")]:
|
||||
rows = self.driver.find_elements(
|
||||
By.XPATH,
|
||||
f"//h4[text()='{status_label}']/following::table[1]/tbody/tr"
|
||||
)
|
||||
for row in rows:
|
||||
cells = row.find_elements(By.TAG_NAME, "td")
|
||||
if len(cells) < 3:
|
||||
continue
|
||||
member_number = self._cell_text(cells[2])
|
||||
norm_cell = self._normalize_id(member_number)
|
||||
norm_self = self._normalize_id(self.memberId)
|
||||
if norm_self and norm_cell and (norm_self in norm_cell or norm_cell in norm_self):
|
||||
full_name = self._cell_text(cells[4]) if len(cells) > 4 else ""
|
||||
plan_name = (
|
||||
self._cell_text(cells[6]) if len(cells) > 6
|
||||
else (self._cell_text(cells[-1]) if len(cells) > 4 else "")
|
||||
)
|
||||
name_parts = full_name.split()
|
||||
extracted = {
|
||||
"eligibility": elig_flag,
|
||||
"firstName": name_parts[0] if name_parts else "",
|
||||
"lastName": " ".join(name_parts[1:]) if len(name_parts) > 1 else "",
|
||||
"insurance": plan_name,
|
||||
}
|
||||
print(f"[extraction] MATCHED {status_label} → name='{full_name}' plan='{plan_name}'")
|
||||
return extracted
|
||||
if attempt < 3:
|
||||
print(f"[extraction] cells not yet populated (attempt {attempt + 1}), retrying...")
|
||||
time.sleep(1.5)
|
||||
|
||||
print(f"[extraction] No matching row for memberId='{self.memberId}'")
|
||||
return {"eligibility": None}
|
||||
except Exception as e:
|
||||
print("Extraction error:", e)
|
||||
return {"eligibility": None}
|
||||
# Waits until the matching row's name and plan are filled in
|
||||
return extract_eligibility_row(self.driver, self.memberId)
|
||||
|
||||
# ── step 2 — print eligibility PDF, stay on results tab ─────────────────────
|
||||
|
||||
@@ -333,6 +278,16 @@ class AutomationMassHealthEligibilityHistoryCheck:
|
||||
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||||
time.sleep(2)
|
||||
|
||||
# Gender, address and phone from the Member Detail block
|
||||
details = extract_member_demographics(self.driver)
|
||||
member_name = details.pop("memberName", "")
|
||||
# Results table never filled the name in — use the Member Detail heading
|
||||
if member_name and not self.extracted_data.get("firstName"):
|
||||
parts = member_name.split()
|
||||
details["firstName"] = parts[0]
|
||||
details["lastName"] = " ".join(parts[1:])
|
||||
self.extracted_data.update(details)
|
||||
|
||||
safe_member = "".join(c for c in str(self.memberId) if c.isalnum() or c in "-_.")
|
||||
pdf_filename = f"eligibility_member_details_{safe_member}.pdf"
|
||||
pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", {"printBackground": True})
|
||||
|
||||
@@ -9,6 +9,8 @@ import time
|
||||
import os
|
||||
import base64
|
||||
|
||||
from mh_eligibility_results import extract_eligibility_row
|
||||
|
||||
class AutomationMassHealthEligibilityCheck:
|
||||
def __init__(self, data):
|
||||
self.headless = False
|
||||
@@ -245,124 +247,9 @@ class AutomationMassHealthEligibilityCheck:
|
||||
print(f"[step1] Could not save screenshot: {ss_err}")
|
||||
return f"ERROR:STEP1:{substep}"
|
||||
|
||||
def _cell_text(self, cell):
|
||||
"""Get text from a cell, falling back to JS innerText if .text is empty."""
|
||||
text = cell.text.strip()
|
||||
if not text:
|
||||
try:
|
||||
text = (self.driver.execute_script("return arguments[0].innerText;", cell) or "").strip()
|
||||
except Exception:
|
||||
pass
|
||||
return text
|
||||
|
||||
def _normalize_id(self, s):
|
||||
"""Strip all non-alphanumeric characters and lowercase for robust ID matching."""
|
||||
return ''.join(c for c in str(s) if c.isalnum()).lower()
|
||||
|
||||
def _extract_data_from_page(self):
|
||||
wait = WebDriverWait(self.driver, 15)
|
||||
extracted = {}
|
||||
|
||||
try:
|
||||
# Wait until at least one table row appears under Eligible or Ineligible —
|
||||
# the h4 header renders before the rows are populated, so we must wait for rows.
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, "//h4[text()='Eligible' or text()='Ineligible']/following::table[1]/tbody/tr")
|
||||
)
|
||||
)
|
||||
|
||||
eligible_rows = self.driver.find_elements(
|
||||
By.XPATH,
|
||||
"//h4[text()='Eligible']/following::table[1]/tbody/tr"
|
||||
)
|
||||
|
||||
if eligible_rows:
|
||||
for row in eligible_rows:
|
||||
cells = row.find_elements(By.TAG_NAME, "td")
|
||||
|
||||
if len(cells) < 3:
|
||||
continue
|
||||
|
||||
member_number = self._cell_text(cells[2])
|
||||
norm_cell = self._normalize_id(member_number)
|
||||
norm_self = self._normalize_id(self.memberId)
|
||||
print(f"[eligible] cells count={len(cells)}, memberId check: '{norm_self}' vs '{norm_cell}' (raw: '{member_number}')")
|
||||
if len(cells) >= 5:
|
||||
print(f" cells[3]='{self._cell_text(cells[3])}' cells[4]='{self._cell_text(cells[4])}'", end="")
|
||||
if len(cells) > 5:
|
||||
print(f" cells[5]='{self._cell_text(cells[5])}'", end="")
|
||||
if len(cells) > 6:
|
||||
print(f" cells[6]='{self._cell_text(cells[6])}'", end="")
|
||||
print()
|
||||
|
||||
if norm_self and norm_cell and (norm_self in norm_cell or norm_cell in norm_self):
|
||||
# name is in cells[4], insurance in cells[6] (fallback to last cell)
|
||||
full_name = self._cell_text(cells[4]) if len(cells) > 4 else ""
|
||||
plan_name = self._cell_text(cells[6]) if len(cells) > 6 else (self._cell_text(cells[-1]) if len(cells) > 4 else "")
|
||||
|
||||
name_parts = full_name.split()
|
||||
first_name = name_parts[0] if name_parts else ""
|
||||
last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else ""
|
||||
|
||||
print(f"[eligible] MATCHED → name='{full_name}' plan='{plan_name}'")
|
||||
|
||||
extracted = {
|
||||
"eligibility": "Y",
|
||||
"firstName": first_name,
|
||||
"lastName": last_name,
|
||||
"insurance": plan_name
|
||||
}
|
||||
|
||||
return extracted
|
||||
|
||||
ineligible_rows = self.driver.find_elements(
|
||||
By.XPATH,
|
||||
"//h4[text()='Ineligible']/following::table[1]/tbody/tr"
|
||||
)
|
||||
|
||||
if ineligible_rows:
|
||||
for row in ineligible_rows:
|
||||
cells = row.find_elements(By.TAG_NAME, "td")
|
||||
|
||||
if len(cells) < 3:
|
||||
continue
|
||||
|
||||
member_number = self._cell_text(cells[2])
|
||||
norm_cell = self._normalize_id(member_number)
|
||||
norm_self = self._normalize_id(self.memberId)
|
||||
print(f"[ineligible] cells count={len(cells)}, memberId check: '{norm_self}' vs '{norm_cell}' (raw: '{member_number}')")
|
||||
if len(cells) >= 5:
|
||||
print(f" cells[3]='{self._cell_text(cells[3])}' cells[4]='{self._cell_text(cells[4])}'", end="")
|
||||
if len(cells) > 5:
|
||||
print(f" cells[5]='{self._cell_text(cells[5])}'", end="")
|
||||
print()
|
||||
|
||||
if norm_self and norm_cell and (norm_self in norm_cell or norm_cell in norm_self):
|
||||
full_name = self._cell_text(cells[4]) if len(cells) > 4 else ""
|
||||
plan_name = self._cell_text(cells[5]) if len(cells) > 5 else (self._cell_text(cells[-1]) if len(cells) > 4 else "")
|
||||
|
||||
name_parts = full_name.split()
|
||||
first_name = name_parts[0] if name_parts else ""
|
||||
last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else ""
|
||||
|
||||
print(f"[ineligible] MATCHED → name='{full_name}' plan='{plan_name}'")
|
||||
|
||||
extracted = {
|
||||
"eligibility": "N",
|
||||
"firstName": first_name,
|
||||
"lastName": last_name,
|
||||
"insurance": plan_name
|
||||
}
|
||||
|
||||
return extracted
|
||||
|
||||
print(f"[extraction] No matching row found for memberId='{self.memberId}'")
|
||||
return {"eligibility": None}
|
||||
|
||||
except Exception as e:
|
||||
print("Extraction error:", e)
|
||||
return {"eligibility": None}
|
||||
# Waits until the matching row's name and plan are filled in
|
||||
return extract_eligibility_row(self.driver, self.memberId)
|
||||
|
||||
def _print_current_page_as_pdf(self):
|
||||
"""Generate PDF from the currently active tab using CDP."""
|
||||
|
||||
Reference in New Issue
Block a user