From 3d956f3fd27106f464763b54bff16c84b4fc61cf Mon Sep 17 00:00:00 2001 From: nmemmert <67961292+nmemmert@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:47:56 -0400 Subject: [PATCH] Support CVFCU Account Details (recent activity) PDFs; never double-count overlapping statements Co-Authored-By: Claude Sonnet 5.5 --- README.md | 2 +- app.py | 8 +++++-- statement_loader.py | 55 ++++++++++++++++++++++++++++++++++++++++----- 3 files changed, 56 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 1f911cc..5fae278 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # money -Local tool that loads bank/card statements (Chase Visa, Ally, CVFCU PDFs; Capital One CSV) into `Bills_Tracker.xlsx` and shows a web dashboard. +Local tool that loads bank/card statements (Chase Visa, Ally, CVFCU PDFs; CVFCU "Account Details" page saved as PDF for recent activity; Capital One CSV) into `Bills_Tracker.xlsx` and shows a web dashboard. Everything runs locally; statements and the workbook are **never** committed or baked into the image. ## Run on ZimaOS / CasaOS diff --git a/app.py b/app.py index 1416e8b..78b74ac 100644 --- a/app.py +++ b/app.py @@ -23,9 +23,12 @@ def coverage(sts): today = dt.date.today() cfg = {a[0]: a[2] for a in L.ACCOUNTS} by = {} - for st in sts: by.setdefault(st.account, []).append(st) + recent = {} # "Account Details" pages: partial months, never count as a full statement + for st in sts: + if st.kind == "activity": recent[st.account] = max(recent.get(st.account, st.end), st.end) + else: by.setdefault(st.account, []).append(st) if not by: return dict(months=[], accounts=[]) - lo = min(s.start or s.end for s in sts); months = [] + lo = min(s.start or s.end for s in sum(by.values(), [])); months = [] y, m = lo.year, lo.month while (y, m) <= (today.year, today.month): months.append(f"{y}-{m:02d}"); m += 1 @@ -52,6 +55,7 @@ def coverage(sts): if cfg.get(acct) is None and last.day != eom(last.year, last.month): ny, nm = last.year, last.month nd = dt.date(ny, nm, cfg.get(acct) or eom(ny, nm)) nxt = f"Next statement closes ~{nd:%b %-d, %Y}" + (" (ready to download)" if nd <= today else "") + if acct in recent and recent[acct] > last: nxt += f" · recent activity loaded through {recent[acct]:%b %-d}" out.append(dict(account=acct, cells=cells, count=len(lst), through=f"{last:%b %-d, %Y}", next=nxt, kind="csv" if csv_only else "pdf")) return dict(months=months, accounts=out) diff --git a/statement_loader.py b/statement_loader.py index 32cef4d..970d381 100644 --- a/statement_loader.py +++ b/statement_loader.py @@ -162,6 +162,45 @@ def parse_cvfcu(path, pages): t.extra = f"Principal ${pr:,.2f} / Interest ${fin:,.2f}" return st +# Online "Account Details" page saved as PDF (recent activity, no statement dates). Account is named by its product line. +ACTIVITY_ACCOUNTS = {"PREFERRED CHECKING": "CVFCU Checking"} +ACT_TX = re.compile(r"^([A-Z][a-z]{2}) (\d\d) (.+?) ([+-])\$([\d,]+\.\d\d) \$(-?[\d,]+\.\d\d)$") +MONTHS = {m: i for i, m in enumerate(["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"], 1)} + +def identify_activity(pages): + first = pages[0] + if "Account Details" in first and "Past Transactions" in first: + for k, name in ACTIVITY_ACCOUNTS.items(): + if k in first.upper(): return name + return None + +def parse_activity(path, pages, account): + """Recent-activity page: newest first, signed amounts and a running balance. Every row is checked against the balance chain.""" + full = "\n".join(pages) + rng = re.search(r"(\d\d/\d\d/\d{4}) (\d\d/\d\d/\d{4})", full) + lo, hi = (pdate(rng.group(1)), pdate(rng.group(2))) if rng else (None, None) + cur = re.search(r"Current Balance \$(-?[\d,]+\.\d\d)", full) + rows = [] + for line in full.split("Past Transactions", 1)[-1].splitlines(): + m = ACT_TX.match(line.strip()) + if not m: continue + mon, day = MONTHS[m.group(1)], int(m.group(2)) + yr = (hi or dt.date.today()).year + d = dt.date(yr, mon, day) + if hi and d > hi: d = dt.date(yr - 1, mon, day) + amt = money(m.group(5)) * (1 if m.group(4) == "+" else -1) # signed: + money in + rows.append((d, m.group(3).strip(), amt, money(m.group(6)))) + if not rows: raise ValueError("no transactions found") + if cur and abs(rows[0][3] - money(cur.group(1))) > 0.005: raise ValueError("latest balance doesn't match Current Balance") + for (d, _, a, bal), (_, _, _, nxt) in zip(rows, rows[1:]): + if abs(nxt - (bal - a)) > 0.005: raise ValueError(f"balance doesn't add up near {d:%b %-d}: a row is missing or misread") + st = Statement(Path(path).name, account, min(r[0] for r in rows), max(r[0] for r in rows), kind="activity") + for d, desc, a, _ in rows: + desc = re.sub(r"^ACH TRANS : ", "EFT ", desc) + st.txns.append(Txn(account, d, desc, "", round(-a, 2), st.file)) # app convention: + = money out + st.stated_out = round(sum(-a for _, _, a, _ in rows if a < 0), 2) + return st + def parse_capone_csv(path): rows = list(csv.DictReader(open(path, newline="", encoding="utf-8-sig"))) rows = [r for r in rows if r.get("Transaction Date") and pdate(r["Transaction Date"])] @@ -195,8 +234,9 @@ def load_statements(folder=DATA): st, err = None, None try: if p.suffix.lower() == ".pdf": - pages = _pages(p); name = identify_pdf(pages) - if not name: err = "not recognized (new account? see ACCOUNTS in statement_loader.py)" + pages = _pages(p); name = identify_pdf(pages); act = None if name else identify_activity(pages) + if act: st = parse_activity(p, pages, act) + elif not name: err = "not recognized (new account? see ACCOUNTS in statement_loader.py)" else: st = {"Chase Visa": parse_chase, "Ally Checking": parse_ally, "CVFCU Checking": parse_cvfcu}[name](p, pages) elif p.suffix.lower() == ".csv": head = open(p, encoding="utf-8-sig").readline() @@ -293,7 +333,7 @@ def classify(t, rules, known_payees, cats=None): row = Row(t, "log") if payee: row.payee, row.what = payee, what else: - row.payee = (sm[0] if sm and sm[0] else " ".join(w.capitalize() for w in (norm(t.desc) or t.desc.upper()).split())[:40]) or "Unknown" + row.payee = (sm[0] if sm and sm[0] else " ".join(w.capitalize() for w in re.sub(r"^EFT ", "", norm(t.desc) or t.desc.upper()).split())[:40]) or "Unknown" row.what = sm[3] if sm else "NEW – please categorize" row.is_new_payee = True if not sm: row.needs_review = True; row.note = "NEW – review payee/category" @@ -327,10 +367,13 @@ def plan(wb, statements): known = {r[0] for r in lists.iter_rows(min_row=4, max_row=153, max_col=1, values_only=True) if r[0]} cats = _categories(wb) have = existing_keys(wb) - rows = [] - for st in statements: + rows, taken = [], collections.Counter() + for st in sorted(statements, key=lambda x: x.kind == "activity"): # real statements win over recent-activity pages + local = collections.Counter() for t in st.txns: - rows.append(classify(t, rules, known, cats)) + k = (t.account, t.date, round(abs(t.amount), 2)); local[k] += 1 + if local[k] > taken[k]: rows.append(classify(t, rules, known, cats)) # same charge in two overlapping files counts once + for k, n in local.items(): taken[k] = max(taken[k], n) new, already = [], 0 for r in rows: t = r.txn; k = (t.account, t.date, round(abs(t.amount), 2))