mirror of
https://github.com/Matthww/TuneBot.git
synced 2026-09-21 22:47:51 +00:00
113 lines
3.9 KiB
Python
113 lines
3.9 KiB
Python
import typing
|
|
|
|
from discord import app_commands
|
|
from discord import Interaction
|
|
|
|
from bot import config
|
|
from cogs.music import helper
|
|
from utils.embed import create_embed
|
|
|
|
if typing.TYPE_CHECKING:
|
|
from bot import TuneBot
|
|
|
|
QUEUE_SIZE = config["queue_buffer_size"] + 1
|
|
|
|
|
|
class MusicCommands(app_commands.Group):
|
|
def __init__(self, bot: "TuneBot"):
|
|
super().__init__(name="radio", description="Never-ending stream of good vibes")
|
|
self.bot: "TuneBot" = bot
|
|
|
|
@app_commands.command(name="connect", description="Start the radio")
|
|
async def connect(self, ctx: Interaction):
|
|
embed = create_embed(ctx.user)
|
|
player = helper.get_player(ctx.client, ctx.guild_id)
|
|
|
|
try:
|
|
await helper.ensure_voice(
|
|
permissions=ctx.app_permissions,
|
|
player=player,
|
|
author=ctx.user,
|
|
voice_client=ctx.guild.voice_client,
|
|
channel_id=ctx.channel.id,
|
|
)
|
|
except Exception as exc:
|
|
embed.title = str(exc)
|
|
await ctx.response.send_message(embed=embed)
|
|
return
|
|
|
|
if player.is_connected:
|
|
embed.title = "Already connected."
|
|
await ctx.response.send_message(embed=embed)
|
|
return
|
|
|
|
await helper.fill_player_queue(ctx.client, player, QUEUE_SIZE)
|
|
|
|
if not player.is_playing:
|
|
await player.play()
|
|
|
|
embed.title = "*⃣ | Connected."
|
|
await ctx.response.send_message(embed=embed)
|
|
|
|
@app_commands.command(
|
|
name="disconnect", description="I've had enough music for a while"
|
|
)
|
|
async def disconnect(self, ctx: Interaction):
|
|
embed = create_embed(ctx.user)
|
|
player = helper.get_player(ctx.client, ctx.guild_id)
|
|
|
|
if not ctx.user.voice or (
|
|
player.is_connected and ctx.user.voice.channel.id != int(player.channel_id)
|
|
):
|
|
embed.title = "You're not in my voicechannel!"
|
|
await ctx.response.send_message(embed=embed)
|
|
return
|
|
|
|
# Clear the queue to ensure old tracks don't start playing
|
|
# when someone else queues something.
|
|
player.queue.clear()
|
|
# Stop the current track so Lavalink consumes less resources.
|
|
await player.stop()
|
|
# Disconnect from the voice channel.
|
|
await ctx.guild.voice_client.disconnect(force=True)
|
|
embed.title = "*⃣ | Disconnected."
|
|
await ctx.response.send_message(embed=embed)
|
|
|
|
@app_commands.command(
|
|
name="current", description="Gives more details about what's currently playing."
|
|
)
|
|
async def current(self, ctx: Interaction):
|
|
player = helper.get_player(ctx.client, ctx.guild_id)
|
|
if not player.current:
|
|
embed = create_embed(ctx.user)
|
|
embed.title = "You're not listening to anything right now."
|
|
await ctx.response.send_message(embed=embed)
|
|
return
|
|
|
|
embed = helper.create_track_embed(player.current)
|
|
await ctx.response.send_message(embed=embed)
|
|
|
|
@app_commands.command(name="queue", description="See what's ahead")
|
|
async def queue(self, ctx: Interaction):
|
|
player = helper.get_player(ctx.client, ctx.guild_id)
|
|
embed = create_embed(ctx.user)
|
|
embed.title = "Coming up..."
|
|
if len(player.queue) > 0:
|
|
embed.description = "\n".join(
|
|
f"{index}. [{track.title}]({track.uri})"
|
|
for index, track in enumerate(player.queue, 1)
|
|
)
|
|
else:
|
|
embed.description = "We are still determining a playlist"
|
|
await ctx.response.send_message(embed=embed)
|
|
|
|
@app_commands.command(
|
|
name="skip", description="You might like the next track better"
|
|
)
|
|
async def skip(self, ctx: Interaction):
|
|
player = helper.get_player(ctx.client, ctx.guild_id)
|
|
embed = create_embed(ctx.user)
|
|
embed.title = "Skipped current song"
|
|
await player.skip()
|
|
await ctx.response.send_message(embed=embed)
|