321 lines
11 KiB
Python
321 lines
11 KiB
Python
"""Voice channel music handler with YTDLSource, FFmpegOpusAudio, queue, and controls."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import discord
|
|
from discord.ext import commands
|
|
from discord.ext.commands import Cog, Context
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ─── Configuration ───────────────────────────────────────────────────────────
|
|
|
|
PREFIX = "!"
|
|
GUILD_IDS = [] # Set in bot.json or env var
|
|
CHANNEL_ID = None # Set in bot.json or env var
|
|
|
|
# ─── Music Cog ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
class MusicCog(Cog):
|
|
"""Music commands for voice channels."""
|
|
|
|
def __init__(self, bot: commands.Bot) -> None:
|
|
self.bot = bot
|
|
self._current_queue: list[discord.AudioTrack] = []
|
|
self._is_playing: bool = False
|
|
self._voice_channel: Optional[discord.VoiceChannel] = None
|
|
self._playing_lock: asyncio.Lock = asyncio.Lock()
|
|
self._stop_requested: bool = False
|
|
self._current_volume: int = 100
|
|
self._is_paused: bool = False
|
|
|
|
def cog_unload(self) -> None:
|
|
"""Clean up on cog unload."""
|
|
if self._voice_channel:
|
|
self._voice_channel.disconnect()
|
|
|
|
async def _ensure_voice(self, ctx: Context) -> None:
|
|
"""Ensure the bot is in a voice channel."""
|
|
if self._voice_channel is None:
|
|
channel = await ctx.author.voice.channel.connect()
|
|
self._voice_channel = channel
|
|
self._is_playing = True
|
|
elif not self._voice_channel.is_connected():
|
|
self._voice_channel = await ctx.author.voice.channel.connect()
|
|
|
|
async def _disconnect(self) -> None:
|
|
"""Disconnect from the voice channel."""
|
|
if self._voice_channel:
|
|
try:
|
|
await self._voice_channel.disconnect()
|
|
except discord.DiscordError:
|
|
pass
|
|
self._voice_channel = None
|
|
self._is_playing = False
|
|
self._stop_requested = False
|
|
|
|
@commands.command()
|
|
async def join(self, ctx: Context, c=None):
|
|
if c is None:
|
|
# Check if the user is in a voice channel
|
|
if ctx.author.voice:
|
|
channel = ctx.author.voice.channel
|
|
self._voice_channel = await channel.connect()
|
|
await ctx.send(f"Joined {channel.name}!")
|
|
else:
|
|
await ctx.send("You must be in a voice channel first!")
|
|
channel = discord.utils.get(ctx.guild.voice_channels, name=c)
|
|
print(channel)
|
|
await channel.connect()
|
|
|
|
@commands.command()
|
|
async def leave(self, ctx):
|
|
# Check if the bot is in a voice channel in this server
|
|
if ctx.voice_client:
|
|
await ctx.voice_client.disconnect()
|
|
self._voice_channel = None
|
|
await ctx.send("Disconnected from the voice channel.")
|
|
else:
|
|
await ctx.send("I am not in a voice channel.")
|
|
|
|
@commands.command()
|
|
async def play(self, ctx: Context, source: str) -> None:
|
|
"""Play a track from a source (URL or file)."""
|
|
if not self._is_playing:
|
|
|
|
try:
|
|
track = await YTDLSource.from_url(source, source_type="youtube", loop=False)
|
|
self._current_queue.append(track)
|
|
await self._play_next()
|
|
except Exception as e:
|
|
logger.error(f"Error playing source {source}: {e}")
|
|
await ctx.send(f"Error playing: {e}")
|
|
|
|
async def _play_next(self) -> None:
|
|
"""Play the next track in the queue."""
|
|
if not self._current_queue:
|
|
await self._disconnect()
|
|
return
|
|
|
|
track = self._current_queue.pop(0)
|
|
|
|
try:
|
|
if self._voice_channel:
|
|
await self._voice_channel.play(track, volume=self._current_volume / 100.0)
|
|
self._is_playing = True
|
|
except Exception as e:
|
|
logger.error(f"Error playing track: {e}")
|
|
await self._play_next()
|
|
|
|
async def skip(self, ctx: Context) -> None:
|
|
"""Skip the current track."""
|
|
if not self._is_playing:
|
|
await ctx.send("Not playing anything.")
|
|
return
|
|
|
|
try:
|
|
await self._voice_channel.stop()
|
|
self._current_queue.clear()
|
|
self._is_playing = False
|
|
await ctx.send("Skipped.")
|
|
except Exception as e:
|
|
logger.error(f"Error skipping: {e}")
|
|
await ctx.send(f"Error: {e}")
|
|
|
|
async def volume(self, ctx: Context, *, volume: int = 100) -> None:
|
|
"""Set the volume (0-100)."""
|
|
if not self._is_playing:
|
|
await ctx.send("Not playing anything.")
|
|
return
|
|
|
|
if volume < 0 or volume > 100:
|
|
await ctx.send("Volume must be between 0 and 100.")
|
|
return
|
|
|
|
self._current_volume = volume
|
|
await ctx.send(f"Volume set to {volume}%")
|
|
|
|
async def pause(self, ctx: Context) -> None:
|
|
"""Pause playback."""
|
|
if not self._is_playing:
|
|
await ctx.send("Not playing anything.")
|
|
return
|
|
|
|
self._is_paused = True
|
|
await ctx.send("Playback paused.")
|
|
|
|
async def resume(self, ctx: Context) -> None:
|
|
"""Resume playback."""
|
|
if self._is_paused:
|
|
self._is_paused = False
|
|
await ctx.send("Playback resumed.")
|
|
else:
|
|
await ctx.send("Already playing.")
|
|
|
|
async def stop(self, ctx: Context) -> None:
|
|
"""Stop and disconnect."""
|
|
await self._disconnect()
|
|
await ctx.send("Stopped and disconnected.")
|
|
|
|
async def queue(self, ctx: Context) -> None:
|
|
"""Show the current queue."""
|
|
if not self._current_queue:
|
|
await ctx.send("Queue is empty.")
|
|
return
|
|
|
|
tracks = []
|
|
for i, track in enumerate(self._current_queue):
|
|
tracks.append(f"{i+1}. {track.title} — {track.author}")
|
|
await ctx.send("\n".join(tracks))
|
|
|
|
|
|
# ─── YTDLSource ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
class YTDLSource(discord.PCMVolumeTransformer):
|
|
"""Audio source from youtube-dl."""
|
|
|
|
def __init__(self, data, *, is_live=False, loop=False):
|
|
super().__init__()
|
|
self.title = data["title"]
|
|
self.url = data["url"]
|
|
self.is_live = is_live
|
|
self.duration = data["duration"] if "duration" in data else None
|
|
self.author = data.get("uploader", "Unknown")
|
|
self.loop = loop
|
|
self._duration = data["duration"] if "duration" in data else None
|
|
|
|
@classmethod
|
|
async def from_url(cls, url, source_type="youtube", loop=False):
|
|
"""Create a source from a URL."""
|
|
import yt_dlp as youtube_dl
|
|
|
|
ydl_opts = {
|
|
"format": "bestaudio/best",
|
|
"noplaylist": True,
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
}
|
|
|
|
with youtube_dl.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(url, download=False)
|
|
|
|
if info is None:
|
|
raise ValueError("No video info found.")
|
|
|
|
if info.get("url") is None:
|
|
raise ValueError("No URL found in video info.")
|
|
|
|
if info.get("duration") is None:
|
|
raise ValueError("No duration found in video info.")
|
|
|
|
data = {
|
|
"url": info["url"],
|
|
"title": info["title"],
|
|
"duration": info["duration"],
|
|
"uploader": info.get("uploader", "Unknown"),
|
|
}
|
|
|
|
return cls(data, is_live=info.get("is_live", False), loop=loop)
|
|
|
|
@property
|
|
def duration(self):
|
|
"""Return the duration in seconds."""
|
|
return self._duration
|
|
|
|
@duration.setter
|
|
def duration(self, value):
|
|
"""Set the duration."""
|
|
self._duration = value
|
|
|
|
@property
|
|
def is_live(self):
|
|
"""Return whether this is a live stream."""
|
|
return self._is_live
|
|
|
|
@is_live.setter
|
|
def is_live(self, value):
|
|
"""Set whether this is a live stream."""
|
|
self._is_live = value
|
|
|
|
|
|
# ─── FFmpegOpusAudio ─────────────────────────────────────────────────────────
|
|
|
|
|
|
class FFmpegOpusAudio(discord.PCMVolumeTransformer):
|
|
"""Audio source from FFmpegOpusAudio."""
|
|
|
|
def __init__(self, data, *, is_live=False, loop=False):
|
|
super().__init__()
|
|
self.title = data["title"]
|
|
self.url = data["url"]
|
|
self.is_live = is_live
|
|
self.duration = data["duration"] if "duration" in data else None
|
|
self.author = data.get("uploader", "Unknown")
|
|
self.loop = loop
|
|
self._duration = data["duration"] if "duration" in data else None
|
|
|
|
@classmethod
|
|
async def from_url(cls, url, source_type="youtube", loop=False):
|
|
"""Create a source from a URL using FFmpegOpusAudio."""
|
|
import youtube_dl
|
|
|
|
ydl_opts = {
|
|
"format": "bestaudio/best",
|
|
"noplaylist": True,
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
}
|
|
|
|
with youtube_dl.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(url, download=False)
|
|
|
|
if info is None:
|
|
raise ValueError("No video info found.")
|
|
|
|
if info.get("url") is None:
|
|
raise ValueError("No URL found in video info.")
|
|
|
|
if info.get("duration") is None:
|
|
raise ValueError("No duration found in video info.")
|
|
|
|
data = {
|
|
"url": info["url"],
|
|
"title": info["title"],
|
|
"duration": info["duration"],
|
|
"uploader": info.get("uploader", "Unknown"),
|
|
}
|
|
|
|
return cls(data, is_live=info.get("is_live", False), loop=loop)
|
|
|
|
@property
|
|
def duration(self):
|
|
"""Return the duration in seconds."""
|
|
return self._duration
|
|
|
|
@duration.setter
|
|
def duration(self, value):
|
|
"""Set the duration."""
|
|
self._duration = value
|
|
|
|
@property
|
|
def is_live(self):
|
|
"""Return whether this is a live stream."""
|
|
return self._is_live
|
|
|
|
@is_live.setter
|
|
def is_live(self, value):
|
|
"""Set whether this is a live stream."""
|
|
self._is_live = value
|
|
|
|
|
|
# ─── Bot Integration ─────────────────────────────────────────────────────────
|
|
|
|
|
|
async def setup(bot: commands.Bot) -> None:
|
|
"""Register the music cog with the bot."""
|
|
await bot.add_cog(MusicCog(bot))
|