File size: 4,532 Bytes
29a88f8
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
b8912e3
 
 
29a88f8
b8912e3
29a88f8
 
 
 
 
 
 
 
 
b8912e3
29a88f8
 
 
 
b8912e3
29a88f8
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
"""
Player stats — stored in SQLite.

On HF Spaces (Docker mode), persistent storage mounts at /data.
Enable it in the Space settings: Settings → Persistent Storage → Enable.

For local dev, falls back to ./data/stats.db (relative to this file).
Override with env var STATS_DB_PATH.
"""

from __future__ import annotations

import os
import sqlite3
import time
from pathlib import Path
from typing import Literal, Optional

EventType = Literal["tutorial_complete", "game_won"]

_DEFAULT_DB = "/data/stats.db" if Path("/data").exists() else str(Path(__file__).parent / "data" / "stats.db")
DB_PATH = os.environ.get("STATS_DB_PATH", _DEFAULT_DB)


def _connect() -> sqlite3.Connection:
    Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    return conn


def init_db() -> None:
    with _connect() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS player_stats (
                id            INTEGER PRIMARY KEY AUTOINCREMENT,
                player_name   TEXT    NOT NULL,
                event_type    TEXT    NOT NULL,
                duration_s    INTEGER,
                opponent_name TEXT,
                recorded_at   INTEGER NOT NULL
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_event ON player_stats(event_type)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_recorded ON player_stats(recorded_at DESC)")
        # Migrate: add opponent_name if missing (schema upgrade)
        try:
            conn.execute("ALTER TABLE player_stats ADD COLUMN opponent_name TEXT")
        except Exception:
            pass
        conn.commit()


def record(
    player_name: str,
    event_type: EventType,
    duration_s: int | None,
    opponent_name: Optional[str] = None,
) -> None:
    with _connect() as conn:
        conn.execute(
            """INSERT INTO player_stats
               (player_name, event_type, duration_s, opponent_name, recorded_at)
               VALUES (?, ?, ?, ?, ?)""",
            (player_name, event_type, duration_s, opponent_name, int(time.time())),
        )
        conn.commit()


def get_tutorial_leaderboard(limit: int = 50) -> list[dict]:
    """Fastest tutorial completions, one entry per player (personal best)."""
    with _connect() as conn:
        rows = conn.execute(
            """
            SELECT player_name,
                   MIN(duration_s) AS best_s,
                   COUNT(*)        AS attempts,
                   MAX(recorded_at) AS last_at
            FROM player_stats
            WHERE event_type = 'tutorial_complete'
            GROUP BY player_name
            ORDER BY best_s ASC NULLS LAST
            LIMIT ?
            """,
            (limit,),
        ).fetchall()
    return [dict(r) for r in rows]


_BOT_PLAYER_NAME = "Bot IA"


def get_wins_leaderboard(limit: int = 50) -> list[dict]:
    """Most 1v1 wins, ordered by win count then best time. Bots are excluded."""
    with _connect() as conn:
        rows = conn.execute(
            """
            SELECT player_name,
                   COUNT(*)        AS wins,
                   MIN(duration_s) AS best_s,
                   MAX(recorded_at) AS last_at
            FROM player_stats
            WHERE event_type = 'game_won'
              AND player_name != ?
            GROUP BY player_name
            ORDER BY wins DESC, best_s ASC NULLS LAST
            LIMIT ?
            """,
            (_BOT_PLAYER_NAME, limit),
        ).fetchall()
    return [dict(r) for r in rows]


def get_recent_matches(limit: int = 20) -> list[dict]:
    """Recent 1v1 match results."""
    with _connect() as conn:
        rows = conn.execute(
            """
            SELECT player_name, opponent_name, duration_s, recorded_at
            FROM player_stats
            WHERE event_type = 'game_won'
            ORDER BY recorded_at DESC
            LIMIT ?
            """,
            (limit,),
        ).fetchall()
    return [dict(r) for r in rows]


def get_recent_events(limit: int = 10) -> list[dict]:
    """Latest events of any type — used for landing-page toast notifications."""
    with _connect() as conn:
        rows = conn.execute(
            """
            SELECT player_name, event_type, duration_s, opponent_name, recorded_at
            FROM player_stats
            ORDER BY recorded_at DESC
            LIMIT ?
            """,
            (limit,),
        ).fetchall()
    return [dict(r) for r in rows]