SowinskiBraeden/boker
1
0 Fork 0
Code Issues Packages Projects Releases Wiki Activity

backend overhaul

Browse Source
This commit is contained in:
SowinskiBraeden committed 2026-06-22 14:17:10 -07:00
1 parent 70e62342ee
commit 3ad833b4e7
18 files changed
+1823 -1746
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+19 -4
View File
@@ -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
app.py
+21 -847
View File
@@ -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}"
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]] = {}
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]:
@app.context_processor
def inject_globals() -> dict:
return {"is_admin": is_admin()}
app.register_blueprint(public_bp)
app.register_blueprint(admin_bp)
@app.get("/")
def home() -> str:
return redirect(url_for("leaderboard"))
return app
@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/<session_id>")
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/<player_name>")
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)
auth.py
+13
View File
@@ -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"))
charts.py
+149
View File
@@ -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],
}
config.py
+20
View File
@@ -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
models.py
+366
View File
@@ -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
routes/__init__.py
View File
Whitespace-only changes.
routes/admin.py
+583
View File
@@ -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,
)
routes/public.py
+186
View File
@@ -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/<session_id>")
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/<player_name>")
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,
)
services.py
+343
View File
@@ -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,
)
stats.py
-857
View File
@@ -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'<span class="currency-symbol">$</span>{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,
)
templates/admin_dashboard.html
+11 -11
View File
@@ -45,7 +45,7 @@
<!-- Admin workspace -->
<div class="admin-workspace">
<form class="panel append" method="post" action="{{ url_for('admin_dashboard') }}">
<form class="panel append" method="post" action="{{ url_for('admin.dashboard') }}">
<div class="panel__head">
<h2 class="panel__title">Append ledger event</h2>
<span class="panel__tag">fast entry</span>
@@ -104,7 +104,7 @@
<span>{{ item.player_name }}</span>
<strong>{{ item.amount_cents | money }}</strong>
{% if live_session and item.player_name in live_session_player_names %}
<form method="post" action="{{ url_for('admin_apply_payout_carry_in') }}">
<form method="post" action="{{ url_for('admin.apply_payout_carry_in') }}">
<input type="hidden" name="session_id" value="{{ live_session.session_id }}">
<input type="hidden" name="player_name" value="{{ item.player_name }}">
<button class="table-action-button" type="submit">Apply</button>
@@ -146,7 +146,7 @@
<div><div class="micro">Seated</div><div class="mono16">{{ live_session.entries|length }}</div></div>
<div><div class="micro">Open</div><div class="mono16 mono16--warn">{{ live_unresolved_count }}</div></div>
</div>
<form method="post" action="{{ url_for('admin_session_state') }}"
<form method="post" action="{{ url_for('admin.session_state') }}"
{% if live_unresolved_count > 0 %}data-close-warning="true"{% endif %}>
<input type="hidden" name="session_id" value="{{ live_session.session_id }}">
<button class="btn-close" type="submit" name="state" value="closed">Close session</button>
@@ -167,12 +167,12 @@
</div>
{% endif %}
<form method="post" action="{{ url_for('admin_open_session') }}">
<form method="post" action="{{ url_for('admin.open_session') }}">
<input type="hidden" name="session_date" value="{{ admin_today }}">
<button class="btn-newsession {{ 'btn-newsession--primary' if not live_session else '' }}" type="submit">+ Open new session</button>
</form>
<form method="post" action="{{ url_for('admin_prune_empty_sessions') }}" class="prune-session-form">
<form method="post" action="{{ url_for('admin.prune_empty_sessions') }}" class="prune-session-form">
<button class="secondary-button" type="submit" {% if empty_session_count == 0 %}disabled{% endif %}>
Prune empty sessions{% if empty_session_count > 0 %} ({{ empty_session_count }}){% endif %}
</button>
@@ -182,9 +182,9 @@
<div class="panel panel--pad import-export-panel">
<h2 class="panel__title" style="margin:0 0 4px">Import / export</h2>
<p class="csv__desc">Back up or migrate the raw event ledger.</p>
<form method="post" action="{{ url_for('admin_import_csv') }}" enctype="multipart/form-data" class="import-form">
<form method="post" action="{{ url_for('admin.import_csv') }}" enctype="multipart/form-data" class="import-form">
<div class="csv__btns">
<a class="csv-btn-ghost" href="{{ url_for('admin_export_csv') }}">&darr; Export CSV</a>
<a class="csv-btn-ghost" href="{{ url_for('admin.export_csv') }}">&darr; Export CSV</a>
<label class="csv-btn-ghost" for="csvTopImport">&uarr; Import CSV</label>
</div>
<input id="csvTopImport" type="file" accept=".csv,text/csv" name="csv_file" hidden onchange="this.form.submit()">
@@ -223,7 +223,7 @@
</div>
<div class="debt-action-grid">
<form class="form-card debt-form" method="post" action="{{ url_for('admin_collect_front') }}">
<form class="form-card debt-form" method="post" action="{{ url_for('admin.collect_front') }}">
<h3>Collect debt</h3>
<label>
<span>Outstanding debt</span>
@@ -248,7 +248,7 @@
<button class="primary-button" type="submit">Mark collected</button>
</form>
<form class="form-card debt-form" method="post" action="{{ url_for('admin_write_off_front') }}">
<form class="form-card debt-form" method="post" action="{{ url_for('admin.write_off_front') }}">
<h3>Write off debt</h3>
<label>
<span>Outstanding debt</span>
@@ -308,7 +308,7 @@
<tbody>
{% for s in recent_sessions %}
<tr>
<td><a href="{{ url_for('session_detail', session_id=s.session_id) }}">{{ session_label(s) }}</a></td>
<td><a href="{{ url_for('public.session_detail', session_id=s.session_id) }}">{{ session_label(s) }}</a></td>
<td style="font-family:var(--font-mono);color:var(--num)">{{ s.entries|length }}</td>
<td style="font-family:var(--font-mono);color:var(--num)">{{ s.total_banker_cash_in_cents | money }}</td>
<td style="font-family:var(--font-mono);color:var(--pos)">{{ s.total_banker_cash_out_cents | money }}</td>
@@ -317,7 +317,7 @@
<td class="{% if s.total_net_book_position_cents > 0 %}positive{% elif s.total_net_book_position_cents < 0 %}negative{% endif %}" style="font-family:var(--font-mono)">{{ s.total_net_book_position_cents | money }}</td>
<td><span class="status-badge status-{{ s.status }}">{{ s.status }}</span></td>
<td>
<form method="post" action="{{ url_for('admin_session_state') }}" class="inline-action-form">
<form method="post" action="{{ url_for('admin.session_state') }}" class="inline-action-form">
<input type="hidden" name="session_id" value="{{ s.session_id }}">
{% if s.status == "open" %}
<button class="table-action-button" type="submit" name="state" value="closed">Close</button>
templates/base.html
+6 -6
View File
@@ -11,23 +11,23 @@
<body>
<header class="topbar">
<div class="topbar__inner">
<a class="brand" href="{{ url_for('leaderboard') }}">
<a class="brand" href="{{ url_for('public.leaderboard') }}">
<span class="brand__mark"></span>
<span class="brand__name">myboker.org</span>
</a>
<nav class="mainnav">
<a class="navlink {{ 'is-active' if request.endpoint == 'leaderboard' }}" href="{{ url_for('leaderboard') }}">Leaderboard</a>
<a class="navlink {{ 'is-active' if request.endpoint in ('sessions', 'session_detail') }}" href="{{ url_for('sessions') }}">Sessions</a>
<a class="navlink {{ 'is-active' if request.endpoint == 'public.leaderboard' }}" href="{{ url_for('public.leaderboard') }}">Leaderboard</a>
<a class="navlink {{ 'is-active' if request.endpoint in ('public.sessions', 'public.session_detail') }}" href="{{ url_for('public.sessions') }}">Sessions</a>
{% if is_admin %}
<a class="navlink {{ 'is-active' if request.endpoint == 'admin_dashboard' }}" href="{{ url_for('admin_dashboard') }}">Admin</a>
<a class="navlink {{ 'is-active' if request.endpoint == 'admin.dashboard' }}" href="{{ url_for('admin.dashboard') }}">Admin</a>
{% else %}
<a class="navlink {{ 'is-active' if request.endpoint == 'admin_login' }}" href="{{ url_for('admin_login') }}">Admin</a>
<a class="navlink {{ 'is-active' if request.endpoint == 'admin.login' }}" href="{{ url_for('admin.login') }}">Admin</a>
{% endif %}
</nav>
<span class="topbar__spacer"></span>
{% if is_admin %}
<div class="topbar__admin-actions">
<form method="post" action="{{ url_for('admin_logout') }}" style="margin:0">
<form method="post" action="{{ url_for('admin.logout') }}" style="margin:0">
<button class="btn btn--ghost btn--sm" type="submit">Logout</button>
</form>
</div>
templates/leaderboard.html
+4 -4
View File
@@ -72,7 +72,7 @@
{% endfor %}
</select>
{% if selected_session_id %}
<a class="btn btn--ghost btn--sm" href="{{ url_for('leaderboard', mode=mode) }}">Reset</a>
<a class="btn btn--ghost btn--sm" href="{{ url_for('public.leaderboard', mode=mode) }}">Reset</a>
{% endif %}
</form>
</div>
@@ -101,12 +101,12 @@
{% for player in main_board %}
{% set idx = loop.index %}
<tr data-player-row="true" data-original-rank="{{ idx }}"
onclick="window.location='{{ url_for('player_detail', player_name=player.player_name) }}'">
onclick="window.location='{{ url_for('public.player_detail', player_name=player.player_name) }}'">
<td class="col-rank" data-sort-value="{{ idx }}">
<span class="rank-num {{ 'rank-1' if idx==1 else 'rank-2' if idx==2 else 'rank-3' if idx==3 else '' }}" data-rank-cell>#{{ idx }}</span>
</td>
<td class="col-player" data-sort-value="{{ player.player_name|lower }}">
<a href="{{ url_for('player_detail', player_name=player.player_name) }}">{{ player.player_name }}</a>
<a href="{{ url_for('public.player_detail', player_name=player.player_name) }}">{{ player.player_name }}</a>
</td>
<td class="col-rank-delta" data-sort-value="{{ player.rank_change }}">
{% if player.rank_change > 0 %}
@@ -146,7 +146,7 @@
</div>
<div class="provisional-grid">
{% for player in provisional_board %}
<a class="provisional-card" href="{{ url_for('player_detail', player_name=player.player_name) }}"
<a class="provisional-card" href="{{ url_for('public.player_detail', player_name=player.player_name) }}"
style="display:block;text-decoration:none;">
<p class="provisional-card__name">{{ player.player_name }}</p>
<div class="provisional-card__meta">
templates/session_detail.html
+5 -5
View File
@@ -4,15 +4,15 @@
{% block content %}
<div class="breadcrumb">
<a href="{{ url_for('sessions') }}">Sessions</a>
<a href="{{ url_for('public.sessions') }}">Sessions</a>
<span class="breadcrumb__sep">/</span>
<span>{{ session_label(session) }}</span>
<span style="flex:1"></span>
{% if prev_session %}
<a class="btn btn--ghost btn--sm" href="{{ url_for('session_detail', session_id=prev_session.session_id) }}">← Prev</a>
<a class="btn btn--ghost btn--sm" href="{{ url_for('public.session_detail', session_id=prev_session.session_id) }}">← Prev</a>
{% endif %}
{% if next_session %}
<a class="btn btn--ghost btn--sm" href="{{ url_for('session_detail', session_id=next_session.session_id) }}">Next →</a>
<a class="btn btn--ghost btn--sm" href="{{ url_for('public.session_detail', session_id=next_session.session_id) }}">Next →</a>
{% endif %}
</div>
@@ -111,9 +111,9 @@
</thead>
<tbody>
{% for entry in session.entries %}
<tr onclick="window.location='{{ url_for('player_detail', player_name=entry.player_name) }}'">
<tr onclick="window.location='{{ url_for('public.player_detail', player_name=entry.player_name) }}'">
<td class="col-player">
<a href="{{ url_for('player_detail', player_name=entry.player_name) }}">{{ entry.player_name }}</a>
<a href="{{ url_for('public.player_detail', player_name=entry.player_name) }}">{{ entry.player_name }}</a>
</td>
<td>{{ entry.invested_cents | money }}</td>
<td>{{ entry.cash_out_cents | money }}</td>
templates/sessions.html
+2 -2
View File
@@ -29,9 +29,9 @@
</thead>
<tbody>
{% for s in sessions %}
<tr onclick="window.location='{{ url_for('session_detail', session_id=s.session_id) }}'">
<tr onclick="window.location='{{ url_for('public.session_detail', session_id=s.session_id) }}'">
<td class="col-left">
<a href="{{ url_for('session_detail', session_id=s.session_id) }}">{{ session_label(s) }}</a>
<a href="{{ url_for('public.session_detail', session_id=s.session_id) }}">{{ session_label(s) }}</a>
</td>
<td><span class="badge badge--{{ s.status }}">{{ s.status }}</span></td>
<td>{{ s.entries|length }}</td>
tests/test_accounting.py
+11 -10
View File
@@ -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)
utils.py
+84
View File
@@ -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'<span class="currency-symbol">$</span>{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"