Auto-categorize with built-in merchant rules, payee category editor, income/net on dashboard
build-container / build (push) Successful in 35s
build-container / build (push) Successful in 35s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
6073279a1b
commit
837f89e563
7 files changed
+191
-17
No files matched your search
+77
-8
@@ -11,6 +11,7 @@ import re, csv, sys, shutil, zipfile, datetime as dt, calendar, collections
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
import pdfplumber
|
||||
from seed_rules import seed_match
|
||||
|
||||
import os
|
||||
HERE = Path(__file__).parent # code lives here
|
||||
@@ -260,6 +261,8 @@ class Row:
|
||||
note: str = ""
|
||||
is_new_payee: bool = False
|
||||
needs_review: bool = False
|
||||
category: str = ""
|
||||
type: str = ""
|
||||
|
||||
def learn_rules(wb):
|
||||
"""payee/what-for per normalized description, learned from rows already in the Bill Log."""
|
||||
@@ -269,7 +272,7 @@ def learn_rules(wb):
|
||||
c[norm(r[7])][(r[1], r[4])] += 1
|
||||
return {k: v.most_common(1)[0][0] for k, v in c.items() if k}
|
||||
|
||||
def classify(t, rules, known_payees):
|
||||
def classify(t, rules, known_payees, cats=None):
|
||||
ex = exclusion(t)
|
||||
if ex: return Row(t, "excluded", reason=ex[0], direction=ex[1])
|
||||
d = (t.desc + " " + t.detail).upper()
|
||||
@@ -286,18 +289,28 @@ def classify(t, rules, known_payees):
|
||||
cands = {v for kk, v in rules.items() if " ".join(kk.split()[:2]) == two}
|
||||
if len(cands) == 1: hit = next(iter(cands))
|
||||
if hit: payee, what = hit
|
||||
sm = seed_match(d) # built-in generic merchant rules
|
||||
row = Row(t, "log")
|
||||
if payee: row.payee, row.what = payee, what
|
||||
else:
|
||||
row.payee = " ".join(w.capitalize() for w in (norm(t.desc) or t.desc.upper()).split())[:40] or "Unknown"
|
||||
row.what = "NEW – please categorize"; row.is_new_payee = True; row.needs_review = True
|
||||
row.note = "NEW – review payee/category"
|
||||
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.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"
|
||||
if payee and payee not in known_payees: row.is_new_payee = True
|
||||
if row.is_new_payee and row.payee in known_payees: row.is_new_payee = False
|
||||
if row.is_new_payee:
|
||||
row.category, row.type = ("Other / Unsorted", "Spending")
|
||||
if sm and (cats is None or sm[1] in cats): row.category, row.type = sm[1], sm[2]
|
||||
if t.amount < 0: row.note = (row.note + "; " if row.note else "") + "Refund / credit"
|
||||
if t.extra: row.note = (row.note + "; " if row.note else "") + t.extra
|
||||
return row
|
||||
|
||||
# ---------------------------------------------------------------- workbook access
|
||||
def _categories(wb):
|
||||
return [r[0] for r in wb["Lists"].iter_rows(min_row=4, max_row=40, min_col=5, max_col=5, values_only=True)
|
||||
if r[0] and not str(r[0]).startswith("Slots")]
|
||||
|
||||
def existing_keys(wb):
|
||||
"""Multiset of (account, date, |amount|) already in Bill Log + Excluded Transfers."""
|
||||
keys = collections.Counter()
|
||||
@@ -312,11 +325,12 @@ def plan(wb, statements):
|
||||
rules = learn_rules(wb)
|
||||
lists = wb["Lists"]
|
||||
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:
|
||||
for t in st.txns:
|
||||
rows.append(classify(t, rules, known))
|
||||
rows.append(classify(t, rules, known, cats))
|
||||
new, already = [], 0
|
||||
for r in rows:
|
||||
t = r.txn; k = (t.account, t.date, round(abs(t.amount), 2))
|
||||
@@ -363,7 +377,7 @@ def apply(new_rows, workbook=WORKBOOK):
|
||||
"""Back up, then append new rows. Returns a summary dict."""
|
||||
from openpyxl import load_workbook
|
||||
BACKUPS.mkdir(exist_ok=True)
|
||||
backup = BACKUPS / f"Bills_Tracker_{dt.datetime.now():%Y%m%d_%H%M%S}.xlsx"
|
||||
backup = BACKUPS / f"Bills_Tracker_{dt.datetime.now():%Y%m%d_%H%M%S_%f}.xlsx"
|
||||
shutil.copy2(workbook, backup)
|
||||
wb = load_workbook(workbook)
|
||||
bl, ex, ls, rc, rv = wb["Bill Log"], wb["Excluded Transfers"], wb["Lists"], wb["Reconciliation"], wb["Review Items"]
|
||||
@@ -399,7 +413,7 @@ def apply(new_rows, workbook=WORKBOOK):
|
||||
for row in log_rows:
|
||||
if row.is_new_payee and row.payee not in added_payees and row.payee not in {ls.cell(i, 1).value for i in range(4, lr)}:
|
||||
if lr > 153: break
|
||||
ls.cell(lr, 1).value, ls.cell(lr, 2).value, ls.cell(lr, 3).value = row.payee, "Other / Unsorted", "Spending"
|
||||
ls.cell(lr, 1).value, ls.cell(lr, 2).value, ls.cell(lr, 3).value = row.payee, row.category or "Other / Unsorted", row.type or "Spending"
|
||||
for c in (1, 2, 3): _copy_style(ls.cell(lr - 1, c), ls.cell(lr, c))
|
||||
added_payees.append(row.payee); lr += 1
|
||||
# Reconciliation: statement money-out added per account (statements are verified against their printed totals before import)
|
||||
@@ -425,7 +439,7 @@ def apply(new_rows, workbook=WORKBOOK):
|
||||
n = max([x[0] for x in rv.iter_rows(min_row=5, max_col=1, values_only=True) if isinstance(x[0], int)] + [0]) + 1
|
||||
rr = rv.max_row + 1
|
||||
for c, v in enumerate((n, f"{len(added_payees)} new payee(s) from imported statements: " + ", ".join(added_payees[:12]) + ("…" if len(added_payees) > 12 else ""),
|
||||
"Added to the Lists tab as 'Other / Unsorted' (Spending) so they appear in every summary.",
|
||||
"Added to the Lists tab (category from the built-in merchant rules, or 'Other / Unsorted' when unknown) so they appear in every summary.",
|
||||
"Set the right Category/Type on the Lists tab (and rename the payee if you like)."), 1):
|
||||
rv.cell(rr, c).value = v; _copy_style(rv.cell(rr - 1, c), rv.cell(rr, c))
|
||||
wb.calculation.fullCalcOnLoad = True
|
||||
@@ -435,6 +449,61 @@ def apply(new_rows, workbook=WORKBOOK):
|
||||
tmp.replace(workbook)
|
||||
return dict(backup=str(backup), log_rows=len(log_rows), excluded_rows=len(ex_rows), new_payees=added_payees)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- categorizing existing payees
|
||||
def _backup(workbook):
|
||||
BACKUPS.mkdir(exist_ok=True)
|
||||
b = BACKUPS / f"Bills_Tracker_{dt.datetime.now():%Y%m%d_%H%M%S_%f}.xlsx"
|
||||
shutil.copy2(workbook, b); return b
|
||||
|
||||
def _save_workbook(wb, workbook):
|
||||
wb.calculation.fullCalcOnLoad = True
|
||||
tmp = Path(str(workbook) + ".new"); wb.save(tmp); _restore_charts(workbook, tmp); tmp.replace(workbook)
|
||||
|
||||
def uncategorized(workbook=WORKBOOK):
|
||||
"""Payees on the Lists tab still set to Other / Unsorted, with how much they account for."""
|
||||
from openpyxl import load_workbook
|
||||
wb = load_workbook(workbook, read_only=True)
|
||||
tot, cnt = collections.Counter(), collections.Counter()
|
||||
for r in wb["Bill Log"].iter_rows(min_row=6, max_col=6, values_only=True):
|
||||
if r[0] and r[1]: tot[r[1]] += r[5] or 0; cnt[r[1]] += 1
|
||||
cats = _categories(wb)
|
||||
out = [dict(payee=r[0], category=r[1] or "Other / Unsorted", type=r[2] or "Spending", n=cnt[r[0]], total=round(tot[r[0]], 2))
|
||||
for r in wb["Lists"].iter_rows(min_row=4, max_row=153, max_col=3, values_only=True)
|
||||
if r[0] and (not r[1] or r[1] == "Other / Unsorted")]
|
||||
wb.close()
|
||||
return dict(payees=sorted(out, key=lambda x: -x["total"]), categories=cats, types=["Bill", "Spending"])
|
||||
|
||||
def set_categories(changes, workbook=WORKBOOK):
|
||||
"""changes: [{payee, category, type}] -> written to the Lists tab (backup first)."""
|
||||
from openpyxl import load_workbook
|
||||
b = _backup(workbook); wb = load_workbook(workbook); ls = wb["Lists"]
|
||||
cats = set(_categories(wb)); rows = {ls.cell(r, 1).value: r for r in range(4, 154) if ls.cell(r, 1).value}
|
||||
n = 0
|
||||
for ch in changes:
|
||||
r = rows.get(ch.get("payee"))
|
||||
if r and ch.get("category") in cats and ch.get("type") in ("Bill", "Spending"):
|
||||
ls.cell(r, 2).value, ls.cell(r, 3).value = ch["category"], ch["type"]; n += 1
|
||||
_save_workbook(wb, workbook)
|
||||
return dict(updated=n, backup=b.name)
|
||||
|
||||
def autocategorize(workbook=WORKBOOK):
|
||||
"""Apply the built-in merchant rules to payees still set to Other / Unsorted."""
|
||||
from openpyxl import load_workbook
|
||||
b = _backup(workbook); wb = load_workbook(workbook); ls = wb["Lists"]; cats = set(_categories(wb))
|
||||
descs = collections.defaultdict(list)
|
||||
for r in wb["Bill Log"].iter_rows(min_row=6, max_col=8, values_only=True):
|
||||
if r[0] and r[1] and r[7] and len(descs[r[1]]) < 3: descs[r[1]].append(str(r[7]))
|
||||
n = left = 0
|
||||
for r in range(4, 154):
|
||||
p = ls.cell(r, 1).value
|
||||
if not p or (ls.cell(r, 2).value and ls.cell(r, 2).value != "Other / Unsorted"): continue
|
||||
sm = seed_match(" ".join([p] + descs[p]))
|
||||
if sm and sm[1] in cats: ls.cell(r, 2).value, ls.cell(r, 3).value = sm[1], sm[2]; n += 1
|
||||
else: left += 1
|
||||
_save_workbook(wb, workbook)
|
||||
return dict(updated=n, remaining=left, backup=b.name)
|
||||
|
||||
def summarize(rows):
|
||||
return dict(log=sum(r.kind == "log" for r in rows), excluded=sum(r.kind == "excluded" for r in rows),
|
||||
new_payees=sorted({r.payee for r in rows if r.is_new_payee}))
|
||||
|
||||
Reference in new issue
Block a user