From fde236118d56f24177e71ea58a576b727ab2b5a1 Mon Sep 17 00:00:00 2001 From: owen Date: Thu, 17 Sep 2026 17:00:42 -0500 Subject: [PATCH] Add YTAudio cog: YouTube playback via yt-dlp -> mp3 -> Lavalink Standalone cog that owns the Red-Lavalink client connection (initialize on startup, close on unload) and plays YouTube/yt-dlp sources by downloading audio, transcoding to mp3, and playing through Lavalink's local source. Per-guild queues, multi-VC capable. Commands: play, skip, stop, pause, resume, queue, nowplaying, volume, join, leave. --- ytaudio/__init__.py | 7 + ytaudio/info.json | 10 ++ ytaudio/ytaudio.py | 421 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 438 insertions(+) create mode 100644 ytaudio/__init__.py create mode 100644 ytaudio/info.json create mode 100644 ytaudio/ytaudio.py diff --git a/ytaudio/__init__.py b/ytaudio/__init__.py new file mode 100644 index 0000000..4e9e7a6 --- /dev/null +++ b/ytaudio/__init__.py @@ -0,0 +1,7 @@ +from .ytaudio import YTAudio + + +async def setup(bot): + cog = YTAudio(bot) + await bot.add_cog(cog) + cog.start_up_task() diff --git a/ytaudio/info.json b/ytaudio/info.json new file mode 100644 index 0000000..fe5464d --- /dev/null +++ b/ytaudio/info.json @@ -0,0 +1,10 @@ +{ + "author": ["Kingston"], + "install_msg": "YTAudio loaded. It owns the Lavalink client connection (initialize/close) and plays YouTube via yt-dlp -> mp3 -> Lavalink local source. Ensure the standalone Lavalink systemd node is running on localhost:2333.", + "name": "YTAudio", + "short": "YouTube audio via yt-dlp -> mp3 -> Lavalink, independent of the built-in Audio cog.", + "description": "Plays YouTube (and yt-dlp-supported sites) by downloading audio with yt-dlp, transcoding to mp3, and playing through Lavalink's local source. Owns the shared Red-Lavalink client connection so ttstoy and other cogs can use get_player() without the built-in Audio cog. Per-guild queues; multi-voice-channel capable.", + "requirements": ["yt-dlp"], + "tags": ["audio", "music", "youtube", "yt-dlp", "lavalink", "voice"], + "min_bot_version": "3.5.0" +} diff --git a/ytaudio/ytaudio.py b/ytaudio/ytaudio.py new file mode 100644 index 0000000..2793141 --- /dev/null +++ b/ytaudio/ytaudio.py @@ -0,0 +1,421 @@ +""" +YTAudio — YouTube audio for Red-DiscordBot via yt-dlp -> mp3 -> Lavalink (local source). + +This cog OWNS the shared Red-Lavalink client connection: + * On startup it calls ``lavalink.initialize()`` (once the bot is ready), + connecting the in-process Lavalink websocket client to the standalone + Lavalink node running on localhost:2333 (managed by systemd). + * On unload it calls ``lavalink.close()``. + +Because the Red-Lavalink client is a process-global singleton, exactly ONE cog +should own initialize()/close(). Other cogs (e.g. ttstoy) simply call +``lavalink.get_player(guild_id)`` and share this connection. + +Playback path: + yt-dlp downloads bestaudio -> ffmpeg transcodes to mp3 -> the mp3 file is + loaded through Lavalink's ``local`` source and played. The mp3 is deleted + when the track finishes / is skipped / on stop. +""" + +import asyncio +import functools +import logging +import os +import shutil +from typing import Dict, List, Optional + +import discord +import lavalink +import yt_dlp +from redbot.core import Config, commands +from redbot.core.bot import Red +from redbot.core.data_manager import cog_data_path + +log = logging.getLogger("red.scrapyard.ytaudio") + +# --- Lavalink node connection (must match the standalone application.yml) --- +LL_HOST = "localhost" +LL_PORT = 2333 +LL_PASSWORD = "youshallnotpass" +LL_SECURED = False + +MP3_BITRATE = "192" + +YTDLP_BASE_OPTS = { + "format": "bestaudio/best", + "quiet": True, + "no_warnings": True, + "noplaylist": True, + "ignoreerrors": False, + "default_search": "ytsearch", + "source_address": "0.0.0.0", + "postprocessors": [ + { + "key": "FFmpegExtractAudio", + "preferredcodec": "mp3", + "preferredquality": MP3_BITRATE, + } + ], +} + + +class QueuedTrack: + """A track queued for a guild, resolved lazily at play time.""" + + __slots__ = ("query", "requester", "title", "mp3_path") + + def __init__(self, query: str, requester: discord.abc.User): + self.query = query + self.requester = requester + self.title: Optional[str] = None + self.mp3_path: Optional[str] = None + + +class GuildPlayer: + """Per-guild queue + player-loop state. One per guild.""" + + def __init__(self, cog: "YTAudio", guild_id: int): + self.cog = cog + self.guild_id = guild_id + self.queue: "asyncio.Queue[Optional[QueuedTrack]]" = asyncio.Queue() + self.current: Optional[QueuedTrack] = None + self.task: Optional[asyncio.Task] = None + self.next_event = asyncio.Event() + self.volume = 100 + + def start(self): + if self.task is None or self.task.done(): + self.task = asyncio.create_task(self._player_loop()) + + async def _player_loop(self): + while True: + self.next_event.clear() + try: + track = await self.queue.get() + except asyncio.CancelledError: + return + if track is None: # shutdown sentinel + return + + self.current = track + try: + await self.cog._play_track(self.guild_id, track, self.next_event) + # Wait until the track-end handler fires next_event + await self.next_event.wait() + except asyncio.CancelledError: + self.cog._cleanup_file(track) + return + except Exception: + log.exception("Error playing track in guild %s", self.guild_id) + finally: + self.cog._cleanup_file(track) + self.current = None + + +class YTAudio(commands.Cog): + """YouTube audio via yt-dlp -> mp3 -> Lavalink.""" + + def __init__(self, bot: Red): + self.bot = bot + self.config = Config.get_conf(self, identifier=0x59544155_44494F, force_registration=True) + self.config.register_guild(volume=100) + self.players: Dict[int, GuildPlayer] = {} + self._ll_ready = False + self._connect_task: Optional[asyncio.Task] = None + self.mp3_dir = cog_data_path(self) / "mp3cache" + self.mp3_dir.mkdir(parents=True, exist_ok=True) + self._ytdlp_opts = dict(YTDLP_BASE_OPTS) + self._ytdlp_opts["outtmpl"] = str(self.mp3_dir / "%(id)s.%(ext)s") + + # ------------------------------------------------------------------ + # Lavalink client ownership + # ------------------------------------------------------------------ + def start_up_task(self): + """Called from setup() after add_cog.""" + self._connect_task = asyncio.create_task(self._connect_lavalink()) + + async def _connect_lavalink(self): + await self.bot.wait_until_red_ready() + # If a node is already connected (e.g. built-in Audio still loaded), + # don't double-initialize — just share it. + if lavalink.get_all_nodes(): + self._ll_ready = True + log.info("YTAudio: a Lavalink node is already connected; sharing it.") + return + max_retries = 5 + for attempt in range(1, max_retries + 1): + try: + await lavalink.initialize( + bot=self.bot, + host=LL_HOST, + password=LL_PASSWORD, + port=LL_PORT, + timeout=30, + secured=LL_SECURED, + ) + self._ll_ready = True + lavalink.register_event_listener(self._track_event_handler) + log.info("YTAudio: connected Lavalink client to %s:%s", LL_HOST, LL_PORT) + return + except lavalink.AbortingNodeConnection: + await self._safe_close() + log.warning("YTAudio: Lavalink connection aborted.") + return + except asyncio.TimeoutError: + await self._safe_close() + log.warning( + "YTAudio: Lavalink connect timed out (attempt %s/%s), retrying...", + attempt, + max_retries, + ) + await asyncio.sleep(2) + except Exception: + log.exception("YTAudio: unexpected error connecting to Lavalink.") + await self._safe_close() + return + log.critical("YTAudio: failed to connect to Lavalink node after %s attempts.", max_retries) + + async def _safe_close(self): + try: + await lavalink.close(self.bot) + except Exception: + pass + + # ------------------------------------------------------------------ + # Track end handling + # ------------------------------------------------------------------ + async def _track_event_handler(self, player, event, extra): + if event in ( + lavalink.LavalinkEvents.TRACK_END, + lavalink.LavalinkEvents.TRACK_EXCEPTION, + lavalink.LavalinkEvents.TRACK_STUCK, + ): + gp = self.players.get(player.guild.id) + if gp is not None: + gp.next_event.set() + + # ------------------------------------------------------------------ + # yt-dlp resolution (blocking -> executor) + # ------------------------------------------------------------------ + def _blocking_extract(self, query: str) -> dict: + with yt_dlp.YoutubeDL(self._ytdlp_opts) as ydl: + info = ydl.extract_info(query, download=True) + if "entries" in info: # search/playlist -> first entry + info = info["entries"][0] + return info + + async def _resolve_to_mp3(self, track: QueuedTrack): + loop = asyncio.get_running_loop() + info = await loop.run_in_executor( + None, functools.partial(self._blocking_extract, track.query) + ) + track.title = info.get("title", track.query) + vid = info.get("id") + mp3_path = self.mp3_dir / f"{vid}.mp3" + track.mp3_path = str(mp3_path) + return track + + def _cleanup_file(self, track: Optional[QueuedTrack]): + if track and track.mp3_path and os.path.isfile(track.mp3_path): + try: + os.remove(track.mp3_path) + except OSError: + log.warning("Could not delete mp3: %s", track.mp3_path) + + # ------------------------------------------------------------------ + # Playback + # ------------------------------------------------------------------ + async def _play_track(self, guild_id: int, track: QueuedTrack, done_event: asyncio.Event): + await self._resolve_to_mp3(track) + player = lavalink.get_player(guild_id) + gp = self.players.get(guild_id) + if gp: + await player.set_volume(gp.volume) + loaded = await player.load_tracks(track.mp3_path) + if loaded.has_error or loaded.load_type != lavalink.enums.LoadType.TRACK_LOADED: + log.error("YTAudio: failed to load mp3 into Lavalink: %s", track.mp3_path) + done_event.set() + return + player.add(requester=track.requester, track=loaded.tracks[0]) + if not player.is_playing: + await player.play() + + def _get_gp(self, guild_id: int) -> GuildPlayer: + gp = self.players.get(guild_id) + if gp is None: + gp = GuildPlayer(self, guild_id) + self.players[guild_id] = gp + gp.start() + return gp + + async def _ensure_connected(self, ctx: commands.Context) -> bool: + if not self._ll_ready and not lavalink.get_all_nodes(): + await ctx.send("❌ Lavalink is not connected yet. Try again in a moment.") + return False + if ctx.author.voice is None or ctx.author.voice.channel is None: + await ctx.send("❌ You need to be in a voice channel.") + return False + me_voice = ctx.guild.me.voice + if me_voice is None or me_voice.channel != ctx.author.voice.channel: + await lavalink.connect(ctx.author.voice.channel) + return True + + # ------------------------------------------------------------------ + # Commands + # ------------------------------------------------------------------ + @commands.command() + @commands.guild_only() + async def play(self, ctx: commands.Context, *, query: str): + """Play a YouTube URL or search query.""" + if not await self._ensure_connected(ctx): + return + gp = self._get_gp(ctx.guild.id) + gp.volume = await self.config.guild(ctx.guild).volume() + track = QueuedTrack(query, ctx.author) + await gp.queue.put(track) + await ctx.send(f"➕ Queued: `{query}`") + + @commands.command() + @commands.guild_only() + async def skip(self, ctx: commands.Context): + """Skip the current track.""" + try: + player = lavalink.get_player(ctx.guild.id) + except Exception: + return await ctx.send("❌ Nothing is playing.") + await player.stop() + gp = self.players.get(ctx.guild.id) + if gp: + gp.next_event.set() + await ctx.send("⏭️ Skipped.") + + @commands.command() + @commands.guild_only() + async def stop(self, ctx: commands.Context): + """Stop playback and clear the queue.""" + gp = self.players.get(ctx.guild.id) + if gp: + while not gp.queue.empty(): + try: + t = gp.queue.get_nowait() + self._cleanup_file(t) + except asyncio.QueueEmpty: + break + gp.next_event.set() + try: + player = lavalink.get_player(ctx.guild.id) + await player.stop() + except Exception: + pass + await ctx.send("⏹️ Stopped and cleared the queue.") + + @commands.command() + @commands.guild_only() + async def pause(self, ctx: commands.Context): + """Pause playback.""" + try: + player = lavalink.get_player(ctx.guild.id) + except Exception: + return await ctx.send("❌ Nothing is playing.") + await player.pause(True) + await ctx.send("⏸️ Paused.") + + @commands.command() + @commands.guild_only() + async def resume(self, ctx: commands.Context): + """Resume playback.""" + try: + player = lavalink.get_player(ctx.guild.id) + except Exception: + return await ctx.send("❌ Nothing is playing.") + await player.pause(False) + await ctx.send("▶️ Resumed.") + + @commands.command(name="queue") + @commands.guild_only() + async def queue_cmd(self, ctx: commands.Context): + """Show the queue.""" + gp = self.players.get(ctx.guild.id) + if not gp or (gp.current is None and gp.queue.empty()): + return await ctx.send("The queue is empty.") + lines = [] + if gp.current: + lines.append(f"**Now:** {gp.current.title or gp.current.query}") + pending: List[QueuedTrack] = list(gp.queue._queue) # snapshot + for i, t in enumerate(pending, 1): + lines.append(f"{i}. {t.title or t.query}") + await ctx.send("\n".join(lines[:20])) + + @commands.command(name="nowplaying", aliases=["np"]) + @commands.guild_only() + async def nowplaying(self, ctx: commands.Context): + """Show the current track.""" + gp = self.players.get(ctx.guild.id) + if not gp or gp.current is None: + return await ctx.send("Nothing is playing.") + await ctx.send(f"🎵 {gp.current.title or gp.current.query}") + + @commands.command() + @commands.guild_only() + async def volume(self, ctx: commands.Context, level: int): + """Set volume (0-150).""" + level = max(0, min(150, level)) + await self.config.guild(ctx.guild).volume.set(level) + gp = self.players.get(ctx.guild.id) + if gp: + gp.volume = level + try: + player = lavalink.get_player(ctx.guild.id) + await player.set_volume(level) + except Exception: + pass + await ctx.send(f"🔊 Volume set to {level}.") + + @commands.command() + @commands.guild_only() + async def join(self, ctx: commands.Context): + """Join your voice channel.""" + if await self._ensure_connected(ctx): + await ctx.send("✅ Joined.") + + @commands.command(aliases=["disconnect", "dc"]) + @commands.guild_only() + async def leave(self, ctx: commands.Context): + """Leave the voice channel and clear state.""" + gp = self.players.get(ctx.guild.id) + if gp: + while not gp.queue.empty(): + try: + self._cleanup_file(gp.queue.get_nowait()) + except asyncio.QueueEmpty: + break + gp.next_event.set() + try: + player = lavalink.get_player(ctx.guild.id) + await player.stop() + await player.disconnect() + except Exception: + pass + await ctx.send("👋 Left the channel.") + + # ------------------------------------------------------------------ + # Cleanup + # ------------------------------------------------------------------ + async def cog_unload(self): + if self._connect_task: + self._connect_task.cancel() + for gp in self.players.values(): + if gp.task: + gp.task.cancel() + try: + lavalink.unregister_event_listener(self._track_event_handler) + except Exception: + pass + # This cog owns the connection: close it on unload. + await self._safe_close() + # Sweep leftover mp3s. + try: + shutil.rmtree(self.mp3_dir, ignore_errors=True) + self.mp3_dir.mkdir(parents=True, exist_ok=True) + except Exception: + pass