VasanthandClaude Opus 5.5
fd480fa874
- mek_recon.py: Playwright download for both sites in parallel (headless on server) - consolidate.py: Patient Name + DOS match, per-location and All Locations Excel - app.py: sign-in (server-side password), Run button, per-file and ZIP downloads - Dockerfile for Dokploy Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
326 lines
13 KiB
Python
326 lines
13 KiB
Python
"""MEK Charge reconciliation process.
|
|
|
|
Waystar (ZirMed): for each location -> switch account -> Claims -> Status All,
|
|
Transaction Date All -> Search -> Download CSV -> processing/Waystar/<location>.csv
|
|
Aquilaehr: Reports -> Encounter & Claim Monitor -> pick practice -> Export CSV
|
|
-> processing/Aquilaehr/<location>.csv (same file name as Waystar so the pair lines up)
|
|
Then consolidate.py matches Patient Name + DOS per location -> output/<location>.xlsx
|
|
|
|
Run: python mek_recon.py both sites at the same time, all locations
|
|
python mek_recon.py Aquilaehr only Aquilaehr
|
|
python mek_recon.py Waystar "Healing Bridge" only Waystar, only that location
|
|
Login is automatic from .env (WAYSTAR_USERNAME, WAYSTAR_PASSWORD, WAYSTAR_ANSWER_CAR,
|
|
WAYSTAR_ANSWER_JOB, AQUILAEHR_USERNAME, AQUILAEHR_PASSWORD). If something is
|
|
missing/unexpected, finish the login by hand in Chrome; the script waits.
|
|
"""
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
from multiprocessing import Process
|
|
from pathlib import Path
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
import consolidate
|
|
|
|
ROOT = Path(__file__).parent
|
|
WAYSTAR, AQUILAEHR = "Waystar", "Aquilaehr"
|
|
OUT = {site: ROOT / "processing" / site for site in (WAYSTAR, AQUILAEHR)}
|
|
|
|
# Waystar account name -> Aquilaehr practice dropdown names.
|
|
# Several Aquilaehr practices are exported and merged into one file.
|
|
# Adelman Health is skipped on both sites (no Aquilaehr practice).
|
|
LOCATIONS = {
|
|
"Alpha Well-Being LLC": ["ALPHA WELL-BEING LLC"],
|
|
"ARC Wound Care": ["ARC Woundcare LLC"],
|
|
"Cynthia DM Concierge NP Services LLC": ["Moona Concierge Nurse Practitioner Services LLC"],
|
|
"Healing Bridge": ["HEALING BRIDGE LLC"],
|
|
"Joy Solutions LLC": ["Joy Solutions CORP"],
|
|
"Lunas Home NP LLC": ["LUNAS HOME NP LLC"],
|
|
"Stem Regen NJ": ["Stem Regen Medical"],
|
|
"Sunwave Woundcare": ["Sunwave Woundcare"],
|
|
"WoundDocs LLC": ["WOUNDOCS LLC", "WOUNDOCS LLC - 2"],
|
|
"Wound Pros of FL": ["Wound Professionals of Florida LLC"],
|
|
}
|
|
|
|
WAYSTAR_LOGIN = "https://login.zirmed.com/ui"
|
|
WAYSTAR_CLAIMS = "https://claims.zirmed.com/Claims/Listing/Index?appid=1"
|
|
AQUILAEHR_REPORTS = "https://platform.aquilaehr.com/#/reports/clinical"
|
|
# time to finish a login by hand; headless (server) runs have no hand, so fail sooner
|
|
LOGIN_WAIT_MS = (3 if os.environ.get("HEADLESS") == "1" else 10) * 60 * 1000
|
|
|
|
|
|
def load_env():
|
|
env = ROOT / ".env"
|
|
if env.exists():
|
|
for line in env.read_text(encoding="utf-8").splitlines():
|
|
if "=" in line and not line.lstrip().startswith("#"):
|
|
k, v = line.split("=", 1)
|
|
os.environ.setdefault(k.strip(), v.strip().strip('"'))
|
|
|
|
|
|
def log(site, msg):
|
|
print(f"{site:9} {msg}", flush=True)
|
|
|
|
|
|
# ---------------- Waystar ----------------
|
|
|
|
def waystar_login(page):
|
|
page.goto(WAYSTAR_LOGIN)
|
|
account_box = page.locator("input.header-account-search-text")
|
|
if os.environ.get("WAYSTAR_USERNAME") and page.locator("#loginName").is_visible():
|
|
page.fill("#loginName", os.environ["WAYSTAR_USERNAME"])
|
|
page.fill("#password", os.environ.get("WAYSTAR_PASSWORD", ""))
|
|
page.click("#loginButton")
|
|
log(WAYSTAR, "login submitted")
|
|
|
|
# pages Waystar may show after "Log in" (ids taken from the G2 ChargeBilledAutomation bot)
|
|
answers = {"first car": os.environ.get("WAYSTAR_ANSWER_CAR"), "first job": os.environ.get("WAYSTAR_ANSWER_JOB")}
|
|
question = page.locator("#authDiv strong").first
|
|
alert = page.locator("#continueButton") # "Proactive Alert" page
|
|
sec_info = page.locator("h2", has_text="Security Information")
|
|
end = time.time() + LOGIN_WAIT_MS / 1000
|
|
while time.time() < end:
|
|
if account_box.is_visible():
|
|
log(WAYSTAR, "logged in")
|
|
return
|
|
if alert.is_visible():
|
|
alert.click()
|
|
log(WAYSTAR, "proactive alert continued")
|
|
elif question.is_visible():
|
|
text = question.inner_text().lower()
|
|
key = next((k for k in answers if k in text), None)
|
|
if not key or not answers[key]:
|
|
raise RuntimeError(f"unknown Waystar security question: {text}")
|
|
page.fill("#verifyAnswer", answers[key])
|
|
page.click("#VerifyButton")
|
|
log(WAYSTAR, f"security question ({key}) answered")
|
|
elif sec_info.is_visible():
|
|
page.click("#NextButton")
|
|
page.get_by_text("Skip This Step", exact=True).click()
|
|
log(WAYSTAR, "security information page skipped")
|
|
page.wait_for_timeout(1000)
|
|
raise RuntimeError("Waystar login did not finish in time")
|
|
|
|
|
|
def waystar_switch_account(page, location):
|
|
box = page.locator("input.header-account-search-text")
|
|
box.click()
|
|
box.fill("")
|
|
box.press_sequentially(location, delay=50)
|
|
items = page.locator("ul.ui-autocomplete li:visible")
|
|
# dropdown text looks like "Healing Bridge (313858)"; it shows the first 25
|
|
# accounts before the typed filter kicks in, so wait for the actual match
|
|
match = items.filter(has_text=location).first
|
|
try:
|
|
match.wait_for(timeout=15000)
|
|
except Exception:
|
|
raise RuntimeError(f"no account match in dropdown: {items.all_inner_texts()}")
|
|
match.click()
|
|
page.wait_for_url("**/SuccessfulAccountChange/**", timeout=30000)
|
|
|
|
|
|
def waystar_download(page, location):
|
|
waystar_switch_account(page, location)
|
|
page.goto(WAYSTAR_CLAIMS)
|
|
page.select_option("#SearchCriteria_Status", label="All")
|
|
page.select_option("#SearchCriteria_TransDate", label="All")
|
|
page.click("#ClaimListingSearchButtonTop")
|
|
page.get_by_text("Transaction Date: All").first.wait_for(timeout=60000)
|
|
page.wait_for_load_state("networkidle")
|
|
with page.expect_download(timeout=120000) as dl:
|
|
page.click("#downloadCsvBtn")
|
|
out = OUT[WAYSTAR] / f"{location}.csv"
|
|
dl.value.save_as(out)
|
|
return out
|
|
|
|
|
|
# ---------------- Aquilaehr ----------------
|
|
|
|
AQUILAEHR_PRACTICE = ".ant-select[style*='200px']" # practice dropdown at the top of the page
|
|
AQUILAEHR_EXPORT = "a[download]" # <a> inside the "Export CSV" button (blob URL)
|
|
|
|
|
|
def aquilaehr_login(page):
|
|
page.goto(AQUILAEHR_REPORTS)
|
|
email = page.locator("#normal_login_email")
|
|
logged_in = page.get_by_text("Reports", exact=True).first
|
|
submitted = False
|
|
end = time.time() + LOGIN_WAIT_MS / 1000
|
|
# SPA: either the login form or the app shows up after a few seconds
|
|
while time.time() < end:
|
|
if logged_in.is_visible():
|
|
log(AQUILAEHR, "logged in")
|
|
return
|
|
if not submitted and os.environ.get("AQUILAEHR_USERNAME") and email.is_visible():
|
|
email.fill(os.environ["AQUILAEHR_USERNAME"])
|
|
page.fill("#normal_login_password", os.environ.get("AQUILAEHR_PASSWORD", ""))
|
|
page.click("button[type=submit]")
|
|
submitted = True
|
|
log(AQUILAEHR, "login submitted")
|
|
page.wait_for_timeout(1000)
|
|
raise RuntimeError("Aquilaehr login did not finish in time")
|
|
|
|
|
|
def aquilaehr_download(page, location):
|
|
"""Export every practice for this location and merge into one CSV (header once)."""
|
|
out = OUT[AQUILAEHR] / f"{location}.csv"
|
|
lines = []
|
|
for practice in LOCATIONS[location]:
|
|
part = aquilaehr_export(page, practice, OUT[AQUILAEHR] / f"{location}.part.csv")
|
|
rows = part.read_text(encoding="utf-8-sig").splitlines()
|
|
part.unlink()
|
|
lines += rows if not lines else rows[1:]
|
|
log(AQUILAEHR, f" {practice}: {len(rows) - 1} rows")
|
|
out.write_text("\n".join(lines) + "\n", encoding="utf-8-sig")
|
|
return out
|
|
|
|
|
|
def norm(s):
|
|
return " ".join(s.split()).lower()
|
|
|
|
|
|
def aquilaehr_select_practice(page, practice):
|
|
# practice dropdown is a virtual list: scroll the open one until the option renders
|
|
page.locator(AQUILAEHR_PRACTICE).click()
|
|
page.locator(".ant-select-dropdown:not(.ant-select-dropdown-hidden)").wait_for()
|
|
found = page.evaluate(
|
|
"""async (title) => {
|
|
const norm = s => (s || '').split(/\\s+/).join(' ').trim().toLowerCase();
|
|
const dd = document.querySelector('.ant-select-dropdown:not(.ant-select-dropdown-hidden)');
|
|
const h = dd.querySelector('.rc-virtual-list-holder');
|
|
for (let y = 0; y <= h.scrollHeight + 100; y += 80) {
|
|
h.scrollTop = y;
|
|
await new Promise(r => setTimeout(r, 150));
|
|
const o = [...dd.querySelectorAll('.ant-select-item-option')]
|
|
.find(o => norm(o.getAttribute('title')) === norm(title));
|
|
if (o) { o.click(); return true; }
|
|
}
|
|
return false;
|
|
}""",
|
|
practice,
|
|
)
|
|
if not found:
|
|
raise RuntimeError(f"practice '{practice}' not in dropdown")
|
|
|
|
|
|
def aquilaehr_export(page, practice, out):
|
|
# practice ids whose report data has arrived (GET .../reports/clinical_notes?practice_id=<id>)
|
|
loaded = []
|
|
|
|
def on_response(r):
|
|
m = re.search(r"/reports/clinical_notes\?practice_id=([\w-]+)", r.url)
|
|
if m and r.ok:
|
|
loaded.append(m.group(1))
|
|
|
|
page.on("response", on_response)
|
|
try:
|
|
# Reports -> Encounter & Claim Monitor tab
|
|
page.goto(AQUILAEHR_REPORTS)
|
|
page.get_by_text("Encounter & Claim Monitor", exact=True).first.click()
|
|
page.wait_for_load_state("networkidle")
|
|
current = page.locator(f"{AQUILAEHR_PRACTICE} .ant-select-selection-item")
|
|
if norm(current.inner_text()) != norm(practice):
|
|
before = page.url
|
|
aquilaehr_select_practice(page, practice)
|
|
page.wait_for_url(lambda u: u != before and "practiceId=" in u, timeout=30000)
|
|
if norm(current.inner_text()) != norm(practice):
|
|
raise RuntimeError(f"dropdown shows '{current.inner_text()}', expected '{practice}'")
|
|
|
|
# wait for THIS practice's data; data for the previous practice can still be arriving
|
|
pid = re.search(r"practiceId=([\w-]+)", page.url)
|
|
end = time.time() + 120
|
|
while not (pid.group(1) in loaded if pid else loaded):
|
|
if time.time() > end:
|
|
raise RuntimeError(f"report data for '{practice}' did not load")
|
|
page.wait_for_timeout(500)
|
|
page.wait_for_load_state("networkidle")
|
|
page.wait_for_timeout(2000) # let the table + export link re-render
|
|
|
|
with page.expect_download(timeout=120000) as dl:
|
|
page.locator(AQUILAEHR_EXPORT).click()
|
|
dl.value.save_as(out)
|
|
finally:
|
|
page.remove_listener("response", on_response)
|
|
return out
|
|
|
|
|
|
# ---------------- runner ----------------
|
|
|
|
SITES = {
|
|
WAYSTAR: (waystar_login, waystar_download),
|
|
AQUILAEHR: (aquilaehr_login, aquilaehr_download),
|
|
}
|
|
|
|
|
|
def run_site(site, locations):
|
|
"""One site in its own Chrome (own profile), so both sites can run at the same time."""
|
|
load_env()
|
|
OUT[site].mkdir(parents=True, exist_ok=True)
|
|
login, download = SITES[site]
|
|
failed = []
|
|
seen = {} # file content -> location; two locations with identical exports = stale data
|
|
with sync_playwright() as p:
|
|
# server/Docker: HEADLESS=1 and CHROME_CHANNEL="" (Playwright's bundled Chromium)
|
|
headless = os.environ.get("HEADLESS") == "1"
|
|
ctx = p.chromium.launch_persistent_context(
|
|
str(ROOT / f".chrome_profile_{site.lower()}"),
|
|
channel=os.environ.get("CHROME_CHANNEL", "chrome") or None, headless=headless,
|
|
accept_downloads=True, no_viewport=not headless,
|
|
viewport={"width": 1440, "height": 900} if headless else None,
|
|
)
|
|
page = ctx.pages[0] if ctx.pages else ctx.new_page()
|
|
login(page)
|
|
for loc in locations:
|
|
try:
|
|
out = download(page, loc)
|
|
content = out.read_bytes()
|
|
if content in seen:
|
|
log(site, f"WARNING {loc}: same data as {seen[content]} - stale export?")
|
|
seen[content] = loc
|
|
log(site, f"OK {loc} -> {out.name} ({content.count(b'\n')} lines)")
|
|
except Exception as e:
|
|
failed.append(loc)
|
|
log(site, f"FAIL {loc}: {e}")
|
|
ctx.close()
|
|
log(site, f"failed: {failed or 'none'}")
|
|
|
|
|
|
def locked_files():
|
|
"""Files open in Excel can't be overwritten -> the run would silently keep old data."""
|
|
locked = []
|
|
for f in [*ROOT.glob("processing/*/*.csv"), *ROOT.glob("output/*.xlsx")]:
|
|
try:
|
|
with open(f, "a"):
|
|
pass
|
|
except PermissionError:
|
|
locked.append(f.relative_to(ROOT))
|
|
return locked
|
|
|
|
|
|
def main():
|
|
locked = locked_files()
|
|
if locked:
|
|
print("Close these files in Excel first, then run again:")
|
|
for f in locked:
|
|
print(f" {f}")
|
|
sys.exit(1)
|
|
args = sys.argv[1:]
|
|
sites = [args.pop(0)] if args and args[0] in SITES else list(SITES)
|
|
locations = [loc for loc in LOCATIONS if not args or loc in args]
|
|
if len(sites) == 1:
|
|
run_site(sites[0], locations)
|
|
else:
|
|
procs = [Process(target=run_site, args=(site, locations)) for site in sites]
|
|
for pr in procs:
|
|
pr.start()
|
|
for pr in procs:
|
|
pr.join()
|
|
print("Consolidating -> output/<Location>.xlsx", flush=True)
|
|
consolidate.main()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|