307 lines
10 KiB
Python
307 lines
10 KiB
Python
import discord
|
|
from discord import app_commands
|
|
from discord.ext import commands
|
|
import yt_dlp as youtube_dl
|
|
import asyncio
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
TOKEN = os.getenv('DISCORD_TOKEN')
|
|
GUILD_ID = os.getenv('GUILD_ID') # Optional: For testing
|
|
|
|
ytdl_format_options = {
|
|
'format': 'bestaudio/best',
|
|
'outtmpl': '%(extractor)s-%(id)s-%(title)s.%(ext)s',
|
|
'restrictfilenames': True,
|
|
'noplaylist': True,
|
|
'nocheckcertificate': True,
|
|
'ignoreerrors': False,
|
|
'logtostderr': False,
|
|
'quiet': True,
|
|
'no_warnings': True,
|
|
'default_search': 'auto',
|
|
'source_address': '0.0.0.0',
|
|
}
|
|
|
|
ffmpeg_options = {
|
|
'options': '-vn',
|
|
'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5'
|
|
}
|
|
|
|
ytdl = youtube_dl.YoutubeDL(ytdl_format_options)
|
|
|
|
|
|
class YTDLSource(discord.PCMVolumeTransformer):
|
|
def __init__(self, source, *, data, volume=0.5):
|
|
super().__init__(source, volume)
|
|
self.data = data
|
|
self.title = data.get('title')
|
|
self.url = data.get('url')
|
|
|
|
@classmethod
|
|
async def from_url(cls, url, *, loop=None, stream=False):
|
|
loop = loop or asyncio.get_event_loop()
|
|
data = await loop.run_in_executor(None, lambda: ytdl.extract_info(url, download=not stream))
|
|
|
|
if 'entries' in data:
|
|
data = data['entries'][0]
|
|
|
|
filename = data['url'] if stream else ytdl.prepare_filename(data)
|
|
return cls(discord.FFmpegPCMAudio(filename, **ffmpeg_options), data=data)
|
|
|
|
|
|
class MusicBot(commands.Cog):
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
self.queue = []
|
|
self.current = None
|
|
self.volume = 0.5
|
|
|
|
async def ensure_voice(self, interaction):
|
|
if not interaction.user.voice:
|
|
await interaction.response.send_message(
|
|
"You are not connected to a voice channel",
|
|
ephemeral=True
|
|
)
|
|
return False
|
|
|
|
if interaction.guild.voice_client is None:
|
|
await interaction.user.voice.channel.connect()
|
|
elif interaction.guild.voice_client.channel != interaction.user.voice.channel:
|
|
await interaction.response.send_message(
|
|
"I'm already in another voice channel",
|
|
ephemeral=True
|
|
)
|
|
return False
|
|
|
|
return True
|
|
|
|
@app_commands.command(name="play", description="Play a song from YouTube")
|
|
@app_commands.describe(query="YouTube URL or search query")
|
|
async def play(self, interaction: discord.Interaction, query: str):
|
|
"""Play a song from YouTube"""
|
|
if not await self.ensure_voice(interaction):
|
|
return
|
|
|
|
await interaction.response.defer()
|
|
|
|
try:
|
|
player = await YTDLSource.from_url(query, loop=self.bot.loop, stream=True)
|
|
|
|
if interaction.guild.voice_client.is_playing():
|
|
self.queue.append(player)
|
|
await interaction.followup.send(
|
|
f"Added to queue: {player.title}",
|
|
ephemeral=False
|
|
)
|
|
else:
|
|
self.current = player
|
|
interaction.guild.voice_client.play(
|
|
player,
|
|
after=lambda e: asyncio.run_coroutine_threadsafe(
|
|
self.play_next(interaction),
|
|
self.bot.loop
|
|
)
|
|
)
|
|
await interaction.followup.send(
|
|
f"Now playing: {player.title}",
|
|
ephemeral=False
|
|
)
|
|
except Exception as e:
|
|
await interaction.followup.send(
|
|
f"An error occurred: {e}",
|
|
ephemeral=True
|
|
)
|
|
|
|
async def play_next(self, interaction):
|
|
if len(self.queue) > 0:
|
|
self.current = self.queue.pop(0)
|
|
interaction.guild.voice_client.play(
|
|
self.current,
|
|
after=lambda e: asyncio.run_coroutine_threadsafe(
|
|
self.play_next(interaction),
|
|
self.bot.loop
|
|
)
|
|
)
|
|
await interaction.followup.send(
|
|
f"Now playing: {self.current.title}",
|
|
ephemeral=False
|
|
)
|
|
else:
|
|
self.current = None
|
|
await interaction.followup.send(
|
|
"Queue is empty",
|
|
ephemeral=False
|
|
)
|
|
|
|
@app_commands.command(name="queue", description="Show the current queue")
|
|
async def show_queue(self, interaction: discord.Interaction):
|
|
"""Show the current queue"""
|
|
if not self.queue:
|
|
await interaction.response.send_message(
|
|
"The queue is empty",
|
|
ephemeral=True
|
|
)
|
|
return
|
|
|
|
queue_list = "\n".join([f"{i + 1}. {item.title}" for i, item in enumerate(self.queue)])
|
|
await interaction.response.send_message(
|
|
f"Current queue:\n{queue_list}",
|
|
ephemeral=False
|
|
)
|
|
|
|
@app_commands.command(name="skip", description="Skip the current song")
|
|
async def skip(self, interaction: discord.Interaction):
|
|
"""Skip the current song"""
|
|
if interaction.guild.voice_client.is_playing():
|
|
interaction.guild.voice_client.stop()
|
|
await interaction.response.send_message(
|
|
"Skipped current song",
|
|
ephemeral=False
|
|
)
|
|
else:
|
|
await interaction.response.send_message(
|
|
"Not playing anything",
|
|
ephemeral=True
|
|
)
|
|
|
|
@app_commands.command(name="pause", description="Pause the current song")
|
|
async def pause(self, interaction: discord.Interaction):
|
|
"""Pause the current song"""
|
|
if interaction.guild.voice_client.is_playing():
|
|
interaction.guild.voice_client.pause()
|
|
await interaction.response.send_message(
|
|
"Paused",
|
|
ephemeral=False
|
|
)
|
|
else:
|
|
await interaction.response.send_message(
|
|
"Not playing anything",
|
|
ephemeral=True
|
|
)
|
|
|
|
@app_commands.command(name="resume", description="Resume the current song")
|
|
async def resume(self, interaction: discord.Interaction):
|
|
"""Resume the current song"""
|
|
if interaction.guild.voice_client.is_paused():
|
|
interaction.guild.voice_client.resume()
|
|
await interaction.response.send_message(
|
|
"Resumed",
|
|
ephemeral=False
|
|
)
|
|
else:
|
|
await interaction.response.send_message(
|
|
"Not paused",
|
|
ephemeral=True
|
|
)
|
|
|
|
@app_commands.command(name="stop", description="Stop playback and clear queue")
|
|
async def stop(self, interaction: discord.Interaction):
|
|
"""Stop playback and clear queue"""
|
|
if interaction.guild.voice_client.is_playing():
|
|
interaction.guild.voice_client.stop()
|
|
self.queue.clear()
|
|
self.current = None
|
|
await interaction.response.send_message(
|
|
"Stopped and cleared queue",
|
|
ephemeral=False
|
|
)
|
|
|
|
@app_commands.command(name="volume", description="Change the player volume")
|
|
@app_commands.describe(volume="The volume level (1-100)")
|
|
async def volume(self, interaction: discord.Interaction, volume: int):
|
|
"""Change the player volume"""
|
|
if interaction.guild.voice_client is None:
|
|
return await interaction.response.send_message(
|
|
"Not connected to a voice channel",
|
|
ephemeral=True
|
|
)
|
|
|
|
if 0 < volume <= 100:
|
|
self.volume = volume / 100
|
|
interaction.guild.voice_client.source.volume = self.volume
|
|
await interaction.response.send_message(
|
|
f"Changed volume to {volume}%",
|
|
ephemeral=False
|
|
)
|
|
return None
|
|
else:
|
|
await interaction.response.send_message(
|
|
"Please enter a value between 1 and 100",
|
|
ephemeral=True
|
|
)
|
|
return None
|
|
|
|
@app_commands.command(name="nowplaying", description="Show the currently playing song")
|
|
async def now_playing(self, interaction: discord.Interaction):
|
|
"""Show the currently playing song"""
|
|
if self.current:
|
|
await interaction.response.send_message(
|
|
f"Now playing: {self.current.title}",
|
|
ephemeral=False
|
|
)
|
|
else:
|
|
await interaction.response.send_message(
|
|
"Nothing is currently playing",
|
|
ephemeral=True
|
|
)
|
|
|
|
@app_commands.command(name="leave", description="Leave the voice channel")
|
|
async def leave(self, interaction: discord.Interaction):
|
|
"""Leave the voice channel"""
|
|
if interaction.guild.voice_client is not None:
|
|
await interaction.guild.voice_client.disconnect()
|
|
self.queue.clear()
|
|
self.current = None
|
|
await interaction.response.send_message(
|
|
"Left the voice channel",
|
|
ephemeral=False
|
|
)
|
|
else:
|
|
await interaction.response.send_message(
|
|
"I'm not in a voice channel",
|
|
ephemeral=True
|
|
)
|
|
|
|
|
|
class MyBot(commands.Bot):
|
|
def __init__(self):
|
|
intents = discord.Intents.default()
|
|
intents.message_content = True
|
|
intents.voice_states = True
|
|
super().__init__(command_prefix="!", intents=intents)
|
|
|
|
async def setup_hook(self):
|
|
await self.add_cog(MusicBot(self))
|
|
|
|
if GUILD_ID:
|
|
guild = discord.Object(id=int(GUILD_ID))
|
|
self.tree.copy_global_to(guild=guild)
|
|
await self.tree.sync(guild=guild)
|
|
print(f"Synced commands to guild {GUILD_ID}")
|
|
else:
|
|
await self.tree.sync()
|
|
print("Synced commands globally")
|
|
|
|
async def on_ready(self):
|
|
print(f'Logged in as {self.user.name} ({self.user.id})')
|
|
print('------')
|
|
|
|
|
|
bot = MyBot()
|
|
|
|
|
|
@bot.event
|
|
async def on_command_error(ctx, error):
|
|
print(f"Error: {error}")
|
|
|
|
|
|
async def main():
|
|
async with bot:
|
|
await bot.start(TOKEN)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |