"""Per-location reconciliation: Aquilaehr visits vs Waystar claims, matched on Patient Name + Date of Service. Writes output/.xlsx. Aquilaehr: Patient ("Mary Lopez Garcia") + Visit Date Waystar: Patient Name ("LOPEZ, MARY") + From Date Match = same date, same first name, and at least one last-name word in common. Run: uv run consolidate.py """ from pathlib import Path import pandas as pd ROOT = Path(__file__).parent WAYSTAR_DIR = ROOT / "processing" / "Waystar" AQUILAEHR_DIR = ROOT / "processing" / "Aquilaehr" OUT_DIR = ROOT / "output" WATERMARK = str(ROOT / "static" / "report-watermark.png") SUFFIXES = {"JR", "SR", "II", "III", "IV"} def words(s): return [w for w in str(s).upper().replace(",", " ").replace(".", " ").split() if w not in SUFFIXES] def join(values): return ", ".join(dict.fromkeys(v for v in values if isinstance(v, str) and v.strip())) def load_aquilaehr(path): a = pd.read_csv(path, dtype=str).fillna("") a["DOS"] = pd.to_datetime(a["Visit Date"], errors="coerce").dt.date a["Name"] = a["Patient"].str.split().str.join(" ") # one row per note/template -> one row per patient visit v = a.groupby(["Account #", "Name", "DOS"], as_index=False).agg( **{"Aquilaehr Claim ID": ("Claim ID", join), "Note Status": ("Note Status", join), "Claim Status": ("Claim Status", join), "Provider": ("Provider", lambda s: join(" ".join(x.split()) for x in s))} ) t = v["Name"].map(words) v["first"] = t.str[0].fillna("") v["last"] = t.str[1:].map(set) return v def load_waystar(path): w = pd.read_csv(path, skiprows=7, dtype=str, index_col=False).fillna("") for c in w.columns: # Excel-friendly CSV wraps values like ="..." w[c] = w[c].str.replace(r'^="(.*)"$', r"\1", regex=True) w["DOS"] = pd.to_datetime(w["From Date"], errors="coerce").dt.date w["Charges"] = pd.to_numeric(w["Charges"].str.replace(r"[$,]", "", regex=True), errors="coerce").fillna(0) w["Claim ID ref"] = w["Claim Number"].str.split().str[0] v = w.groupby(["Patient Name", "DOS"], as_index=False).agg( **{"Waystar Claim Number": ("Claim Number", join), "Charges": ("Charges", "sum"), "Waystar Status": ("Status", join), "Claim ID ref": ("Claim ID ref", join)} ) parts = v["Patient Name"].str.partition(",") # "LAST, FIRST MI" v["first"] = [(words(f) or [""])[0] for f in parts[2]] v["last"] = [set(words(n)) for n in parts[0]] return v def reconcile(a, w): rows, used = [], set() a_by_key = {} for i, r in a.iterrows(): a_by_key.setdefault((r["first"], r["DOS"]), []).append(i) a_claim_dates = {} for _, r in a.iterrows(): for cid in r["Aquilaehr Claim ID"].split(", "): if cid: a_claim_dates.setdefault(cid, set()).add(r["DOS"]) w_claim_dates = {} for _, r in w.iterrows(): for cid in r["Claim ID ref"].split(", "): if cid: w_claim_dates.setdefault(cid, set()).add(r["DOS"]) def fmt(dates): return ", ".join(d.strftime("%m/%d/%Y") for d in sorted(dates) if pd.notna(d)) for _, r in w.iterrows(): same = [i for i in a_by_key.get((r["first"], r["DOS"]), []) if a.at[i, "last"] & r["last"]] hit = next((i for i in same if i not in used), None) row = {"Patient Name": r["Patient Name"], "DOS": r["DOS"], "Waystar Claim Number": r["Waystar Claim Number"], "Charges": r["Charges"], "Waystar Status": r["Waystar Status"]} if hit is not None: used.add(hit) row.update(Result="Matched", **a.loc[hit, ["Aquilaehr Claim ID", "Note Status", "Claim Status", "Provider"]]) row["Aquilaehr Patient"] = a.at[hit, "Name"] elif same: # visit already matched to another Waystar claim row.update(Result="Duplicate in Waystar", **a.loc[same[0], ["Aquilaehr Claim ID", "Claim Status"]]) row["Aquilaehr Patient"] = a.at[same[0], "Name"] row["Note"] = "Same patient + DOS already matched to another Waystar claim" else: row["Result"] = "Missing in Aquilaehr" other = set().union(*(a_claim_dates.get(c, set()) for c in r["Claim ID ref"].split(", "))) if other: row["Note"] = f"Claim ID {r['Claim ID ref']} is in Aquilaehr with Visit Date {fmt(other)}" rows.append(row) for i, r in a.iterrows(): if i in used: continue row = {"Patient Name": r["Name"], "DOS": r["DOS"], "Result": "Missing in Waystar", "Aquilaehr Patient": r["Name"], **r[["Aquilaehr Claim ID", "Note Status", "Claim Status", "Provider"]]} other = set().union(*(w_claim_dates.get(c, set()) for c in r["Aquilaehr Claim ID"].split(", ") if c)) if other: row["Note"] = f"Claim ID {r['Aquilaehr Claim ID']} is in Waystar with DOS {fmt(other)}" rows.append(row) cols = ["Result", "Patient Name", "DOS", "Aquilaehr Patient", "Aquilaehr Claim ID", "Note Status", "Claim Status", "Provider", "Waystar Claim Number", "Charges", "Waystar Status", "Note"] order = {"Missing in Waystar": 0, "Missing in Aquilaehr": 1, "Duplicate in Waystar": 2, "Matched": 3} df = pd.DataFrame(rows).reindex(columns=cols) return df.sort_values(["Result", "Patient Name", "DOS"], key=lambda s: s.map(order) if s.name == "Result" else s) def write_report(df, location, out): counts = df["Result"].value_counts() summary = pd.DataFrame({ "Item": ["Location", "Aquilaehr visits", "Waystar visits", "Matched", "Missing in Waystar", "Missing in Aquilaehr", "Duplicate in Waystar"], "Value": [location, int(counts.get("Matched", 0) + counts.get("Missing in Waystar", 0)), int((df["Result"] != "Missing in Waystar").sum()), int(counts.get("Matched", 0)), int(counts.get("Missing in Waystar", 0)), int(counts.get("Missing in Aquilaehr", 0)), int(counts.get("Duplicate in Waystar", 0))], }) write_xlsx(out, summary, df) return summary def write_xlsx(out, summary, df): # xlsxwriter, not openpyxl: only it can set a sheet background (the eCare/Waystar/Aquilaehr watermark). with pd.ExcelWriter(out, engine="xlsxwriter", date_format="mm/dd/yyyy") as xw: money = xw.book.add_format({"num_format": '"$"#,##0.00'}) for name, data in (("Summary", summary), ("Reconciliation", df)): data.to_excel(xw, sheet_name=name, index=False) ws = xw.sheets[name] ws.set_background(WATERMARK) ws.freeze_panes(1, 0) ws.autofilter(0, 0, len(data), len(data.columns) - 1) for i, col in enumerate(data.columns): width = max([len(col)] + [len(str(v)) for v in data[col].head(199) if pd.notna(v)]) ws.set_column(i, i, min(max(width + 2, 10), 60), money if col == "Charges" else None) ALL_FILE = "All Locations.xlsx" def main(): OUT_DIR.mkdir(exist_ok=True) summaries, frames = [], [] for wf in sorted(WAYSTAR_DIR.glob("*.csv")): af = AQUILAEHR_DIR / wf.name if not af.exists(): print(f"SKIP {wf.stem}: no Aquilaehr file") continue df = reconcile(load_aquilaehr(af), load_waystar(wf)) summary = write_report(df, wf.stem, OUT_DIR / f"{wf.stem}.xlsx") s = summary["Value"].tolist() print(f"{wf.stem:40} Aquilaehr={s[1]:5} Waystar={s[2]:5} Matched={s[3]:5} " f"Missing in Waystar={s[4]:4} Missing in Aquilaehr={s[5]:4} Duplicate in Waystar={s[6]:3}") if s[1] and s[2] and s[3] < 0.5 * min(s[1], s[2]): print(f" WARNING {wf.stem}: under half the visits match - wrong practice data downloaded?") summaries.append(summary.set_index("Item")["Value"]) frames.append(df.assign(Location=wf.stem)) if frames: # one overall workbook: a row per location + every location's rows overall = pd.DataFrame(summaries).reset_index(drop=True) total = overall.drop(columns="Location").sum().to_dict() overall = pd.concat([overall, pd.DataFrame([{"Location": "TOTAL", **total}])], ignore_index=True) rows = pd.concat(frames, ignore_index=True) write_xlsx(OUT_DIR / ALL_FILE, overall, rows[["Location", *rows.columns[:-1]]]) print(f"{'ALL LOCATIONS':40} -> output/{ALL_FILE}") if __name__ == "__main__": main()