1098 lines
50 KiB
Python
1098 lines
50 KiB
Python
"""The Button - a single-file, dependency-free game server (Python 3.9+)."""
|
|
|
|
import csv
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from functools import lru_cache
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
STATIC_DIR = ROOT / "static"
|
|
DATA_FILE = Path(os.environ.get("BUTTON_DATA_FILE", ROOT / "data.json"))
|
|
DATABASE_FILE = Path(os.environ.get("BUTTON_DATABASE_FILE", DATA_FILE.with_suffix(".sqlite3")))
|
|
MAILING_LIST_FILE = Path(os.environ.get("BUTTON_MAILING_LIST_FILE", ROOT / "mailing_list.csv"))
|
|
MAILING_LIST_FIELDS = ["subscribed_at", "player", "first_name", "last_name", "email", "phone"]
|
|
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
|
PHONE_RE = re.compile(r"^\+?[0-9 ()\-.]{7,25}$")
|
|
HOST = os.environ.get("BUTTON_HOST", "127.0.0.1")
|
|
PORT = int(os.environ.get("BUTTON_PORT", "8000"))
|
|
DURATION_MS = int(float(os.environ.get("BUTTON_DURATION_SECONDS", 24 * 60 * 60)) * 1000)
|
|
MAX_BODY_BYTES = 1024
|
|
MAX_NAME_LEN = 24
|
|
MIN_PASSWORD_LEN = 8
|
|
MAX_PASSWORD_LEN = 128
|
|
PBKDF2_ITERATIONS = 600_000
|
|
MAX_SESSIONS_PER_PLAYER = 10
|
|
LOGIN_MAX_FAILURES = 5
|
|
LOGIN_LOCKOUT_MS = 15 * 60 * 1000
|
|
MAX_CHAT_LENGTH = 200
|
|
CHAT_HISTORY_LIMIT = 100
|
|
CHAT_COOLDOWN_MS = 2000
|
|
PRESENCE_TTL_SECONDS = 45
|
|
MAX_PRESENCE_TABS_PER_SESSION = 10
|
|
|
|
# Ordered from a freshly reset countdown to an almost-expired one.
|
|
# "share" is each zone's relative slice of the timer, so high-value zones are brief.
|
|
ZONES = [
|
|
{"name": "green", "points": 0, "share": 32},
|
|
{"name": "blue", "points": 2, "share": 16},
|
|
{"name": "yellow", "points": 4, "share": 8},
|
|
{"name": "red", "points": 8, "share": 4},
|
|
{"name": "black", "points": 16, "share": 2},
|
|
{"name": "white", "points": 32, "share": 1},
|
|
]
|
|
_total_share = sum(z["share"] for z in ZONES)
|
|
_upper = DURATION_MS
|
|
for _z in ZONES:
|
|
_z["remaining_from_ms"] = _upper
|
|
_upper = round(_upper - DURATION_MS * _z["share"] / _total_share)
|
|
_z["remaining_to_ms"] = max(_upper, 0)
|
|
ZONES[-1]["remaining_to_ms"] = 0
|
|
|
|
STATIC_FILES = {
|
|
**{f"/favicon-{zone['name']}.svg": (f"favicon-{zone['name']}.svg", "image/svg+xml") for zone in ZONES},
|
|
"/": ("index.html", "text/html; charset=utf-8"),
|
|
"/index.html": ("index.html", "text/html; charset=utf-8"),
|
|
"/app.js": ("app.js", "text/javascript; charset=utf-8"),
|
|
"/style.css": ("style.css", "text/css; charset=utf-8"),
|
|
"/favicon.ico": ("favicon.ico", "image/x-icon"),
|
|
"/poster.png": ("poster.png", "image/png"),
|
|
}
|
|
|
|
lock = threading.Lock()
|
|
|
|
|
|
def now_ms():
|
|
return int(time.time() * 1000)
|
|
|
|
|
|
@lru_cache(maxsize=64)
|
|
def season_boundaries(year):
|
|
"""UTC midnights of astronomical events (Meeus, Astronomical Algorithms ch. 27).
|
|
|
|
Uses the 2000-era equinox/solstice polynomials and periodic correction, then
|
|
a Delta-T estimate to convert terrestrial time to UTC before taking the date.
|
|
Dates, not the event's hour, define the game boundary. Supports 2000-2100;
|
|
2026/2027 UTC dates are regression checked against USNO data.
|
|
"""
|
|
if not 1999 <= year <= 2101:
|
|
raise ValueError("Season calendar supports years 2000-2100")
|
|
y = (year - 2000) / 1000
|
|
coefficients = (
|
|
(2451623.80984, 365242.37404, .05169, -.00411, -.00057),
|
|
(2451716.56767, 365241.62603, .00325, .00888, -.00030),
|
|
(2451810.21715, 365242.01767, -.11575, .00337, .00078),
|
|
(2451900.05952, 365242.74049, -.06223, -.00823, .00032),
|
|
)
|
|
terms = (
|
|
(485, 324.96, 1934.136), (203, 337.23, 32964.467), (199, 342.08, 20.186),
|
|
(182, 27.85, 445267.112), (156, 73.14, 45036.886), (136, 171.52, 22518.443),
|
|
(77, 222.54, 65928.934), (74, 296.72, 3034.906), (70, 243.58, 9037.513),
|
|
(58, 119.81, 33718.147), (52, 297.17, 150.678), (50, 21.02, 2281.226),
|
|
(45, 247.54, 29929.562), (44, 325.15, 31555.956), (29, 60.93, 4443.417),
|
|
(18, 155.12, 67555.328), (17, 288.79, 4562.452), (16, 198.04, 62894.029),
|
|
(14, 199.76, 31436.921), (12, 95.39, 14577.848), (12, 287.11, 31931.756),
|
|
(12, 320.81, 34777.259), (9, 227.73, 1222.114), (8, 15.45, 16859.074),
|
|
)
|
|
result = []
|
|
for index, coefficients_for_event in enumerate(coefficients):
|
|
jde = sum(c * y ** power for power, c in enumerate(coefficients_for_event))
|
|
t = (jde - 2451545) / 36525
|
|
w = math.radians(35999.373 * t - 2.47)
|
|
correction = sum(a * math.cos(math.radians(b + c * t)) for a, b, c in terms)
|
|
jde += .00001 * correction / (1 + .0334 * math.cos(w) + .0007 * math.cos(2 * w))
|
|
decimal_year = year + (index * 3 + 2.5) / 12
|
|
u = decimal_year - 2000
|
|
if decimal_year < 2005:
|
|
delta_t = 63.86 + .3345 * u - .060374 * u**2 + .0017275 * u**3 + .000651814 * u**4 + .00002373599 * u**5
|
|
elif decimal_year < 2050:
|
|
delta_t = 62.92 + .32217 * u + .005589 * u**2
|
|
else:
|
|
delta_t = -20 + 32 * ((decimal_year - 1820) / 100)**2 - .5628 * (2150 - decimal_year)
|
|
event = datetime.fromtimestamp((jde - 2440587.5) * 86400 - delta_t, timezone.utc)
|
|
midnight = event.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
result.append({"id": midnight.strftime("%Y-%m-%d"),
|
|
"name": f"{('March equinox', 'June solstice', 'September equinox', 'December solstice')[index]} {year}",
|
|
"starts_at": int(midnight.timestamp() * 1000)})
|
|
return tuple(result)
|
|
|
|
|
|
def season_for_time(at):
|
|
year = datetime.fromtimestamp(at / 1000, timezone.utc).year
|
|
if not 2000 <= year <= 2100:
|
|
raise ValueError("Season calendar supports years 2000-2100")
|
|
boundaries = [entry for y in (year - 1, year, year + 1) for entry in season_boundaries(y)]
|
|
index = max(i for i, entry in enumerate(boundaries) if entry["starts_at"] <= at)
|
|
return {**boundaries[index], "ends_at": boundaries[index + 1]["starts_at"]}
|
|
|
|
|
|
def ensure_season(connection, at=None):
|
|
"""Adopt the existing game into the current season without deleting its history."""
|
|
if connection.execute("SELECT 1 FROM season_clock WHERE singleton = 1").fetchone() is None:
|
|
season = season_for_time(now_ms() if at is None else at)
|
|
connection.execute(
|
|
"INSERT INTO season_clock (singleton, id, name, starts_at, ends_at) VALUES (1, ?, ?, ?, ?)",
|
|
(season["id"], season["name"], season["starts_at"], season["ends_at"]),
|
|
)
|
|
connection.execute("""INSERT OR IGNORE INTO season_numbering (singleton, origin_at)
|
|
SELECT 1, MIN(starts_at) FROM (
|
|
SELECT starts_at FROM season_clock UNION ALL SELECT starts_at FROM season_results
|
|
)""")
|
|
|
|
|
|
def roman_numeral(number):
|
|
if number < 1:
|
|
raise ValueError("Season numbers must be positive")
|
|
result = []
|
|
for value, symbol in ((1000, "M"), (900, "CM"), (500, "D"), (400, "CD"), (100, "C"),
|
|
(90, "XC"), (50, "L"), (40, "XL"), (10, "X"), (9, "IX"),
|
|
(5, "V"), (4, "IV"), (1, "I")):
|
|
count, number = divmod(number, value)
|
|
result.append(symbol * count)
|
|
return "".join(result)
|
|
|
|
|
|
def numbered_season(connection, season):
|
|
origin = connection.execute("SELECT origin_at FROM season_numbering WHERE singleton = 1").fetchone()[0]
|
|
def ordinal(at):
|
|
date = datetime.fromtimestamp(at / 1000, timezone.utc)
|
|
return date.year * 4 + (date.month - 3) // 3
|
|
number = ordinal(season["starts_at"]) - ordinal(origin) + 1
|
|
return {**season, "number": number, "name": f"Season {roman_numeral(number)}"}
|
|
|
|
|
|
def current_season(connection):
|
|
row = dict(connection.execute("SELECT * FROM season_clock WHERE singleton = 1").fetchone())
|
|
del row["singleton"]
|
|
return numbered_season(connection, row)
|
|
|
|
|
|
def hash_token(token):
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def hash_password(password, salt=None, iterations=PBKDF2_ITERATIONS):
|
|
salt = salt or secrets.token_hex(16)
|
|
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), bytes.fromhex(salt), iterations).hex()
|
|
return f"pbkdf2_sha256${iterations}${salt}${digest}"
|
|
|
|
|
|
def verify_password(password, stored):
|
|
_, iterations, salt, _ = stored.split("$")
|
|
return hmac.compare_digest(hash_password(password, salt, int(iterations)), stored)
|
|
|
|
|
|
# Verified against when the name doesn't exist so response timing doesn't reveal valid names.
|
|
DUMMY_PASSWORD_HASH = hash_password(secrets.token_hex(16))
|
|
|
|
|
|
def zone_for_remaining(remaining_ms):
|
|
for zone in ZONES:
|
|
if remaining_ms > zone["remaining_to_ms"]:
|
|
return zone
|
|
return ZONES[-1]
|
|
|
|
|
|
def new_game():
|
|
t = now_ms()
|
|
return {
|
|
"duration_ms": DURATION_MS,
|
|
"started_at": t,
|
|
"deadline": t + DURATION_MS,
|
|
"ended": False,
|
|
"ended_at": None,
|
|
"players": {}, # id -> {name, password_hash, token_hashes, created_at}
|
|
"presses": [], # {player_id, at, remaining_ms, zone, points}
|
|
}
|
|
|
|
|
|
@contextmanager
|
|
def database():
|
|
"""Use a separate connection per operation, with atomic writes and FK checks."""
|
|
connection = sqlite3.connect(DATABASE_FILE)
|
|
connection.row_factory = sqlite3.Row
|
|
try:
|
|
connection.execute("PRAGMA foreign_keys = ON")
|
|
with connection:
|
|
yield connection
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def write_state(connection, state):
|
|
"""Persist a complete snapshot inside the caller's transaction."""
|
|
connection.execute("DELETE FROM presses")
|
|
connection.execute("DELETE FROM sessions")
|
|
connection.execute("DELETE FROM players")
|
|
connection.executemany(
|
|
"INSERT INTO players (id, name, password_hash, created_at, subscribed) VALUES (?, ?, ?, ?, ?)",
|
|
[(pid, p["name"], p.get("password_hash"), p["created_at"], p.get("subscribed", False))
|
|
for pid, p in state["players"].items()],
|
|
)
|
|
connection.executemany(
|
|
"INSERT INTO sessions (player_id, position, token_hash) VALUES (?, ?, ?)",
|
|
[(pid, position, token_hash) for pid, p in state["players"].items()
|
|
for position, token_hash in enumerate(p["token_hashes"])],
|
|
)
|
|
connection.executemany(
|
|
"INSERT INTO presses (position, player_id, at, remaining_ms, zone, points) VALUES (?, ?, ?, ?, ?, ?)",
|
|
[(position, p["player_id"], p["at"], p["remaining_ms"], p["zone"], p["points"])
|
|
for position, p in enumerate(state["presses"])],
|
|
)
|
|
connection.execute(
|
|
"""INSERT INTO game (id, duration_ms, started_at, deadline, ended, ended_at)
|
|
VALUES (1, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(id) DO UPDATE SET duration_ms = excluded.duration_ms,
|
|
started_at = excluded.started_at, deadline = excluded.deadline,
|
|
ended = excluded.ended, ended_at = excluded.ended_at""",
|
|
(state["duration_ms"], state["started_at"], state["deadline"], state["ended"], state["ended_at"]),
|
|
)
|
|
|
|
|
|
def milestone_status(game_state, at):
|
|
"""Eight fixed community goals; elapsed survival stops at the deadline."""
|
|
until = min(at, game_state["ended_at"] if game_state["ended_at"] is not None else game_state["deadline"])
|
|
elapsed = max(0, until - game_state["started_at"])
|
|
players = len(game_state["players"])
|
|
presses = len(game_state["presses"])
|
|
goals = [
|
|
("day_one", "One day together", "The community kept the game alive for one day!", elapsed, 86400000),
|
|
("one_week", "A week of nerve", "Seven days together. The clock is still part of our story!", elapsed, 7 * 86400000),
|
|
("one_month", "Thirty days strong", "The game survived 30 days. What a community!", elapsed, 30 * 86400000),
|
|
("ten_players", "A growing crowd", "Ten players have joined the experiment!", players, 10),
|
|
("fifty_players", "Fifty strong", "Fifty players, one shared clock. Welcome aboard!", players, 50),
|
|
("hundred_presses", "A hundred second chances", "Together we've pressed the button 100 times!", presses, 100),
|
|
("thousand_presses", "A thousand moments", "1,000 presses. A thousand decisions to keep us going!", presses, 1000),
|
|
("first_photo_finish", "Our first photo finish", "A community first: someone pressed with less than ten seconds left!", int(any(0 < p["remaining_ms"] < 10000 for p in game_state["presses"])), 1),
|
|
]
|
|
return [{"id": key, "name": name, "message": message, "progress": min(progress, target), "target": target}
|
|
for key, name, message, progress, target in goals]
|
|
|
|
|
|
def sync_milestones(connection, game_state, at=None):
|
|
"""Commit unlocks with game changes. Existing goals are silently baselined once."""
|
|
at = now_ms() if at is None else at
|
|
initialized = connection.execute("SELECT 1 FROM community_milestones LIMIT 1").fetchone() is not None
|
|
for goal in milestone_status(game_state, at):
|
|
connection.execute("INSERT OR IGNORE INTO community_milestones (id) VALUES (?)", (goal["id"],))
|
|
if goal["progress"] >= goal["target"]:
|
|
connection.execute(
|
|
"""UPDATE community_milestones SET reached_at = ?, announced_at = ?, message = ?
|
|
WHERE id = ? AND reached_at IS NULL""",
|
|
(at, at if initialized else None, f"Milestone: {goal['name']}. {goal['message']}", goal["id"]),
|
|
)
|
|
|
|
|
|
def load_state():
|
|
if DATABASE_FILE.resolve() == DATA_FILE.resolve():
|
|
raise ValueError("BUTTON_DATABASE_FILE must differ from the legacy BUTTON_DATA_FILE")
|
|
DATABASE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with database() as connection:
|
|
connection.executescript("""
|
|
CREATE TABLE IF NOT EXISTS game (
|
|
id INTEGER PRIMARY KEY CHECK (id = 1),
|
|
duration_ms INTEGER NOT NULL,
|
|
started_at INTEGER NOT NULL,
|
|
deadline INTEGER NOT NULL,
|
|
ended INTEGER NOT NULL CHECK (ended IN (0, 1)),
|
|
ended_at INTEGER
|
|
);
|
|
CREATE TABLE IF NOT EXISTS players (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
password_hash TEXT,
|
|
created_at INTEGER NOT NULL,
|
|
subscribed INTEGER NOT NULL CHECK (subscribed IN (0, 1))
|
|
);
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
player_id TEXT NOT NULL REFERENCES players(id),
|
|
position INTEGER NOT NULL,
|
|
token_hash TEXT NOT NULL UNIQUE,
|
|
PRIMARY KEY (player_id, position)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS presses (
|
|
position INTEGER PRIMARY KEY,
|
|
player_id TEXT NOT NULL REFERENCES players(id),
|
|
at INTEGER NOT NULL,
|
|
remaining_ms INTEGER NOT NULL,
|
|
zone TEXT NOT NULL,
|
|
points INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS chat_messages (
|
|
id INTEGER PRIMARY KEY,
|
|
player_id TEXT NOT NULL REFERENCES players(id) DEFERRABLE INITIALLY DEFERRED,
|
|
name TEXT NOT NULL,
|
|
at INTEGER NOT NULL,
|
|
message TEXT NOT NULL CHECK (length(message) BETWEEN 1 AND 200)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS chat_player_messages ON chat_messages (player_id, id);
|
|
CREATE TABLE IF NOT EXISTS community_milestones (
|
|
id TEXT PRIMARY KEY,
|
|
reached_at INTEGER,
|
|
announced_at INTEGER,
|
|
message TEXT
|
|
);
|
|
CREATE TABLE IF NOT EXISTS season_clock (
|
|
singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
|
|
id TEXT NOT NULL, name TEXT NOT NULL,
|
|
starts_at INTEGER NOT NULL, ends_at INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS season_results (
|
|
id TEXT PRIMARY KEY, name TEXT NOT NULL,
|
|
starts_at INTEGER NOT NULL, ends_at INTEGER NOT NULL,
|
|
played_from INTEGER NOT NULL, finished_at INTEGER NOT NULL,
|
|
reason TEXT NOT NULL, leaderboard TEXT NOT NULL, presses TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS season_milestone_messages (
|
|
id TEXT PRIMARY KEY, at INTEGER NOT NULL, message TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS season_numbering (
|
|
singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
|
|
origin_at INTEGER NOT NULL
|
|
);
|
|
""")
|
|
columns = {row["name"] for row in connection.execute("PRAGMA table_info(chat_messages)")}
|
|
if "kind" not in columns:
|
|
connection.execute("ALTER TABLE chat_messages ADD COLUMN kind TEXT NOT NULL DEFAULT 'message'")
|
|
# Serialize initialization. A committed game row is the migration marker,
|
|
# not merely the existence of a possibly empty or interrupted database.
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
game = connection.execute("SELECT * FROM game WHERE id = 1").fetchone()
|
|
if game is None:
|
|
if DATA_FILE.exists():
|
|
with DATA_FILE.open("r", encoding="utf-8") as f:
|
|
loaded = json.load(f)
|
|
for player in loaded["players"].values():
|
|
if "token_hash" in player: # migrate pre-login data as well
|
|
player["token_hashes"] = [player.pop("token_hash")]
|
|
else:
|
|
loaded = new_game()
|
|
# Any read/validation/SQL failure aborts startup without consuming the
|
|
# JSON source or committing a partial game. The next start can retry.
|
|
write_state(connection, loaded)
|
|
sync_milestones(connection, loaded)
|
|
ensure_season(connection)
|
|
return loaded
|
|
|
|
loaded = dict(game)
|
|
del loaded["id"]
|
|
loaded["ended"] = bool(loaded["ended"])
|
|
loaded["players"] = {}
|
|
for row in connection.execute("SELECT * FROM players ORDER BY rowid"):
|
|
player = dict(row)
|
|
pid = player.pop("id")
|
|
player["subscribed"] = bool(player["subscribed"])
|
|
player["token_hashes"] = []
|
|
loaded["players"][pid] = player
|
|
for row in connection.execute("SELECT * FROM sessions ORDER BY player_id, position"):
|
|
loaded["players"][row["player_id"]]["token_hashes"].append(row["token_hash"])
|
|
loaded["presses"] = []
|
|
for row in connection.execute("SELECT * FROM presses ORDER BY position"):
|
|
press = dict(row)
|
|
del press["position"]
|
|
loaded["presses"].append(press)
|
|
sync_milestones(connection, loaded)
|
|
ensure_season(connection)
|
|
return loaded
|
|
|
|
|
|
def save_state(state):
|
|
with database() as connection:
|
|
write_state(connection, state)
|
|
sync_milestones(connection, state)
|
|
|
|
|
|
state = load_state()
|
|
token_index = {th: pid for pid, p in state["players"].items() for th in p["token_hashes"]}
|
|
failed_logins = {} # lowercase name -> (failure count, first failure ms)
|
|
presence = {} # (token hash, tab id) -> (monotonic heartbeat seconds, viewing chat)
|
|
|
|
|
|
def prune_presence():
|
|
"""Must be called with the lock held. Presence is never persisted."""
|
|
cutoff = time.monotonic() - PRESENCE_TTL_SECONDS
|
|
for key, (seen, _) in list(presence.items()):
|
|
if seen <= cutoff or key[0] not in token_index:
|
|
del presence[key]
|
|
|
|
|
|
def public_presence():
|
|
"""Public aggregates and opted-in chat names only; no session identifiers."""
|
|
prune_presence()
|
|
online = {token_index[th] for th, _ in presence}
|
|
chatting = {token_index[th] for (th, _), (_, in_chat) in presence.items() if in_chat}
|
|
return {
|
|
"online_count": len(online),
|
|
"chat_players": sorted((state["players"][pid]["name"] for pid in chatting), key=str.lower),
|
|
}
|
|
|
|
|
|
def chat_messages():
|
|
with database() as connection:
|
|
rows = connection.execute(
|
|
"""SELECT id, name, at, message, kind FROM (
|
|
SELECT id, name, at, message, kind, 0 AS source, id AS position FROM chat_messages
|
|
UNION ALL
|
|
SELECT 'milestone:' || id, 'The Button', announced_at, message, 'milestone', 1, rowid
|
|
FROM community_milestones WHERE announced_at IS NOT NULL
|
|
UNION ALL
|
|
SELECT id, 'The Button', at, message, 'milestone', 1, rowid
|
|
FROM season_milestone_messages
|
|
) ORDER BY at DESC, source DESC, position DESC LIMIT ?""",
|
|
(CHAT_HISTORY_LIMIT,),
|
|
).fetchall()
|
|
return [dict(row) for row in reversed(rows)]
|
|
|
|
|
|
def find_player_by_name(name):
|
|
for pid, p in state["players"].items():
|
|
if p["name"].lower() == name.lower():
|
|
return pid
|
|
return None
|
|
|
|
|
|
def issue_token(pid):
|
|
"""Must be called with the lock held."""
|
|
token = secrets.token_urlsafe(32)
|
|
th = hash_token(token)
|
|
hashes = state["players"][pid]["token_hashes"]
|
|
hashes.append(th)
|
|
token_index[th] = pid
|
|
while len(hashes) > MAX_SESSIONS_PER_PLAYER:
|
|
token_index.pop(hashes.pop(0), None)
|
|
save_state(state)
|
|
return token
|
|
|
|
|
|
def validate_credentials(data):
|
|
"""Returns (name, password, error)."""
|
|
name, password = data.get("name"), data.get("password")
|
|
if not isinstance(name, str) or not (1 <= len(name.strip()) <= MAX_NAME_LEN):
|
|
return None, None, f"Name must be 1-{MAX_NAME_LEN} characters"
|
|
name = " ".join(name.split())
|
|
if not name.isprintable():
|
|
return None, None, "Name contains invalid characters"
|
|
if not isinstance(password, str) or not (MIN_PASSWORD_LEN <= len(password) <= MAX_PASSWORD_LEN):
|
|
return None, None, f"Password must be {MIN_PASSWORD_LEN}-{MAX_PASSWORD_LEN} characters"
|
|
return name, password, None
|
|
|
|
|
|
def archive_season(connection, season, game_state, finished_at, reason):
|
|
"""Immutable score/history snapshots have no FK to mutable player rows."""
|
|
if connection.execute("SELECT 1 FROM season_results WHERE id = ?", (season["id"],)).fetchone():
|
|
return
|
|
scores = {pid: {"name": p["name"], "total_points": 0, "press_count": 0}
|
|
for pid, p in game_state["players"].items()}
|
|
for press in game_state["presses"]:
|
|
scores[press["player_id"]]["total_points"] += press["points"]
|
|
scores[press["player_id"]]["press_count"] += 1
|
|
standings = sorted(scores.values(), key=lambda p: (-p["total_points"], p["name"].lower()))
|
|
connection.execute(
|
|
"""INSERT OR IGNORE INTO season_results
|
|
(id, name, starts_at, ends_at, played_from, finished_at, reason, leaderboard, presses)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(season["id"], season["name"], season["starts_at"], season["ends_at"],
|
|
game_state["started_at"], max(game_state["started_at"], finished_at), reason,
|
|
json.dumps(standings), json.dumps(game_state["presses"])),
|
|
)
|
|
|
|
|
|
def check_ended(at=None):
|
|
"""Advance season and game atomically; caller must hold the game lock."""
|
|
t = now_ms() if at is None else at
|
|
replacement = None
|
|
with database() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
season = current_season(connection)
|
|
while t >= season["ends_at"]:
|
|
previous = replacement if replacement is not None else state
|
|
finish = min(previous["ended_at"] if previous["ended_at"] is not None else previous["deadline"], season["ends_at"])
|
|
archive_season(connection, season, previous, finish,
|
|
"countdown" if finish < season["ends_at"] else "boundary")
|
|
# Preserve old milestone announcements before resetting seasonal goals.
|
|
connection.execute(
|
|
"""INSERT OR IGNORE INTO season_milestone_messages (id, at, message)
|
|
SELECT ? || id, announced_at, message FROM community_milestones
|
|
WHERE announced_at IS NOT NULL""", (f"milestone:{season['id']}:",),
|
|
)
|
|
connection.execute("DELETE FROM community_milestones")
|
|
season = season_for_time(season["ends_at"])
|
|
connection.execute(
|
|
"UPDATE season_clock SET id = ?, name = ?, starts_at = ?, ends_at = ? WHERE singleton = 1",
|
|
(season["id"], season["name"], season["starts_at"], season["ends_at"]),
|
|
)
|
|
replacement = {**previous, "duration_ms": DURATION_MS, "started_at": season["starts_at"],
|
|
"deadline": season["starts_at"] + DURATION_MS, "ended": False,
|
|
"ended_at": None, "presses": []}
|
|
current = replacement if replacement is not None else state
|
|
if not current["ended"] and current["deadline"] > season["ends_at"]:
|
|
current = {**current, "deadline": season["ends_at"]}
|
|
replacement = current
|
|
if not current["ended"] and t >= current["deadline"]:
|
|
current = {**current, "ended": True, "ended_at": current["deadline"]}
|
|
replacement = current
|
|
if replacement is not None:
|
|
write_state(connection, current)
|
|
sync_milestones(connection, current, min(t, season["ends_at"]))
|
|
if current["ended"]:
|
|
archive_season(connection, season, current, current["ended_at"],
|
|
"boundary" if current["ended_at"] == season["ends_at"] else "countdown")
|
|
if replacement is not None:
|
|
state.clear()
|
|
state.update(replacement)
|
|
|
|
|
|
def seasonal_leaderboard():
|
|
with database() as connection:
|
|
season = current_season(connection)
|
|
rows = connection.execute("""SELECT id, name, starts_at, ends_at, played_from,
|
|
finished_at, reason, leaderboard FROM season_results ORDER BY starts_at DESC""").fetchall()
|
|
rows = [numbered_season(connection, dict(row)) for row in rows]
|
|
archives = []
|
|
for row in rows:
|
|
standings = json.loads(row["leaderboard"])
|
|
participants = [player for player in standings if player["press_count"] > 0]
|
|
top = max((p["total_points"] for p in participants), default=None)
|
|
archives.append({
|
|
"id": row["id"], "name": row["name"], "number": row["number"], "starts_at": row["starts_at"],
|
|
"ends_at": row["ends_at"], "played_from": row["played_from"], "finished_at": row["finished_at"],
|
|
"duration_ms": max(0, row["finished_at"] - row["played_from"]), "reason": row["reason"],
|
|
"winners": [p for p in participants if p["total_points"] == top],
|
|
"leaderboard": standings,
|
|
})
|
|
return {"current": {**season, "played_from": state["started_at"], "ended": state["ended"],
|
|
"leaderboard": public_leaderboard()}, "seasons": archives}
|
|
|
|
|
|
def season_summary():
|
|
with database() as connection:
|
|
return {**current_season(connection), "played_from": state["started_at"], "ended": state["ended"]}
|
|
|
|
|
|
def empty_zone_counts():
|
|
return {z["name"]: 0 for z in ZONES}
|
|
|
|
|
|
def player_achievements(pid):
|
|
"""Derive private achievements from recorded history, including legacy games."""
|
|
quiet_target = max(1, state["duration_ms"] // 2)
|
|
close_at = spectrum_at = quiet_at = None
|
|
visited = set()
|
|
longest_quiet = 0
|
|
previous_at = state["started_at"]
|
|
zone_names = {zone["name"] for zone in ZONES}
|
|
extra_rules = [
|
|
("red_alert", "Red alert", "Press in the red zone.", "red", None),
|
|
("into_the_dark", "Into the dark", "Press in the black zone.", "black", None),
|
|
("white_knuckle", "White knuckle", "Press in the white zone.", "white", None),
|
|
("photo_finish", "Photo finish", "Press with less than ten seconds remaining.", None, 10000),
|
|
("last_second", "Last second", "Press with less than one second remaining.", None, 1000),
|
|
]
|
|
extra_unlocks = {rule[0]: None for rule in extra_rules}
|
|
for press in state["presses"]:
|
|
quiet_ms = max(0, press["at"] - previous_at)
|
|
previous_at = press["at"] # measure the global gap, not this player's gap
|
|
if press["player_id"] != pid:
|
|
continue
|
|
for achievement_id, _, _, zone, threshold in extra_rules:
|
|
qualifies = press["zone"] == zone if zone is not None else 0 < press["remaining_ms"] < threshold
|
|
if extra_unlocks[achievement_id] is None and qualifies:
|
|
extra_unlocks[achievement_id] = press["at"]
|
|
if close_at is None and 0 < press["remaining_ms"] < 60000:
|
|
close_at = press["at"]
|
|
if press["zone"] in zone_names:
|
|
visited.add(press["zone"])
|
|
if spectrum_at is None and visited == zone_names:
|
|
spectrum_at = press["at"]
|
|
longest_quiet = max(longest_quiet, quiet_ms)
|
|
if quiet_at is None and quiet_ms >= quiet_target:
|
|
quiet_at = press["at"]
|
|
return [
|
|
{"id": "close_call", "name": "Close call",
|
|
"description": "Press with less than one minute remaining.",
|
|
"unlocked": close_at is not None, "unlocked_at": close_at,
|
|
"progress": int(close_at is not None), "target": 1},
|
|
{"id": "full_spectrum", "name": "Full spectrum",
|
|
"description": "Press in all six scoring zones.",
|
|
"unlocked": spectrum_at is not None, "unlocked_at": spectrum_at,
|
|
"progress": len(visited), "target": len(zone_names)},
|
|
{"id": "keeping_us_alive", "name": "Keeping us alive",
|
|
"description": "Press after nobody has pressed for at least half the game's countdown.",
|
|
"unlocked": quiet_at is not None, "unlocked_at": quiet_at,
|
|
"progress": min(longest_quiet, quiet_target), "target": quiet_target},
|
|
] + [
|
|
{"id": achievement_id, "name": name, "description": description,
|
|
"unlocked": extra_unlocks[achievement_id] is not None,
|
|
"unlocked_at": extra_unlocks[achievement_id],
|
|
"progress": int(extra_unlocks[achievement_id] is not None), "target": 1}
|
|
for achievement_id, name, description, _, _ in extra_rules
|
|
]
|
|
|
|
|
|
def player_summary(pid):
|
|
presses = [p for p in state["presses"] if p["player_id"] == pid]
|
|
counts = empty_zone_counts()
|
|
for p in presses:
|
|
counts[p["zone"]] += 1
|
|
standings = ranked_players()
|
|
position = next(i for i, (player_id, _) in enumerate(standings) if player_id == pid)
|
|
points = standings[position][1]["total_points"]
|
|
rival = standings[position - 1][1] if position else None
|
|
|
|
def record(press):
|
|
return {key: press[key] for key in ("at", "remaining_ms", "zone", "points")} if press else None
|
|
|
|
closest = min(presses, key=lambda p: (p["remaining_ms"], p["at"]), default=None)
|
|
best = min(presses, key=lambda p: (-p["points"], p["remaining_ms"], p["at"]), default=None)
|
|
return {
|
|
"name": state["players"][pid]["name"],
|
|
"subscribed": state["players"][pid].get("subscribed", False),
|
|
"total_players": len(state["players"]),
|
|
"total_presses": len(state["presses"]),
|
|
"total_points": sum(p["points"] for p in presses),
|
|
"press_count": len(presses),
|
|
"zone_counts": counts,
|
|
"best_remaining_ms": min((p["remaining_ms"] for p in presses), default=None),
|
|
"personal_bests": {"closest_call": record(closest), "highest_scoring_press": record(best)},
|
|
"achievements": player_achievements(pid),
|
|
"ranking": {
|
|
"position": position + 1,
|
|
"total_players": len(standings),
|
|
"tied_for_lead": len(standings) > 1 and points == standings[0][1]["total_points"]
|
|
and sum(score["total_points"] == points for _, score in standings) > 1,
|
|
"rival": {
|
|
"name": rival["name"],
|
|
"total_points": rival["total_points"],
|
|
"points_to_tie": rival["total_points"] - points,
|
|
"points_to_pass": rival["total_points"] - points + 1,
|
|
} if rival else None,
|
|
},
|
|
"presses": [
|
|
{k: p[k] for k in ("at", "remaining_ms", "zone", "points")} for p in presses
|
|
],
|
|
}
|
|
|
|
|
|
def public_press(p):
|
|
return {
|
|
"name": state["players"][p["player_id"]]["name"],
|
|
"at": p["at"],
|
|
"remaining_ms": p["remaining_ms"],
|
|
"zone": p["zone"],
|
|
"points": p["points"],
|
|
}
|
|
|
|
|
|
def ranked_players():
|
|
"""Internal standings retain IDs to identify players even in legacy data."""
|
|
scores = {
|
|
pid: {"name": player["name"], "total_points": 0, "press_count": 0}
|
|
for pid, player in state["players"].items()
|
|
}
|
|
for press in state["presses"]:
|
|
score = scores[press["player_id"]]
|
|
score["total_points"] += press["points"]
|
|
score["press_count"] += 1
|
|
return sorted(scores.items(), key=lambda item: (-item[1]["total_points"], item[1]["name"].lower()))
|
|
|
|
|
|
def public_leaderboard():
|
|
"""Public score totals only; do not expose sessions or individual presses."""
|
|
return [score for _, score in ranked_players()]
|
|
|
|
|
|
def global_stats():
|
|
counts = empty_zone_counts()
|
|
for p in state["presses"]:
|
|
counts[p["zone"]] += 1
|
|
leaderboard = []
|
|
for pid in state["players"]:
|
|
s = player_summary(pid)
|
|
del s["presses"], s["subscribed"], s["personal_bests"], s["ranking"], s["achievements"]
|
|
leaderboard.append(s)
|
|
leaderboard.sort(key=lambda s: (-s["total_points"], s["name"].lower()))
|
|
presses = state["presses"]
|
|
lowest = min(presses, key=lambda p: p["remaining_ms"], default=None)
|
|
return {
|
|
"started_at": state["started_at"],
|
|
"ended_at": state["ended_at"],
|
|
"total_players": len(state["players"]),
|
|
"total_presses": len(presses),
|
|
"zone_counts": counts,
|
|
"leaderboard": leaderboard,
|
|
"last_press": public_press(presses[-1]) if presses else None,
|
|
"lowest_press": public_press(lowest) if lowest else None,
|
|
"presses": [public_press(p) for p in presses],
|
|
}
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
server_version = "TheButton/1.0"
|
|
|
|
def log_message(self, fmt, *args):
|
|
pass
|
|
|
|
def send_json(self, status, payload):
|
|
body = json.dumps(payload).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_common_headers()
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def send_common_headers(self):
|
|
self.send_header("X-Content-Type-Options", "nosniff")
|
|
self.send_header("X-Frame-Options", "DENY")
|
|
self.send_header("Referrer-Policy", "no-referrer")
|
|
self.send_header(
|
|
"Content-Security-Policy",
|
|
"default-src 'self'; script-src 'self' https://js.stripe.com; "
|
|
"frame-src 'self' https://js.stripe.com; frame-ancestors 'none'; "
|
|
"base-uri 'none'; form-action 'self'",
|
|
)
|
|
|
|
def read_json(self):
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
if length > MAX_BODY_BYTES:
|
|
raise ValueError("Request body too large")
|
|
raw = self.rfile.read(length) if length else b"{}"
|
|
data = json.loads(raw.decode("utf-8"))
|
|
if not isinstance(data, dict):
|
|
raise ValueError("Expected a JSON object")
|
|
return data
|
|
|
|
def bearer_hash(self):
|
|
auth = self.headers.get("Authorization", "")
|
|
if not auth.startswith("Bearer "):
|
|
return None
|
|
return hash_token(auth[7:].strip())
|
|
|
|
def authed_player(self):
|
|
th = self.bearer_hash()
|
|
return token_index.get(th) if th else None
|
|
|
|
def do_GET(self):
|
|
path = self.path.split("?", 1)[0]
|
|
if path == "/api/state":
|
|
with lock:
|
|
check_ended()
|
|
return self.send_json(HTTPStatus.OK, {
|
|
"now": now_ms(),
|
|
"started_at": state["started_at"],
|
|
"deadline": state["deadline"],
|
|
"duration_ms": DURATION_MS,
|
|
"ended": state["ended"],
|
|
"total_players": len(state["players"]),
|
|
"total_presses": len(state["presses"]),
|
|
"zones": ZONES,
|
|
"season": season_summary(),
|
|
})
|
|
if path == "/api/seasons":
|
|
with lock:
|
|
check_ended()
|
|
return self.send_json(HTTPStatus.OK, seasonal_leaderboard())
|
|
if path == "/api/leaderboard":
|
|
with lock:
|
|
check_ended()
|
|
return self.send_json(HTTPStatus.OK, {"leaderboard": public_leaderboard()})
|
|
if path == "/api/chat":
|
|
return self.send_json(HTTPStatus.OK, {"messages": chat_messages()})
|
|
if path == "/api/presence":
|
|
with lock:
|
|
return self.send_json(HTTPStatus.OK, public_presence())
|
|
if path == "/api/me":
|
|
with lock:
|
|
check_ended()
|
|
pid = self.authed_player()
|
|
if pid is None:
|
|
return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Not registered"})
|
|
return self.send_json(HTTPStatus.OK, player_summary(pid))
|
|
if path == "/api/stats":
|
|
with lock:
|
|
check_ended()
|
|
if not state["ended"]:
|
|
return self.send_json(HTTPStatus.FORBIDDEN, {"error": "Stats are published when the game ends"})
|
|
return self.send_json(HTTPStatus.OK, global_stats())
|
|
if path in STATIC_FILES:
|
|
filename, ctype = STATIC_FILES[path]
|
|
body = (STATIC_DIR / filename).read_bytes()
|
|
self.send_response(HTTPStatus.OK)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_common_headers()
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
return
|
|
self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"})
|
|
|
|
def do_POST(self):
|
|
path = self.path.split("?", 1)[0]
|
|
try:
|
|
data = self.read_json()
|
|
except (ValueError, UnicodeDecodeError):
|
|
return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid JSON body"})
|
|
|
|
if path == "/api/presence":
|
|
with lock:
|
|
th = self.bearer_hash()
|
|
if th not in token_index:
|
|
return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to share your presence"})
|
|
tab = data.get("tab_id")
|
|
active, in_chat = data.get("active"), data.get("in_chat", False)
|
|
if (not isinstance(tab, str) or re.fullmatch(r"[a-f0-9]{32}", tab) is None
|
|
or not isinstance(active, bool) or not isinstance(in_chat, bool)):
|
|
return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid presence update"})
|
|
prune_presence()
|
|
key = (th, tab)
|
|
if not active:
|
|
presence.pop(key, None)
|
|
else:
|
|
if key not in presence and sum(k[0] == th for k in presence) >= MAX_PRESENCE_TABS_PER_SESSION:
|
|
return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Too many active tabs"})
|
|
presence[key] = (time.monotonic(), in_chat)
|
|
return self.send_json(HTTPStatus.OK, public_presence())
|
|
|
|
if path == "/api/chat":
|
|
with lock:
|
|
pid = self.authed_player()
|
|
if pid is None:
|
|
return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to send messages"})
|
|
message = data.get("message")
|
|
if not isinstance(message, str) or not (1 <= len(message.strip()) <= MAX_CHAT_LENGTH):
|
|
return self.send_json(HTTPStatus.BAD_REQUEST, {"error": f"Message must be 1-{MAX_CHAT_LENGTH} characters"})
|
|
message = message.strip()
|
|
if not message.isprintable():
|
|
return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Message contains invalid characters"})
|
|
with database() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
t = now_ms()
|
|
last = connection.execute(
|
|
"SELECT at FROM chat_messages WHERE player_id = ? AND kind = 'message' ORDER BY id DESC LIMIT 1", (pid,),
|
|
).fetchone()
|
|
if last is not None and t - last["at"] < CHAT_COOLDOWN_MS:
|
|
return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Please wait two seconds between messages"})
|
|
name = state["players"][pid]["name"]
|
|
cursor = connection.execute(
|
|
"INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
|
|
(pid, name, t, message),
|
|
)
|
|
posted = {"id": cursor.lastrowid, "name": name, "at": t, "message": message, "kind": "message"}
|
|
return self.send_json(HTTPStatus.CREATED, posted)
|
|
|
|
if path == "/api/register":
|
|
name, password, error = validate_credentials(data)
|
|
if error:
|
|
return self.send_json(HTTPStatus.BAD_REQUEST, {"error": error})
|
|
password_hash = hash_password(password) # slow; done outside the lock
|
|
with lock:
|
|
check_ended()
|
|
if state["ended"]:
|
|
return self.send_json(HTTPStatus.CONFLICT, {"error": "The game is over"})
|
|
if find_player_by_name(name) is not None:
|
|
return self.send_json(HTTPStatus.CONFLICT, {"error": "Name already taken"})
|
|
pid = secrets.token_hex(8)
|
|
state["players"][pid] = {
|
|
"name": name,
|
|
"password_hash": password_hash,
|
|
"token_hashes": [],
|
|
"created_at": now_ms(),
|
|
}
|
|
token = issue_token(pid)
|
|
return self.send_json(HTTPStatus.CREATED, {"name": name, "token": token})
|
|
|
|
if path == "/api/login":
|
|
name, password = data.get("name"), data.get("password")
|
|
if not isinstance(name, str) or not isinstance(password, str) or len(password) > MAX_PASSWORD_LEN:
|
|
return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Name and password required"})
|
|
name = " ".join(name.split())
|
|
key = name.lower()
|
|
with lock:
|
|
count, first = failed_logins.get(key, (0, 0))
|
|
if count >= LOGIN_MAX_FAILURES and now_ms() - first < LOGIN_LOCKOUT_MS:
|
|
return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Too many failed attempts; try again later"})
|
|
pid = find_player_by_name(name)
|
|
stored = state["players"][pid].get("password_hash") if pid else None
|
|
ok = verify_password(password, stored or DUMMY_PASSWORD_HASH) and stored is not None
|
|
with lock:
|
|
if not ok:
|
|
count, first = failed_logins.get(key, (0, 0))
|
|
if now_ms() - first >= LOGIN_LOCKOUT_MS:
|
|
count, first = 0, now_ms()
|
|
failed_logins[key] = (count + 1, first)
|
|
return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Invalid name or password"})
|
|
failed_logins.pop(key, None)
|
|
token = issue_token(pid)
|
|
return self.send_json(HTTPStatus.OK, {"name": state["players"][pid]["name"], "token": token})
|
|
|
|
if path == "/api/logout":
|
|
with lock:
|
|
th = self.bearer_hash()
|
|
pid = token_index.pop(th, None) if th else None
|
|
prune_presence()
|
|
if pid is not None:
|
|
state["players"][pid]["token_hashes"].remove(th)
|
|
save_state(state)
|
|
return self.send_json(HTTPStatus.OK, {"ok": True})
|
|
|
|
if path == "/api/subscribe":
|
|
with lock:
|
|
pid = self.authed_player()
|
|
player_name = state["players"][pid]["name"] if pid else None
|
|
already_subscribed = bool(pid and state["players"][pid].get("subscribed"))
|
|
if pid is None:
|
|
return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to join the mailing list"})
|
|
if already_subscribed:
|
|
return self.send_json(HTTPStatus.OK, {"ok": True})
|
|
entry, error = validate_subscription(data)
|
|
if error:
|
|
return self.send_json(HTTPStatus.BAD_REQUEST, {"error": error})
|
|
entry["player"] = player_name
|
|
with mailing_lock:
|
|
# Same response for repeat emails so the endpoint doesn't reveal who is subscribed.
|
|
if entry["email"].lower() not in subscribed_emails:
|
|
add_subscriber(entry)
|
|
with lock:
|
|
state["players"][pid]["subscribed"] = True
|
|
save_state(state)
|
|
return self.send_json(HTTPStatus.OK, {"ok": True})
|
|
|
|
if path == "/api/press":
|
|
with lock:
|
|
pid = self.authed_player()
|
|
if pid is None:
|
|
return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Not registered"})
|
|
t = now_ms()
|
|
check_ended(t)
|
|
if state["ended"]:
|
|
return self.send_json(HTTPStatus.CONFLICT, {"error": "The game is over"})
|
|
remaining = state["deadline"] - t
|
|
zone = zone_for_remaining(remaining)
|
|
press = {
|
|
"player_id": pid,
|
|
"at": t,
|
|
"remaining_ms": remaining,
|
|
"zone": zone["name"],
|
|
"points": zone["points"],
|
|
}
|
|
state["presses"].append(press)
|
|
with database() as connection:
|
|
season_end = current_season(connection)["ends_at"]
|
|
state["deadline"] = min(t + DURATION_MS, season_end)
|
|
hours, remainder = divmod(max(0, remaining), 3600000)
|
|
minutes, remainder = divmod(remainder, 60000)
|
|
seconds, milliseconds = divmod(remainder, 1000)
|
|
time_left = f"{hours:02d}:{minutes:02d}:{seconds:02d}.{milliseconds:03d}"
|
|
name = state["players"][pid]["name"]
|
|
points = press["points"]
|
|
announcement = f"{name} pushed the button with {time_left} remaining and received {points} {'point' if points == 1 else 'points'}."
|
|
try:
|
|
with database() as connection:
|
|
write_state(connection, state)
|
|
connection.execute(
|
|
"INSERT INTO chat_messages (player_id, name, at, message, kind) VALUES (?, ?, ?, ?, 'press')",
|
|
(pid, "The Button", t, announcement),
|
|
)
|
|
sync_milestones(connection, state, t)
|
|
except Exception:
|
|
# Keep memory consistent with the rolled-back database snapshot.
|
|
state["presses"].pop()
|
|
state["deadline"] = t + remaining
|
|
raise
|
|
return self.send_json(HTTPStatus.OK, {
|
|
"press": {k: press[k] for k in ("at", "remaining_ms", "zone", "points")},
|
|
"deadline": state["deadline"],
|
|
"now": t,
|
|
})
|
|
|
|
self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"})
|
|
|
|
|
|
mailing_lock = threading.Lock()
|
|
|
|
|
|
def load_subscribed_emails():
|
|
if not MAILING_LIST_FILE.exists():
|
|
return set()
|
|
with MAILING_LIST_FILE.open("r", newline="", encoding="utf-8") as f:
|
|
return {row["email"].lower() for row in csv.DictReader(f)}
|
|
|
|
|
|
subscribed_emails = load_subscribed_emails()
|
|
|
|
|
|
def clean_text(value, max_len):
|
|
"""Returns a whitespace-normalized string, "" for missing values, or None if invalid."""
|
|
if value is None:
|
|
return ""
|
|
if not isinstance(value, str):
|
|
return None
|
|
value = " ".join(value.split())
|
|
if len(value) > max_len or not value.isprintable():
|
|
return None
|
|
return value
|
|
|
|
|
|
def validate_subscription(data):
|
|
first = clean_text(data.get("first_name"), 50)
|
|
last = clean_text(data.get("last_name"), 50)
|
|
email = clean_text(data.get("email"), 254)
|
|
phone = clean_text(data.get("phone"), 25)
|
|
if not first:
|
|
return None, "First name is required (max 50 characters)"
|
|
if last is None:
|
|
return None, "Last name must be at most 50 characters"
|
|
if not email or not EMAIL_RE.match(email):
|
|
return None, "A valid email address is required"
|
|
if phone is None or (phone and not PHONE_RE.match(phone)):
|
|
return None, "Phone number is invalid"
|
|
return {"first_name": first, "last_name": last, "email": email, "phone": phone}, None
|
|
|
|
|
|
def csv_safe(value):
|
|
# Stops spreadsheet apps from interpreting user input as a formula.
|
|
return "'" + value if value[:1] in ("=", "+", "-", "@") else value
|
|
|
|
|
|
def add_subscriber(entry):
|
|
"""Must be called with mailing_lock held."""
|
|
is_new_file = not MAILING_LIST_FILE.exists()
|
|
row = {k: csv_safe(v) for k, v in entry.items()}
|
|
row["subscribed_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
with MAILING_LIST_FILE.open("a", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=MAILING_LIST_FIELDS)
|
|
if is_new_file:
|
|
writer.writeheader()
|
|
writer.writerow(row)
|
|
subscribed_emails.add(entry["email"].lower())
|
|
|
|
|
|
def end_watcher():
|
|
while True:
|
|
with lock:
|
|
check_ended()
|
|
with database() as connection:
|
|
next_boundary = current_season(connection)["ends_at"]
|
|
wake_at = next_boundary if state["ended"] else min(state["deadline"], next_boundary)
|
|
wait = (wake_at - now_ms()) / 1000
|
|
time.sleep(min(max(wait, 0.05), 60))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
threading.Thread(target=end_watcher, daemon=True).start()
|
|
print(f"The Button running on http://{HOST}:{PORT} (timer: {DURATION_MS / 1000:.0f}s)")
|
|
ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()
|