From ce61bf6bd5949046d15f58609f572fedcb8a073e Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 4 Feb 2026 16:16:26 +0100 Subject: refactor music --- cogs/music/audio_playback.py | 179 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 cogs/music/audio_playback.py (limited to 'cogs/music/audio_playback.py') diff --git a/cogs/music/audio_playback.py b/cogs/music/audio_playback.py new file mode 100644 index 0000000..3d4f453 --- /dev/null +++ b/cogs/music/audio_playback.py @@ -0,0 +1,179 @@ +# This file is part of Senator Sperma +# Licensed under the GPL-3.0 License +# Copyright (C) 2026 Salmonik @ salmonik@tuta.io +# https://codeberg.org/salmonik/senator-sperma + +import os +import logging +from pathlib import Path + +import discord +import mutagen + +logger = logging.getLogger(__name__) + +class AudioPlaybackManager: + """Handles audio playback, voice connections, and source preparation.""" + + @staticmethod + async def cleanup_temp_file(source): + """Clean up temporary file if it exists.""" + if source and os.path.exists(source): + try: + os.remove(source) + logger.info(f"Cleaned up temporary file: {source}") + except Exception as e: + logger.error(f"Error cleaning up file {source}: {str(e)}") + + @staticmethod + async def validate_voice_channel(ctx): + """Validate user is in a voice channel. Returns channel or None.""" + if not ctx.author.voice or not ctx.author.voice.channel: + await ctx.respond("Musisz być na kanale głosowym, żebym mógł dołączyć!", ephemeral=True) + return None + return ctx.author.voice.channel + + @staticmethod + async def prepare_file_source(source): + """Prepare and validate local file source. Returns (display_name, absolute_path) or (None, None).""" + logger.info(f"Attempting to play local file: {source}") + + # Get absolute path and verify file exists + if not os.path.isabs(source): + source = os.path.abspath(source) + logger.debug(f"Converted to absolute path: {source}") + + if not os.path.exists(source): + logger.error(f"File not found: {source}") + return None, None + + filename = os.path.basename(source) + display_name = os.path.splitext(filename)[0] + return display_name, source + + @staticmethod + async def connect_to_voice_channel(ctx, vc, target_channel): + """Connect or move to voice channel. Returns voice client.""" + try: + # Connect to voice channel if not connected + if vc is None: + logger.info(f"Connecting to voice channel: {target_channel.name}") + vc = await target_channel.connect() + logger.info(f"Connected to voice channel: {target_channel.name}") + # Move to user's channel if in a different one + elif vc.channel != target_channel: + logger.info(f"Moving to voice channel: {target_channel.name}") + await vc.move_to(target_channel) + logger.info(f"Moved to voice channel: {target_channel.name}") + return vc + except Exception as e: + logger.error(f"Error connecting to voice channel: {str(e)}") + raise + + @staticmethod + async def prepare_audio_source(source, is_url, display_name): + """Prepare audio source for playback. Returns (source, final_display_name).""" + if is_url: + # For URLs, use FFmpeg with the direct URL + ffmpeg_options = { + 'options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5', + 'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5' + } + audio_source = discord.FFmpegPCMAudio(source, **ffmpeg_options) + return audio_source, display_name + else: + # For local files, use default options + audio_source = discord.FFmpegPCMAudio(source=source) + + # Try to get metadata for local files + try: + metadata = mutagen.File(source) + if metadata: + title = metadata.get('title', [None])[0] if hasattr(metadata, 'get') else None + if title: + display_name = title + logger.debug(f"Read metadata - title: {title}") + except Exception as e: + logger.warning(f"Could not read metadata from {source}: {str(e)}") + + return audio_source, display_name + + @staticmethod + async def cleanup_if_needed(source, delete_after_play, is_url): + """Clean up temporary file if needed.""" + if delete_after_play and not is_url: + await AudioPlaybackManager.cleanup_temp_file(source) + + @staticmethod + async def prepare_source(ctx, source): + """Prepare source and get display name. Returns (source, display_name, is_url) or (None, None, None).""" + is_url = source.startswith(('http://', 'https://')) + + if is_url: + logger.info(f"Attempting to play stream from URL: {source}") + return source, source, is_url + + display_name, absolute_source = await AudioPlaybackManager.prepare_file_source(source) + if not display_name: + return None, None, None + + return absolute_source, display_name, is_url + + @staticmethod + async def execute_playback(ctx, source, display_name, is_url): + """Execute the actual audio playback.""" + vc = ctx.voice_client + target_channel = await AudioPlaybackManager.validate_voice_channel(ctx) + if not target_channel: + return False + + try: + vc = await AudioPlaybackManager.connect_to_voice_channel(ctx, vc, target_channel) + + # Stop any currently playing audio + if vc.is_playing() or vc.is_paused(): + vc.stop() + + # Prepare audio source + audio_source, final_display_name = await AudioPlaybackManager.prepare_audio_source(source, is_url, display_name) + + # Play the audio + vc.play(audio_source) + logger.info(f"Playing: {final_display_name}") + await ctx.respond(f"grom: {final_display_name}") + return True + + except Exception as e: + logger.error(f"Error in play: {str(e)}", exc_info=True) + await ctx.respond(f"eror: {str(e)}", ephemeral=True) + return False + + @staticmethod + async def validate_and_prepare_source(ctx, source): + """Validate and prepare source for playback. Returns (source, display_name, is_url) or (None, None, None).""" + prepared_source, display_name, is_url = await AudioPlaybackManager.prepare_source(ctx, source) + if not prepared_source: + return None, None, None + return prepared_source, display_name, is_url + + @staticmethod + async def play(ctx, source, delete_after_play=False): + """Play audio from a local file or URL. + + Args: + ctx: Discord context + source: Either a local file path or a URL to stream from + delete_after_play: If True, delete the file after playing (only for local files) + """ + # Validate and prepare source + prepared_source, display_name, is_url = await AudioPlaybackManager.validate_and_prepare_source(ctx, source) + if not prepared_source: + await AudioPlaybackManager.cleanup_if_needed(source, delete_after_play, False) + return + + # Execute playback + success = await AudioPlaybackManager.execute_playback(ctx, prepared_source, display_name, is_url) + + # Cleanup if needed + if not success: + await AudioPlaybackManager.cleanup_if_needed(prepared_source, delete_after_play, is_url) -- cgit v1.2.3