From 1df408fb6cb0e775d2a99765b8d7e2c26cdfa650 Mon Sep 17 00:00:00 2001 From: Jan Tuomi Date: Fri, 26 Jun 2026 21:17:43 +0300 Subject: Initial commit --- .gitignore | 2 + app.py | 444 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 446 insertions(+) create mode 100644 .gitignore create mode 100644 app.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..fc18853 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__ +*.db diff --git a/app.py b/app.py new file mode 100644 index 0000000..e572915 --- /dev/null +++ b/app.py @@ -0,0 +1,444 @@ +#!/usr/bin/env python3 +"""Minimal personal expense tracker. Python 3 stdlib only.""" + +import os +import random +import re +import sqlite3 +import string +from datetime import datetime, timezone +from http.server import BaseHTTPRequestHandler, HTTPServer +from urllib.parse import parse_qs, urlparse + +DB = "expenses.db" +PAGE_SIZE = 20 +BASE_URL = os.environ.get("EXPENSES_BASE_URL", "http://localhost:8000").rstrip("/") + + +def get_db(): + conn = sqlite3.connect(DB) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + return conn + + +def init_db(): + conn = get_db() + conn.executescript(""" + CREATE TABLE IF NOT EXISTS groups ( + id TEXT PRIMARY KEY + ); + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + group_id TEXT NOT NULL REFERENCES groups(id), + name TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS expenses ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + group_id TEXT NOT NULL REFERENCES groups(id), + paid_by INTEGER NOT NULL REFERENCES users(id), + description TEXT NOT NULL, + amount REAL NOT NULL, + created_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS expense_shares ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + expense_id INTEGER NOT NULL REFERENCES expenses(id), + user_id INTEGER NOT NULL REFERENCES users(id), + percentage REAL NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_expenses_group ON expenses(group_id); + CREATE INDEX IF NOT EXISTS idx_shares_expense ON expense_shares(expense_id); + """) + conn.commit() + conn.close() + + +def gen_id(): + return "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) + + +def compute_debts(group_id): + """Compute net balances: positive = owed money, negative = owes money.""" + conn = get_db() + users = conn.execute("SELECT * FROM users WHERE group_id=?", (group_id,)).fetchall() + + balances = {u["id"]: 0.0 for u in users} + + # Credit payers + for row in conn.execute( + "SELECT paid_by, SUM(amount) as total FROM expenses WHERE group_id=? GROUP BY paid_by", + (group_id,), + ).fetchall(): + balances[row["paid_by"]] += row["total"] + + # Debit shares + for row in conn.execute( + "SELECT es.user_id, SUM(e.amount * es.percentage / 100.0) as owed FROM expense_shares es JOIN expenses e ON es.expense_id=e.id WHERE e.group_id=? GROUP BY es.user_id", + (group_id,), + ).fetchall(): + balances[row["user_id"]] -= row["owed"] + + conn.close() + + # Simplify debts + user_map = {u["id"]: u["name"] for u in users} + debts = [] + creditors = [(uid, bal) for uid, bal in balances.items() if bal > 0.01] + debtors = [(uid, -bal) for uid, bal in balances.items() if bal < -0.01] + creditors.sort(key=lambda x: -x[1]) + debtors.sort(key=lambda x: -x[1]) + + i, j = 0, 0 + while i < len(debtors) and j < len(creditors): + amount = min(debtors[i][1], creditors[j][1]) + debts.append((user_map[debtors[i][0]], user_map[creditors[j][0]], amount)) + debtors[i] = (debtors[i][0], debtors[i][1] - amount) + creditors[j] = (creditors[j][0], creditors[j][1] - amount) + if debtors[i][1] < 0.01: + i += 1 + if creditors[j][1] < 0.01: + j += 1 + return debts + + +CSS = """ +*{box-sizing:border-box;margin:0;} +body{font-family:system-ui,sans-serif;max-width:900px;margin:0 auto;padding:1rem;color:#222;background:#fafafa} +h1,h2{margin-bottom:.5rem} +a{color:#0066cc} +form{margin-bottom:1rem} +label{display:block;margin:.3rem 0 .1rem;font-size:1rem} +input[type=text],input[type=number],input[type=date],select{width:100%;padding:.4rem;border:1px solid #ccc;border-radius:4px;font-size:1rem} +input[type=date]{min-width:100%;height:33px;} +button{padding:.5rem 1rem;background:#0066cc;color:#fff;border:none;border-radius:4px;cursor:pointer;font-size:1rem;margin-top:.5rem} +button:hover{background:#0052a3} +.section{background:#fff;border:1px solid #e0e0e0;border-radius:6px;padding:1rem;margin-bottom:1rem} +table{width:100%;border-collapse:collapse;font-size:1rem} +th,td{text-align:left;padding:.2rem .4rem .2rem 0;white-space:nowrap} +tr{border-bottom:1px solid #eee} +td:nth-child(3),th:nth-child(3){max-width:300px;white-space:normal;word-break:break-word} +.shares{display:flex;flex-direction:column;gap:.3rem} +.share-item{display:flex;align-items:center;gap:.3rem} +.share-item input[type=number]{width:4rem;margin-left:auto} +.row{display:flex;gap:.5rem}.row>div{flex:1} +.btn-del{background:none;border:none;color:#c00;cursor:pointer;padding:0;margin:0;} +.actions{white-space:nowrap;vertical-align:middle} +.actions a,.actions button{padding:8px;font-size:1.2rem;text-decoration:none;} +.actions form{margin:0;} +.page-nav{display:flex;gap:1rem;justify-content:center;margin-top:.5rem} +.debt{padding:.3rem 0} +.nav{margin-bottom:1rem;display:flex;align-items:center} +.nav h1{margin:0} +.nav .links{margin-left:auto;display:flex;gap:1rem} +.landing{margin:0 -1rem} +@media(min-width:700px){.landing{display:grid;grid-template-columns:1fr 1fr;gap:1rem;margin:0}.landing .log{grid-column:1/3}} +select{height:33px;} +""" + + +def html(title, body, group_id=None): + nav = "" + if group_id: + nav = f'' + return f"""{title} - {group_id}{nav}{body}""" + + +def escape(s): + return ( + s.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + ) + + +class Handler(BaseHTTPRequestHandler): + def do_GET(self): + parsed = urlparse(self.path) + path = parsed.path.rstrip("/") or "/" + qs = parse_qs(parsed.query) + + if path == "/new": + return self.new_group() + + m = re.match(r"^/([a-z0-9]+)$", path) + if m: + return self.landing(m.group(1), qs) + + m = re.match(r"^/([a-z0-9]+)/settings$", path) + if m: + return self.settings(m.group(1)) + + self.send_response(302) + self.send_header("Location", "/new") + self.end_headers() + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length).decode() + data = parse_qs(body) + + path = urlparse(self.path).path.rstrip("/") + + m = re.match(r"^/([a-z0-9]+)/add$", path) + if m: + return self.add_expense(m.group(1), data) + + m = re.match(r"^/([a-z0-9]+)/add-user$", path) + if m: + return self.add_user(m.group(1), data) + + m = re.match(r"^/([a-z0-9]+)/delete/(\d+)$", path) + if m: + return self.delete_expense(m.group(1), int(m.group(2))) + + self.send_response(400) + self.end_headers() + + def respond(self, code, content): + self.send_response(code) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.end_headers() + self.wfile.write(content.encode()) + + def new_group(self): + gid = gen_id() + conn = get_db() + conn.execute("INSERT INTO groups(id) VALUES(?)", (gid,)) + conn.commit() + conn.close() + self.send_response(302) + self.send_header("Location", f"/{gid}/settings") + self.end_headers() + + def landing(self, gid, qs): + conn = get_db() + group = conn.execute("SELECT * FROM groups WHERE id=?", (gid,)).fetchone() + if not group: + conn.close() + return self.respond(404, html("Not Found", "

Group not found

")) + + users = conn.execute("SELECT * FROM users WHERE group_id=?", (gid,)).fetchall() + page = int(qs.get("page", ["1"])[0]) + offset = (page - 1) * PAGE_SIZE + total = conn.execute( + "SELECT COUNT(*) as c FROM expenses WHERE group_id=?", (gid,) + ).fetchone()["c"] + expenses = conn.execute( + "SELECT e.*, u.name as payer FROM expenses e JOIN users u ON e.paid_by=u.id WHERE e.group_id=? ORDER BY e.created_at DESC LIMIT ? OFFSET ?", + (gid, PAGE_SIZE, offset), + ).fetchall() + + # Get shares for prefilling (from copy param or last expense) + copy_id = qs.get("copy", [None])[0] + copy_exp = None + prefill_shares = {} + prefill_desc = "" + prefill_amount = "" + prefill_payer = None + prefill_date = datetime.now(timezone.utc).strftime("%Y-%m-%d") + + if copy_id: + copy_exp = conn.execute( + "SELECT * FROM expenses WHERE id=? AND group_id=?", (copy_id, gid) + ).fetchone() + if copy_exp: + prefill_desc = copy_exp["description"] + prefill_amount = f"{copy_exp['amount']:.2f}" + prefill_payer = copy_exp["paid_by"] + prefill_date = copy_exp["created_at"][:10] + for s in conn.execute( + "SELECT user_id, percentage FROM expense_shares WHERE expense_id=?", + (copy_exp["id"],), + ).fetchall(): + prefill_shares[s["user_id"]] = s["percentage"] + else: + last_exp = conn.execute( + "SELECT id FROM expenses WHERE group_id=? ORDER BY created_at DESC LIMIT 1", + (gid,), + ).fetchone() + if last_exp: + for s in conn.execute( + "SELECT user_id, percentage FROM expense_shares WHERE expense_id=?", + (last_exp["id"],), + ).fetchall(): + prefill_shares[s["user_id"]] = s["percentage"] + + # Add expense form + form = '

Add expense

' + if not users: + form += ( + "

Add users in settings first.

".format( + gid + ) + ) + else: + form += f'
' + form += '" + form += f'' + form += f'
' + form += f'
' + form += '
' + for u in users: + checked = ( + "checked" + if (not prefill_shares or u["id"] in prefill_shares) + else "" + ) + if u["id"] in prefill_shares: + p = prefill_shares[u["id"]] + pv = int(p) if p == int(p) else f"{p:.2f}" + pct_val = f' value="{pv}"' + else: + pct_val = "" + form += f'' + form += "
" + form += "
" + form += "
" + + # Debt status + debts = compute_debts(gid) + debt_html = '

Balances

' + if not debts: + debt_html += "

All settled up!

" + else: + for frm, to, amt in debts: + debt_html += f'
{escape(frm)} owes {escape(to)} {amt:.2f}
' + debt_html += "
" + + # Expense log + log_html = '

Expenses

' + if expenses: + log_html += "" + for e in expenses: + d = e["created_at"][:10] + log_html += f"" + log_html += ( + f'' + log_html += "
DatePayerDescriptionAmount
{d}{escape(e['payer'])}{escape(e['description'])}{e['amount']:.2f} ' + ) + log_html += f'
" + # Pagination + pages = (total + PAGE_SIZE - 1) // PAGE_SIZE + if pages > 1: + log_html += '" + else: + log_html += "

No expenses yet.

" + log_html += "
" + + body = f'
{form}{debt_html}{log_html}
' + conn.close() + self.respond(200, html("Expenses", body, gid)) + + def settings(self, gid): + conn = get_db() + group = conn.execute("SELECT * FROM groups WHERE id=?", (gid,)).fetchone() + if not group: + conn.close() + return self.respond(404, html("Not Found", "

Group not found

")) + users = conn.execute("SELECT * FROM users WHERE group_id=?", (gid,)).fetchall() + conn.close() + + user_list = "" + for u in users: + user_list += f"
  • {escape(u['name'])}
  • " + + body = f"""

    Members

    + +
    +
    +
    +

    Share Link

    Anyone with this link can view, add and delete expenses in this group:

    + {BASE_URL}/{gid}
    +

    New Group

    Start a separate expense group with its own members and history.

    Create new group
    """ + self.respond(200, html("Settings", body, gid)) + + def add_expense(self, gid, data): + paid_by = int(data["paid_by"][0]) + description = data["description"][0].strip() + amount = round(float(data["amount"][0]), 2) + + conn = get_db() + users = conn.execute("SELECT * FROM users WHERE group_id=?", (gid,)).fetchall() + + # Determine participants and percentages + participants = [] + for u in users: + if f"share_{u['id']}" in data: + pct_val = data.get(f"pct_{u['id']}", [""])[0].strip() + participants.append((u["id"], float(pct_val) if pct_val else 0)) + + if not participants: + conn.close() + self.send_response(302) + self.send_header("Location", f"/{gid}") + self.end_headers() + return + + final = [(uid, pct) for uid, pct in participants if pct > 0] + + now = data.get("date", [datetime.now(timezone.utc).strftime("%Y-%m-%d")])[0] + cur = conn.execute( + "INSERT INTO expenses(group_id, paid_by, description, amount, created_at) VALUES(?,?,?,?,?)", + (gid, paid_by, description, amount, now), + ) + exp_id = cur.lastrowid + for uid, pct in final: + conn.execute( + "INSERT INTO expense_shares(expense_id, user_id, percentage) VALUES(?,?,?)", + (exp_id, uid, pct), + ) + conn.commit() + conn.close() + + self.send_response(302) + self.send_header("Location", f"/{gid}") + self.end_headers() + + def add_user(self, gid, data): + name = data["name"][0].strip() + if name: + conn = get_db() + conn.execute("INSERT INTO users(group_id, name) VALUES(?,?)", (gid, name)) + conn.commit() + conn.close() + self.send_response(302) + self.send_header("Location", f"/{gid}/settings") + self.end_headers() + + def delete_expense(self, gid, exp_id): + conn = get_db() + conn.execute("DELETE FROM expense_shares WHERE expense_id=?", (exp_id,)) + conn.execute("DELETE FROM expenses WHERE id=? AND group_id=?", (exp_id, gid)) + conn.commit() + conn.close() + self.send_response(302) + self.send_header("Location", f"/{gid}") + self.end_headers() + + def address_string(self): + return self.client_address[0] + + def log_message(self, format, *args): + pass # Suppress request logs + + +if __name__ == "__main__": + init_db() + port = 8000 + print(f"Running on http://localhost:{port}") + print("Visit http://localhost:{}/new to create a group".format(port)) + HTTPServer(("", port), Handler).serve_forever() -- cgit v1.3