perefouras/database.py

# database.py
import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Tuple

DB_FILE = "data/database.db"


def ensure_db() -> None:
    """Initialize SQLite database and create tables if needed."""
    Path(DB_FILE).parent.mkdir(parents=True, exist_ok=True)

    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS guild_state (
                guild_id TEXT PRIMARY KEY,
                guild_name TEXT NOT NULL DEFAULT '',
                cooldown_until TEXT NOT NULL DEFAULT '1970-01-01T00:00:00',
                cooldown_ratio REAL NOT NULL DEFAULT 1.0,
                self_control REAL NOT NULL DEFAULT 1.0,
                last_updated TEXT NOT NULL DEFAULT (datetime('now'))
            )
        """)
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS active_riddles (
                channel_id TEXT PRIMARY KEY,
                riddle_index INTEGER NOT NULL,
                nb_clues INTEGER NOT NULL DEFAULT -1,
                message_id INTEGER NOT NULL,
                solver_id TEXT,
                created_at TEXT NOT NULL DEFAULT (datetime('now')),
                updated_at TEXT NOT NULL DEFAULT (datetime('now'))
            )
        """)
        conn.commit()


def get_connection() -> sqlite3.Connection:
    """Return SQLite connection with row factory for named column access."""
    conn = sqlite3.connect(DB_FILE)
    conn.row_factory = sqlite3.Row
    return conn


def get_guild_state(guild_id: str) -> Dict[str, Any]:
    """Retrieve guild state from database."""
    with get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            """
            SELECT guild_id, guild_name, cooldown_until, cooldown_ratio, self_control, last_updated
            FROM guild_state WHERE guild_id = ?
        """,
            (guild_id,),
        )
        row = cursor.fetchone()

        if row:
            return {
                "guild_id": row["guild_id"],
                "guild_name": row["guild_name"],
                "cooldown_until": row["cooldown_until"],
                "cooldown_ratio": row["cooldown_ratio"],
                "self_control": row["self_control"],
                "last_updated": row["last_updated"],
            }
        else:
            return {
                "guild_id": guild_id,
                "guild_name": "",
                "cooldown_until": "1970-01-01T00:00:00",
                "cooldown_ratio": 1.0,
                "self_control": 1.0,
                "last_updated": datetime.now().isoformat(),
            }


def update_guild_state(
    guild_id: str, guild_name: str, cooldown_until: str, cooldown_ratio: float, self_control: float
) -> None:
    """Update guild state in database."""
    with get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            """
            INSERT OR REPLACE INTO guild_state (guild_id, guild_name, cooldown_until, cooldown_ratio, self_control, last_updated)
            VALUES (?, ?, ?, ?, ?, ?)
        """,
            (
                guild_id,
                guild_name,
                cooldown_until,
                cooldown_ratio,
                self_control,
                datetime.now().isoformat(),
            ),
        )
        conn.commit()


def delete_guild_state(guild_id: str) -> bool:
    """Delete guild state from database."""
    with get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("DELETE FROM guild_state WHERE guild_id = ?", (guild_id,))
        conn.commit()
        return cursor.rowcount > 0


def get_all_guild_states() -> Dict[str, Dict[str, Any]]:
    """Retrieve all guild states (for debug purposes)."""
    with get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT guild_id, guild_name, cooldown_until, cooldown_ratio, self_control, last_updated FROM guild_state"
        )
        return {
            row["guild_id"]: {
                "guild_name": row["guild_name"],
                "cooldown_until": row["cooldown_until"],
                "cooldown_ratio": row["cooldown_ratio"],
                "self_control": row["self_control"],
                "last_updated": row["last_updated"],
            }
            for row in cursor.fetchall()
        }


def get_active_riddle(channel_id: str):
    """Récupérer une énigme active pour un canal spécifique."""
    with get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT * FROM active_riddles WHERE channel_id = ?", (str(channel_id),)
        )
        row = cursor.fetchone()

        if row:
            return {
                "channel_id": row["channel_id"],
                "riddle_index": row["riddle_index"],
                "nb_clues": row["nb_clues"],
                "message_id": row["message_id"],
                "solver_id": row["solver_id"],
                "created_at": row["created_at"],
                "updated_at": row["updated_at"],
            }
        return None


def save_active_riddle(
    channel_id: str,
    riddle_index: int,
    nb_clues: int,
    message_id: int,
    solver_id: str = None,
):
    """Sauvegarder ou mettre à jour une énigme active."""
    with get_connection() as conn:
        cursor = conn.cursor()
        # Si solver_id existe, on considère que c'est résolu et on garde l'entrée pour historique
        # Sinon on met à jour l'énigme en cours

        cursor.execute(
            """
            INSERT OR REPLACE INTO active_riddles
            (channel_id, riddle_index, nb_clues, message_id, solver_id, updated_at)
            VALUES (?, ?, ?, ?, ?, datetime('now'))
        """,
            (str(channel_id), riddle_index, nb_clues, message_id, solver_id),
        )
        conn.commit()


def remove_active_riddle(channel_id: str) -> bool:
    """Supprimer une énigme active du suivi."""
    with get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "DELETE FROM active_riddles WHERE channel_id = ?", (str(channel_id),)
        )
        conn.commit()
        return cursor.rowcount > 0


def get_all_active_riddles():
    """Récupérer toutes les énigmes actives (pour debug)."""
    with get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM active_riddles")
        return [
            {
                "channel_id": row["channel_id"],
                "riddle_index": row["riddle_index"],
                "nb_clues": row["nb_clues"],
                "message_id": row["message_id"],
                "solver_id": row["solver_id"],
                "created_at": row["created_at"],
                "updated_at": row["updated_at"],
            }
            for row in cursor.fetchall()
        ]