bot.py

from __future__ import annotations

import logging
import os
import sqlite3
import json
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path

import discord
from discord import app_commands
from discord.ext import commands
from dotenv import load_dotenv


BASE_DIR = Path(__file__).resolve().parent
LOCALES_DIR = BASE_DIR / "locales"


def parse_channel_id(value: str, env_name: str) -> int:
    try:
        return int(value)
    except ValueError as exc:
        raise ValueError(f"{env_name} must be a valid Discord channel ID.") from exc


def str_to_bool(value: str | None, default: bool = False) -> bool:
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def parse_channel_ids() -> set[int]:
    channel_ids: set[int] = set()

    for index in range(1, 11):
        value = os.getenv(f"CHANNEL_{index}")
        if not value:
            continue
        channel_ids.add(parse_channel_id(value, f"CHANNEL_{index}"))

    extra_channels = os.getenv("MONITORED_CHANNEL_IDS", "")
    for raw_channel_id in extra_channels.split(","):
        candidate = raw_channel_id.strip()
        if candidate:
            channel_ids.add(parse_channel_id(candidate, "MONITORED_CHANNEL_IDS"))

    return channel_ids


def load_locale(language_code: str) -> dict[str, str]:
    locale_path = LOCALES_DIR / f"{language_code}.json"
    fallback_path = LOCALES_DIR / "en_US.json"

    try:
        return json.loads(locale_path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return json.loads(fallback_path.read_text(encoding="utf-8"))


@dataclass(slots=True)
class Settings:
    token: str
    notification_channel_id: int
    monitored_channel_ids: set[int]
    locale_code: str
    rich_presence: str
    activity_type: str
    show_log: bool
    mention_everyone_on_empty_channel: bool
    database_path: Path

    @classmethod
    def from_env(cls) -> "Settings":
        load_dotenv()

        token = os.getenv("DISCORD_BOT_TOKEN")
        notification_channel = os.getenv("NOTIFICATION_CHANNEL")

        if not token:
            raise ValueError(
                "DISCORD_BOT_TOKEN is required. Create a .env file before starting the bot."
            )

        if not notification_channel:
            raise ValueError(
                "NOTIFICATION_CHANNEL is required. Set the channel that will receive notifications."
            )

        monitored_channel_ids = parse_channel_ids()
        if not monitored_channel_ids:
            raise ValueError(
                "At least one monitored voice channel is required. "
                "Set CHANNEL_1 or MONITORED_CHANNEL_IDS in the .env file."
            )

        return cls(
            token=token,
            notification_channel_id=parse_channel_id(
                notification_channel, "NOTIFICATION_CHANNEL"
            ),
            monitored_channel_ids=monitored_channel_ids,
            locale_code=os.getenv("BOT_LOCALE", "en_US"),
            rich_presence=os.getenv("RICH_PRESENCE", "Katchau!"),
            activity_type=os.getenv("ACTIVITY", "playing").lower(),
            show_log=str_to_bool(os.getenv("SHOW_LOG"), default=False),
            mention_everyone_on_empty_channel=str_to_bool(
                os.getenv("MENTION_EVERYONE_ON_EMPTY_CHANNEL"), default=True
            ),
            database_path=BASE_DIR / os.getenv("DATABASE_PATH", "bot_database.db"),
        )


class StatsRepository:
    def __init__(self, database_path: Path) -> None:
        self.connection = sqlite3.connect(database_path)
        self.connection.row_factory = sqlite3.Row
        self._ensure_schema()

    def _ensure_schema(self) -> None:
        with self.connection:
            self.connection.execute(
                """
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY,
                    name TEXT NOT NULL,
                    entry_count INTEGER NOT NULL DEFAULT 0,
                    last_call_time TEXT
                )
                """
            )
            self.connection.execute(
                """
                CREATE TABLE IF NOT EXISTS bot_settings (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )
                """
            )

    def record_entry(self, member: discord.Member) -> None:
        now = datetime.now(UTC).isoformat(timespec="seconds")
        with self.connection:
            self.connection.execute(
                """
                INSERT INTO users (id, name, entry_count, last_call_time)
                VALUES (?, ?, 1, ?)
                ON CONFLICT(id) DO UPDATE SET
                    name = excluded.name,
                    entry_count = users.entry_count + 1,
                    last_call_time = excluded.last_call_time
                """,
                (member.id, member.display_name, now),
            )

    def get_leaderboard(self, limit: int = 10) -> list[sqlite3.Row]:
        cursor = self.connection.execute(
            """
            SELECT id, name, entry_count, last_call_time
            FROM users
            ORDER BY entry_count DESC, last_call_time DESC
            LIMIT ?
            """,
            (limit,),
        )
        return cursor.fetchall()

    def get_user_stats(self, user_id: int) -> sqlite3.Row | None:
        cursor = self.connection.execute(
            """
            SELECT id, name, entry_count, last_call_time
            FROM users
            WHERE id = ?
            """,
                (user_id,),
            )
        return cursor.fetchone()

    def get_notifications_enabled(self, default: bool = True) -> bool:
        cursor = self.connection.execute(
            """
            SELECT value
            FROM bot_settings
            WHERE key = 'notifications_enabled'
            """
        )
        row = cursor.fetchone()
        if row is None:
            return default
        return str_to_bool(row["value"], default=default)

    def set_notifications_enabled(self, enabled: bool) -> None:
        with self.connection:
            self.connection.execute(
                """
                INSERT INTO bot_settings (key, value)
                VALUES ('notifications_enabled', ?)
                ON CONFLICT(key) DO UPDATE SET
                    value = excluded.value
                """,
                ("true" if enabled else "false",),
            )

    def close(self) -> None:
        self.connection.close()


class EveryoneShouldKnowBot(commands.Bot):
    def __init__(self, settings: Settings, locale: dict[str, str], repository: StatsRepository):
        intents = discord.Intents.default()
        intents.voice_states = True
        intents.members = True

        super().__init__(command_prefix=commands.when_mentioned, intents=intents)

        self.settings = settings
        self.locale = locale
        self.repository = repository
        self.send_message_enabled = repository.get_notifications_enabled(default=True)
        self.logger = logging.getLogger("esk")

    async def setup_hook(self) -> None:
        self.tree.add_command(leaders)
        self.tree.add_command(toggle_notifications)
        self.tree.add_command(help_command)
        self.tree.add_command(stats)
        await self.tree.sync()

    async def close(self) -> None:
        self.repository.close()
        await super().close()

    async def on_ready(self) -> None:
        await self.change_presence(
            status=discord.Status.online,
            activity=self._build_activity(),
        )

        self.logger.info("Bot online as %s", self.user)
        self.logger.info("Monitoring %s voice channel(s)", len(self.settings.monitored_channel_ids))
        self.logger.info("Notifications enabled: %s", self.send_message_enabled)

    async def on_voice_state_update(
        self,
        member: discord.Member,
        before: discord.VoiceState,
        after: discord.VoiceState,
    ) -> None:
        if member.bot or not self.send_message_enabled:
            return

        if before.channel == after.channel:
            return

        before_channel = before.channel
        after_channel = after.channel
        monitors = self.settings.monitored_channel_ids

        was_monitored = before_channel is not None and before_channel.id in monitors
        is_monitored = after_channel is not None and after_channel.id in monitors

        if not was_monitored and not is_monitored:
            return

        notification_channel = self.get_channel(self.settings.notification_channel_id)
        if not isinstance(notification_channel, discord.abc.Messageable):
            self.logger.warning(
                "Notification channel %s is not available.",
                self.settings.notification_channel_id,
            )
            return

        joined_monitored_from_outside = is_monitored and not was_monitored
        if joined_monitored_from_outside:
            self.repository.record_entry(member)

        if is_monitored and after_channel is not None:
            if was_monitored and before_channel is not None:
                await notification_channel.send(
                    self.locale["user_moved_to_channel"].format(
                        member=member.display_name,
                        old_channel=before_channel.name,
                        new_channel=after_channel.name,
                    )
                )
            elif len(after_channel.members) == 1:
                await notification_channel.send(
                    self.locale["user_joined_call_everyone"].format(member=member.mention)
                    if self.settings.mention_everyone_on_empty_channel
                    else self.locale["user_joined_call"].format(member=member.mention)
                )
            else:
                await notification_channel.send(
                    self.locale["user_in_channel"].format(
                        member=member.display_name,
                        channel=after_channel.name,
                    )
                )
            return

        if was_monitored and before_channel is not None:
            await notification_channel.send(
                self.locale["user_left_channel"].format(
                    member=member.display_name,
                    channel=before_channel.name,
                )
            )

    def _build_activity(self) -> discord.BaseActivity:
        activity_map: dict[str, discord.ActivityType] = {
            "playing": discord.ActivityType.playing,
            "listening": discord.ActivityType.listening,
            "watching": discord.ActivityType.watching,
            "streaming": discord.ActivityType.streaming,
            "competing": discord.ActivityType.competing,
        }

        activity_type = activity_map.get(self.settings.activity_type, discord.ActivityType.playing)
        if activity_type is discord.ActivityType.playing:
            return discord.Game(name=self.settings.rich_presence)

        return discord.Activity(name=self.settings.rich_presence, type=activity_type)


def command_channel_scope(bot: EveryoneShouldKnowBot) -> str:
    channel_count = len(bot.settings.monitored_channel_ids)
    return bot.locale["stats_scope"].format(count=channel_count)


@app_commands.command(
    name="leaders",
    description="Show the ranking of users who joined monitored voice channels the most.",
)
async def leaders(interaction: discord.Interaction) -> None:
    bot = interaction.client
    assert isinstance(bot, EveryoneShouldKnowBot)

    leaderboard = bot.repository.get_leaderboard()

    if not leaderboard:
        await interaction.response.send_message(bot.locale["leaderboard_empty"], ephemeral=True)
        return

    lines = [bot.locale["leaderboard_title"], command_channel_scope(bot), ""]
    for position, row in enumerate(leaderboard, start=1):
        lines.append(
            bot.locale["leaderboard_entry"].format(
                position=position,
                name=row["name"],
                count=row["entry_count"],
            )
        )

    await interaction.response.send_message("\n".join(lines))


@app_commands.command(
    name="toggle",
    description="Enable or disable call notifications for the monitored channels.",
)
@app_commands.default_permissions(manage_guild=True)
async def toggle_notifications(interaction: discord.Interaction) -> None:
    bot = interaction.client
    assert isinstance(bot, EveryoneShouldKnowBot)

    if interaction.guild and not interaction.user.guild_permissions.manage_guild:
        await interaction.response.send_message(
            bot.locale["missing_manage_guild_permission"],
            ephemeral=True,
        )
        return

    bot.send_message_enabled = not bot.send_message_enabled
    bot.repository.set_notifications_enabled(bot.send_message_enabled)
    status_key = "toggle_enabled" if bot.send_message_enabled else "toggle_disabled"
    await interaction.response.send_message(bot.locale[status_key], ephemeral=True)


@app_commands.command(
    name="help",
    description="Show the available slash commands and what each one does.",
)
async def help_command(interaction: discord.Interaction) -> None:
    bot = interaction.client
    assert isinstance(bot, EveryoneShouldKnowBot)

    commands_list = [
        bot.locale["help_leaders"],
        bot.locale["help_stats"],
        bot.locale["help_toggle"],
        bot.locale["help_help"],
    ]
    body = "\n".join(commands_list)
    message = bot.locale["help_title"].format(author=interaction.user.display_name)
    await interaction.response.send_message(f"{message}\n\n{body}", ephemeral=True)


@app_commands.command(
    name="stats",
    description="Show your call stats or the stats for another member.",
)
@app_commands.describe(member="Optional member to inspect")
async def stats(
    interaction: discord.Interaction,
    member: discord.Member | None = None,
) -> None:
    bot = interaction.client
    assert isinstance(bot, EveryoneShouldKnowBot)

    target = member or interaction.user
    row = bot.repository.get_user_stats(target.id)
    if row is None:
        await interaction.response.send_message(
            bot.locale["stats_empty"].format(member=target.display_name),
            ephemeral=True,
        )
        return

    last_call_time = row["last_call_time"] or bot.locale["stats_never"]
    message = bot.locale["stats_message"].format(
        member=row["name"],
        count=row["entry_count"],
        last_call_time=last_call_time,
    )
    await interaction.response.send_message(message, ephemeral=True)


def configure_logging(enabled: bool) -> None:
    level = logging.INFO if enabled else logging.WARNING
    logging.basicConfig(
        level=level,
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    )


def main() -> None:
    settings = Settings.from_env()
    configure_logging(settings.show_log)
    locale = load_locale(settings.locale_code)
    repository = StatsRepository(settings.database_path)
    bot = EveryoneShouldKnowBot(settings, locale, repository)
    bot.run(settings.token, log_handler=None)


if __name__ == "__main__":
    main()