app.py

import asyncio
import logging
import os
import threading
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Callable

import discord
from discord import app_commands
from discord.ext import commands
from dotenv import load_dotenv
import gradio as gr
from huggingface_hub import hf_hub_download

# Configure logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


class LoopMode(str, Enum):
    NONE = "none"


@dataclass
class BotConfig:
    ASSETS_DIR: str = "assets"
    SONG_FILE: str = "lofi.mp3"
    HF_REPO: str = "not-lain/assets"

    @property
    def song_path(self) -> str:
        return f"{self.ASSETS_DIR}/{self.SONG_FILE}"


@dataclass
class QueueItem:
    url: str
    title: Optional[str] = None
    file_path: Optional[str] = None


class VoiceStateError(Exception):
    """Custom exception for voice state errors"""

    pass


class MusicBot:
    def __init__(self, config: BotConfig):
        self.config = config
        self.is_playing: bool = False
        self.voice_client: Optional[discord.VoiceClient] = None
        self.last_context: Optional[commands.Context] = None
        self.loop_mode: LoopMode = LoopMode.NONE
        self.current_song: Optional[QueueItem] = None

    async def ensure_voice_state(self, ctx: commands.Context) -> None:
        """Validate voice state and raise appropriate errors"""
        if not ctx.author.voice:
            raise VoiceStateError("You need to be in a voice channel!")

        if self.voice_client and ctx.author.voice.channel != self.voice_client.channel:
            raise VoiceStateError("You must be in the same voice channel as the bot!")

    async def play_next(self, ctx: commands.Context) -> None:
        if self.is_playing:
            return

        try:
            if not self.current_song:
                self.current_song = QueueItem(
                    url=self.config.song_path,
                    title="Lofi Music",
                    file_path=self.config.song_path,
                )

            self.is_playing = True
            audio_source = discord.FFmpegPCMAudio(self.current_song.file_path)

            def after_playing(error):
                if error:
                    logger.error(f"Error in playback: {error}")
                self.is_playing = False
                asyncio.run_coroutine_threadsafe(self.handle_song_end(ctx), bot.loop)

            self.voice_client.play(audio_source, after=after_playing)

        except Exception as e:
            logger.error(f"Error playing file: {e}")
            self.is_playing = False
            raise

    async def handle_song_end(self, ctx: commands.Context) -> None:
        if self.loop_mode == LoopMode.NONE:
            self.current_song = None

        if not self.is_playing:
            await self.play_next(ctx)

    async def join_voice(self, ctx: commands.Context) -> None:
        if not ctx.author.voice:
            await ctx.send("You need to be in a voice channel!")
            return

        channel = ctx.author.voice.channel
        if self.voice_client is None:
            self.voice_client = await channel.connect()
            self.last_context = ctx
        else:
            await self.voice_client.move_to(channel)

    async def stop_playing(self, ctx: commands.Context) -> bool:
        try:
            if self.voice_client:
                if self.voice_client.is_playing():
                    self.voice_client.stop()
                if self.voice_client.is_connected():
                    await self.voice_client.disconnect(force=False)
                self._reset_state()
                return True
            return False
        except Exception as e:
            logger.error(f"Error during cleanup: {e}")
            self._reset_state()
            return False

    def _reset_state(self) -> None:
        self.is_playing = False
        self.voice_client = None
        self.last_context = None
        self.loop_mode = LoopMode.NONE
        self.current_song = None


async def handle_voice_command(
    interaction: discord.Interaction, action: Callable, defer: bool = True
) -> None:
    """Generic handler for voice-related commands"""
    try:
        if defer:
            await interaction.response.defer()
        ctx = await bot.get_context(interaction)
        await music_bot.ensure_voice_state(ctx)
        await action(ctx, interaction)
    except VoiceStateError as e:
        if not interaction.response.is_done():
            await interaction.response.send_message(str(e))
        else:
            await interaction.followup.send(str(e))
    except Exception as e:
        logger.error(f"Command error: {e}")
        if not interaction.response.is_done():
            await interaction.response.send_message("An error occurred!")
        else:
            await interaction.followup.send("An error occurred!")


# Initialize bot and music bot instance
config = BotConfig()
intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(command_prefix="!", intents=intents)
music_bot = MusicBot(config)


@bot.event
async def on_ready():
    print(f"Bot is ready! Logged in as {bot.user}")
    print("Syncing commands...")
    try:
        await bot.tree.sync(guild=None)  # Set to None for global sync
        print("Successfully synced commands globally!")
    except discord.app_commands.errors.CommandSyncFailure as e:
        print(f"Failed to sync commands: {e}")
    except Exception as e:
        print(f"An error occurred while syncing commands: {e}")


@bot.tree.command(name="lofi", description="Play lofi music")
async def lofi(interaction: discord.Interaction):
    async def play_lofi(ctx, interaction: discord.Interaction):
        await music_bot.join_voice(ctx)
        if not music_bot.is_playing:
            await music_bot.play_next(ctx)
            await interaction.followup.send("Playing lofi music! 🎵")
        else:
            await interaction.followup.send("Already playing!")

    await handle_voice_command(interaction, play_lofi)


@bot.tree.command(name="stop", description="Stop the lofi music")
async def stop(interaction: discord.Interaction):
    if not interaction.user.voice:
        await interaction.response.send_message(
            "You must be in a voice channel to use this command!"
        )
        return

    if not music_bot.voice_client:
        await interaction.response.send_message("No music is currently playing!")
        return

    if interaction.user.voice.channel != music_bot.voice_client.channel:
        await interaction.response.send_message(
            "You must be in the same voice channel as the bot!"
        )
        return

    await interaction.response.defer()
    ctx = await bot.get_context(interaction)

    try:
        success = await music_bot.stop_playing(ctx)
        if success:
            await interaction.followup.send("Music stopped and bot disconnected! 👋")
        else:
            await interaction.followup.send(
                "Failed to stop properly. Please try again."
            )
    except Exception as e:
        print(f"Error during stop command: {e}")
        await interaction.followup.send("An error occurred while trying to stop.")


def initialize_assets() -> None:
    if not os.path.exists(config.ASSETS_DIR):
        os.makedirs(config.ASSETS_DIR, exist_ok=True)
        hf_hub_download(
            config.HF_REPO,
            config.SONG_FILE,
            repo_type="dataset",
            local_dir=config.ASSETS_DIR,
        )


def run_discord_bot() -> None:
    """Run the Discord bot with the token from environment variables."""
    load_dotenv()
    bot.run(os.getenv("DISCORD_TOKEN"))


if __name__ == "__main__":
    initialize_assets()
    bot_thread = threading.Thread(target=run_discord_bot, daemon=True)
    bot_thread.start()

    with gr.Blocks() as iface:
        gr.Markdown("# Discord Music Bot Control Panel")
        gr.Markdown("Bot is running in background")

    iface.launch(debug=True)