storage: SQLite s arkádovou identitou hráčů
Ručně psané SQL nad pěti tabulkami, žádné ORM. - hry ukazují na players.id, nikdy na jméno, takže se přejmenování propíše i do starých her - párování hráčů napříč hrami přes normalizovaný klíč (trim, casefold), plus slučování pro případ překlepu - zapisuje se každý tah hned, ne až na konci — rozehraná hra se po pádu serveru najde podle stavu a dá se v ní pokračovat - součty se neukládají nikde, stav hry se rekonstruuje načtením tahů - časy jako ISO text, sqlite3 vlastní adaptéry pro datetime zavrhl Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
gitea-actionsandClaude Opus 5 committed
1 parent
4743ea74b9
commit
a45a82ae0e
2 files changed
+423
No files matched your search
+289
@@ -0,0 +1,289 @@
|
||||
"""Trvalé úložiště nad SQLite.
|
||||
|
||||
Ručně psané SQL nad pěti tabulkami, žádné ORM. Hry ukazují na `players.id`,
|
||||
nikdy na jméno — přejmenování hráče se tím propíše i do starých her.
|
||||
|
||||
Součty se nikde neukládají; stav hry se rekonstruuje načtením tahů do
|
||||
`core.Game`.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from core import FINAL_SCORE, Game, Turn
|
||||
|
||||
DB_PATH = Path(os.environ.get("DICE_DB", "dice.db"))
|
||||
|
||||
STATUS_RUNNING = "probíhá"
|
||||
STATUS_FINISHED = "dohráno"
|
||||
STATUS_ABANDONED = "opuštěno"
|
||||
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS players (
|
||||
id INTEGER PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
key TEXT NOT NULL UNIQUE,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS games (
|
||||
id INTEGER PRIMARY KEY,
|
||||
final_score INTEGER NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
winner_id INTEGER REFERENCES players(id),
|
||||
started_at TEXT NOT NULL,
|
||||
finished_at TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS game_players (
|
||||
game_id INTEGER NOT NULL REFERENCES games(id),
|
||||
player_id INTEGER NOT NULL REFERENCES players(id),
|
||||
seat INTEGER NOT NULL,
|
||||
PRIMARY KEY (game_id, player_id)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS turns (
|
||||
id INTEGER PRIMARY KEY,
|
||||
game_id INTEGER NOT NULL REFERENCES games(id),
|
||||
player_id INTEGER NOT NULL REFERENCES players(id),
|
||||
round INTEGER NOT NULL,
|
||||
value INTEGER NOT NULL,
|
||||
reset INTEGER NOT NULL DEFAULT 0,
|
||||
at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS turns_game ON turns(game_id, id);
|
||||
"""
|
||||
|
||||
|
||||
def connect(path: Path | str | None = None) -> sqlite3.Connection:
|
||||
conn = sqlite3.connect(path or DB_PATH)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA foreign_keys = ON")
|
||||
conn.executescript(SCHEMA)
|
||||
return conn
|
||||
|
||||
|
||||
def now() -> str:
|
||||
"""Časy ukládáme jako ISO text — sqlite3 vlastní adaptéry pro datetime
|
||||
mezitím zavrhl."""
|
||||
return datetime.now().isoformat(timespec="seconds")
|
||||
|
||||
|
||||
# --- hráči ------------------------------------------------------------------
|
||||
|
||||
def normalize(name: str) -> str:
|
||||
"""Klíč pro párování jmen napříč hrami: Adam, adam i " Adam " je týž hráč."""
|
||||
return re.sub(r"\s+", " ", name).strip().casefold()
|
||||
|
||||
|
||||
def get_or_create_player(conn: sqlite3.Connection, name: str) -> int:
|
||||
name = re.sub(r"\s+", " ", name).strip()
|
||||
if not name:
|
||||
raise ValueError("Jméno nesmí být prázdné.")
|
||||
|
||||
key = normalize(name)
|
||||
row = conn.execute("SELECT id FROM players WHERE key = ?", (key,)).fetchone()
|
||||
if row:
|
||||
return row["id"]
|
||||
|
||||
cursor = conn.execute(
|
||||
"INSERT INTO players (name, key, created_at) VALUES (?, ?, ?)",
|
||||
(name, key, now()),
|
||||
)
|
||||
conn.commit()
|
||||
return cursor.lastrowid
|
||||
|
||||
|
||||
def known_players(conn: sqlite3.Connection) -> list[sqlite3.Row]:
|
||||
"""Známí hráči k odklikání, naposledy hrající první."""
|
||||
return conn.execute("""
|
||||
SELECT p.id, p.name, MAX(t.at) AS last_played
|
||||
FROM players p
|
||||
LEFT JOIN turns t ON t.player_id = p.id
|
||||
GROUP BY p.id
|
||||
ORDER BY last_played DESC, p.name COLLATE NOCASE
|
||||
""").fetchall()
|
||||
|
||||
|
||||
def rename_player(conn: sqlite3.Connection, player_id: int, name: str) -> None:
|
||||
name = re.sub(r"\s+", " ", name).strip()
|
||||
if not name:
|
||||
raise ValueError("Jméno nesmí být prázdné.")
|
||||
|
||||
conn.execute("UPDATE players SET name = ?, key = ? WHERE id = ?",
|
||||
(name, normalize(name), player_id))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def merge_players(conn: sqlite3.Connection, source_id: int, target_id: int) -> None:
|
||||
"""Sloučí hráče vzniklého překlepem do toho správného."""
|
||||
if source_id == target_id:
|
||||
return
|
||||
|
||||
# Kdyby oba omylem hráli tutéž hru, druhé sedadlo zahodíme.
|
||||
conn.execute("""
|
||||
DELETE FROM game_players
|
||||
WHERE player_id = ?
|
||||
AND game_id IN (SELECT game_id FROM game_players WHERE player_id = ?)
|
||||
""", (source_id, target_id))
|
||||
|
||||
conn.execute("UPDATE game_players SET player_id = ? WHERE player_id = ?",
|
||||
(target_id, source_id))
|
||||
conn.execute("UPDATE turns SET player_id = ? WHERE player_id = ?",
|
||||
(target_id, source_id))
|
||||
conn.execute("UPDATE games SET winner_id = ? WHERE winner_id = ?",
|
||||
(target_id, source_id))
|
||||
conn.execute("DELETE FROM players WHERE id = ?", (source_id,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
# --- hry --------------------------------------------------------------------
|
||||
|
||||
def create_game(conn: sqlite3.Connection, names: list[str],
|
||||
final_score: int = FINAL_SCORE) -> int:
|
||||
"""Založí hru. Pořadí jmen je pořadí u stolu."""
|
||||
ids = [get_or_create_player(conn, name) for name in names]
|
||||
if len(set(ids)) != len(ids):
|
||||
raise ValueError("Každý hráč může u stolu sedět jen jednou.")
|
||||
|
||||
cursor = conn.execute(
|
||||
"INSERT INTO games (final_score, status, started_at) VALUES (?, ?, ?)",
|
||||
(final_score, STATUS_RUNNING, now()),
|
||||
)
|
||||
game_id = cursor.lastrowid
|
||||
|
||||
conn.executemany(
|
||||
"INSERT INTO game_players (game_id, player_id, seat) VALUES (?, ?, ?)",
|
||||
[(game_id, player_id, seat) for seat, player_id in enumerate(ids)],
|
||||
)
|
||||
conn.commit()
|
||||
return game_id
|
||||
|
||||
|
||||
def game_row(conn: sqlite3.Connection, game_id: int) -> sqlite3.Row | None:
|
||||
return conn.execute("""
|
||||
SELECT g.*, w.name AS winner_name
|
||||
FROM games g
|
||||
LEFT JOIN players w ON w.id = g.winner_id
|
||||
WHERE g.id = ?
|
||||
""", (game_id,)).fetchone()
|
||||
|
||||
|
||||
def seating(conn: sqlite3.Connection, game_id: int) -> list[sqlite3.Row]:
|
||||
return conn.execute("""
|
||||
SELECT p.id, p.name
|
||||
FROM game_players gp
|
||||
JOIN players p ON p.id = gp.player_id
|
||||
WHERE gp.game_id = ?
|
||||
ORDER BY gp.seat
|
||||
""", (game_id,)).fetchall()
|
||||
|
||||
|
||||
def load_game(conn: sqlite3.Connection, game_id: int) -> Game:
|
||||
"""Rekonstruuje `core.Game` z uložených tahů."""
|
||||
row = game_row(conn, game_id)
|
||||
if row is None:
|
||||
raise LookupError(f"Hra {game_id} neexistuje.")
|
||||
|
||||
names = [player["name"] for player in seating(conn, game_id)]
|
||||
|
||||
turn_rows = conn.execute("""
|
||||
SELECT p.name, t.round, t.value, t.reset, t.at
|
||||
FROM turns t
|
||||
JOIN players p ON p.id = t.player_id
|
||||
WHERE t.game_id = ?
|
||||
ORDER BY t.id
|
||||
""", (game_id,)).fetchall()
|
||||
|
||||
turns = [
|
||||
Turn(player=t["name"], round=t["round"], value=t["value"],
|
||||
reset=bool(t["reset"]), at=datetime.fromisoformat(t["at"]))
|
||||
for t in turn_rows
|
||||
]
|
||||
|
||||
return Game(names, final_score=row["final_score"], turns=turns)
|
||||
|
||||
|
||||
def running_game(conn: sqlite3.Connection) -> int | None:
|
||||
"""Poslední rozehraná hra — díky ní se dá po pádu serveru pokračovat."""
|
||||
row = conn.execute(
|
||||
"SELECT id FROM games WHERE status = ? ORDER BY id DESC LIMIT 1",
|
||||
(STATUS_RUNNING,),
|
||||
).fetchone()
|
||||
return row["id"] if row else None
|
||||
|
||||
|
||||
def recent_games(conn: sqlite3.Connection, limit: int = 20) -> list[sqlite3.Row]:
|
||||
return conn.execute("""
|
||||
SELECT g.id, g.final_score, g.status, g.started_at, g.finished_at,
|
||||
w.name AS winner_name,
|
||||
(SELECT COUNT(*) FROM game_players gp WHERE gp.game_id = g.id) AS players,
|
||||
(SELECT COUNT(*) FROM turns t WHERE t.game_id = g.id) AS turns
|
||||
FROM games g
|
||||
LEFT JOIN players w ON w.id = g.winner_id
|
||||
ORDER BY g.id DESC
|
||||
LIMIT ?
|
||||
""", (limit,)).fetchall()
|
||||
|
||||
|
||||
# --- tahy -------------------------------------------------------------------
|
||||
|
||||
def add_turn(conn: sqlite3.Connection, game_id: int, turn: Turn) -> None:
|
||||
"""Zapisuje se hned po každém tahu, ne až na konci hry."""
|
||||
row = conn.execute("""
|
||||
SELECT p.id
|
||||
FROM game_players gp
|
||||
JOIN players p ON p.id = gp.player_id
|
||||
WHERE gp.game_id = ? AND p.name = ?
|
||||
""", (game_id, turn.player)).fetchone()
|
||||
|
||||
if row is None:
|
||||
raise LookupError(f"Hráč {turn.player} v této hře nehraje.")
|
||||
|
||||
conn.execute("""
|
||||
INSERT INTO turns (game_id, player_id, round, value, reset, at)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""", (game_id, row["id"], turn.round, turn.value, int(turn.reset),
|
||||
turn.at.isoformat(timespec="seconds")))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def undo_turn(conn: sqlite3.Connection, game_id: int) -> bool:
|
||||
"""Zahodí poslední zápis. Dá se opakovat."""
|
||||
row = conn.execute(
|
||||
"SELECT id FROM turns WHERE game_id = ? ORDER BY id DESC LIMIT 1",
|
||||
(game_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return False
|
||||
|
||||
conn.execute("DELETE FROM turns WHERE id = ?", (row["id"],))
|
||||
conn.execute(
|
||||
"UPDATE games SET status = ?, winner_id = NULL, finished_at = NULL WHERE id = ?",
|
||||
(STATUS_RUNNING, game_id),
|
||||
)
|
||||
conn.commit()
|
||||
return True
|
||||
|
||||
|
||||
def finish_game(conn: sqlite3.Connection, game_id: int, winner: str) -> None:
|
||||
row = conn.execute("SELECT id FROM players WHERE key = ?",
|
||||
(normalize(winner),)).fetchone()
|
||||
conn.execute(
|
||||
"UPDATE games SET status = ?, winner_id = ?, finished_at = ? WHERE id = ?",
|
||||
(STATUS_FINISHED, row["id"] if row else None, now(), game_id),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def abandon_game(conn: sqlite3.Connection, game_id: int) -> None:
|
||||
"""Nedohrané hry se nepočítají do statistik výher."""
|
||||
conn.execute(
|
||||
"UPDATE games SET status = ?, finished_at = ? WHERE id = ?",
|
||||
(STATUS_ABANDONED, now(), game_id),
|
||||
)
|
||||
conn.commit()
|
||||
@@ -0,0 +1,134 @@
|
||||
import pytest
|
||||
|
||||
import storage
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def conn(tmp_path):
|
||||
connection = storage.connect(tmp_path / "test.db")
|
||||
yield connection
|
||||
connection.close()
|
||||
|
||||
|
||||
def test_stejne_jmeno_je_stejny_hrac(conn):
|
||||
"""Arkádová identita: Adam z jedné hry je Adam i v další."""
|
||||
first = storage.create_game(conn, ["Adam", "Eva"])
|
||||
second = storage.create_game(conn, ["adam", "Petr"])
|
||||
|
||||
adam_first = [p["id"] for p in storage.seating(conn, first) if p["name"] == "Adam"]
|
||||
adam_second = [p["id"] for p in storage.seating(conn, second)][0]
|
||||
|
||||
assert adam_first[0] == adam_second
|
||||
|
||||
|
||||
def test_normalizace_jmen():
|
||||
assert storage.normalize(" Adam ") == storage.normalize("adam")
|
||||
assert storage.normalize("Jan Novák") == storage.normalize("jan novák")
|
||||
assert storage.normalize("Adam") != storage.normalize("Ádám")
|
||||
|
||||
|
||||
def test_prejmenovani_se_propise_do_starych_her(conn):
|
||||
game_id = storage.create_game(conn, ["Adam", "Eva"])
|
||||
adam = storage.seating(conn, game_id)[0]["id"]
|
||||
|
||||
storage.rename_player(conn, adam, "Adam Š.")
|
||||
|
||||
assert [p["name"] for p in storage.seating(conn, game_id)] == ["Adam Š.", "Eva"]
|
||||
|
||||
|
||||
def test_slouceni_hracu_po_preklepu(conn):
|
||||
good = storage.create_game(conn, ["Adam", "Eva"])
|
||||
typo = storage.create_game(conn, ["Adm", "Eva"])
|
||||
|
||||
adam = storage.get_or_create_player(conn, "Adam")
|
||||
adm = storage.get_or_create_player(conn, "Adm")
|
||||
|
||||
game = storage.load_game(conn, typo)
|
||||
storage.add_turn(conn, typo, game.add_score(150).turn)
|
||||
|
||||
storage.merge_players(conn, adm, adam)
|
||||
|
||||
assert [p["name"] for p in storage.seating(conn, typo)] == ["Adam", "Eva"]
|
||||
assert storage.load_game(conn, typo).totals()["Adam"] == 150
|
||||
assert [p["id"] for p in storage.known_players(conn)].count(adm) == 0
|
||||
assert storage.load_game(conn, good).totals() == {"Adam": 0, "Eva": 0}
|
||||
|
||||
|
||||
def test_hra_se_nacte_zpatky_i_s_tahy(conn):
|
||||
game_id = storage.create_game(conn, ["Adam", "Eva"], final_score=500)
|
||||
game = storage.load_game(conn, game_id)
|
||||
|
||||
for value in (100, 50, 0, 30):
|
||||
storage.add_turn(conn, game_id, game.add_score(value).turn)
|
||||
|
||||
loaded = storage.load_game(conn, game_id)
|
||||
|
||||
assert loaded.names == ["Adam", "Eva"]
|
||||
assert loaded.final_score == 500
|
||||
assert loaded.totals() == {"Adam": 100, "Eva": 80}
|
||||
assert loaded.current_player == "Adam"
|
||||
assert loaded.round_number == 3
|
||||
|
||||
|
||||
def test_vynulovani_se_uklada_jako_nula_s_priznakem(conn):
|
||||
game_id = storage.create_game(conn, ["Adam"], final_score=500)
|
||||
game = storage.load_game(conn, game_id)
|
||||
|
||||
for value in (300, 0, 0, 0):
|
||||
storage.add_turn(conn, game_id, game.add_score(value).turn)
|
||||
|
||||
rows = conn.execute("SELECT value, reset FROM turns ORDER BY id").fetchall()
|
||||
|
||||
assert [r["value"] for r in rows] == [300, 0, 0, 0]
|
||||
assert [r["reset"] for r in rows] == [0, 0, 0, 1]
|
||||
assert storage.load_game(conn, game_id).totals()["Adam"] == 0
|
||||
|
||||
|
||||
def test_rozehrana_hra_se_najde_a_dokonci(conn):
|
||||
game_id = storage.create_game(conn, ["Adam", "Eva"], final_score=100)
|
||||
|
||||
assert storage.running_game(conn) == game_id
|
||||
|
||||
game = storage.load_game(conn, game_id)
|
||||
storage.add_turn(conn, game_id, game.add_score(100).turn)
|
||||
result = game.add_score(10)
|
||||
storage.add_turn(conn, game_id, result.turn)
|
||||
storage.finish_game(conn, game_id, result.winner)
|
||||
|
||||
assert storage.running_game(conn) is None
|
||||
assert storage.game_row(conn, game_id)["status"] == storage.STATUS_FINISHED
|
||||
assert storage.game_row(conn, game_id)["winner_name"] == "Adam"
|
||||
|
||||
|
||||
def test_undo_smaze_tah_a_vrati_hru_do_hry(conn):
|
||||
game_id = storage.create_game(conn, ["Adam", "Eva"], final_score=100)
|
||||
game = storage.load_game(conn, game_id)
|
||||
storage.add_turn(conn, game_id, game.add_score(100).turn)
|
||||
result = game.add_score(10)
|
||||
storage.add_turn(conn, game_id, result.turn)
|
||||
storage.finish_game(conn, game_id, result.winner)
|
||||
|
||||
assert storage.undo_turn(conn, game_id)
|
||||
|
||||
row = storage.game_row(conn, game_id)
|
||||
assert row["status"] == storage.STATUS_RUNNING
|
||||
assert row["winner_id"] is None
|
||||
assert storage.load_game(conn, game_id).current_player == "Eva"
|
||||
|
||||
|
||||
def test_opustena_hra(conn):
|
||||
game_id = storage.create_game(conn, ["Adam", "Eva"])
|
||||
storage.abandon_game(conn, game_id)
|
||||
|
||||
assert storage.running_game(conn) is None
|
||||
assert storage.game_row(conn, game_id)["status"] == storage.STATUS_ABANDONED
|
||||
|
||||
|
||||
def test_hrac_nesmi_sedet_u_stolu_dvakrat(conn):
|
||||
with pytest.raises(ValueError):
|
||||
storage.create_game(conn, ["Adam", "adam"])
|
||||
|
||||
|
||||
def test_prazdne_jmeno(conn):
|
||||
with pytest.raises(ValueError):
|
||||
storage.get_or_create_player(conn, " ")
|
||||
Reference in new issue
Block a user