From 3ad833b4e7318776fbcc76e647f9ef7168bca435 Mon Sep 17 00:00:00 2001 From: SowinskiBraeden Date: Mon, 22 Jun 2026 14:17:10 -0700 Subject: [PATCH] backend overhaul --- .gitignore | 23 +- app.py | 870 +-------------------------------- auth.py | 13 + charts.py | 149 ++++++ config.py | 20 + models.py | 366 ++++++++++++++ routes/__init__.py | 0 routes/admin.py | 583 ++++++++++++++++++++++ routes/public.py | 186 +++++++ services.py | 343 +++++++++++++ stats.py | 857 -------------------------------- templates/admin_dashboard.html | 22 +- templates/base.html | 12 +- templates/leaderboard.html | 8 +- templates/session_detail.html | 10 +- templates/sessions.html | 4 +- tests/test_accounting.py | 21 +- utils.py | 84 ++++ 18 files changed, 1824 insertions(+), 1747 deletions(-) create mode 100644 auth.py create mode 100644 charts.py create mode 100644 config.py create mode 100644 models.py create mode 100644 routes/__init__.py create mode 100644 routes/admin.py create mode 100644 routes/public.py create mode 100644 services.py delete mode 100644 stats.py create mode 100644 utils.py diff --git a/.gitignore b/.gitignore index c6de75f..c6f6011 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,20 @@ +# secrets .env -/__pycache__/* -/.venv/* -/data/* -test/* + +# Python cache — matches any depth, no need to list per-directory +__pycache__/ +*.pyc +*.pyo +*.pyd + +# test artifacts +.pytest_cache/ +.coverage +htmlcov/ + +# virtualenv +.venv/ + +# data + exports +data/ +*.csv diff --git a/app.py b/app.py index db1ce04..5198539 100644 --- a/app.py +++ b/app.py @@ -1,61 +1,17 @@ #!/usr/bin/env python3 from __future__ import annotations -import csv import os -from datetime import datetime, timezone from pathlib import Path -from flask import ( - Flask, - flash, - redirect, - render_template, - request, - send_file, - session, - url_for, -) +from flask import Flask -from stats import ( - apply_rank_changes, - build_leaderboard, - build_session_summaries, - cents_to_dollars, - cumulative_profit_series, - player_session_series, - safe_date_label, - session_breakdown_series, - session_events, - session_label, - session_sort_key, - unique_player_names, -) -from storage import ( - CSV_HEADERS, - EventRow, - append_event, - ensure_data_file, - load_events, - write_events, -) - -BASE_DIR = Path(__file__).resolve().parent -DATA_PATH = BASE_DIR / "data" / "entries.csv" -ENV_PATH = BASE_DIR / ".env" - -ELIGIBLE_MIN_SESSIONS = 3 -ADMIN_EVENT_TYPES = [ - ("buyin", "Buy-in", "var(--accent)"), - ("front", "Front", "var(--warn)"), - ("rollover_in", "Rollover-in", "var(--rolled)"), - ("payout_carry_in", "Payout carry-in", "var(--rolled)"), - ("cashout", "Cashout result", "var(--rank-2)"), - ("paid_out", "Paid out cash", "var(--pos)"), - ("rollover_out", "Rollover-out", "var(--rolled)"), - ("note", "Note", "var(--muted)"), -] -SESSION_MARKER_TYPES = {"session_open", "session_close"} +from auth import is_admin +from config import Config +from routes.admin import admin_bp +from routes.public import public_bp +from storage import ensure_data_file +from utils import cents_to_dollars, safe_date_label def load_local_env(env_path: Path) -> None: @@ -73,811 +29,29 @@ def load_local_env(env_path: Path) -> None: os.environ.setdefault(key, value) -def next_session_id(sessions, session_date: str) -> str: - matching = [session for session in sessions if session.session_date == session_date] +def create_app() -> Flask: + app = Flask(__name__) + app.config.from_object(Config) - highest = 0 - for s in matching: - if s.session_id == session_date: - highest = max(highest, 1) - continue + ensure_data_file(app.config["DATA_PATH"]) - suffix = s.session_id.replace(f"{session_date}-", "") - if suffix.isdigit(): - highest = max(highest, int(suffix)) + app.jinja_env.filters["money"] = cents_to_dollars + app.jinja_env.filters["pretty_date"] = safe_date_label - return f"{session_date}-{highest + 1:02d}" + @app.context_processor + def inject_globals() -> dict: + return {"is_admin": is_admin()} + app.register_blueprint(public_bp) + app.register_blueprint(admin_bp) -def prunable_empty_session_ids(events: list[EventRow], sessions) -> set[str]: - empty_session_ids = { - session.session_id for session in sessions if len(session.entries) == 0 - } - session_events: dict[str, list[EventRow]] = {} + return app - for event in events: - session_id = event["session_id"].strip() or event["session_date"].strip() - if session_id in empty_session_ids: - session_events.setdefault(session_id, []).append(event) - return { - session_id - for session_id, rows in session_events.items() - if rows - and all( - row["event_type"] in SESSION_MARKER_TYPES - and not row["player_name"].strip() - and row["amount_cents"] == 0 - for row in rows - ) - } - - -def pending_payout_carry_items(sessions) -> list[dict[str, object]]: - carried_out: dict[str, int] = {} - carried_in: dict[str, int] = {} - - for session_summary in sessions: - for entry in session_summary.entries: - carried_out[entry.player_name] = ( - carried_out.get(entry.player_name, 0) + entry.rollover_out_cents - ) - carried_in[entry.player_name] = ( - carried_in.get(entry.player_name, 0) - + entry.rollover_in_cents - + entry.payout_carry_in_cents - ) - - items = [] - for player_name, amount_out in carried_out.items(): - pending_cents = amount_out - carried_in.get(player_name, 0) - if pending_cents > 0: - items.append({"player_name": player_name, "amount_cents": pending_cents}) - - return sorted(items, key=lambda item: str(item["player_name"]).casefold()) - - -load_local_env(ENV_PATH) - -app = Flask(__name__) -app.config["SECRET_KEY"] = os.getenv("SECRET_KEY", "change-this-before-deploying") -app.config["ADMIN_USERNAME"] = os.getenv("ADMIN_USERNAME", "admin") -app.config["ADMIN_PASSWORD"] = os.getenv("ADMIN_PASSWORD", "change-me") -app.config["SESSION_COOKIE_HTTPONLY"] = True -app.config["SESSION_COOKIE_SAMESITE"] = "Lax" -app.config["SESSION_COOKIE_NAME"] = "poker_portal_session" - -ensure_data_file(DATA_PATH) -app.jinja_env.filters["money"] = cents_to_dollars -app.jinja_env.filters["pretty_date"] = safe_date_label - - -def is_admin() -> bool: - return bool(session.get("is_admin")) - - -@app.context_processor -def inject_globals() -> dict[str, object]: - return {"is_admin": is_admin()} - - -@app.get("/") -def home() -> str: - return redirect(url_for("leaderboard")) - - -@app.get("/leaderboard") -def leaderboard() -> str: - events = load_events(DATA_PATH) - all_sessions = build_session_summaries(events) - ordered_sessions = sorted(all_sessions, key=session_sort_key) - - session_ids = [s.session_id for s in ordered_sessions] - selected_session_id = request.args.get("through_session", "").strip() - mode = request.args.get("mode", "eligible").strip() - if mode not in ("eligible", "all", "recent"): - mode = "eligible" - label = "" - cutoff_index = len(session_ids) - 1 - - if session_ids: - if selected_session_id in session_ids: - cutoff_index = session_ids.index(selected_session_id) - label = session_label(ordered_sessions[cutoff_index]) - else: - label = session_label(ordered_sessions[cutoff_index]) - - filtered_sessions = ordered_sessions[: cutoff_index + 1] - previous_sessions = ordered_sessions[:cutoff_index] - else: - filtered_sessions = [] - previous_sessions = [] - - # Build full board (all filtered sessions) for rank changes and counts - board = build_leaderboard(filtered_sessions) - previous_board = build_leaderboard(previous_sessions) - board = apply_rank_changes(board, previous_board) - - # Segmented-control counts - eligible_count = sum(1 for p in board if p.sessions_played >= ELIGIBLE_MIN_SESSIONS) - all_count = len(board) - recent_sessions_slice = filtered_sessions[-5:] - recent_count = len({ - entry.player_name - for s in recent_sessions_slice - for entry in s.entries - }) - - # Mode-specific boards - if mode == "recent": - mode_board = build_leaderboard(recent_sessions_slice) - main_board = mode_board - provisional_board = [] - elif mode == "eligible": - main_board = [p for p in board if p.sessions_played >= ELIGIBLE_MIN_SESSIONS] - provisional_board = [p for p in board if p.sessions_played < ELIGIBLE_MIN_SESSIONS] - else: - main_board = board - provisional_board = [] - - chart_data = cumulative_profit_series(filtered_sessions) - cash_paid_out_cents = sum(session.total_paid_out_cents for session in all_sessions) - - return render_template( - "leaderboard.html", - main_board=main_board, - provisional_board=provisional_board, - mode=mode, - eligible_count=eligible_count, - all_count=all_count, - recent_count=recent_count, - eligible_min_sessions=ELIGIBLE_MIN_SESSIONS, - session_count=len(filtered_sessions), - total_session_count=len(all_sessions), - cash_paid_out_cents=cash_paid_out_cents, - chart_data=chart_data, - available_sessions=all_sessions, - selected_session_id=selected_session_id, - selected_session_label=(label if selected_session_id else "Latest session"), - selected_session_date=( - ordered_sessions[cutoff_index].session_date - if selected_session_id and session_ids - else "" - ), - session_label=session_label, - ) - - -@app.get("/sessions") -def sessions(): - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - - return render_template( - "sessions.html", - sessions=sessions, - session_label=session_label, - ) - - -@app.get("/sessions/") -def session_detail(session_id: str) -> str: - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - target_session = next( - ( - session_summary - for session_summary in sessions - if session_summary.session_id == session_id - ), - None, - ) - if target_session is None: - flash("That session was not found.", "error") - return redirect(url_for("sessions")) - - chronological_sessions = sorted(sessions, key=session_sort_key) - chronological_index = next( - index - for index, session in enumerate(chronological_sessions) - if session.session_id == session_id - ) - target_session = chronological_sessions[chronological_index] - session_number = chronological_index + 1 - prev_session = ( - chronological_sessions[chronological_index - 1] - if chronological_index > 0 - else None - ) - next_session = ( - chronological_sessions[chronological_index + 1] - if chronological_index < len(chronological_sessions) - 1 - else None - ) - - return render_template( - "session_detail.html", - session=target_session, - next_session=next_session, - prev_session=prev_session, - session_number=session_number, - raw_events=session_events(events, session_id), - chart_data=session_breakdown_series(target_session), - session_label=session_label, - ) - - -@app.get("/players/") -def player_detail(player_name: str) -> str: - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - board = build_leaderboard(sessions) - player_stats = next( - (player for player in board if player.player_name == player_name), None - ) - if player_stats is None: - flash("That player was not found.", "error") - return redirect(url_for("leaderboard")) - - player_rank = next( - ( - index - for index, ranked_player in enumerate(board, start=1) - if ranked_player.player_name == player_name - ), - None, - ) - chart_data = player_session_series(sessions, player_name) - return render_template( - "player_detail.html", - player=player_stats, - player_rank=player_rank, - chart_data=chart_data, - ) - - -@app.route("/admin/login", methods=["GET", "POST"]) -def admin_login() -> str: - if request.method == "POST": - username = request.form.get("username", "").strip() - password = request.form.get("password", "") - - if ( - username == app.config["ADMIN_USERNAME"] - and password == app.config["ADMIN_PASSWORD"] - ): - session["is_admin"] = True - flash("Admin login successful.", "success") - return redirect(url_for("admin_dashboard")) - - flash("Invalid admin credentials.", "error") - - return render_template("admin_login.html") - - -@app.post("/admin/logout") -def admin_logout() -> str: - session.clear() - flash("Logged out.", "success") - return redirect(url_for("leaderboard")) - - -@app.post("/admin/session-state") -def admin_session_state() -> str: - if not is_admin(): - flash("Admin login required.", "error") - return redirect(url_for("admin_login")) - - session_id = request.form.get("session_id", "").strip() - state = request.form.get("state", "").strip() - - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - by_id = {session.session_id: session for session in sessions} - target = by_id.get(session_id) - - if target is None: - flash("Session not found.", "error") - return redirect(url_for("admin_dashboard")) - - if state not in {"open", "closed"}: - flash("Invalid session state.", "error") - return redirect(url_for("admin_dashboard")) - - append_event( - DATA_PATH, - session_id=target.session_id, - session_date=target.session_date, - amount_cents=0, - player_name="", - event_type="session_open" if state == "open" else "session_close", - note=f"Session marked {state}.", - actor=app.config["ADMIN_USERNAME"], - ) - - flash(f"Session marked {state}.", "success") - return redirect(url_for("admin_dashboard")) - - -@app.get("/admin/export") -@app.get("/admin/export.csv") -def admin_export_csv(): - if not is_admin(): - flash("Admin login required.", "error") - return redirect(url_for("admin_login")) - - export_name = ( - f"entries_export_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.csv" - ) - - return send_file( - DATA_PATH, - as_attachment=True, - download_name=export_name, - mimetype="text/csv", - ) - - -@app.post("/admin/import") -def admin_import_csv(): - if not is_admin(): - flash("Admin login required.", "error") - return redirect(url_for("admin_login")) - - uploaded_file = request.files.get("csv_file") - if uploaded_file is None or uploaded_file.filename == "": - flash("Choose a CSV file to import.", "error") - return redirect(url_for("admin_dashboard")) - - if not uploaded_file.filename.lower().endswith(".csv"): - flash("Only CSV files are allowed.", "error") - return redirect(url_for("admin_dashboard")) - - try: - uploaded_text = uploaded_file.stream.read().decode("utf-8-sig") - except UnicodeDecodeError: - flash("CSV file must be valid UTF-8 text.", "error") - return redirect(url_for("admin_dashboard")) - - uploaded_lines = uploaded_text.splitlines() - if not uploaded_lines: - flash("CSV file is empty.", "error") - return redirect(url_for("admin_dashboard")) - - try: - uploaded_header = next(csv.reader(uploaded_lines)) - except Exception: - flash("Could not read the uploaded CSV header.", "error") - return redirect(url_for("admin_dashboard")) - - if uploaded_header != CSV_HEADERS: - flash("CSV header does not match the current audit format.", "error") - return redirect(url_for("admin_dashboard")) - - try: - existing_ids = {event["id"] for event in load_events(DATA_PATH)} - reader = csv.DictReader(uploaded_lines) - new_rows = [] - skipped_count = 0 - for row in reader: - event_id = row.get("id", "") - if not event_id or event_id in existing_ids: - skipped_count += 1 - continue - - new_rows.append({header: row.get(header, "") for header in CSV_HEADERS}) - existing_ids.add(event_id) - - if new_rows: - with open(DATA_PATH, "a", encoding="utf-8", newline="") as handle: - writer = csv.DictWriter(handle, fieldnames=CSV_HEADERS) - writer.writerows(new_rows) - - flash( - f"Imported {len(new_rows)} new events, skipped {skipped_count} duplicates.", - "success", - ) - except Exception: - flash("Import failed. Existing CSV was not updated.", "error") - - return redirect(url_for("admin_dashboard")) - - -@app.post("/admin/write-off-front") -def admin_write_off_front() -> str: - if not is_admin(): - flash("Admin login required.", "error") - return redirect(url_for("admin_login")) - - debt_key = request.form.get("debt_key", "").strip() - note = request.form.get("note", "").strip() - amount_raw = request.form.get("amount", "0").strip() - - try: - amount_cents = int(round(float(amount_raw) * 100)) - except ValueError: - flash("Amount must be a number.", "error") - return redirect(url_for("admin_dashboard")) - - if amount_cents <= 0: - flash("Write-off amount must be greater than zero.", "error") - return redirect(url_for("admin_dashboard")) - - try: - session_id, player_name = debt_key.split("||", 1) - except ValueError: - flash("Select a valid player debt.", "error") - return redirect(url_for("admin_dashboard")) - - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - target = next( - (session for session in sessions if session.session_id == session_id), - None, - ) - - if target is None: - flash("Session not found.", "error") - return redirect(url_for("admin_dashboard")) - - entry = next( - (entry for entry in target.entries if entry.player_name == player_name), - None, - ) - - if entry is None or entry.player_owes_cents <= 0: - flash("That player does not have an outstanding debt.", "error") - return redirect(url_for("admin_dashboard")) - - if amount_cents > entry.player_owes_cents: - flash( - f"Write-off cannot exceed {cents_to_dollars(entry.player_owes_cents)}.", - "error", - ) - return redirect(url_for("admin_dashboard")) - - append_event( - DATA_PATH, - session_id=target.session_id, - session_date=target.session_date, - player_name=entry.player_name, - event_type="writeoff", - amount_cents=amount_cents, - note=note or "Debt written off.", - actor=app.config["ADMIN_USERNAME"], - ) - - flash("Debt written off.", "success") - return redirect(url_for("admin_dashboard")) - - -@app.post("/admin/collect-front") -def admin_collect_front() -> str: - if not is_admin(): - flash("Admin login required.", "error") - return redirect(url_for("admin_login")) - - debt_key = request.form.get("debt_key", "").strip() - note = request.form.get("note", "").strip() - amount_raw = request.form.get("amount", "0").strip() - - try: - amount_cents = int(round(float(amount_raw) * 100)) - except ValueError: - flash("Amount must be a number.", "error") - return redirect(url_for("admin_dashboard")) - - if amount_cents <= 0: - flash("Collected amount must be greater than zero.", "error") - return redirect(url_for("admin_dashboard")) - - try: - session_id, player_name = debt_key.split("||", 1) - except ValueError: - flash("Select a valid player debt.", "error") - return redirect(url_for("admin_dashboard")) - - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - target = next( - (session for session in sessions if session.session_id == session_id), - None, - ) - - if target is None: - flash("Session not found.", "error") - return redirect(url_for("admin_dashboard")) - - entry = next( - (entry for entry in target.entries if entry.player_name == player_name), - None, - ) - - if entry is None or entry.player_owes_cents <= 0: - flash("That player does not have an outstanding debt.", "error") - return redirect(url_for("admin_dashboard")) - - if amount_cents > entry.player_owes_cents: - flash( - f"Collection cannot exceed {cents_to_dollars(entry.player_owes_cents)}.", - "error", - ) - return redirect(url_for("admin_dashboard")) - - append_event( - DATA_PATH, - session_id=target.session_id, - session_date=target.session_date, - player_name=entry.player_name, - event_type="debt_repayment", - amount_cents=amount_cents, - note=note or "Debt repayment collected.", - actor=app.config["ADMIN_USERNAME"], - ) - - flash("Debt repayment collected.", "success") - return redirect(url_for("admin_dashboard")) - - -@app.post("/admin/open-session") -def admin_open_session() -> str: - if not is_admin(): - flash("Admin login required.", "error") - return redirect(url_for("admin_login")) - - session_date = request.form.get("session_date", "").strip() - if not session_date: - flash("Session date is required.", "error") - return redirect(url_for("admin_dashboard")) - - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - session_id = next_session_id(sessions, session_date) - - append_event( - DATA_PATH, - session_id=session_id, - session_date=session_date, - amount_cents=0, - player_name="", - event_type="session_open", - note="Session opened.", - actor=app.config["ADMIN_USERNAME"], - ) - - flash(f"Opened session {session_id}.", "success") - return redirect(url_for("admin_dashboard")) - - -@app.post("/admin/prune-empty-sessions") -def admin_prune_empty_sessions() -> str: - if not is_admin(): - flash("Admin login required.", "error") - return redirect(url_for("admin_login")) - - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - prunable_session_ids = prunable_empty_session_ids(events, sessions) - - if not prunable_session_ids: - flash("No empty marker-only sessions to prune.", "success") - return redirect(url_for("admin_dashboard")) - - pruned_events = [ - event - for event in events - if (event["session_id"].strip() or event["session_date"].strip()) - not in prunable_session_ids - ] - write_events(DATA_PATH, pruned_events) - - flash( - f"Pruned {len(prunable_session_ids)} empty session" - f"{'s' if len(prunable_session_ids) != 1 else ''}.", - "success", - ) - return redirect(url_for("admin_dashboard")) - - -@app.post("/admin/apply-payout-carry-in") -def admin_apply_payout_carry_in() -> str: - if not is_admin(): - flash("Admin login required.", "error") - return redirect(url_for("admin_login")) - - session_id = request.form.get("session_id", "").strip() - player_name = request.form.get("player_name", "").strip() - - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - by_id = {session.session_id: session for session in sessions} - target = by_id.get(session_id) - - if target is None or target.status != "open": - flash("Select a valid open session.", "error") - return redirect(url_for("admin_dashboard")) - - if not any(entry.player_name == player_name for entry in target.entries): - flash("Add that player to the open session before applying carry-in.", "error") - return redirect(url_for("admin_dashboard")) - - pending_by_player = { - str(item["player_name"]): int(item["amount_cents"]) - for item in pending_payout_carry_items(sessions) - } - amount_cents = pending_by_player.get(player_name, 0) - - if amount_cents <= 0: - flash("No pending payout carry-in found for that player.", "error") - return redirect(url_for("admin_dashboard")) - - append_event( - DATA_PATH, - session_id=target.session_id, - session_date=target.session_date, - player_name=player_name, - event_type="payout_carry_in", - amount_cents=amount_cents, - note="Payout carry-in from prior carry-out.", - actor=app.config["ADMIN_USERNAME"], - ) - - flash( - f"Applied {cents_to_dollars(amount_cents)} payout carry-in for {player_name}.", - "success", - ) - return redirect(url_for("admin_dashboard")) - - -@app.route("/admin", methods=["GET", "POST"]) -def admin_dashboard(): - if not is_admin(): - return redirect(url_for("admin_login")) - - if request.method == "POST": - session_id = request.form.get("session_id", "").strip() - player_name = request.form.get("player_name", "").strip() - event_type = request.form.get("event_type", "").strip() - note = request.form.get("note", "").strip() - - amount_raw = request.form.get("amount", "0").strip() - try: - amount_cents = int(round(float(amount_raw) * 100)) - except ValueError: - flash("Amount must be a number.", "error") - return redirect(url_for("admin_dashboard")) - - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - by_id = {session.session_id: session for session in sessions} - target = by_id.get(session_id) - - if target is None: - flash("Select a valid open session.", "error") - return redirect(url_for("admin_dashboard")) - - if target.status != "open": - flash( - "That session is closed. Reopen it first if you need to add events.", - "error", - ) - return redirect(url_for("admin_dashboard")) - - append_event( - DATA_PATH, - session_id=target.session_id, - session_date=target.session_date, - player_name=player_name, - event_type=event_type, - amount_cents=amount_cents, - note=note, - actor=app.config["ADMIN_USERNAME"], - ) - - flash("Event added.", "success") - return redirect( - url_for( - "admin_dashboard", - session_id=session_id, - player_name=player_name, - event_type=event_type, - ) - ) - - events = load_events(DATA_PATH) - sessions = build_session_summaries(events) - empty_session_count = len(prunable_empty_session_ids(events, sessions)) - pending_carry_items = pending_payout_carry_items(sessions) - pending_carry_total_cents = sum( - int(item["amount_cents"]) for item in pending_carry_items - ) - open_sessions = [session for session in sessions if session.status == "open"] - live_session = open_sessions[0] if open_sessions else None - live_session_player_names = ( - {entry.player_name for entry in live_session.entries} - if live_session is not None - else set() - ) - live_session_no = "" - live_unresolved_count = 0 - if live_session is not None: - live_session_no = f"#{len(sessions):03d}" - live_unresolved_count = sum( - 1 - for entry in live_session.entries - if entry.current_due_cents > 0 or entry.player_owes_cents > 0 - ) - - recent_sessions = sessions - recent_events = list(reversed(events[-20:])) - event_type_colors = {value: color for value, _label, color in ADMIN_EVENT_TYPES} - event_type_labels = {value: label for value, label, _color in ADMIN_EVENT_TYPES} - selected_event_type = request.args.get("event_type", "buyin").strip() - if selected_event_type not in event_type_colors: - selected_event_type = "buyin" - append_form = { - "session_id": request.args.get("session_id", "").strip(), - "player_name": request.args.get("player_name", "").strip(), - "event_type": selected_event_type, - } - debt_entries = [ - {"session": session, "entry": entry} - for session in sessions - for entry in session.entries - if entry.player_owes_cents > 0 - ] - - cash_in_cents = sum(session.total_banker_cash_in_cents for session in sessions) - cash_out_cents = sum(session.total_banker_cash_out_cents for session in sessions) - house_owes_players_cents = ( - sum(session.total_current_due_to_player_cents for session in sessions) - + pending_carry_total_cents - ) - players_owe_house_cents = sum( - session.total_current_due_to_house_cents for session in sessions - ) - admin_totals = { - "cash_in_cents": cash_in_cents, - "cash_out_cents": cash_out_cents, - "house_owes_players_cents": house_owes_players_cents, - "players_owe_house_cents": players_owe_house_cents, - "pending_carry_cents": pending_carry_total_cents, - "collected_cents": sum( - session.total_debt_repayment_cents for session in sessions - ), - "written_off_cents": sum( - session.total_writeoff_cents for session in sessions - ), - "net_book_position_cents": ( - cash_in_cents - - cash_out_cents - + players_owe_house_cents - - house_owes_players_cents - ), - } - - return render_template( - "admin_dashboard.html", - open_sessions=open_sessions, - recent_sessions=recent_sessions, - recent_events=recent_events, - debt_entries=debt_entries, - live_session=live_session, - live_session_player_names=live_session_player_names, - live_session_no=live_session_no, - live_unresolved_count=live_unresolved_count, - pending_carry_items=pending_carry_items, - player_names=unique_player_names(events), - append_form=append_form, - event_types=ADMIN_EVENT_TYPES, - event_type_colors=event_type_colors, - event_type_labels=event_type_labels, - admin_today=datetime.now().date().isoformat(), - session_label=session_label, - admin_totals=admin_totals, - empty_session_count=empty_session_count, - ) +_env_path = Path(__file__).resolve().parent / ".env" +load_local_env(_env_path) +app = create_app() if __name__ == "__main__": app.run(debug=True) diff --git a/auth.py b/auth.py new file mode 100644 index 0000000..19da07b --- /dev/null +++ b/auth.py @@ -0,0 +1,13 @@ +#!/usr/bin/env python3 +"""Authentication helpers. + +Kept in its own module so both app.py (context processor) and routes/admin.py +(route guards) can import is_admin() without circular imports. +""" +from __future__ import annotations + +from flask import session as flask_session + + +def is_admin() -> bool: + return bool(flask_session.get("is_admin")) diff --git a/charts.py b/charts.py new file mode 100644 index 0000000..0a8d074 --- /dev/null +++ b/charts.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""Chart data series builders for the frontend.""" +from __future__ import annotations + +from typing import Any + +from models import SessionEntry, SessionSummary +from utils import net_result_bucket, session_chart_label, session_sort_key + +PLAYER_PALETTE = [ + "#9b8cf0", # --line-1 violet + "#6fc093", # --line-2 green + "#e0b15c", # --line-3 gold + "#cf6f86", # --line-4 pink + "#8f93c2", # --line-5 lavender + "#7cb9e0", + "#f4a261", + "#a78bb0", + "#5cb8a0", + "#e58f3a", + "#84cc16", + "#06b6d4", +] + + +def color_for_name(name: str, names: list[str]) -> str: + try: + index = sorted(names, key=str.casefold).index(name) + except ValueError: + index = abs(hash(name)) + return PLAYER_PALETTE[index % len(PLAYER_PALETTE)] + + +def net_tone(value_cents: int) -> str: + bucket = net_result_bucket(value_cents) + + if bucket == "win": + return "#6fc093" # --pos + if bucket == "loss": + return "#e0758a" # --neg + return "#84828e" # --muted-2 + + +def cumulative_profit_series(sessions: list[SessionSummary]) -> dict[str, object]: + ordered_sessions = sorted(sessions, key=session_sort_key) + + all_players = sorted( + { + entry.player_name + for session in ordered_sessions + for entry in session.entries + }, + key=str.casefold, + ) + + labels = [session_chart_label(session) for session in ordered_sessions] + datasets = [] + + for player_name in all_players: + running_total = 0 + seen_player = False + sessions_played = 0 + data: list[float | None] = [] + + for session in ordered_sessions: + matching_entry = next( + ( + entry + for entry in session.entries + if entry.player_name == player_name + ), + None, + ) + + if matching_entry is None: + data.append(running_total / 100 if seen_player else None) + continue + + seen_player = True + sessions_played += 1 + running_total += matching_entry.net_cents + data.append(round(running_total / 100, 2)) + + datasets.append( + { + "label": player_name, + "data": data, + "sessions_played": sessions_played, + } + ) + + return { + "labels": labels, + "datasets": datasets, + } + + +def player_session_series( + sessions: list[SessionSummary], player_name: str +) -> dict[str, Any]: + ordered_sessions = sorted(sessions, key=session_sort_key) + labels: list[str] = [] + net_values: list[float] = [] + cumulative_values: list[float] = [] + running_total = 0 + + all_player_names = sorted( + { + entry.player_name + for session in ordered_sessions + for entry in session.entries + }, + key=str.casefold, + ) + + for session in ordered_sessions: + matching_entry = next( + (entry for entry in session.entries if entry.player_name == player_name), + None, + ) + if matching_entry is None: + continue + + labels.append(session_chart_label(session)) + net_values.append(round(matching_entry.net_cents / 100, 2)) + running_total += matching_entry.net_cents + cumulative_values.append(round(running_total / 100, 2)) + + return { + "labels": labels, + "color": color_for_name(player_name, all_player_names), + "net_values": net_values, + "net_colors": [net_tone(round(value * 100)) for value in net_values], + "cumulative_values": cumulative_values, + } + + +def session_breakdown_series(session: SessionSummary) -> dict[str, Any]: + ordered_entries = sorted( + session.entries, + key=lambda entry: (entry.net_cents, entry.player_name.casefold()), + reverse=True, + ) + + return { + "labels": [entry.player_name for entry in ordered_entries], + "net_values": [round(entry.net_cents / 100, 2) for entry in ordered_entries], + "net_colors": [net_tone(entry.net_cents) for entry in ordered_entries], + } diff --git a/config.py b/config.py new file mode 100644 index 0000000..442fec3 --- /dev/null +++ b/config.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import os +from pathlib import Path + +BASE_DIR = Path(__file__).resolve().parent +DATA_PATH = BASE_DIR / "data" / "entries.csv" + +ELIGIBLE_MIN_SESSIONS = 3 + + +class Config: + SECRET_KEY: str = os.getenv("SECRET_KEY", "change-this-before-deploying") + ADMIN_USERNAME: str = os.getenv("ADMIN_USERNAME", "admin") + ADMIN_PASSWORD: str = os.getenv("ADMIN_PASSWORD", "change-me") + SESSION_COOKIE_HTTPONLY: bool = True + SESSION_COOKIE_SAMESITE: str = "Lax" + SESSION_COOKIE_NAME: str = "poker_portal_session" + DATA_PATH: Path = DATA_PATH diff --git a/models.py b/models.py new file mode 100644 index 0000000..223e4e5 --- /dev/null +++ b/models.py @@ -0,0 +1,366 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class SessionEntry: + session_id: str + session_date: str + player_name: str + buy_in_cents: int = 0 + front_cents: int = 0 + front_collected_cents: int = 0 + front_writeoff_cents: int = 0 + cash_out_cents: int = 0 + paid_cents: int = 0 + rollover_in_cents: int = 0 + payout_carry_in_cents: int = 0 + rollover_out_cents: int = 0 + notes: list[str] = field(default_factory=list) + + @property + def invested_cents(self) -> int: + # Poker performance: money/chips that entered play. This is intentionally + # broader than real cash because fronts and rollover-ins affect results. + return self.buy_in_cents + self.front_cents + self.rollover_in_cents + + @property + def net_cents(self) -> int: + return self.cash_out_cents - self.invested_cents + + @property + def paid_out_cents(self) -> int: + return self.paid_cents + + @property + def debt_repayment_cents(self) -> int: + return self.front_collected_cents + + @property + def writeoff_cents(self) -> int: + return self.front_writeoff_cents + + @property + def gross_payout_cents(self) -> int: + # Settlement: cashout is a gross chip count/result, not proof that cash + # was paid. Fronts are recovered from the cashout before the player has + # a claim against the house. + return max(self.cash_out_cents - self.front_cents, 0) + + @property + def player_owes_gross_cents(self) -> int: + return max(self.front_cents - self.cash_out_cents, 0) + + @property + def gross_due_to_house_cents(self) -> int: + return self.player_owes_gross_cents + + @property + def settled_to_player_cents(self) -> int: + # Rollover-out resolves the source session's payable even though it is + # not real cash out. The destination session records it as rollover-in. + return self.paid_out_cents + self.rollover_out_cents + + @property + def current_due_to_player_cents(self) -> int: + return max( + self.gross_payout_cents + + self.payout_carry_in_cents + - self.settled_to_player_cents, + 0, + ) + + @property + def settled_to_house_cents(self) -> int: + # Front collections are real cash in. Writeoffs are not cash, but they + # do resolve the receivable. + return self.debt_repayment_cents + self.writeoff_cents + + @property + def current_due_to_house_cents(self) -> int: + return max(self.gross_due_to_house_cents - self.settled_to_house_cents, 0) + + @property + def real_cash_in_cents(self) -> int: + return self.buy_in_cents + self.debt_repayment_cents + + @property + def real_cash_out_cents(self) -> int: + return self.paid_out_cents + + @property + def payout_due_cents(self) -> int: + return self.gross_payout_cents + + @property + def payout_remaining_cents(self) -> int: + return self.current_due_to_player_cents + + @property + def settled_cents(self) -> int: + return self.settled_to_player_cents + + @property + def current_due_cents(self) -> int: + return self.current_due_to_player_cents + + @property + def player_owes_cents(self) -> int: + return self.current_due_to_house_cents + + @property + def raw_player_owes_cents(self) -> int: + return self.gross_due_to_house_cents + + @property + def front_shortfall_cents(self) -> int: + return self.gross_due_to_house_cents + + @property + def overpaid_front_cents(self) -> int: + return 0 + + @property + def front_writeoff_applied_cents(self) -> int: + return min( + self.writeoff_cents, + max(self.gross_due_to_house_cents - self.front_collected_applied_cents, 0), + ) + + @property + def front_collected_applied_cents(self) -> int: + return min(self.debt_repayment_cents, self.gross_due_to_house_cents) + + @property + def front_resolved_cents(self) -> int: + return min(self.settled_to_house_cents, self.gross_due_to_house_cents) + + @property + def payout_status(self) -> str: + if self.current_due_to_house_cents > 0: + return "owes" + if self.writeoff_cents > 0: + return "written_off" + if self.debt_repayment_cents > 0: + return "collected" + if self.gross_payout_cents <= 0: + return "none" + if self.current_due_to_player_cents <= 0: + if self.paid_out_cents < self.gross_payout_cents: + return "settled" + return "paid" + if self.settled_to_player_cents <= 0: + return "unpaid" + return "partial" + + +@dataclass +class SessionSummary: + session_id: str + session_date: str + entries: list[SessionEntry] + status: str = "closed" + opened_at: str = "" + + @property + def is_open(self) -> bool: + return self.status == "open" + + @property + def total_buy_in_cents(self) -> int: + return sum(entry.buy_in_cents for entry in self.entries) + + @property + def total_front_cents(self) -> int: + return sum(entry.front_cents for entry in self.entries) + + @property + def total_rollover_in_cents(self) -> int: + return sum(entry.rollover_in_cents for entry in self.entries) + + @property + def total_payout_carry_in_cents(self) -> int: + return sum(entry.payout_carry_in_cents for entry in self.entries) + + @property + def total_invested_cents(self) -> int: + return sum(entry.invested_cents for entry in self.entries) + + @property + def total_cash_out_cents(self) -> int: + return sum(entry.cash_out_cents for entry in self.entries) + + @property + def total_payout_due_cents(self) -> int: + return self.total_gross_payout_cents + + @property + def total_remaining_cents(self) -> int: + return self.total_current_due_to_player_cents + + @property + def total_net_cents(self) -> int: + return sum(entry.net_cents for entry in self.entries) + + @property + def total_gross_payout_cents(self) -> int: + return sum(entry.gross_payout_cents for entry in self.entries) + + @property + def total_paid_cents(self) -> int: + return self.total_paid_out_cents + + @property + def total_paid_out_cents(self) -> int: + return sum(entry.paid_out_cents for entry in self.entries) + + @property + def total_rollover_out_cents(self) -> int: + return sum(entry.rollover_out_cents for entry in self.entries) + + @property + def total_settled_to_player_cents(self) -> int: + return sum(entry.settled_to_player_cents for entry in self.entries) + + @property + def total_current_due_to_player_cents(self) -> int: + return sum(entry.current_due_to_player_cents for entry in self.entries) + + @property + def total_player_owes_gross_cents(self) -> int: + return sum(entry.gross_due_to_house_cents for entry in self.entries) + + @property + def total_gross_due_to_house_cents(self) -> int: + return self.total_player_owes_gross_cents + + @property + def total_settled_to_house_cents(self) -> int: + return sum(entry.settled_to_house_cents for entry in self.entries) + + @property + def total_current_due_to_house_cents(self) -> int: + return sum(entry.current_due_to_house_cents for entry in self.entries) + + @property + def total_front_writeoff_cents(self) -> int: + return self.total_writeoff_cents + + @property + def total_front_collected_cents(self) -> int: + return self.total_debt_repayment_cents + + @property + def total_debt_repayment_cents(self) -> int: + return sum(entry.debt_repayment_cents for entry in self.entries) + + @property + def total_writeoff_cents(self) -> int: + return sum(entry.writeoff_cents for entry in self.entries) + + @property + def total_cash_in_cents(self) -> int: + return sum(entry.real_cash_in_cents for entry in self.entries) + + @property + def total_real_cash_in_cents(self) -> int: + return self.total_cash_in_cents + + @property + def total_real_cash_out_cents(self) -> int: + return self.total_paid_out_cents + + @property + def total_banker_cash_in_cents(self) -> int: + return self.total_real_cash_in_cents + + @property + def total_banker_cash_out_cents(self) -> int: + return self.total_real_cash_out_cents + + @property + def total_current_due_cents(self) -> int: + return self.total_current_due_to_player_cents + + @property + def total_player_owes_cents(self) -> int: + return self.total_current_due_to_house_cents + + @property + def total_open_balance_cents(self) -> int: + return self.total_current_due_to_player_cents - self.total_current_due_to_house_cents + + @property + def total_net_book_position_cents(self) -> int: + # Banker view: cash currently held, plus collectible receivables, minus + # unpaid player claims. Rollover-outs reduce payables but are not cash. + return ( + self.total_cash_in_cents + - self.total_paid_out_cents + + self.total_current_due_to_house_cents + - self.total_current_due_to_player_cents + ) + + +@dataclass +class PlayerStats: + player_name: str + sessions_played: int + winning_sessions: int + losing_sessions: int + break_even_sessions: int + win_pct: float + avg_win_cents: int + avg_loss_cents: int + biggest_win_cents: int + biggest_loss_cents: int + total_buy_in_cents: int + total_front_cents: int + total_front_collected_cents: int + total_front_writeoff_cents: int + current_player_owes_cents: int + total_rollover_in_cents: int + total_payout_carry_in_cents: int + total_invested_cents: int + total_cash_out_cents: int + total_gross_payout_cents: int + total_paid_cents: int + total_rollover_out_cents: int + current_due_to_player_cents: int + total_net_cents: int + roi_pct: float + current_win_streak: int + current_loss_streak: int + longest_win_streak: int + longest_loss_streak: int + best_session_date: str | None + best_session_net_cents: int + worst_session_date: str | None + worst_session_net_cents: int + rank_change: int = 0 + + @property + def total_debt_repayment_cents(self) -> int: + return self.total_front_collected_cents + + @property + def total_writeoff_cents(self) -> int: + return self.total_front_writeoff_cents + + @property + def current_due_to_house_cents(self) -> int: + return self.current_player_owes_cents + + @property + def total_paid_out_cents(self) -> int: + return self.total_paid_cents + + @property + def total_real_cash_out_cents(self) -> int: + return self.total_paid_out_cents + + @property + def total_real_cash_in_cents(self) -> int: + return self.total_buy_in_cents + self.total_debt_repayment_cents diff --git a/routes/__init__.py b/routes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/routes/admin.py b/routes/admin.py new file mode 100644 index 0000000..08f44ab --- /dev/null +++ b/routes/admin.py @@ -0,0 +1,583 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import csv +from datetime import datetime, timezone + +from flask import ( + Blueprint, + current_app, + flash, + redirect, + render_template, + request, + send_file, + session as flask_session, + url_for, +) + +from auth import is_admin +from services import ( + build_session_summaries, + next_session_id, + pending_payout_carry_items, + prunable_empty_session_ids, + unique_player_names, +) +from storage import CSV_HEADERS, append_event, load_events, write_events +from utils import cents_to_dollars, session_label + +admin_bp = Blueprint("admin", __name__) + +ADMIN_EVENT_TYPES = [ + ("buyin", "Buy-in", "var(--accent)"), + ("front", "Front", "var(--warn)"), + ("rollover_in", "Rollover-in", "var(--rolled)"), + ("payout_carry_in", "Payout carry-in", "var(--rolled)"), + ("cashout", "Cashout result", "var(--rank-2)"), + ("paid_out", "Paid out cash", "var(--pos)"), + ("rollover_out", "Rollover-out", "var(--rolled)"), + ("note", "Note", "var(--muted)"), +] + + +@admin_bp.route("/admin/login", methods=["GET", "POST"]) +def login(): + if request.method == "POST": + username = request.form.get("username", "").strip() + password = request.form.get("password", "") + + if ( + username == current_app.config["ADMIN_USERNAME"] + and password == current_app.config["ADMIN_PASSWORD"] + ): + flask_session["is_admin"] = True + flash("Admin login successful.", "success") + return redirect(url_for("admin.dashboard")) + + flash("Invalid admin credentials.", "error") + + return render_template("admin_login.html") + + +@admin_bp.post("/admin/logout") +def logout(): + flask_session.clear() + flash("Logged out.", "success") + return redirect(url_for("public.leaderboard")) + + +@admin_bp.post("/admin/session-state") +def session_state(): + if not is_admin(): + flash("Admin login required.", "error") + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + session_id = request.form.get("session_id", "").strip() + state = request.form.get("state", "").strip() + + events = load_events(data_path) + sessions = build_session_summaries(events) + by_id = {s.session_id: s for s in sessions} + target = by_id.get(session_id) + + if target is None: + flash("Session not found.", "error") + return redirect(url_for("admin.dashboard")) + + if state not in {"open", "closed"}: + flash("Invalid session state.", "error") + return redirect(url_for("admin.dashboard")) + + append_event( + data_path, + session_id=target.session_id, + session_date=target.session_date, + amount_cents=0, + player_name="", + event_type="session_open" if state == "open" else "session_close", + note=f"Session marked {state}.", + actor=current_app.config["ADMIN_USERNAME"], + ) + + flash(f"Session marked {state}.", "success") + return redirect(url_for("admin.dashboard")) + + +@admin_bp.get("/admin/export") +@admin_bp.get("/admin/export.csv") +def export_csv(): + if not is_admin(): + flash("Admin login required.", "error") + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + export_name = ( + f"entries_export_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.csv" + ) + + return send_file( + data_path, + as_attachment=True, + download_name=export_name, + mimetype="text/csv", + ) + + +@admin_bp.post("/admin/import") +def import_csv(): + if not is_admin(): + flash("Admin login required.", "error") + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + uploaded_file = request.files.get("csv_file") + if uploaded_file is None or uploaded_file.filename == "": + flash("Choose a CSV file to import.", "error") + return redirect(url_for("admin.dashboard")) + + if not uploaded_file.filename.lower().endswith(".csv"): + flash("Only CSV files are allowed.", "error") + return redirect(url_for("admin.dashboard")) + + try: + uploaded_text = uploaded_file.stream.read().decode("utf-8-sig") + except UnicodeDecodeError: + flash("CSV file must be valid UTF-8 text.", "error") + return redirect(url_for("admin.dashboard")) + + uploaded_lines = uploaded_text.splitlines() + if not uploaded_lines: + flash("CSV file is empty.", "error") + return redirect(url_for("admin.dashboard")) + + try: + uploaded_header = next(csv.reader(uploaded_lines)) + except Exception: + flash("Could not read the uploaded CSV header.", "error") + return redirect(url_for("admin.dashboard")) + + if uploaded_header != CSV_HEADERS: + flash("CSV header does not match the current audit format.", "error") + return redirect(url_for("admin.dashboard")) + + try: + existing_ids = {event["id"] for event in load_events(data_path)} + reader = csv.DictReader(uploaded_lines) + new_rows = [] + skipped_count = 0 + for row in reader: + event_id = row.get("id", "") + if not event_id or event_id in existing_ids: + skipped_count += 1 + continue + + new_rows.append({header: row.get(header, "") for header in CSV_HEADERS}) + existing_ids.add(event_id) + + if new_rows: + with open(data_path, "a", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=CSV_HEADERS) + writer.writerows(new_rows) + + flash( + f"Imported {len(new_rows)} new events, skipped {skipped_count} duplicates.", + "success", + ) + except Exception: + flash("Import failed. Existing CSV was not updated.", "error") + + return redirect(url_for("admin.dashboard")) + + +@admin_bp.post("/admin/write-off-front") +def write_off_front(): + if not is_admin(): + flash("Admin login required.", "error") + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + debt_key = request.form.get("debt_key", "").strip() + note = request.form.get("note", "").strip() + amount_raw = request.form.get("amount", "0").strip() + + try: + amount_cents = int(round(float(amount_raw) * 100)) + except ValueError: + flash("Amount must be a number.", "error") + return redirect(url_for("admin.dashboard")) + + if amount_cents <= 0: + flash("Write-off amount must be greater than zero.", "error") + return redirect(url_for("admin.dashboard")) + + try: + session_id, player_name = debt_key.split("||", 1) + except ValueError: + flash("Select a valid player debt.", "error") + return redirect(url_for("admin.dashboard")) + + events = load_events(data_path) + sessions = build_session_summaries(events) + target = next((s for s in sessions if s.session_id == session_id), None) + + if target is None: + flash("Session not found.", "error") + return redirect(url_for("admin.dashboard")) + + entry = next( + (e for e in target.entries if e.player_name == player_name), + None, + ) + + if entry is None or entry.player_owes_cents <= 0: + flash("That player does not have an outstanding debt.", "error") + return redirect(url_for("admin.dashboard")) + + if amount_cents > entry.player_owes_cents: + flash( + f"Write-off cannot exceed {cents_to_dollars(entry.player_owes_cents)}.", + "error", + ) + return redirect(url_for("admin.dashboard")) + + append_event( + data_path, + session_id=target.session_id, + session_date=target.session_date, + player_name=entry.player_name, + event_type="writeoff", + amount_cents=amount_cents, + note=note or "Debt written off.", + actor=current_app.config["ADMIN_USERNAME"], + ) + + flash("Debt written off.", "success") + return redirect(url_for("admin.dashboard")) + + +@admin_bp.post("/admin/collect-front") +def collect_front(): + if not is_admin(): + flash("Admin login required.", "error") + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + debt_key = request.form.get("debt_key", "").strip() + note = request.form.get("note", "").strip() + amount_raw = request.form.get("amount", "0").strip() + + try: + amount_cents = int(round(float(amount_raw) * 100)) + except ValueError: + flash("Amount must be a number.", "error") + return redirect(url_for("admin.dashboard")) + + if amount_cents <= 0: + flash("Collected amount must be greater than zero.", "error") + return redirect(url_for("admin.dashboard")) + + try: + session_id, player_name = debt_key.split("||", 1) + except ValueError: + flash("Select a valid player debt.", "error") + return redirect(url_for("admin.dashboard")) + + events = load_events(data_path) + sessions = build_session_summaries(events) + target = next((s for s in sessions if s.session_id == session_id), None) + + if target is None: + flash("Session not found.", "error") + return redirect(url_for("admin.dashboard")) + + entry = next( + (e for e in target.entries if e.player_name == player_name), + None, + ) + + if entry is None or entry.player_owes_cents <= 0: + flash("That player does not have an outstanding debt.", "error") + return redirect(url_for("admin.dashboard")) + + if amount_cents > entry.player_owes_cents: + flash( + f"Collection cannot exceed {cents_to_dollars(entry.player_owes_cents)}.", + "error", + ) + return redirect(url_for("admin.dashboard")) + + append_event( + data_path, + session_id=target.session_id, + session_date=target.session_date, + player_name=entry.player_name, + event_type="debt_repayment", + amount_cents=amount_cents, + note=note or "Debt repayment collected.", + actor=current_app.config["ADMIN_USERNAME"], + ) + + flash("Debt repayment collected.", "success") + return redirect(url_for("admin.dashboard")) + + +@admin_bp.post("/admin/open-session") +def open_session(): + if not is_admin(): + flash("Admin login required.", "error") + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + session_date = request.form.get("session_date", "").strip() + if not session_date: + flash("Session date is required.", "error") + return redirect(url_for("admin.dashboard")) + + events = load_events(data_path) + sessions = build_session_summaries(events) + new_session_id = next_session_id(sessions, session_date) + + append_event( + data_path, + session_id=new_session_id, + session_date=session_date, + amount_cents=0, + player_name="", + event_type="session_open", + note="Session opened.", + actor=current_app.config["ADMIN_USERNAME"], + ) + + flash(f"Opened session {new_session_id}.", "success") + return redirect(url_for("admin.dashboard")) + + +@admin_bp.post("/admin/prune-empty-sessions") +def prune_empty_sessions(): + if not is_admin(): + flash("Admin login required.", "error") + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + events = load_events(data_path) + sessions = build_session_summaries(events) + prunable_ids = prunable_empty_session_ids(events, sessions) + + if not prunable_ids: + flash("No empty marker-only sessions to prune.", "success") + return redirect(url_for("admin.dashboard")) + + pruned_events = [ + event + for event in events + if (event["session_id"].strip() or event["session_date"].strip()) + not in prunable_ids + ] + write_events(data_path, pruned_events) + + flash( + f"Pruned {len(prunable_ids)} empty session" + f"{'s' if len(prunable_ids) != 1 else ''}.", + "success", + ) + return redirect(url_for("admin.dashboard")) + + +@admin_bp.post("/admin/apply-payout-carry-in") +def apply_payout_carry_in(): + if not is_admin(): + flash("Admin login required.", "error") + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + session_id = request.form.get("session_id", "").strip() + player_name = request.form.get("player_name", "").strip() + + events = load_events(data_path) + sessions = build_session_summaries(events) + by_id = {s.session_id: s for s in sessions} + target = by_id.get(session_id) + + if target is None or target.status != "open": + flash("Select a valid open session.", "error") + return redirect(url_for("admin.dashboard")) + + if not any(e.player_name == player_name for e in target.entries): + flash("Add that player to the open session before applying carry-in.", "error") + return redirect(url_for("admin.dashboard")) + + pending_by_player = { + str(item["player_name"]): int(item["amount_cents"]) + for item in pending_payout_carry_items(sessions) + } + amount_cents = pending_by_player.get(player_name, 0) + + if amount_cents <= 0: + flash("No pending payout carry-in found for that player.", "error") + return redirect(url_for("admin.dashboard")) + + append_event( + data_path, + session_id=target.session_id, + session_date=target.session_date, + player_name=player_name, + event_type="payout_carry_in", + amount_cents=amount_cents, + note="Payout carry-in from prior carry-out.", + actor=current_app.config["ADMIN_USERNAME"], + ) + + flash( + f"Applied {cents_to_dollars(amount_cents)} payout carry-in for {player_name}.", + "success", + ) + return redirect(url_for("admin.dashboard")) + + +@admin_bp.route("/admin", methods=["GET", "POST"]) +def dashboard(): + if not is_admin(): + return redirect(url_for("admin.login")) + + data_path = current_app.config["DATA_PATH"] + + if request.method == "POST": + session_id = request.form.get("session_id", "").strip() + player_name = request.form.get("player_name", "").strip() + event_type = request.form.get("event_type", "").strip() + note = request.form.get("note", "").strip() + + amount_raw = request.form.get("amount", "0").strip() + try: + amount_cents = int(round(float(amount_raw) * 100)) + except ValueError: + flash("Amount must be a number.", "error") + return redirect(url_for("admin.dashboard")) + + events = load_events(data_path) + sessions = build_session_summaries(events) + by_id = {s.session_id: s for s in sessions} + target = by_id.get(session_id) + + if target is None: + flash("Select a valid open session.", "error") + return redirect(url_for("admin.dashboard")) + + if target.status != "open": + flash( + "That session is closed. Reopen it first if you need to add events.", + "error", + ) + return redirect(url_for("admin.dashboard")) + + append_event( + data_path, + session_id=target.session_id, + session_date=target.session_date, + player_name=player_name, + event_type=event_type, + amount_cents=amount_cents, + note=note, + actor=current_app.config["ADMIN_USERNAME"], + ) + + flash("Event added.", "success") + return redirect( + url_for( + "admin.dashboard", + session_id=session_id, + player_name=player_name, + event_type=event_type, + ) + ) + + events = load_events(data_path) + sessions = build_session_summaries(events) + empty_session_count = len(prunable_empty_session_ids(events, sessions)) + pending_carry_items = pending_payout_carry_items(sessions) + pending_carry_total_cents = sum( + int(item["amount_cents"]) for item in pending_carry_items + ) + open_sessions = [s for s in sessions if s.status == "open"] + live_session = open_sessions[0] if open_sessions else None + live_session_player_names = ( + {e.player_name for e in live_session.entries} + if live_session is not None + else set() + ) + live_session_no = "" + live_unresolved_count = 0 + if live_session is not None: + live_session_no = f"#{len(sessions):03d}" + live_unresolved_count = sum( + 1 + for e in live_session.entries + if e.current_due_cents > 0 or e.player_owes_cents > 0 + ) + + recent_events = list(reversed(events[-20:])) + event_type_colors = {value: color for value, _label, color in ADMIN_EVENT_TYPES} + event_type_labels = {value: label for value, label, _color in ADMIN_EVENT_TYPES} + selected_event_type = request.args.get("event_type", "buyin").strip() + if selected_event_type not in event_type_colors: + selected_event_type = "buyin" + append_form = { + "session_id": request.args.get("session_id", "").strip(), + "player_name": request.args.get("player_name", "").strip(), + "event_type": selected_event_type, + } + debt_entries = [ + {"session": s, "entry": e} + for s in sessions + for e in s.entries + if e.player_owes_cents > 0 + ] + + cash_in_cents = sum(s.total_banker_cash_in_cents for s in sessions) + cash_out_cents = sum(s.total_banker_cash_out_cents for s in sessions) + house_owes_players_cents = ( + sum(s.total_current_due_to_player_cents for s in sessions) + + pending_carry_total_cents + ) + players_owe_house_cents = sum( + s.total_current_due_to_house_cents for s in sessions + ) + admin_totals = { + "cash_in_cents": cash_in_cents, + "cash_out_cents": cash_out_cents, + "house_owes_players_cents": house_owes_players_cents, + "players_owe_house_cents": players_owe_house_cents, + "pending_carry_cents": pending_carry_total_cents, + "collected_cents": sum(s.total_debt_repayment_cents for s in sessions), + "written_off_cents": sum(s.total_writeoff_cents for s in sessions), + "net_book_position_cents": ( + cash_in_cents + - cash_out_cents + + players_owe_house_cents + - house_owes_players_cents + ), + } + + return render_template( + "admin_dashboard.html", + open_sessions=open_sessions, + recent_sessions=sessions, + recent_events=recent_events, + debt_entries=debt_entries, + live_session=live_session, + live_session_player_names=live_session_player_names, + live_session_no=live_session_no, + live_unresolved_count=live_unresolved_count, + pending_carry_items=pending_carry_items, + player_names=unique_player_names(events), + append_form=append_form, + event_types=ADMIN_EVENT_TYPES, + event_type_colors=event_type_colors, + event_type_labels=event_type_labels, + admin_today=datetime.now().date().isoformat(), + session_label=session_label, + admin_totals=admin_totals, + empty_session_count=empty_session_count, + ) diff --git a/routes/public.py b/routes/public.py new file mode 100644 index 0000000..93bef28 --- /dev/null +++ b/routes/public.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +from flask import Blueprint, flash, redirect, render_template, request, url_for +from flask import current_app + +from charts import cumulative_profit_series, player_session_series, session_breakdown_series +from config import ELIGIBLE_MIN_SESSIONS +from services import ( + apply_rank_changes, + build_leaderboard, + build_session_summaries, + session_events, +) +from storage import load_events +from utils import session_label, session_sort_key + +public_bp = Blueprint("public", __name__) + + +@public_bp.get("/") +def home(): + return redirect(url_for("public.leaderboard")) + + +@public_bp.get("/leaderboard") +def leaderboard(): + events = load_events(current_app.config["DATA_PATH"]) + all_sessions = build_session_summaries(events) + ordered_sessions = sorted(all_sessions, key=session_sort_key) + + session_ids = [s.session_id for s in ordered_sessions] + selected_session_id = request.args.get("through_session", "").strip() + mode = request.args.get("mode", "eligible").strip() + if mode not in ("eligible", "all", "recent"): + mode = "eligible" + label = "" + cutoff_index = len(session_ids) - 1 + + if session_ids: + if selected_session_id in session_ids: + cutoff_index = session_ids.index(selected_session_id) + label = session_label(ordered_sessions[cutoff_index]) + else: + label = session_label(ordered_sessions[cutoff_index]) + + filtered_sessions = ordered_sessions[: cutoff_index + 1] + previous_sessions = ordered_sessions[:cutoff_index] + else: + filtered_sessions = [] + previous_sessions = [] + + board = build_leaderboard(filtered_sessions) + previous_board = build_leaderboard(previous_sessions) + board = apply_rank_changes(board, previous_board) + + eligible_count = sum(1 for p in board if p.sessions_played >= ELIGIBLE_MIN_SESSIONS) + all_count = len(board) + recent_sessions_slice = filtered_sessions[-5:] + recent_count = len({ + entry.player_name + for s in recent_sessions_slice + for entry in s.entries + }) + + if mode == "recent": + mode_board = build_leaderboard(recent_sessions_slice) + main_board = mode_board + provisional_board = [] + elif mode == "eligible": + main_board = [p for p in board if p.sessions_played >= ELIGIBLE_MIN_SESSIONS] + provisional_board = [p for p in board if p.sessions_played < ELIGIBLE_MIN_SESSIONS] + else: + main_board = board + provisional_board = [] + + chart_data = cumulative_profit_series(filtered_sessions) + cash_paid_out_cents = sum(s.total_paid_out_cents for s in all_sessions) + + return render_template( + "leaderboard.html", + main_board=main_board, + provisional_board=provisional_board, + mode=mode, + eligible_count=eligible_count, + all_count=all_count, + recent_count=recent_count, + eligible_min_sessions=ELIGIBLE_MIN_SESSIONS, + session_count=len(filtered_sessions), + total_session_count=len(all_sessions), + cash_paid_out_cents=cash_paid_out_cents, + chart_data=chart_data, + available_sessions=all_sessions, + selected_session_id=selected_session_id, + selected_session_label=(label if selected_session_id else "Latest session"), + selected_session_date=( + ordered_sessions[cutoff_index].session_date + if selected_session_id and session_ids + else "" + ), + session_label=session_label, + ) + + +@public_bp.get("/sessions") +def sessions(): + events = load_events(current_app.config["DATA_PATH"]) + all_sessions = build_session_summaries(events) + + return render_template( + "sessions.html", + sessions=all_sessions, + session_label=session_label, + ) + + +@public_bp.get("/sessions/") +def session_detail(session_id: str): + events = load_events(current_app.config["DATA_PATH"]) + all_sessions = build_session_summaries(events) + target_session = next( + (s for s in all_sessions if s.session_id == session_id), + None, + ) + if target_session is None: + flash("That session was not found.", "error") + return redirect(url_for("public.sessions")) + + chronological_sessions = sorted(all_sessions, key=session_sort_key) + chronological_index = next( + index + for index, s in enumerate(chronological_sessions) + if s.session_id == session_id + ) + target_session = chronological_sessions[chronological_index] + session_number = chronological_index + 1 + prev_session = ( + chronological_sessions[chronological_index - 1] + if chronological_index > 0 + else None + ) + next_session = ( + chronological_sessions[chronological_index + 1] + if chronological_index < len(chronological_sessions) - 1 + else None + ) + + return render_template( + "session_detail.html", + session=target_session, + next_session=next_session, + prev_session=prev_session, + session_number=session_number, + raw_events=session_events(events, session_id), + chart_data=session_breakdown_series(target_session), + session_label=session_label, + ) + + +@public_bp.get("/players/") +def player_detail(player_name: str): + events = load_events(current_app.config["DATA_PATH"]) + all_sessions = build_session_summaries(events) + board = build_leaderboard(all_sessions) + player_stats = next( + (player for player in board if player.player_name == player_name), None + ) + if player_stats is None: + flash("That player was not found.", "error") + return redirect(url_for("public.leaderboard")) + + player_rank = next( + ( + index + for index, ranked_player in enumerate(board, start=1) + if ranked_player.player_name == player_name + ), + None, + ) + chart_data = player_session_series(all_sessions, player_name) + return render_template( + "player_detail.html", + player=player_stats, + player_rank=player_rank, + chart_data=chart_data, + ) diff --git a/services.py b/services.py new file mode 100644 index 0000000..c364759 --- /dev/null +++ b/services.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python3 +"""Domain logic: session building, leaderboard, and ledger helpers.""" +from __future__ import annotations + +from collections import defaultdict + +from models import PlayerStats, SessionEntry, SessionSummary +from storage import EventRow +from utils import entry_sort_key, net_result_bucket, session_sort_key + +SESSION_MARKER_TYPES = {"session_open", "session_close"} + + +def build_session_summaries(events: list[EventRow]) -> list[SessionSummary]: + grouped: dict[tuple[str, str], SessionEntry] = {} + by_session: dict[str, list[SessionEntry]] = defaultdict(list) + session_status: dict[str, str] = {} + session_dates: dict[str, str] = {} + session_opened_at: dict[str, str] = {} + + for event in events: + session_id = event["session_id"].strip() or event["session_date"].strip() + session_date = event["session_date"].strip() + event_type = event["event_type"].strip() + + if not session_id or not session_date: + continue + + session_dates[session_id] = session_date + session_opened_at.setdefault(session_id, event["created_at"]) + + if event_type == "session_open": + session_status[session_id] = "open" + continue + + if event_type == "session_close": + session_status[session_id] = "closed" + continue + + player_name = event["player_name"].strip() + if not player_name: + continue + + key = (session_id, player_name) + if key not in grouped: + grouped[key] = SessionEntry( + session_id=session_id, + session_date=session_date, + player_name=player_name, + ) + + entry = grouped[key] + + if event_type == "buyin": + entry.buy_in_cents += event["amount_cents"] + elif event_type == "front": + entry.front_cents += event["amount_cents"] + elif event_type == "front_collected": + entry.front_collected_cents += event["amount_cents"] + elif event_type == "debt_repayment": + entry.front_collected_cents += event["amount_cents"] + elif event_type == "front_writeoff": + entry.front_writeoff_cents += event["amount_cents"] + elif event_type == "writeoff": + entry.front_writeoff_cents += event["amount_cents"] + elif event_type == "rollover_in": + entry.rollover_in_cents += event["amount_cents"] + elif event_type == "payout_carry_in": + entry.payout_carry_in_cents += event["amount_cents"] + elif event_type == "cashout": + entry.cash_out_cents += event["amount_cents"] + elif event_type == "paid": + entry.paid_cents += event["amount_cents"] + elif event_type == "paid_out": + entry.paid_cents += event["amount_cents"] + elif event_type == "rollover_out": + entry.rollover_out_cents += event["amount_cents"] + + if event["note"]: + entry.notes.append(event["note"]) + + for entry in grouped.values(): + by_session[entry.session_id].append(entry) + + sessions: list[SessionSummary] = [] + for session_id, session_date in session_dates.items(): + entries = sorted( + by_session.get(session_id, []), + key=lambda entry: entry.player_name.casefold(), + ) + + sessions.append( + SessionSummary( + session_id=session_id, + session_date=session_date, + entries=entries, + status=session_status.get(session_id, "closed"), + opened_at=session_opened_at.get(session_id, ""), + ) + ) + + sessions.sort(key=session_sort_key, reverse=True) + return sessions + + +def summarize_player_runs(entries: list[SessionEntry]) -> dict[str, int | str | None]: + ordered_entries = sorted(entries, key=entry_sort_key) + + longest_win_streak = 0 + longest_loss_streak = 0 + current_run_wins = 0 + current_run_losses = 0 + + best_entry: SessionEntry | None = None + worst_entry: SessionEntry | None = None + + for entry in ordered_entries: + net = entry.net_cents + bucket = net_result_bucket(net) + + if best_entry is None or net > best_entry.net_cents: + best_entry = entry + + if worst_entry is None or net < worst_entry.net_cents: + worst_entry = entry + + if bucket == "win": + current_run_wins += 1 + current_run_losses = 0 + elif bucket == "loss": + current_run_losses += 1 + current_run_wins = 0 + else: + current_run_wins = 0 + current_run_losses = 0 + + longest_win_streak = max(longest_win_streak, current_run_wins) + longest_loss_streak = max(longest_loss_streak, current_run_losses) + + current_win_streak = 0 + current_loss_streak = 0 + + for entry in reversed(ordered_entries): + bucket = net_result_bucket(entry.net_cents) + + if bucket == "win": + if current_loss_streak > 0: + break + current_win_streak += 1 + elif bucket == "loss": + if current_win_streak > 0: + break + current_loss_streak += 1 + else: + break + + return { + "current_win_streak": current_win_streak, + "current_loss_streak": current_loss_streak, + "longest_win_streak": longest_win_streak, + "longest_loss_streak": longest_loss_streak, + "best_session_date": best_entry.session_date if best_entry else None, + "best_session_net_cents": best_entry.net_cents if best_entry else 0, + "worst_session_date": worst_entry.session_date if worst_entry else None, + "worst_session_net_cents": worst_entry.net_cents if worst_entry else 0, + } + + +def build_leaderboard(sessions: list[SessionSummary]) -> list[PlayerStats]: + player_entries: dict[str, list[SessionEntry]] = defaultdict(list) + for session in sessions: + for entry in session.entries: + player_entries[entry.player_name].append(entry) + + leaderboard: list[PlayerStats] = [] + for player_name, entries in player_entries.items(): + nets = [entry.net_cents for entry in entries] + run_summary = summarize_player_runs(entries) + wins = [value for value in nets if net_result_bucket(value) == "win"] + losses = [value for value in nets if net_result_bucket(value) == "loss"] + + sessions_played = len(entries) + winning_sessions = len(wins) + losing_sessions = len(losses) + break_even_sessions = sessions_played - winning_sessions - losing_sessions + win_pct = (winning_sessions / sessions_played * 100) if sessions_played else 0.0 + avg_win = round(sum(wins) / len(wins)) if wins else 0 + avg_loss = ( + round(sum(abs(value) for value in losses) / len(losses)) if losses else 0 + ) + biggest_win = max(wins) if wins else 0 + biggest_loss = min(losses) if losses else 0 + + total_buy_in = sum(entry.buy_in_cents for entry in entries) + total_front = sum(entry.front_cents for entry in entries) + total_front_collected = sum(entry.debt_repayment_cents for entry in entries) + total_front_writeoff = sum(entry.writeoff_cents for entry in entries) + current_player_owes = sum(entry.player_owes_cents for entry in entries) + total_rollover_in = sum(entry.rollover_in_cents for entry in entries) + total_payout_carry_in = sum(entry.payout_carry_in_cents for entry in entries) + total_invested = sum(entry.invested_cents for entry in entries) + total_cash_out = sum(entry.cash_out_cents for entry in entries) + total_gross_payout = sum(entry.gross_payout_cents for entry in entries) + total_paid = sum(entry.paid_out_cents for entry in entries) + total_rollover_out = sum(entry.rollover_out_cents for entry in entries) + current_due_to_player = sum(entry.current_due_to_player_cents for entry in entries) + total_net = sum(entry.net_cents for entry in entries) + roi_pct = (total_net / total_invested * 100) if total_invested else 0.0 + + leaderboard.append( + PlayerStats( + player_name=player_name, + sessions_played=sessions_played, + winning_sessions=winning_sessions, + losing_sessions=losing_sessions, + break_even_sessions=break_even_sessions, + win_pct=win_pct, + avg_win_cents=avg_win, + avg_loss_cents=avg_loss, + biggest_win_cents=biggest_win, + biggest_loss_cents=biggest_loss, + total_buy_in_cents=total_buy_in, + total_front_cents=total_front, + total_front_collected_cents=total_front_collected, + total_front_writeoff_cents=total_front_writeoff, + current_player_owes_cents=current_player_owes, + total_rollover_in_cents=total_rollover_in, + total_payout_carry_in_cents=total_payout_carry_in, + total_invested_cents=total_invested, + total_cash_out_cents=total_cash_out, + total_gross_payout_cents=total_gross_payout, + total_paid_cents=total_paid, + total_rollover_out_cents=total_rollover_out, + current_due_to_player_cents=current_due_to_player, + total_net_cents=total_net, + roi_pct=roi_pct, + current_win_streak=run_summary["current_win_streak"], + current_loss_streak=run_summary["current_loss_streak"], + longest_win_streak=run_summary["longest_win_streak"], + longest_loss_streak=run_summary["longest_loss_streak"], + best_session_date=run_summary["best_session_date"], + best_session_net_cents=run_summary["best_session_net_cents"], + worst_session_date=run_summary["worst_session_date"], + worst_session_net_cents=run_summary["worst_session_net_cents"], + ) + ) + + leaderboard.sort( + key=lambda player: (player.total_net_cents, player.total_cash_out_cents), + reverse=True, + ) + return leaderboard + + +def apply_rank_changes( + current_board: list[PlayerStats], prev_board: list[PlayerStats] +) -> list[PlayerStats]: + prev_ranks = {p.player_name: i for i, p in enumerate(prev_board, start=1)} + + for i, p in enumerate(current_board, start=1): + prev_rank = prev_ranks.get(p.player_name) + + if prev_rank is None: + p.rank_change = 0 + else: + p.rank_change = prev_rank - i + + return current_board + + +def next_session_id(sessions: list[SessionSummary], session_date: str) -> str: + matching = [s for s in sessions if s.session_date == session_date] + + highest = 0 + for s in matching: + if s.session_id == session_date: + highest = max(highest, 1) + continue + + suffix = s.session_id.replace(f"{session_date}-", "") + if suffix.isdigit(): + highest = max(highest, int(suffix)) + + return f"{session_date}-{highest + 1:02d}" + + +def prunable_empty_session_ids( + events: list[EventRow], sessions: list[SessionSummary] +) -> set[str]: + empty_session_ids = {s.session_id for s in sessions if len(s.entries) == 0} + session_event_map: dict[str, list[EventRow]] = {} + + for event in events: + session_id = event["session_id"].strip() or event["session_date"].strip() + if session_id in empty_session_ids: + session_event_map.setdefault(session_id, []).append(event) + + return { + session_id + for session_id, rows in session_event_map.items() + if rows + and all( + row["event_type"] in SESSION_MARKER_TYPES + and not row["player_name"].strip() + and row["amount_cents"] == 0 + for row in rows + ) + } + + +def pending_payout_carry_items(sessions: list[SessionSummary]) -> list[dict[str, object]]: + carried_out: dict[str, int] = {} + carried_in: dict[str, int] = {} + + for session_summary in sessions: + for entry in session_summary.entries: + carried_out[entry.player_name] = ( + carried_out.get(entry.player_name, 0) + entry.rollover_out_cents + ) + carried_in[entry.player_name] = ( + carried_in.get(entry.player_name, 0) + + entry.rollover_in_cents + + entry.payout_carry_in_cents + ) + + items = [] + for player_name, amount_out in carried_out.items(): + pending_cents = amount_out - carried_in.get(player_name, 0) + if pending_cents > 0: + items.append({"player_name": player_name, "amount_cents": pending_cents}) + + return sorted(items, key=lambda item: str(item["player_name"]).casefold()) + + +def session_events(events: list[EventRow], session_id: str) -> list[EventRow]: + return [event for event in events if event["session_id"] == session_id] + + +def unique_player_names(events: list[EventRow]) -> list[str]: + return sorted( + {event["player_name"] for event in events if event["player_name"].strip()}, + key=str.casefold, + ) diff --git a/stats.py b/stats.py deleted file mode 100644 index f89d4b2..0000000 --- a/stats.py +++ /dev/null @@ -1,857 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -from collections import defaultdict -from dataclasses import dataclass, field -from datetime import datetime -from typing import Any - -from markupsafe import Markup - -from storage import EventRow - -PLAYER_PALETTE = [ - "#9b8cf0", # --line-1 violet - "#6fc093", # --line-2 green - "#e0b15c", # --line-3 gold - "#cf6f86", # --line-4 pink - "#8f93c2", # --line-5 lavender - "#7cb9e0", - "#f4a261", - "#a78bb0", - "#5cb8a0", - "#e58f3a", - "#84cc16", - "#06b6d4", -] - -BREAK_EVEN_TOLERANCE_CENTS = 100 - - -def session_sequence_number(session_id: str, session_date: str) -> int: - suffix = session_id.strip().replace(f"{session_date.strip()}-", "", 1) - - if session_id.strip() == session_date.strip(): - return 1 - if suffix.isdigit(): - return int(suffix) - if suffix.lower().startswith("s") and suffix[1:].isdigit(): - return int(suffix[1:]) - return 9999 - - -@dataclass -class SessionEntry: - session_id: str - session_date: str - player_name: str - buy_in_cents: int = 0 - front_cents: int = 0 - front_collected_cents: int = 0 - front_writeoff_cents: int = 0 - cash_out_cents: int = 0 - paid_cents: int = 0 - rollover_in_cents: int = 0 - payout_carry_in_cents: int = 0 - rollover_out_cents: int = 0 - notes: list[str] = field(default_factory=list) - - @property - def invested_cents(self) -> int: - # Poker performance: money/chips that entered play. This is intentionally - # broader than real cash because fronts and rollover-ins affect results. - return self.buy_in_cents + self.front_cents + self.rollover_in_cents - - @property - def net_cents(self) -> int: - return self.cash_out_cents - self.invested_cents - - @property - def paid_out_cents(self) -> int: - return self.paid_cents - - @property - def debt_repayment_cents(self) -> int: - return self.front_collected_cents - - @property - def writeoff_cents(self) -> int: - return self.front_writeoff_cents - - @property - def gross_payout_cents(self) -> int: - # Settlement: cashout is a gross chip count/result, not proof that cash - # was paid. Fronts are recovered from the cashout before the player has - # a claim against the house. - return max(self.cash_out_cents - self.front_cents, 0) - - @property - def player_owes_gross_cents(self) -> int: - return max(self.front_cents - self.cash_out_cents, 0) - - @property - def gross_due_to_house_cents(self) -> int: - return self.player_owes_gross_cents - - @property - def settled_to_player_cents(self) -> int: - # Rollover-out resolves the source session's payable even though it is - # not real cash out. The destination session records it as rollover-in. - return self.paid_out_cents + self.rollover_out_cents - - @property - def current_due_to_player_cents(self) -> int: - return max( - self.gross_payout_cents - + self.payout_carry_in_cents - - self.settled_to_player_cents, - 0, - ) - - @property - def settled_to_house_cents(self) -> int: - # Front collections are real cash in. Writeoffs are not cash, but they - # do resolve the receivable. - return self.debt_repayment_cents + self.writeoff_cents - - @property - def current_due_to_house_cents(self) -> int: - return max(self.gross_due_to_house_cents - self.settled_to_house_cents, 0) - - @property - def real_cash_in_cents(self) -> int: - return self.buy_in_cents + self.debt_repayment_cents - - @property - def real_cash_out_cents(self) -> int: - return self.paid_out_cents - - @property - def payout_due_cents(self) -> int: - return self.gross_payout_cents - - @property - def payout_remaining_cents(self) -> int: - return self.current_due_to_player_cents - - @property - def settled_cents(self) -> int: - return self.settled_to_player_cents - - @property - def current_due_cents(self) -> int: - return self.current_due_to_player_cents - - @property - def player_owes_cents(self) -> int: - return self.current_due_to_house_cents - - @property - def raw_player_owes_cents(self) -> int: - return self.gross_due_to_house_cents - - @property - def front_shortfall_cents(self) -> int: - return self.gross_due_to_house_cents - - @property - def overpaid_front_cents(self) -> int: - return 0 - - @property - def front_writeoff_applied_cents(self) -> int: - return min( - self.writeoff_cents, - max(self.gross_due_to_house_cents - self.front_collected_applied_cents, 0), - ) - - @property - def front_collected_applied_cents(self) -> int: - return min(self.debt_repayment_cents, self.gross_due_to_house_cents) - - @property - def front_resolved_cents(self) -> int: - return min(self.settled_to_house_cents, self.gross_due_to_house_cents) - - @property - def payout_status(self) -> str: - if self.current_due_to_house_cents > 0: - return "owes" - if self.writeoff_cents > 0: - return "written_off" - if self.debt_repayment_cents > 0: - return "collected" - if self.gross_payout_cents <= 0: - return "none" - if self.current_due_to_player_cents <= 0: - if self.paid_out_cents < self.gross_payout_cents: - return "settled" - return "paid" - if self.settled_to_player_cents <= 0: - return "unpaid" - return "partial" - - -@dataclass -class SessionSummary: - session_id: str - session_date: str - entries: list[SessionEntry] - status: str = "closed" - opened_at: str = "" - - @property - def is_open(self) -> bool: - return self.status == "open" - - @property - def total_buy_in_cents(self) -> int: - return sum(entry.buy_in_cents for entry in self.entries) - - @property - def total_front_cents(self) -> int: - return sum(entry.front_cents for entry in self.entries) - - @property - def total_rollover_in_cents(self) -> int: - return sum(entry.rollover_in_cents for entry in self.entries) - - @property - def total_payout_carry_in_cents(self) -> int: - return sum(entry.payout_carry_in_cents for entry in self.entries) - - @property - def total_invested_cents(self) -> int: - return sum(entry.invested_cents for entry in self.entries) - - @property - def total_cash_out_cents(self) -> int: - return sum(entry.cash_out_cents for entry in self.entries) - - @property - def total_payout_due_cents(self) -> int: - return self.total_gross_payout_cents - - @property - def total_remaining_cents(self) -> int: - return self.total_current_due_to_player_cents - - @property - def total_net_cents(self) -> int: - return sum(entry.net_cents for entry in self.entries) - - @property - def total_gross_payout_cents(self) -> int: - return sum(entry.gross_payout_cents for entry in self.entries) - - @property - def total_paid_cents(self) -> int: - return self.total_paid_out_cents - - @property - def total_paid_out_cents(self) -> int: - return sum(entry.paid_out_cents for entry in self.entries) - - @property - def total_rollover_out_cents(self) -> int: - return sum(entry.rollover_out_cents for entry in self.entries) - - @property - def total_settled_to_player_cents(self) -> int: - return sum(entry.settled_to_player_cents for entry in self.entries) - - @property - def total_current_due_to_player_cents(self) -> int: - return sum(entry.current_due_to_player_cents for entry in self.entries) - - @property - def total_player_owes_gross_cents(self) -> int: - return sum(entry.gross_due_to_house_cents for entry in self.entries) - - @property - def total_gross_due_to_house_cents(self) -> int: - return self.total_player_owes_gross_cents - - @property - def total_settled_to_house_cents(self) -> int: - return sum(entry.settled_to_house_cents for entry in self.entries) - - @property - def total_current_due_to_house_cents(self) -> int: - return sum(entry.current_due_to_house_cents for entry in self.entries) - - @property - def total_front_writeoff_cents(self) -> int: - return self.total_writeoff_cents - - @property - def total_front_collected_cents(self) -> int: - return self.total_debt_repayment_cents - - @property - def total_debt_repayment_cents(self) -> int: - return sum(entry.debt_repayment_cents for entry in self.entries) - - @property - def total_writeoff_cents(self) -> int: - return sum(entry.writeoff_cents for entry in self.entries) - - @property - def total_cash_in_cents(self) -> int: - return sum(entry.real_cash_in_cents for entry in self.entries) - - @property - def total_real_cash_in_cents(self) -> int: - return self.total_cash_in_cents - - @property - def total_real_cash_out_cents(self) -> int: - return self.total_paid_out_cents - - @property - def total_banker_cash_in_cents(self) -> int: - return self.total_real_cash_in_cents - - @property - def total_banker_cash_out_cents(self) -> int: - return self.total_real_cash_out_cents - - @property - def total_current_due_cents(self) -> int: - return self.total_current_due_to_player_cents - - @property - def total_player_owes_cents(self) -> int: - return self.total_current_due_to_house_cents - - @property - def total_open_balance_cents(self) -> int: - return self.total_current_due_to_player_cents - self.total_current_due_to_house_cents - - @property - def total_net_book_position_cents(self) -> int: - # Banker view: cash currently held, plus collectible receivables, minus - # unpaid player claims. Rollover-outs reduce payables but are not cash. - return ( - self.total_cash_in_cents - - self.total_paid_out_cents - + self.total_current_due_to_house_cents - - self.total_current_due_to_player_cents - ) - - -@dataclass -class PlayerStats: - player_name: str - sessions_played: int - winning_sessions: int - losing_sessions: int - break_even_sessions: int - win_pct: float - avg_win_cents: int - avg_loss_cents: int - biggest_win_cents: int - biggest_loss_cents: int - total_buy_in_cents: int - total_front_cents: int - total_front_collected_cents: int - total_front_writeoff_cents: int - current_player_owes_cents: int - total_rollover_in_cents: int - total_payout_carry_in_cents: int - total_invested_cents: int - total_cash_out_cents: int - total_gross_payout_cents: int - total_paid_cents: int - total_rollover_out_cents: int - current_due_to_player_cents: int - total_net_cents: int - roi_pct: float - current_win_streak: int - current_loss_streak: int - longest_win_streak: int - longest_loss_streak: int - best_session_date: str | None - best_session_net_cents: int - worst_session_date: str | None - worst_session_net_cents: int - rank_change: int = 0 - - @property - def total_debt_repayment_cents(self) -> int: - return self.total_front_collected_cents - - @property - def total_writeoff_cents(self) -> int: - return self.total_front_writeoff_cents - - @property - def current_due_to_house_cents(self) -> int: - return self.current_player_owes_cents - - @property - def total_paid_out_cents(self) -> int: - return self.total_paid_cents - - @property - def total_real_cash_out_cents(self) -> int: - return self.total_paid_out_cents - - @property - def total_real_cash_in_cents(self) -> int: - return self.total_buy_in_cents + self.total_debt_repayment_cents - - -def apply_rank_changes( - current_board: list[PlayerStats], prev_board: list[PlayerStats] -) -> list[PlayerStats]: - prev_ranks = {p.player_name: i for i, p in enumerate(prev_board, start=1)} - - for i, p in enumerate(current_board, start=1): - prev_rank = prev_ranks.get(p.player_name) - - if prev_rank is None: - p.rank_change = 0 - else: - p.rank_change = prev_rank - i - - return current_board - - -def cents_to_dollars(cents: int) -> Markup: - value = cents / 100 - return Markup(f'${value:,.2f}') - - -def session_sort_key(session: SessionSummary) -> tuple[str, int, str, str]: - session_id = session.session_id.strip() - session_date = session.session_date.strip() - return ( - session_date, - session_sequence_number(session_id, session_date), - session.opened_at, - session_id, - ) - - -def entry_sort_key(entry: SessionEntry) -> tuple[str, int, str]: - return ( - entry.session_date, - session_sequence_number(entry.session_id, entry.session_date), - entry.session_id, - ) - - -def session_display_suffix(session_id: str, session_date: str) -> str: - suffix = session_id.strip().replace(f"{session_date.strip()}-", "", 1) - - if session_id.strip() == session_date.strip(): - return "" - if suffix.isdigit(): - return f"S{int(suffix)}" - if suffix.lower().startswith("s") and suffix[1:].isdigit(): - return f"S{int(suffix[1:])}" - return suffix - - -def session_chart_label(session: SessionSummary) -> str: - suffix = session_display_suffix(session.session_id, session.session_date) - date_label = safe_date_label(session.session_date) - return f"{date_label} · {suffix}" if suffix else date_label - - -def safe_date_label(raw_date: str) -> str: - try: - return datetime.strptime(raw_date, "%Y-%m-%d").strftime("%b %d, %Y") - except ValueError: - return raw_date - - -def session_label(session: SessionSummary) -> str: - return session_chart_label(session) - - -def color_for_name(name: str, names: list[str]) -> str: - try: - index = sorted(names, key=str.casefold).index(name) - except ValueError: - index = abs(hash(name)) - return PLAYER_PALETTE[index % len(PLAYER_PALETTE)] - - -def net_result_bucket(value_cents: int) -> str: - if value_cents > BREAK_EVEN_TOLERANCE_CENTS: - return "win" - if value_cents < -BREAK_EVEN_TOLERANCE_CENTS: - return "loss" - return "even" - - -def net_tone(value_cents: int) -> str: - bucket = net_result_bucket(value_cents) - - if bucket == "win": - return "#6fc093" # --pos - if bucket == "loss": - return "#e0758a" # --neg - return "#84828e" # --muted-2 - - -def build_session_summaries(events: list[EventRow]) -> list[SessionSummary]: - grouped: dict[tuple[str, str], SessionEntry] = {} - by_session: dict[str, list[SessionEntry]] = defaultdict(list) - session_status: dict[str, str] = {} - session_dates: dict[str, str] = {} - session_opened_at: dict[str, str] = {} - - for event in events: - session_id = event["session_id"].strip() or event["session_date"].strip() - session_date = event["session_date"].strip() - event_type = event["event_type"].strip() - - if not session_id or not session_date: - continue - - session_dates[session_id] = session_date - session_opened_at.setdefault(session_id, event["created_at"]) - - if event_type == "session_open": - session_status[session_id] = "open" - continue - - if event_type == "session_close": - session_status[session_id] = "closed" - continue - - player_name = event["player_name"].strip() - if not player_name: - continue - - key = (session_id, player_name) - if key not in grouped: - grouped[key] = SessionEntry( - session_id=session_id, - session_date=session_date, - player_name=player_name, - ) - - entry = grouped[key] - - if event_type == "buyin": - entry.buy_in_cents += event["amount_cents"] - elif event_type == "front": - entry.front_cents += event["amount_cents"] - elif event_type == "front_collected": - entry.front_collected_cents += event["amount_cents"] - elif event_type == "debt_repayment": - entry.front_collected_cents += event["amount_cents"] - elif event_type == "front_writeoff": - entry.front_writeoff_cents += event["amount_cents"] - elif event_type == "writeoff": - entry.front_writeoff_cents += event["amount_cents"] - elif event_type == "rollover_in": - entry.rollover_in_cents += event["amount_cents"] - elif event_type == "payout_carry_in": - entry.payout_carry_in_cents += event["amount_cents"] - elif event_type == "cashout": - entry.cash_out_cents += event["amount_cents"] - elif event_type == "paid": - entry.paid_cents += event["amount_cents"] - elif event_type == "paid_out": - entry.paid_cents += event["amount_cents"] - elif event_type == "rollover_out": - entry.rollover_out_cents += event["amount_cents"] - - if event["note"]: - entry.notes.append(event["note"]) - - for entry in grouped.values(): - by_session[entry.session_id].append(entry) - - sessions: list[SessionSummary] = [] - for session_id, session_date in session_dates.items(): - entries = sorted( - by_session.get(session_id, []), - key=lambda entry: entry.player_name.casefold(), - ) - - sessions.append( - SessionSummary( - session_id=session_id, - session_date=session_date, - entries=entries, - status=session_status.get(session_id, "closed"), - opened_at=session_opened_at.get(session_id, ""), - ) - ) - - sessions.sort(key=session_sort_key, reverse=True) - return sessions - - -def summarize_player_runs(entries: list[SessionEntry]) -> dict[str, int | str | None]: - ordered_entries = sorted(entries, key=entry_sort_key) - - longest_win_streak = 0 - longest_loss_streak = 0 - current_run_wins = 0 - current_run_losses = 0 - - best_entry: SessionEntry | None = None - worst_entry: SessionEntry | None = None - - for entry in ordered_entries: - net = entry.net_cents - bucket = net_result_bucket(net) - - if best_entry is None or net > best_entry.net_cents: - best_entry = entry - - if worst_entry is None or net < worst_entry.net_cents: - worst_entry = entry - - if bucket == "win": - current_run_wins += 1 - current_run_losses = 0 - elif bucket == "loss": - current_run_losses += 1 - current_run_wins = 0 - else: - current_run_wins = 0 - current_run_losses = 0 - - longest_win_streak = max(longest_win_streak, current_run_wins) - longest_loss_streak = max(longest_loss_streak, current_run_losses) - - current_win_streak = 0 - current_loss_streak = 0 - - for entry in reversed(ordered_entries): - bucket = net_result_bucket(entry.net_cents) - - if bucket == "win": - if current_loss_streak > 0: - break - current_win_streak += 1 - elif bucket == "loss": - if current_win_streak > 0: - break - current_loss_streak += 1 - else: - break - - return { - "current_win_streak": current_win_streak, - "current_loss_streak": current_loss_streak, - "longest_win_streak": longest_win_streak, - "longest_loss_streak": longest_loss_streak, - "best_session_date": best_entry.session_date if best_entry else None, - "best_session_net_cents": best_entry.net_cents if best_entry else 0, - "worst_session_date": worst_entry.session_date if worst_entry else None, - "worst_session_net_cents": worst_entry.net_cents if worst_entry else 0, - } - - -def build_leaderboard(sessions: list[SessionSummary]) -> list[PlayerStats]: - player_entries: dict[str, list[SessionEntry]] = defaultdict(list) - for session in sessions: - for entry in session.entries: - player_entries[entry.player_name].append(entry) - - leaderboard: list[PlayerStats] = [] - for player_name, entries in player_entries.items(): - nets = [entry.net_cents for entry in entries] - run_summary = summarize_player_runs(entries) - wins = [value for value in nets if net_result_bucket(value) == "win"] - losses = [value for value in nets if net_result_bucket(value) == "loss"] - - sessions_played = len(entries) - winning_sessions = len(wins) - losing_sessions = len(losses) - break_even_sessions = sessions_played - winning_sessions - losing_sessions - win_pct = (winning_sessions / sessions_played * 100) if sessions_played else 0.0 - avg_win = round(sum(wins) / len(wins)) if wins else 0 - avg_loss = ( - round(sum(abs(value) for value in losses) / len(losses)) if losses else 0 - ) - biggest_win = max(wins) if wins else 0 - biggest_loss = min(losses) if losses else 0 - - total_buy_in = sum(entry.buy_in_cents for entry in entries) - total_front = sum(entry.front_cents for entry in entries) - total_front_collected = sum(entry.debt_repayment_cents for entry in entries) - total_front_writeoff = sum(entry.writeoff_cents for entry in entries) - current_player_owes = sum(entry.player_owes_cents for entry in entries) - total_rollover_in = sum(entry.rollover_in_cents for entry in entries) - total_payout_carry_in = sum(entry.payout_carry_in_cents for entry in entries) - total_invested = sum(entry.invested_cents for entry in entries) - total_cash_out = sum(entry.cash_out_cents for entry in entries) - total_gross_payout = sum(entry.gross_payout_cents for entry in entries) - total_paid = sum(entry.paid_out_cents for entry in entries) - total_rollover_out = sum(entry.rollover_out_cents for entry in entries) - current_due_to_player = sum(entry.current_due_to_player_cents for entry in entries) - total_net = sum(entry.net_cents for entry in entries) - roi_pct = (total_net / total_invested * 100) if total_invested else 0.0 - - leaderboard.append( - PlayerStats( - player_name=player_name, - sessions_played=sessions_played, - winning_sessions=winning_sessions, - losing_sessions=losing_sessions, - break_even_sessions=break_even_sessions, - win_pct=win_pct, - avg_win_cents=avg_win, - avg_loss_cents=avg_loss, - biggest_win_cents=biggest_win, - biggest_loss_cents=biggest_loss, - total_buy_in_cents=total_buy_in, - total_front_cents=total_front, - total_front_collected_cents=total_front_collected, - total_front_writeoff_cents=total_front_writeoff, - current_player_owes_cents=current_player_owes, - total_rollover_in_cents=total_rollover_in, - total_payout_carry_in_cents=total_payout_carry_in, - total_invested_cents=total_invested, - total_cash_out_cents=total_cash_out, - total_gross_payout_cents=total_gross_payout, - total_paid_cents=total_paid, - total_rollover_out_cents=total_rollover_out, - current_due_to_player_cents=current_due_to_player, - total_net_cents=total_net, - roi_pct=roi_pct, - current_win_streak=run_summary["current_win_streak"], - current_loss_streak=run_summary["current_loss_streak"], - longest_win_streak=run_summary["longest_win_streak"], - longest_loss_streak=run_summary["longest_loss_streak"], - best_session_date=run_summary["best_session_date"], - best_session_net_cents=run_summary["best_session_net_cents"], - worst_session_date=run_summary["worst_session_date"], - worst_session_net_cents=run_summary["worst_session_net_cents"], - ) - ) - - leaderboard.sort( - key=lambda player: (player.total_net_cents, player.total_cash_out_cents), - reverse=True, - ) - return leaderboard - - -def cumulative_profit_series(sessions: list[SessionSummary]) -> dict[str, object]: - ordered_sessions = sorted(sessions, key=session_sort_key) - - all_players = sorted( - { - entry.player_name - for session in ordered_sessions - for entry in session.entries - }, - key=str.casefold, - ) - - labels = [session_chart_label(session) for session in ordered_sessions] - datasets = [] - - for player_name in all_players: - running_total = 0 - seen_player = False - sessions_played = 0 - data: list[float | None] = [] - - for session in ordered_sessions: - matching_entry = next( - ( - entry - for entry in session.entries - if entry.player_name == player_name - ), - None, - ) - - if matching_entry is None: - data.append(running_total / 100 if seen_player else None) - continue - - seen_player = True - sessions_played += 1 - running_total += matching_entry.net_cents - data.append(round(running_total / 100, 2)) - - datasets.append( - { - "label": player_name, - "data": data, - "sessions_played": sessions_played, - } - ) - - return { - "labels": labels, - "datasets": datasets, - } - - -def player_session_series( - sessions: list[SessionSummary], player_name: str -) -> dict[str, Any]: - ordered_sessions = sorted(sessions, key=session_sort_key) - labels: list[str] = [] - net_values: list[float] = [] - cumulative_values: list[float] = [] - running_total = 0 - - all_player_names = sorted( - { - entry.player_name - for session in ordered_sessions - for entry in session.entries - }, - key=str.casefold, - ) - - for session in ordered_sessions: - matching_entry = next( - (entry for entry in session.entries if entry.player_name == player_name), - None, - ) - if matching_entry is None: - continue - - labels.append(session_chart_label(session)) - net_values.append(round(matching_entry.net_cents / 100, 2)) - running_total += matching_entry.net_cents - cumulative_values.append(round(running_total / 100, 2)) - - return { - "labels": labels, - "color": color_for_name(player_name, all_player_names), - "net_values": net_values, - "net_colors": [net_tone(round(value * 100)) for value in net_values], - "cumulative_values": cumulative_values, - } - - -def session_breakdown_series(session: SessionSummary) -> dict[str, Any]: - ordered_entries = sorted( - session.entries, - key=lambda entry: (entry.net_cents, entry.player_name.casefold()), - reverse=True, - ) - - return { - "labels": [entry.player_name for entry in ordered_entries], - "net_values": [round(entry.net_cents / 100, 2) for entry in ordered_entries], - "net_colors": [net_tone(entry.net_cents) for entry in ordered_entries], - } - - -def session_events(events: list[EventRow], session_id: str) -> list[EventRow]: - return [event for event in events if event["session_id"] == session_id] - - -def unique_player_names(events: list[EventRow]) -> list[str]: - return sorted( - {event["player_name"] for event in events if event["player_name"].strip()}, - key=str.casefold, - ) diff --git a/templates/admin_dashboard.html b/templates/admin_dashboard.html index 6064d90..975cd46 100644 --- a/templates/admin_dashboard.html +++ b/templates/admin_dashboard.html @@ -45,7 +45,7 @@
-
+

Append ledger event

fast entry @@ -104,7 +104,7 @@ {{ item.player_name }} {{ item.amount_cents | money }} {% if live_session and item.player_name in live_session_player_names %} - + @@ -146,7 +146,7 @@
Seated
{{ live_session.entries|length }}
Open
{{ live_unresolved_count }}
- 0 %}data-close-warning="true"{% endif %}> @@ -167,12 +167,12 @@
{% endif %} - + -
+ @@ -182,9 +182,9 @@

Import / export

Back up or migrate the raw event ledger.

- +
- ↓ Export CSV + ↓ Export CSV
@@ -223,7 +223,7 @@
- +

Collect debt

{% for player in provisional_board %} -

{{ player.player_name }}

diff --git a/templates/session_detail.html b/templates/session_detail.html index 19bad72..7b6f1bd 100644 --- a/templates/session_detail.html +++ b/templates/session_detail.html @@ -4,15 +4,15 @@ {% block content %}
- Sessions + Sessions / {{ session_label(session) }} {% if prev_session %} - ← Prev + ← Prev {% endif %} {% if next_session %} - Next → + Next → {% endif %}
@@ -111,9 +111,9 @@ {% for entry in session.entries %} - + - {{ entry.player_name }} + {{ entry.player_name }} {{ entry.invested_cents | money }} {{ entry.cash_out_cents | money }} diff --git a/templates/sessions.html b/templates/sessions.html index 6cfd272..1fa76f9 100644 --- a/templates/sessions.html +++ b/templates/sessions.html @@ -29,9 +29,9 @@ {% for s in sessions %} - + - {{ session_label(s) }} + {{ session_label(s) }} {{ s.status }} {{ s.entries|length }} diff --git a/tests/test_accounting.py b/tests/test_accounting.py index 0ccb7ee..f4e2c8c 100644 --- a/tests/test_accounting.py +++ b/tests/test_accounting.py @@ -3,13 +3,14 @@ import tempfile import unittest import app as app_module -from stats import ( +from charts import player_session_series +from services import ( build_leaderboard, build_session_summaries, - player_session_series, - session_sort_key, + pending_payout_carry_items, ) from storage import load_events, write_events +from utils import session_sort_key def cents(amount: float) -> int: @@ -202,13 +203,13 @@ class AccountingTests(unittest.TestCase): with tempfile.TemporaryDirectory() as tmpdir: data_path = Path(tmpdir) / "entries.csv" write_events(data_path, rows) - original_path = app_module.DATA_PATH - app_module.DATA_PATH = data_path + original_path = app_module.app.config["DATA_PATH"] + app_module.app.config["DATA_PATH"] = data_path try: sessions = build_session_summaries(load_events(data_path)) self.assertEqual( - app_module.pending_payout_carry_items(sessions), + pending_payout_carry_items(sessions), [{"player_name": "A", "amount_cents": cents(5)}], ) @@ -221,7 +222,7 @@ class AccountingTests(unittest.TestCase): data={"session_id": "s2", "player_name": "A"}, ) finally: - app_module.DATA_PATH = original_path + app_module.app.config["DATA_PATH"] = original_path self.assertEqual(response.status_code, 302) updated_sessions = build_session_summaries(load_events(data_path)) @@ -260,8 +261,8 @@ class AccountingTests(unittest.TestCase): with tempfile.TemporaryDirectory() as tmpdir: data_path = Path(tmpdir) / "entries.csv" write_events(data_path, rows) - original_path = app_module.DATA_PATH - app_module.DATA_PATH = data_path + original_path = app_module.app.config["DATA_PATH"] + app_module.app.config["DATA_PATH"] = data_path try: with app_module.app.test_client() as client: @@ -270,7 +271,7 @@ class AccountingTests(unittest.TestCase): response = client.post("/admin/prune-empty-sessions") finally: - app_module.DATA_PATH = original_path + app_module.app.config["DATA_PATH"] = original_path self.assertEqual(response.status_code, 302) remaining = load_events(data_path) diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..badd71f --- /dev/null +++ b/utils.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Formatting, sorting, and classification helpers shared across the app.""" +from __future__ import annotations + +from datetime import datetime + +from markupsafe import Markup + +from models import SessionEntry, SessionSummary + +BREAK_EVEN_TOLERANCE_CENTS = 100 + + +def cents_to_dollars(cents: int) -> Markup: + value = cents / 100 + return Markup(f'${value:,.2f}') + + +def safe_date_label(raw_date: str) -> str: + try: + return datetime.strptime(raw_date, "%Y-%m-%d").strftime("%b %d, %Y") + except ValueError: + return raw_date + + +def session_sequence_number(session_id: str, session_date: str) -> int: + suffix = session_id.strip().replace(f"{session_date.strip()}-", "", 1) + + if session_id.strip() == session_date.strip(): + return 1 + if suffix.isdigit(): + return int(suffix) + if suffix.lower().startswith("s") and suffix[1:].isdigit(): + return int(suffix[1:]) + return 9999 + + +def session_display_suffix(session_id: str, session_date: str) -> str: + suffix = session_id.strip().replace(f"{session_date.strip()}-", "", 1) + + if session_id.strip() == session_date.strip(): + return "" + if suffix.isdigit(): + return f"S{int(suffix)}" + if suffix.lower().startswith("s") and suffix[1:].isdigit(): + return f"S{int(suffix[1:])}" + return suffix + + +def session_chart_label(session: SessionSummary) -> str: + suffix = session_display_suffix(session.session_id, session.session_date) + date_label = safe_date_label(session.session_date) + return f"{date_label} · {suffix}" if suffix else date_label + + +def session_label(session: SessionSummary) -> str: + return session_chart_label(session) + + +def session_sort_key(session: SessionSummary) -> tuple[str, int, str, str]: + session_id = session.session_id.strip() + session_date = session.session_date.strip() + return ( + session_date, + session_sequence_number(session_id, session_date), + session.opened_at, + session_id, + ) + + +def entry_sort_key(entry: SessionEntry) -> tuple[str, int, str]: + return ( + entry.session_date, + session_sequence_number(entry.session_id, entry.session_date), + entry.session_id, + ) + + +def net_result_bucket(value_cents: int) -> str: + if value_cents > BREAK_EVEN_TOLERANCE_CENTS: + return "win" + if value_cents < -BREAK_EVEN_TOLERANCE_CENTS: + return "loss" + return "even"