MEK-Charge-Reconciliation/consolidate.py

"""Per-location reconciliation: Aquilaehr visits vs Waystar claims, matched on
Patient Name + Date of Service. Writes output/<Location>.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()