#!/usr/bin/env python3 """Local web front end for the statement loader. Run: .venv/bin/python app.py Opens http://127.0.0.1:8765 - listens on this Mac only; nothing is sent anywhere. """ import os, json, secrets, shutil, subprocess, sys, re, calendar, datetime as dt, threading, webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse, parse_qs import statement_loader as L PORT = int(os.environ.get("PORT", "8765")) BIND = os.environ.get("BIND", "127.0.0.1") # in Docker: 0.0.0.0 (publish the port to 127.0.0.1 only) EXTRA_HOSTS = {h.strip() for h in os.environ.get("ALLOWED_HOSTS", "").split(",") if h.strip()} TOKEN = secrets.token_urlsafe(24) LOCK = threading.Lock() HERE = L.HERE # code and pages DATA = L.DATA # statements and workbook def eom(y, m): return calendar.monthrange(y, m)[1] 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) if not by: return dict(months=[], accounts=[]) lo = min(s.start or s.end for s in sts); months = [] y, m = lo.year, lo.month while (y, m) <= (today.year, today.month): months.append(f"{y}-{m:02d}"); m += 1 if m > 12: y, m = y + 1, 1 out = [] for acct, lst in by.items(): csv_only = all(s.kind == "csv" for s in lst) cells, first = {}, min((s.start or s.end) for s in lst) for mo in months: y, m = int(mo[:4]), int(mo[5:]) if csv_only: have = any((s.start.year, s.start.month) <= (y, m) <= (s.end.year, s.end.month) for s in lst) cells[mo] = "have" if have else ("future" if (y, m) > (max(s.end for s in lst).year, max(s.end for s in lst).month) else ("none" if (y, m) < (first.year, first.month) else "missing")) continue have = any((s.end.year, s.end.month) == (y, m) for s in lst) due = dt.date(y, m, cfg.get(acct) or eom(y, m)) started = (y, m) >= (min(s.end for s in lst).year, min(s.end for s in lst).month) cells[mo] = "have" if have else ("none" if not started else ("future" if due > today else "missing")) last = max(s.end for s in lst) if csv_only: nxt = f"Next download: from {last + dt.timedelta(days=1):%b %-d, %Y}" else: ny, nm = (last.year, last.month + 1) if last.month < 12 else (last.year + 1, 1) 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 "") 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) def state(): res = L.run_preview(DATA) new = res["new"] rows = [dict(date=f"{r.txn.date:%Y-%m-%d}", account=r.txn.account, desc=r.txn.desc[:60], amount=r.txn.amount, kind=r.kind, payee=r.payee, what=r.what, reason=r.reason, new_payee=r.is_new_payee) for r in new] sm = L.summarize(new) return dict(coverage=coverage(res["statements"]), skipped=[dict(file=f, why=w) for f, w in res["skipped"]], duplicates=[dict(file=a, same_as=b) for a, b in res["dups"]], bad=[dict(file=f, parsed=p, stated=s) for f, p, s, ok in res["checks"] if not ok], already=res["already"], new=rows, summary=sm, statements=len(res["statements"]), excel_open=(DATA / "~$Bills_Tracker.xlsx").exists() or (DATA / ".~lock.Bills_Tracker.xlsx#").exists(), backups=sorted(p.name for p in L.BACKUPS.glob("*.xlsx"))[-5:] if L.BACKUPS.exists() else []) _DATA = {} def workbook_data(): """Read-only snapshot of the Bill Log for the web dashboard (computed from raw cells, not cached formulas).""" from openpyxl import load_workbook key = L.WORKBOOK.stat().st_mtime_ns if _DATA.get("key") == key: return _DATA["val"] wb = load_workbook(L.WORKBOOK, read_only=True) pay = {} for r in wb["Lists"].iter_rows(min_row=4, max_row=153, max_col=3, values_only=True): if r[0]: pay[r[0]] = (r[1] or "Other / Unsorted", r[2] or "Spending") cats = [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")] tx = [] for r in wb["Bill Log"].iter_rows(min_row=6, max_col=9, values_only=True): if isinstance(r[0], dt.datetime) and r[1] and r[5] is not None: c, t = pay.get(r[1], ("Other / Unsorted", "Spending")) tx.append([f"{r[0]:%Y-%m-%d}", r[1], c, t, r[4] or "", round(float(r[5]), 2), r[6] or "", r[8] or ""]) ex = sum(1 for r in wb["Excluded Transfers"].iter_rows(min_row=5, max_col=1, values_only=True) if isinstance(r[0], dt.datetime)) wb.close() _DATA["key"], _DATA["val"] = key, dict(tx=tx, categories=cats, excluded=ex, updated=f"{dt.datetime.fromtimestamp(L.WORKBOOK.stat().st_mtime):%b %-d, %Y %-I:%M %p}") return _DATA["val"] def safe_name(name): name = re.sub(r"[^\w.\- ]", "_", Path(name).name) if Path(name).suffix.lower() not in (".pdf", ".csv"): raise ValueError("Only PDF or CSV files can be added.") p = DATA / name; i = 1 while p.exists(): p = DATA / f"{Path(name).stem}_{i}{Path(name).suffix}"; i += 1 return p class H(BaseHTTPRequestHandler): def log_message(self, *a): pass def _send(self, code, body, ctype="application/json"): b = body if isinstance(body, bytes) else json.dumps(body).encode() self.send_response(code); self.send_header("Content-Type", ctype); self.send_header("Content-Length", str(len(b))) self.send_header("Cache-Control", "no-store"); self.end_headers(); self.wfile.write(b) def _ok_host(self): if "*" in EXTRA_HOSTS: return True # LAN/NAS use: any hostname or IP (the app has no login, so keep it on a trusted network) h = self.headers.get("Host", ""); return h.rsplit(":", 1)[0] in {"127.0.0.1", "localhost"} | EXTRA_HOSTS or h in EXTRA_HOSTS def do_GET(self): if not self._ok_host(): return self._send(403, {"error": "forbidden"}) u = urlparse(self.path) if u.path == "/": return self._send(200, PAGE.replace("__TOKEN__", TOKEN).encode(), "text/html; charset=utf-8") if u.path == "/dashboard": f = HERE / "app_dashboard.html" return self._send(200, f.read_bytes() if f.exists() else b"app_dashboard.html missing", "text/html; charset=utf-8") if u.path == "/api/data": with LOCK: try: return self._send(200, workbook_data()) except Exception as e: return self._send(500, {"error": str(e)}) if u.path == "/api/state": with LOCK: try: return self._send(200, state()) except Exception as e: return self._send(500, {"error": str(e)}) self._send(404, {"error": "not found"}) def do_POST(self): if not self._ok_host() or self.headers.get("X-Token") != TOKEN: return self._send(403, {"error": "forbidden"}) u = urlparse(self.path); n = int(self.headers.get("Content-Length") or 0) try: if u.path == "/api/upload": name = parse_qs(u.query).get("name", [""])[0]; data = self.rfile.read(n) with LOCK: p = safe_name(name); p.write_bytes(data) return self._send(200, {"saved": p.name}) self.rfile.read(n) if u.path == "/api/import": with LOCK: res = L.run_preview(DATA) if not res["new"]: return self._send(200, {"message": "Nothing new to import."}) out = L.apply(res["new"]) return self._send(200, {"message": f"Added {out['log_rows']} rows to the Bill Log and {out['excluded_rows']} to Excluded Transfers.", "backup": Path(out["backup"]).name, "new_payees": out["new_payees"]}) if u.path == "/api/undo": with LOCK: b = sorted(L.BACKUPS.glob("*.xlsx")) if L.BACKUPS.exists() else [] if not b: return self._send(200, {"message": "No backups to restore."}) keep = L.BACKUPS / f"before_undo_{dt.datetime.now():%Y%m%d_%H%M%S}.xlsx" shutil.copy2(L.WORKBOOK, keep); shutil.copy2(b[-1], L.WORKBOOK) return self._send(200, {"message": f"Restored the workbook from {b[-1].name}."}) if u.path == "/api/open": if sys.platform != "darwin": return self._send(200, {"message": "Open Bills_Tracker.xlsx from the data folder on the host."}) subprocess.run(["open", str(L.WORKBOOK)], check=False) return self._send(200, {"message": "Opening the workbook…"}) except Exception as e: return self._send(400, {"error": str(e)}) self._send(404, {"error": "not found"}) PAGE = (HERE / "app_page.html").read_text() if (HERE / "app_page.html").exists() else "

app_page.html missing

" if __name__ == "__main__": if not L.WORKBOOK.exists(): print(f"WARNING: {L.WORKBOOK} not found - put Bills_Tracker.xlsx in the data folder.") srv = ThreadingHTTPServer((BIND, PORT), H) url = f"http://127.0.0.1:{PORT}/" print(f"Statement loader running at {url} (Ctrl+C to stop)") if "--no-browser" not in sys.argv and not os.environ.get("DATA_DIR"): threading.Timer(0.8, lambda: webbrowser.open(url)).start() try: srv.serve_forever() except KeyboardInterrupt: print("\nStopped.")