Support CVFCU Account Details (recent activity) PDFs; never double-count overlapping statements

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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

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)
 

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))