218 lines
6.7 KiB
Python
218 lines
6.7 KiB
Python
"""
|
|
SQLite-backed game state persistence for MLP Pony.
|
|
|
|
Provides a persistent store for game sessions, replacing Flask in-memory sessions.
|
|
Each game gets a UUID, stored in a SQLite database.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
|
|
DB_PATH = os.path.join(os.path.dirname(__file__), "..", "game_state.db")
|
|
|
|
|
|
def get_connection():
|
|
"""Get a database connection, creating tables if needed."""
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
_ensure_tables(conn)
|
|
return conn
|
|
|
|
|
|
def _ensure_tables(conn):
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS games (
|
|
game_id TEXT PRIMARY KEY,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
state TEXT NOT NULL
|
|
)
|
|
""")
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS game_stats (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
game_id TEXT NOT NULL,
|
|
pony_type TEXT NOT NULL,
|
|
tema_id TEXT NOT NULL,
|
|
tema_title TEXT NOT NULL,
|
|
successes INTEGER NOT NULL DEFAULT 0,
|
|
failures INTEGER NOT NULL DEFAULT 0,
|
|
total_scenes INTEGER NOT NULL DEFAULT 0,
|
|
dice_rolls TEXT DEFAULT '[]',
|
|
victory INTEGER NOT NULL DEFAULT 0,
|
|
played_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_game_stats_played_at ON game_stats(played_at)")
|
|
conn.commit()
|
|
|
|
|
|
def create_game(game_id, state):
|
|
"""Insert a new game state. Returns True if created."""
|
|
conn = get_connection()
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO games (game_id, state) VALUES (?, ?)",
|
|
(game_id, json.dumps(state))
|
|
)
|
|
conn.commit()
|
|
return True
|
|
except sqlite3.IntegrityError:
|
|
return False
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_game(game_id):
|
|
"""Get game state by ID. Returns dict or None."""
|
|
conn = get_connection()
|
|
row = conn.execute("SELECT state FROM games WHERE game_id = ?", (game_id,)).fetchone()
|
|
conn.close()
|
|
if row:
|
|
return json.loads(row["state"])
|
|
return None
|
|
|
|
|
|
def update_game(game_id, state):
|
|
"""Update game state. Returns True if updated."""
|
|
conn = get_connection()
|
|
result = conn.execute(
|
|
"UPDATE games SET state = ?, updated_at = CURRENT_TIMESTAMP WHERE game_id = ?",
|
|
(json.dumps(state), game_id)
|
|
)
|
|
conn.commit()
|
|
updated = result.rowcount > 0
|
|
conn.close()
|
|
return updated
|
|
|
|
|
|
def delete_game(game_id):
|
|
"""Delete a game. Returns True if deleted."""
|
|
conn = get_connection()
|
|
result = conn.execute("DELETE FROM games WHERE game_id = ?", (game_id,))
|
|
conn.commit()
|
|
deleted = result.rowcount > 0
|
|
conn.close()
|
|
return deleted
|
|
|
|
|
|
def list_games(limit=10):
|
|
"""List recent games. Returns list of dicts with game_id, created_at."""
|
|
conn = get_connection()
|
|
rows = conn.execute(
|
|
"SELECT game_id, created_at, updated_at FROM games ORDER BY updated_at DESC LIMIT ?",
|
|
(limit,)
|
|
).fetchall()
|
|
conn.close()
|
|
return [{"game_id": r["game_id"], "created_at": r["created_at"], "updated_at": r["updated_at"]}
|
|
for r in rows]
|
|
|
|
|
|
def record_game_summary(game):
|
|
"""Record a completed game's stats for analytics.
|
|
|
|
Args:
|
|
game: game state dict (must have 'færdig' == True)
|
|
"""
|
|
pony = game.get("pony", {})
|
|
tema = game.get("tema", {})
|
|
tema_id = tema.get("id", "unknown")
|
|
tema_title = tema.get("titel", "Unknown")
|
|
scenes = tema.get("scener", [])
|
|
|
|
# Collect dice rolls from history
|
|
dice_rolls = []
|
|
for h in game.get("historie", []):
|
|
for d in h.get("dice", []):
|
|
dice_rolls.append(d)
|
|
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"""INSERT INTO game_stats
|
|
(game_id, pony_type, tema_id, tema_title, successes, failures,
|
|
total_scenes, dice_rolls, victory)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
game.get("game_id", ""),
|
|
pony.get("type", "unknown"),
|
|
tema_id,
|
|
tema_title,
|
|
game.get("succeser", 0),
|
|
game.get("fiaskoer", 0),
|
|
len(scenes),
|
|
json.dumps(dice_rolls),
|
|
1 if game.get("succeser", 0) >= len(scenes) else 0,
|
|
)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def get_stats_summary():
|
|
"""Return aggregated game statistics.
|
|
|
|
Returns dict with:
|
|
- total_games, total_victories, win_rate
|
|
- by_pony_type: list of {type, games, avg_successes, win_rate}
|
|
- by_tema: list of {tema_id, tema_title, games, avg_successes, win_rate}
|
|
- recent_games: last 10 completed games
|
|
"""
|
|
conn = get_connection()
|
|
|
|
row = conn.execute(
|
|
"SELECT COUNT(*) as total, SUM(victory) as victories FROM game_stats"
|
|
).fetchone()
|
|
total_games = row["total"] or 0
|
|
total_victories = row["victories"] or 0
|
|
|
|
by_pony = []
|
|
for r in conn.execute(
|
|
"SELECT pony_type, COUNT(*) as games, AVG(successes) as avg_s, "
|
|
"SUM(victory) as wins FROM game_stats GROUP BY pony_type ORDER BY games DESC"
|
|
).fetchall():
|
|
by_pony.append({
|
|
"type": r["pony_type"],
|
|
"games": r["games"],
|
|
"avg_successes": round(r["avg_s"] or 0, 2),
|
|
"win_rate": round((r["wins"] / r["games"] * 100) if r["games"] > 0 else 0, 1),
|
|
})
|
|
|
|
by_tema = []
|
|
for r in conn.execute(
|
|
"SELECT tema_id, tema_title, COUNT(*) as games, AVG(successes) as avg_s, "
|
|
"SUM(victory) as wins FROM game_stats GROUP BY tema_id ORDER BY games DESC"
|
|
).fetchall():
|
|
by_tema.append({
|
|
"tema_id": r["tema_id"],
|
|
"tema_title": r["tema_title"],
|
|
"games": r["games"],
|
|
"avg_successes": round(r["avg_s"] or 0, 2),
|
|
"win_rate": round((r["wins"] / r["games"] * 100) if r["games"] > 0 else 0, 1),
|
|
})
|
|
|
|
recent = []
|
|
for r in conn.execute(
|
|
"SELECT game_id, pony_type, tema_title, successes, failures, "
|
|
"total_scenes, victory, played_at FROM game_stats ORDER BY played_at DESC LIMIT 10"
|
|
).fetchall():
|
|
recent.append({
|
|
"game_id": r["game_id"],
|
|
"pony_type": r["pony_type"],
|
|
"tema": r["tema_title"],
|
|
"successes": r["successes"],
|
|
"failures": r["failures"],
|
|
"total_scenes": r["total_scenes"],
|
|
"victory": r["victory"],
|
|
"played_at": r["played_at"],
|
|
})
|
|
|
|
conn.close()
|
|
return {
|
|
"total_games": total_games,
|
|
"total_victories": total_victories,
|
|
"win_rate": round((total_victories / total_games * 100) if total_games > 0 else 0, 1),
|
|
"by_pony_type": by_pony,
|
|
"by_tema": by_tema,
|
|
"recent_games": recent,
|
|
} |