backend/stats.py

"""
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]