81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
"""Leaderboard — stockage SQLite et requêtes."""
|
|
|
|
import aiosqlite
|
|
from pathlib import Path
|
|
|
|
DB_PATH = str(Path(__file__).parent / "soulgate.db")
|
|
|
|
_CREATE = """
|
|
CREATE TABLE IF NOT EXISTS game_results (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
team_name TEXT NOT NULL,
|
|
score INTEGER NOT NULL,
|
|
time_seconds INTEGER NOT NULL,
|
|
waves_completed INTEGER NOT NULL,
|
|
victory INTEGER NOT NULL DEFAULT 0,
|
|
p1_username TEXT, p1_class TEXT, p1_discord TEXT,
|
|
p2_username TEXT, p2_class TEXT, p2_discord TEXT,
|
|
p3_username TEXT, p3_class TEXT, p3_discord TEXT,
|
|
played_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
)
|
|
"""
|
|
|
|
|
|
async def init_db() -> None:
|
|
async with aiosqlite.connect(DB_PATH) as db:
|
|
await db.execute(_CREATE)
|
|
await db.commit()
|
|
|
|
|
|
async def is_discord_taken(tag: str) -> bool:
|
|
"""Retourne True si ce tag Discord est déjà enregistré dans le leaderboard."""
|
|
if not tag:
|
|
return False
|
|
async with aiosqlite.connect(DB_PATH) as db:
|
|
async with db.execute(
|
|
"SELECT COUNT(*) FROM game_results WHERE p1_discord=? OR p2_discord=? OR p3_discord=?",
|
|
(tag, tag, tag),
|
|
) as cur:
|
|
row = await cur.fetchone()
|
|
return bool(row and row[0] > 0)
|
|
|
|
|
|
async def save_result(data: dict) -> int:
|
|
"""Sauvegarde un résultat de partie. Retourne le rang (1 = meilleur score)."""
|
|
async with aiosqlite.connect(DB_PATH) as db:
|
|
await db.execute(
|
|
"""INSERT INTO game_results
|
|
(team_name, score, time_seconds, waves_completed, victory,
|
|
p1_username, p1_class, p1_discord,
|
|
p2_username, p2_class, p2_discord,
|
|
p3_username, p3_class, p3_discord)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
|
(
|
|
data["team_name"],
|
|
int(data["score"]),
|
|
int(data["time_seconds"]),
|
|
int(data["waves_completed"]),
|
|
1 if data.get("victory") else 0,
|
|
data.get("p1_username"), data.get("p1_class"), data.get("p1_discord") or None,
|
|
data.get("p2_username"), data.get("p2_class"), data.get("p2_discord") or None,
|
|
data.get("p3_username"), data.get("p3_class"), data.get("p3_discord") or None,
|
|
),
|
|
)
|
|
await db.commit()
|
|
async with db.execute(
|
|
"SELECT COUNT(*) FROM game_results WHERE score > ?", (int(data["score"]),)
|
|
) as cur:
|
|
row = await cur.fetchone()
|
|
return (row[0] if row else 0) + 1 # rang 1-based
|
|
|
|
|
|
async def get_leaderboard(limit: int = 20) -> list[dict]:
|
|
"""Retourne les `limit` meilleures parties par score décroissant."""
|
|
async with aiosqlite.connect(DB_PATH) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
async with db.execute(
|
|
"SELECT * FROM game_results ORDER BY score DESC LIMIT ?", (limit,)
|
|
) as cur:
|
|
rows = await cur.fetchall()
|
|
return [dict(r) for r in rows]
|
