xantorn
/
splitbuddy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

audiot update

Browse Source
main
oscar 2 months ago
parent 6392139b0c
commit 903341f901
1 changed files with 170 additions and 8 deletions
Show Stats Download Patch File Download Diff File

178
main.py
Unescape Escape View File

@ -1,7 +1,7 @@
from __future__ import annotations
import os, sqlite3, csv, io, datetime as dt
from typing import Optional
from flask import Flask, g, request, redirect, url_for, render_template, send_file
import os, sqlite3, csv, io, json, datetime as dt
from typing import Optional, Dict, Any
from flask import Flask, g, request, redirect, url_for, render_template, send_file, jsonify
app = Flask(__name__)
@ -11,6 +11,7 @@ CURRENCY = "₪"
PERSON_A = os.environ.get("SPLITBUDDY_ME", "Me") # you
PERSON_B = os.environ.get("SPLITBUDDY_ROOMIE", "Idan") # roommate
DEFAULT_A_SHARE_PCT = 66.6667 # <-- you pay 2/3 by default
DEFAULT_ACTOR = os.environ.get("SPLITBUDDY_ACTOR", "anon") # who is using this device (optional)
# ----- DB helpers -----
def get_db() -> sqlite3.Connection:
@ -27,19 +28,37 @@ def close_db(_=None):
def init_db():
db = get_db()
# entries
db.execute("""
CREATE TABLE IF NOT EXISTS entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
created_at TEXT NOT NULL,
kind TEXT NOT NULL DEFAULT 'bill', -- 'bill' | 'transfer'
total REAL NOT NULL DEFAULT 0, -- amount
payer TEXT NOT NULL DEFAULT 'A', -- 'A' (you) or 'B' (Idan)
total REAL NOT NULL DEFAULT 0,
payer TEXT NOT NULL DEFAULT 'A', -- 'A' or 'B'
a_share REAL NOT NULL DEFAULT 0.666667, -- your share (0..1) for bills
method TEXT NOT NULL DEFAULT 'cash',
note TEXT
)
""")
# Migrations (no-op if columns already exist)
# audits
db.execute("""
CREATE TABLE IF NOT EXISTS audits (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL, -- UTC ISO timestamp
action TEXT NOT NULL, -- 'add' | 'edit' | 'delete'
entry_id INTEGER, -- affected entry id (may be null for failures)
actor TEXT, -- who (env/config or future auth)
ip TEXT,
user_agent TEXT,
device TEXT, -- parsed (macOS, Windows, iPhone, Android, etc.)
old_row TEXT, -- JSON of previous row (if any)
new_row TEXT -- JSON of new row (if any)
)
""")
# (best-effort) migrations, ignore if exists
for ddl in [
"ALTER TABLE entries ADD COLUMN kind TEXT NOT NULL DEFAULT 'bill'",
"ALTER TABLE entries ADD COLUMN total REAL",
@ -50,6 +69,7 @@ def init_db():
db.execute(ddl)
except sqlite3.OperationalError:
pass
db.execute("""
UPDATE entries
SET kind = COALESCE(kind, 'bill'),
@ -63,6 +83,24 @@ def init_db():
def _now_local_iso_min() -> str:
return dt.datetime.now().replace(second=0, microsecond=0).isoformat(timespec="minutes")
def _now_utc_iso() -> str:
return dt.datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
def _client_ip() -> str:
# works behind simple proxy too
xff = request.headers.get("X-Forwarded-For")
return xff.split(",")[0].strip() if xff else (request.remote_addr or "")
def _guess_device(ua: str) -> str:
u = (ua or "").lower()
if "iphone" in u: return "iPhone"
if "ipad" in u: return "iPad"
if "android" in u: return "Android"
if "mac os x" in u or "macintosh" in u: return "macOS"
if "windows" in u: return "Windows"
if "linux" in u and "android" not in u: return "Linux"
return "Unknown"
class ByMethod:
def __init__(self, cash=0.0, transfer=0.0, other=0.0):
self.cash = cash; self.transfer = transfer; self.other = other
@ -82,6 +120,29 @@ def _delta_for_entry(kind: str, total: float, payer: str, a_share: float) -> flo
paid_by_a = total if payer == "A" else 0.0
return a_share * total - paid_by_a
def _row_to_dict(row: sqlite3.Row | Dict[str, Any] | None) -> Dict[str, Any] | None:
if row is None: return None
return dict(row)
def _audit(action: str, entry_id: Optional[int], old_row: Optional[dict], new_row: Optional[dict]):
db = get_db()
ua = request.headers.get("User-Agent", "")
db.execute("""
INSERT INTO audits (ts, action, entry_id, actor, ip, user_agent, device, old_row, new_row)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
_now_utc_iso(),
action,
entry_id,
DEFAULT_ACTOR,
_client_ip(),
ua,
_guess_device(ua),
json.dumps(old_row, ensure_ascii=False) if old_row is not None else None,
json.dumps(new_row, ensure_ascii=False) if new_row is not None else None,
))
db.commit()
# ----- Routes -----
@app.before_request
def _ensure_db():
@ -131,7 +192,8 @@ def add():
note = (request.form.get("note") or "").strip()
created_at = request.form.get("created_at") or _now_local_iso_min()
if total is None or total < 0: return redirect(url_for("index"))
if total is None or total < 0:
return redirect(url_for("index"))
if payer not in ("A","B"): payer = "A"
if kind not in ("bill","transfer"): kind = "bill"
@ -143,18 +205,66 @@ def add():
a_share = DEFAULT_A_SHARE_PCT / 100.0 # stored but ignored for transfers
db = get_db()
db.execute(
cur = db.execute(
"INSERT INTO entries (created_at, kind, total, payer, a_share, method, note) VALUES (?, ?, ?, ?, ?, ?, ?)",
(created_at, kind, total, payer, a_share, method, note),
)
db.commit()
new_id = cur.lastrowid
new_row = _row_to_dict(db.execute("SELECT * FROM entries WHERE id = ?", (new_id,)).fetchone())
_audit("add", new_id, old_row=None, new_row=new_row)
return redirect(url_for("index"))
@app.post("/delete/<int:entry_id>")
def delete(entry_id: int):
db = get_db()
old = db.execute("SELECT * FROM entries WHERE id = ?", (entry_id,)).fetchone()
db.execute("DELETE FROM entries WHERE id = ?", (entry_id,))
db.commit()
_audit("delete", entry_id, old_row=_row_to_dict(old), new_row=None)
return redirect(url_for("index"))
# Minimal edit endpoint (POST-only). Accepts same fields as /add.
@app.post("/edit/<int:entry_id>")
def edit(entry_id: int):
db = get_db()
old = db.execute("SELECT * FROM entries WHERE id = ?", (entry_id,)).fetchone()
if not old:
_audit("edit", entry_id, old_row=None, new_row=None)
return redirect(url_for("index"))
kind = (request.form.get("kind") or old["kind"]).strip().lower()
payer = (request.form.get("payer") or old["payer"]).strip().upper()
total = request.form.get("total", type=float)
a_share_pct = request.form.get("a_share_pct", type=float)
method = (request.form.get("method") or old["method"]).strip().lower()
note = (request.form.get("note") or old["note"] or "").strip()
created_at = request.form.get("created_at") or old["created_at"]
if total is None:
total = float(old["total"])
if payer not in ("A","B"):
payer = old["payer"]
if kind not in ("bill","transfer"):
kind = old["kind"]
if kind == "bill":
if a_share_pct is None:
a_share = float(old["a_share"])
else:
a_share = max(0.0, min(100.0, a_share_pct)) / 100.0
else:
a_share = float(old["a_share"]) # ignored but stored
db.execute("""
UPDATE entries
SET created_at = ?, kind = ?, total = ?, payer = ?, a_share = ?, method = ?, note = ?
WHERE id = ?
""", (created_at, kind, total, payer, a_share, method, note, entry_id))
db.commit()
new = db.execute("SELECT * FROM entries WHERE id = ?", (entry_id,)).fetchone()
_audit("edit", entry_id, old_row=_row_to_dict(old), new_row=_row_to_dict(new))
return redirect(url_for("index"))
@app.get("/export.csv")
@ -177,6 +287,58 @@ def export_csv():
return send_file(io.BytesIO(buff.read().encode("utf-8")), mimetype="text/csv",
as_attachment=True, download_name="splitbuddy_export.csv")
# ----- Audit viewers -----
@app.get("/audit")
def audit_html():
db = get_db()
rows = db.execute("""
SELECT id, ts, action, entry_id, actor, ip, user_agent, device, old_row, new_row
FROM audits
ORDER BY datetime(ts) DESC, id DESC
LIMIT 500
""").fetchall()
# simple inline HTML (you can move to a template later)
html = ["<!doctype html><meta charset='utf-8'><title>Audit</title>",
"<style>body{font-family:Arial;background:#111;color:#eee}table{width:100%;border-collapse:collapse}th,td{padding:8px;border-bottom:1px solid #333}code{white-space:pre-wrap}</style>",
"<h1>Audit (latest 500)</h1><table><tr><th>ts</th><th>action</th><th>entry</th><th>actor</th><th>ip</th><th>device</th><th>old</th><th>new</th></tr>"]
for r in rows:
html.append(f"<tr><td>{r['ts']}</td><td>{r['action']}</td><td>{r['entry_id']}</td>"
f"<td>{r['actor']}</td><td>{r['ip']}</td><td>{r['device']}</td>"
f"<td><code>{(r['old_row'] or '')[:2000]}</code></td>"
f"<td><code>{(r['new_row'] or '')[:2000]}</code></td></tr>")
html.append("</table>")
return "".join(html)
@app.get("/audit.csv")
def audit_csv():
db = get_db()
rows = db.execute("""
SELECT id, ts, action, entry_id, actor, ip, user_agent, device, old_row, new_row
FROM audits
ORDER BY datetime(ts) DESC, id DESC
""").fetchall()
buff = io.StringIO()
w = csv.writer(buff)
w.writerow(["id","ts","action","entry_id","actor","ip","device","user_agent","old_row","new_row"])
for r in rows:
w.writerow([r["id"], r["ts"], r["action"], r["entry_id"], r["actor"], r["ip"], r["device"], r["user_agent"], r["old_row"] or "", r["new_row"] or ""])
buff.seek(0)
return send_file(io.BytesIO(buff.read().encode("utf-8")), mimetype="text/csv",
as_attachment=True, download_name="splitbuddy_audit.csv")
@app.get("/api/audit")
def audit_api():
db = get_db()
rows = db.execute("""
SELECT id, ts, action, entry_id, actor, ip, user_agent, device, old_row, new_row
FROM audits
ORDER BY datetime(ts) DESC, id DESC
LIMIT 200
""").fetchall()
data = [dict(r) for r in rows]
return jsonify(data)
# ----- Main -----
if __name__ == "__main__":
os.makedirs(os.path.dirname(DB_PATH) or ".", exist_ok=True)
with app.app_context():

Write Preview
Loading…
Cancel
Save