zyrarnet/zyrarbot
2
0 Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases 6 Wiki Activity

feat: add beautiful web queue dashboard with loop, reordering, and text channel settings persistence

Browse Source
This commit is contained in:
mmacierzanka committed 2026-05-23 14:32:43 +02:00
1 parent 6382955b3f
commit 71800c8fd5
9 files changed
+2367 -298
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

main.py
+19 -298
View File
@@ -1,306 +1,27 @@
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}")
import uvicorn
from bot import bot, TOKEN
from web import app
async def main():
async with bot:
await bot.start(TOKEN)
config = uvicorn.Config(app, host="0.0.0.0", port=8000, log_level="info")
server = uvicorn.Server(config)
server_task = asyncio.create_task(server.serve())
try:
await bot.start(TOKEN)
except Exception as e:
print(f"Error starting Discord bot: {e}")
print("Keep running FastAPI web server...")
while not server.should_exit:
try:
await asyncio.sleep(1)
except asyncio.CancelledError:
break
finally:
server.should_exit = True
await server_task
if __name__ == "__main__":