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 +++++++++++++++++ cogs/music/media_processing.py | 124 ++++++++++++ cogs/music/play.py | 219 --------------------- cogs/music/play_commands.py | 236 +++++++++++++++++++++++ cogs/music/play_utils.py | 426 ----------------------------------------- cogs/music/search_utils.py | 286 +++++++++++++++++++++++++++ 6 files changed, 825 insertions(+), 645 deletions(-) create mode 100644 cogs/music/audio_playback.py create mode 100644 cogs/music/media_processing.py delete mode 100644 cogs/music/play.py create mode 100644 cogs/music/play_commands.py delete mode 100644 cogs/music/play_utils.py create mode 100644 cogs/music/search_utils.py (limited to 'cogs/music') 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) diff --git a/cogs/music/media_processing.py b/cogs/music/media_processing.py new file mode 100644 index 0000000..430b51f --- /dev/null +++ b/cogs/music/media_processing.py @@ -0,0 +1,124 @@ +# 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 asyncio +import logging +from pathlib import Path + +import yt_dlp +from youtubesearchpython import VideosSearch + +logger = logging.getLogger(__name__) + +class MediaDownloader: + """Handles downloading media from various sources.""" + + @staticmethod + async def download(ctx, url): + """Download audio from YouTube URL.""" + # Create temp directory if it doesn't exist + temp_dir = os.path.join('cogs', 'temp', 'music') + os.makedirs(temp_dir, exist_ok=True) + + ytdl_opts = { + "outtmpl": os.path.join(temp_dir, "%(title)s.%(ext)s"), + "postprocessors": [ + { + "key": "FFmpegExtractAudio", + "preferredcodec": "mp3", + "preferredquality": "192", + } + ], + "force_ipv4": True, + } + + try: + with yt_dlp.YoutubeDL(ytdl_opts) as ydl: + info_dict = ydl.extract_info(url, download=True) + # Get the base filename without extension + base_filename = os.path.splitext(ydl.prepare_filename(info_dict))[0] + # Add .mp3 extension since we're extracting audio + filename = f"{base_filename}.mp3" + logger.info(f"Downloaded file: {filename}") + return filename + except Exception as e: + logger.error(f"Error downloading video: {e}") + await ctx.respond("eror pobirajync wideło", ephemeral=True) + + @staticmethod + async def get_url(ctx, query): + """Search YouTube and get URL for a query.""" + search = VideosSearch(query, limit=1) + if search.result()["result"]: + title = search.result()["result"][0]["title"] + url = search.result()["result"][0]["link"] + logger.info(f"Found video: {title}") + return url, title + else: + logger.warning("No results found for the query.") + await ctx.respond("nic ni ma", ephemeral=True) + return None + +class AudioCompressor: + """Handles audio compression using FFmpeg.""" + + @staticmethod + async def compress(ctx, filename): + """Compress audio file to opus format.""" + if not filename: + await ctx.respond("dzie wideło", ephemeral=True) + return None + + # Create temp directory if it doesn't exist + temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'temp', 'compressed') + os.makedirs(temp_dir, exist_ok=True) + + # Create a temporary output filename in the temp directory + base_name = os.path.basename(filename) + output_filename = os.path.join(temp_dir, f"{os.path.splitext(base_name)[0]}_compressed.opus") + + # Ensure the output directory exists + os.makedirs(os.path.dirname(output_filename) or '.', exist_ok=True) + + # Use proper quoting for filenames with spaces + ffmpeg_command = [ + 'ffmpeg', + '-y', # Overwrite output file if it exists + '-i', filename, + '-ac', '1', # Mono audio + '-b:a', '500', # 500k bitrate + '-acodec', 'libopus', + output_filename + ] + + try: + logger.info(f"Running FFmpeg command: {' '.join(ffmpeg_command)}") + process = await asyncio.create_subprocess_exec( + *ffmpeg_command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + + _, stderr = await process.communicate() + + if process.returncode != 0: + error_msg = stderr.decode() + logger.error(f"FFmpeg error ({process.returncode}): {error_msg}") + await ctx.respond("eror kompresujync pilk", ephemeral=True) + return None + + if not os.path.exists(output_filename): + logger.error(f"Output file was not created: {output_filename}") + await ctx.respond("eror kompresujync pilk - output file not created", ephemeral=True) + return None + + logger.info(f"Successfully compressed file: {output_filename}") + return output_filename + + except Exception as e: + logger.error(f"Error compressing file: {str(e)}", exc_info=True) + await ctx.respond(f"eror kompresujync pilk: {str(e)}", ephemeral=True) + return None diff --git a/cogs/music/play.py b/cogs/music/play.py deleted file mode 100644 index c7dd9f1..0000000 --- a/cogs/music/play.py +++ /dev/null @@ -1,219 +0,0 @@ -# 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 discord -import mutagen -from typing import List, Tuple, Callable, Awaitable -from os import PathLike - -from cogs.music import play_utils -from discord.ext import commands -from discord.commands import OptionChoice -import logging - -logger = logging.getLogger(__name__) - -class Play(commands.Cog): - def __init__(self, bot): - self.bot = bot - self.temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "temp") - self.path_mapping = {} # Maps display names to full paths - - @staticmethod - async def _create_search_embed(candidates: List[str]) -> Tuple[discord.Embed, discord.ui.View]: - """Create an embed and view for search results.""" - embed = discord.Embed( - title="Wyniki wyszukiwania", - description="Wybierz plik do odtworzenia:", - color=discord.Color.blue() - ) - - view = discord.ui.View(disable_on_timeout=True) - - for i, candidate in enumerate(candidates, start=1): - filename = os.path.basename(candidate) - title = None - - # Try to get metadata if it's an audio file - if os.path.splitext(filename)[1].lower() in ['.mp3', '.wav', '.opus']: - try: - metadata = mutagen.File(candidate) - if metadata: - title = metadata.get('title', None) - except Exception as e: - logger.debug(f"Could not read metadata for {filename}: {str(e)}") - - # Add field with title if available, otherwise use filename - display_name = title if title else filename - embed.add_field(name=f"{i}. {display_name}", value="\u200b", inline=False) - - return embed, view - - async def _handle_search_results( - self, - ctx: discord.ApplicationContext, - candidates: List[str], - play_callback: Callable[[discord.Interaction, str], Awaitable[None]] - ) -> None: - """Handle the display of search results and button interactions.""" - if not candidates: - await ctx.respond("Nie znaleziono pliku do odtworzenia", ephemeral=True) - return - - embed, view = await self._create_search_embed(candidates) - - # Add buttons for each candidate - for i, candidate in enumerate(candidates, start=1): - button = discord.ui.Button(label=f"Play {i}") - - async def button_callback(interaction: discord.Interaction, candidate_path=candidate): - await interaction.response.defer(ephemeral=True) - await ctx.interaction.delete_original_response() - try: - await play_callback(interaction, candidate_path) - except Exception as e: - logger.error(f"Error in play callback: {str(e)}", exc_info=True) - await interaction.followup.send(f"Wystąpił błąd podczas odtwarzania: {str(e)}", ephemeral=True) - - button.callback = button_callback - view.add_item(button) - - await ctx.respond(embed=embed, view=view, ephemeral=True) - - async def _search_autocomplete(self, ctx: discord.AutocompleteContext) -> List[OptionChoice]: - """Autocomplete handler for search queries.""" - query = ctx.value.lower() - try: - results = await play_utils.PlayUtils.search_local_autocomplete(query, max_results=25) - choices = [] - self.path_mapping = {} # Reset the mapping for each new query - - for path in results: - filename = os.path.splitext(os.path.basename(path))[0] # Remove extension - display = filename # Default display name - - # Try to get metadata for display - try: - if os.path.splitext(path)[1].lower() in ['.mp3', '.wav', '.opus', '.m4a']: - metadata = mutagen.File(path, easy=True) - - # Get metadata fields, falling back to filename if not available - title = metadata.get('title', [filename])[0] - artist = metadata.get('artist', [''])[0] - album_artist = metadata.get('albumartist', [artist])[0] - - # Build display name based on available metadata - if album_artist and title: - display = f"{album_artist} - {title}" - elif artist and title: - display = f"{artist} - {title}" - elif title: - display = title - - except Exception as e: - logger.debug(f"Could not read metadata for {filename}: {str(e)}") - - # Clean up the display name (remove any extra whitespace or brackets) - display = display.strip() - - # Store the mapping (use the cleaned display name as the key) - self.path_mapping[display] = path - - # Truncate display name if too long - if len(display) > 90: # Leave room for potential numbering - display = display[:87] + '...' - - choices.append(OptionChoice(display[:100], display[:100])) - - # Limit to 25 choices (Discord's limit) - if len(choices) >= 25: - break - - return choices - - except Exception as e: - logger.error(f"Error in autocomplete: {str(e)}", exc_info=True) - return [] - - @discord.slash_command(name="grajcuj", description="grajcujmy wszyscy!") - async def play( - self, - ctx: discord.ApplicationContext, - file: str = discord.Option( - str, - "wyszukaj plik z serwera", - autocomplete=_search_autocomplete, - required=False - ), - stream: str = discord.Option( - str, - "odtwórz stream internetowy (e.g. radio)", - required=False - ), - compressed: bool = discord.Option( - bool, - "Czy skompresować dźwięk? (tylko dla plików lokalnych)", - default=False - ) - ): - """Odtwarza wybrany plik dźwiękowy lub stream internetowy.""" - await ctx.defer() - - # Check if either file or stream is provided - if not file and not stream: - await ctx.respond("Musisz podać plik do odtworzenia lub adres streamu.", ephemeral=True) - return - - if file and stream: - await ctx.respond("Możesz podać tylko plik LUB stream, nie oba naraz.", ephemeral=True) - return - - try: - if stream: - # Handle internet stream - await play_utils.PlayUtils(self.bot).play(ctx, stream) - else: - # Handle local file - full_path = self.path_mapping.get(file, file) - - if not os.path.exists(full_path): - logger.warning(f"File not found: {full_path} (display name: {file})") - await ctx.respond("Wybrany plik nie istnieje.", ephemeral=True) - return - - if compressed: - # Compress the file before playing - compressed_path = await play_utils.PlayUtils(self.bot).compress(ctx, full_path) - if not compressed_path: - await ctx.respond("Nie udało się skompresować pliku.", ephemeral=True) - return - - try: - await play_utils.PlayUtils(self.bot).play(ctx, compressed_path, delete_after_play=True) - except Exception as e: - # Clean up the compressed file if play fails - if os.path.exists(compressed_path): - try: - os.remove(compressed_path) - logger.info(f"Cleaned up temporary file after play error: {compressed_path}") - except Exception as cleanup_error: - logger.error(f"Error cleaning up file {compressed_path}: {str(cleanup_error)}") - raise e - else: - await play_utils.PlayUtils(self.bot).play(ctx, full_path) - except Exception as e: - logger.error(f"Error playing: {str(e)}", exc_info=True) - await ctx.respond(f"Wystąpił błąd podczas odtwarzania: {str(e)}", ephemeral=True) - - @discord.slash_command(name="grajcuj_plik", description="grajcuj plik") - async def play_file(self, ctx: discord.ApplicationContext, attachment: discord.Attachment = discord.Option(discord.Attachment, "co zagrać??", required=True)): - await ctx.defer() - file_path: PathLike[str] | str = os.path.join(self.temp_dir, attachment.filename) - await attachment.save(file_path) - await play_utils.PlayUtils(self.bot).play(ctx, file_path) - -def setup(bot): - bot.add_cog(Play(bot)) \ No newline at end of file diff --git a/cogs/music/play_commands.py b/cogs/music/play_commands.py new file mode 100644 index 0000000..b25ce84 --- /dev/null +++ b/cogs/music/play_commands.py @@ -0,0 +1,236 @@ +# 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 discord +import mutagen +from typing import List, Tuple, Callable, Awaitable +from os import PathLike + +from cogs.music import search_utils, audio_playback, media_processing +from discord.ext import commands +from discord.commands import OptionChoice +import logging + +logger = logging.getLogger(__name__) + +class PlayCommands(commands.Cog): + """Discord slash commands for music playback.""" + + def __init__(self, bot): + self.bot = bot + self.temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "temp") + self.path_mapping = {} # Maps display names to full paths + + @staticmethod + async def _create_search_embed(candidates: List[str]) -> Tuple[discord.Embed, discord.ui.View]: + """Create an embed and view for search results.""" + embed = discord.Embed( + title="Wyniki wyszukiwania", + description="Wybierz plik do odtworzenia:", + color=discord.Color.blue() + ) + + view = discord.ui.View(disable_on_timeout=True) + + for i, candidate in enumerate(candidates, start=1): + filename = os.path.basename(candidate) + title = None + + # Try to get metadata if it's an audio file + if os.path.splitext(filename)[1].lower() in ['.mp3', '.wav', '.opus']: + try: + metadata = mutagen.File(candidate) + if metadata: + title = metadata.get('title', None) + except Exception as e: + logger.debug(f"Could not read metadata for {filename}: {str(e)}") + + # Add field with title if available, otherwise use filename + display_name = title if title else filename + embed.add_field(name=f"{i}. {display_name}", value="\u200b", inline=False) + + return embed, view + + async def _handle_search_results( + self, + ctx: discord.ApplicationContext, + candidates: List[str], + play_callback: Callable[[discord.Interaction, str], Awaitable[None]] + ) -> None: + """Handle the display of search results and button interactions.""" + if not candidates: + await ctx.respond("Nie znaleziono pliku do odtworzenia", ephemeral=True) + return + + embed, view = await self._create_search_embed(candidates) + + # Add buttons for each candidate + for i, candidate in enumerate(candidates, start=1): + button = discord.ui.Button(label=f"Play {i}") + + async def button_callback(interaction: discord.Interaction, candidate_path=candidate): + await interaction.response.defer(ephemeral=True) + await ctx.interaction.delete_original_response() + try: + await play_callback(interaction, candidate_path) + except Exception as e: + logger.error(f"Error in play callback: {str(e)}", exc_info=True) + await interaction.followup.send(f"Wystąpił błąd podczas odtwarzania: {str(e)}", ephemeral=True) + + button.callback = button_callback + view.add_item(button) + + await ctx.respond(embed=embed, view=view, ephemeral=True) + + def _get_display_name_from_metadata(self, path: str, filename: str) -> str: + """Extract display name from audio file metadata.""" + try: + if os.path.splitext(path)[1].lower() in ['.mp3', '.wav', '.opus', '.m4a']: + metadata = mutagen.File(path, easy=True) + + title = metadata.get('title', [filename])[0] + artist = metadata.get('artist', [''])[0] + album_artist = metadata.get('albumartist', [artist])[0] + + if album_artist and title: + return f"{album_artist} - {title}" + elif artist and title: + return f"{artist} - {title}" + elif title: + return title + except Exception as e: + logger.debug(f"Could not read metadata for {filename}: {str(e)}") + + return filename + + async def _search_autocomplete(self, ctx: discord.AutocompleteContext) -> List[OptionChoice]: + """Autocomplete handler for search queries.""" + query = ctx.value.lower() + try: + results = await search_utils.AudioSearcher.search_local_autocomplete(query, max_results=25) + choices = [] + self.path_mapping = {} # Reset the mapping for each new query + + for path in results: + filename = os.path.splitext(os.path.basename(path))[0] # Remove extension + display = self._get_display_name_from_metadata(path, filename) + + # Clean up the display name (remove any extra whitespace or brackets) + display = display.strip() + + # Store the mapping (use the cleaned display name as the key) + self.path_mapping[display] = path + + # Truncate display name if too long + if len(display) > 90: # Leave room for potential numbering + display = display[:87] + '...' + + choices.append(OptionChoice(display[:100], display[:100])) + + # Limit to 25 choices (Discord's limit) + if len(choices) >= 25: + break + + return choices + + except Exception as e: + logger.error(f"Error in autocomplete: {str(e)}", exc_info=True) + return [] + + def _validate_play_inputs(self, file: str, stream: str, ctx: discord.ApplicationContext) -> bool: + """Validate input parameters for play command.""" + if not file and not stream: + ctx.respond("Musisz podać plik do odtworzenia lub adres streamu.", ephemeral=True) + return False + + if file and stream: + ctx.respond("Możesz podać tylko plik LUB stream, nie oba naraz.", ephemeral=True) + return False + + return True + + async def _handle_stream_play(self, ctx: discord.ApplicationContext, stream: str) -> None: + """Handle internet stream playback.""" + await audio_playback.AudioPlaybackManager.play(ctx, stream) + + async def _handle_compressed_file_play(self, ctx: discord.ApplicationContext, full_path: str) -> None: + """Handle compressed file playback with cleanup.""" + compressed_path = await media_processing.AudioCompressor.compress(ctx, full_path) + if not compressed_path: + await ctx.respond("Nie udało się skompresować pliku.", ephemeral=True) + return + + try: + await audio_playback.AudioPlaybackManager.play(ctx, compressed_path, delete_after_play=True) + except Exception as e: + # Clean up the compressed file if play fails + if os.path.exists(compressed_path): + try: + os.remove(compressed_path) + logger.info(f"Cleaned up temporary file after play error: {compressed_path}") + except Exception as cleanup_error: + logger.error(f"Error cleaning up file {compressed_path}: {str(cleanup_error)}") + raise e + + async def _handle_local_file_play(self, ctx: discord.ApplicationContext, file: str, compressed: bool) -> None: + """Handle local file playback.""" + full_path = self.path_mapping.get(file, file) + + if not os.path.exists(full_path): + logger.warning(f"File not found: {full_path} (display name: {file})") + await ctx.respond("Wybrany plik nie istnieje.", ephemeral=True) + return + + if compressed: + await self._handle_compressed_file_play(ctx, full_path) + else: + await audio_playback.AudioPlaybackManager.play(ctx, full_path) + + @discord.slash_command(name="grajcuj", description="grajcujmy wszyscy!") + async def play( + self, + ctx: discord.ApplicationContext, + file: str = discord.Option( + str, + "wyszukaj plik z serwera", + autocomplete=_search_autocomplete, + required=False + ), + stream: str = discord.Option( + str, + "odtwórz stream internetowy (e.g. radio)", + required=False + ), + compressed: bool = discord.Option( + bool, + "Czy skompresować dźwięk? (tylko dla plików lokalnych)", + default=False + ) + ): + """Odtwarza wybrany plik dźwiękowy lub stream internetowy.""" + await ctx.defer() + + if not self._validate_play_inputs(file, stream, ctx): + return + + try: + if stream: + await self._handle_stream_play(ctx, stream) + else: + await self._handle_local_file_play(ctx, file, compressed) + except Exception as e: + logger.error(f"Error playing: {str(e)}", exc_info=True) + await ctx.respond(f"Wystąpił błąd podczas odtwarzania: {str(e)}", ephemeral=True) + + @discord.slash_command(name="grajcuj_plik", description="grajcuj plik") + async def play_file(self, ctx: discord.ApplicationContext, attachment: discord.Attachment = discord.Option(discord.Attachment, "co zagrać??", required=True)): + await ctx.defer() + file_path: PathLike[str] | str = os.path.join(self.temp_dir, attachment.filename) + await attachment.save(file_path) + await audio_playback.AudioPlaybackManager.play(ctx, file_path) + +def setup(bot): + bot.add_cog(PlayCommands(bot)) diff --git a/cogs/music/play_utils.py b/cogs/music/play_utils.py deleted file mode 100644 index ed98395..0000000 --- a/cogs/music/play_utils.py +++ /dev/null @@ -1,426 +0,0 @@ -# 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 json -import logging -import math -import os -import asyncio -import re -from pathlib import Path - -import discord -import mutagen -import yt_dlp -from rapidfuzz import fuzz -from youtubesearchpython import VideosSearch - -logger = logging.getLogger(__name__) - -class Utils: - @staticmethod - def normalize_name(p: Path, keep_ext=False): - """Preprocess filenames: lower, replace separators, strip extensions optionally.""" - name = str(p.name if keep_ext else p.stem) - return name.lower().replace('_', ' ').replace('-', ' ') - - @staticmethod - def length_penalty(query, candidate): - """Penalize tiny filenames when query is long; returns multiplier [0.0,1.0].""" - ql = len(query) - cl = len(candidate) - if ql == 0: - return 1.0 - ratio = cl / ql - # Prefer candidate lengths near query length; gentle falloff otherwise - return math.exp(-0.5 * ((math.log(max(ratio, 1e-6)))**2)) - - @staticmethod - def combined_score(query, candidate): - """Calculate a combined fuzzy matching score using multiple strategies. - - Args: - query: The search query - candidate: The candidate string to score against - - Returns: - float: Score between 0-100, where higher means better match - """ - if not query or not candidate: - return 0.0 - - query = query.lower().strip() - candidate = candidate.lower() - - # First check for direct substring match (fast path) - if query in candidate: - # Base score of 100 for exact match, with bonus for being at word boundary - base_score = 100 - # Check if match is at word boundary - if re.search(rf'\b{re.escape(query)}\b', candidate): - base_score = 110 # Slight bonus for word boundary match - return min(100, base_score) - - # For very short queries (2-3 chars), be more lenient - if len(query) <= 3: - # Check for partial matches at word boundaries - if re.search(rf'\b{re.escape(query)}', candidate): - return 90 # High score for short query matching word start - if query in candidate: - return 80 # Still good for any substring match with short query - - # For longer queries, check word presence - query_words = set(word for word in re.split(r'\W+', query) if len(word) > 1) # Reduced min word length to 1 - if query_words: - candidate_words = set(re.split(r'\W+', candidate)) - # Check if all query words are in candidate (with some leniency) - matching_words = sum(1 for word in query_words if word in candidate_words) - word_ratio = matching_words / len(query_words) - - # If no words match at all, return 0 - if matching_words == 0: - return 0.0 - - # Calculate different similarity metrics - r_ratio = fuzz.ratio(query, candidate) # Overall edit similarity - r_partial = fuzz.partial_ratio(query, candidate) # Best substring match - r_token_set = fuzz.token_set_ratio(query, candidate) # Word overlap - - # Calculate word-based scores - query_tokens = set(query.split()) - candidate_tokens = set(candidate.split()) - word_intersection = len(query_tokens & candidate_tokens) - word_union = len(query_tokens | candidate_tokens) - - # Jaccard similarity for word overlap - jaccard_sim = word_intersection / word_union if word_union > 0 else 0 - - # Calculate prefix score (bonus for matching start of words) - prefix_bonus = 0 - for q_word in query_words: - for c_word in candidate_words: - if c_word.startswith(q_word): - prefix_bonus += (len(q_word) / max(len(c_word), 1)) * 20 # Increased weight - - # Weighted sum with adjusted weights - score = ( - 0.40 * r_token_set + # Increased weight for token set - 0.25 * r_ratio + # Overall similarity - 0.15 * r_partial + # Substring matches - 0.15 * (jaccard_sim * 100) + # Word overlap ratio - 0.05 * min(prefix_bonus, 100) # Prefix bonus (capped at 100) - ) - - # Apply length penalty (less severe) - score *= min(1.0, 0.9 + (0.1 * Utils.length_penalty(query, candidate))) - - # Boost score based on word ratio - score = score * (0.7 + (0.3 * word_ratio)) - - score = int(score) - - # Ensure score is within bounds - return min(100, max(0, score)) - - return 0.0 # No match found - -class PlayUtils: - def __init__(self, bot): - self.bot = bot - - @staticmethod - async def download(ctx, url): - - # Create temp directory if it doesn't exist - temp_dir = os.path.join('cogs', 'temp', 'music') - os.makedirs(temp_dir, exist_ok=True) - - ytdl_opts = { - "outtmpl": os.path.join(temp_dir, "%(title)s.%(ext)s"), - "postprocessors": [ - { - "key": "FFmpegExtractAudio", - "preferredcodec": "mp3", - "preferredquality": "192", - } - ], - "force_ipv4": True, - } - - try: - with yt_dlp.YoutubeDL(ytdl_opts) as ydl: - info_dict = ydl.extract_info(url, download=True) - # Get the base filename without extension - base_filename = os.path.splitext(ydl.prepare_filename(info_dict))[0] - # Add .mp3 extension since we're extracting audio - filename = f"{base_filename}.mp3" - logger.info(f"Downloaded file: {filename}") - return filename - except Exception as e: - logger.error(f"Error downloading video: {e}") - await ctx.respond("eror pobirajync wideło", ephemeral=True) - - @staticmethod - async def get_url(ctx, query): - search = VideosSearch(query, limit=1) - if search.result()["result"]: - title = search.result()["result"][0]["title"] - url = search.result()["result"][0]["link"] - logger.info(f"Found video: {title}") - return url, title - else: - logger.warning("No results found for the query.") - await ctx.respond("nic ni ma", ephemeral=True) - return None - - @staticmethod - async def compress(ctx, filename): - if not filename: - await ctx.respond("dzie wideło", ephemeral=True) - return None - - # Create temp directory if it doesn't exist - temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'temp', 'compressed') - os.makedirs(temp_dir, exist_ok=True) - - # Create a temporary output filename in the temp directory - base_name = os.path.basename(filename) - output_filename = os.path.join(temp_dir, f"{os.path.splitext(base_name)[0]}_compressed.opus") - - # Ensure the output directory exists - os.makedirs(os.path.dirname(output_filename) or '.', exist_ok=True) - - # Use proper quoting for filenames with spaces - ffmpeg_command = [ - 'ffmpeg', - '-y', # Overwrite output file if it exists - '-i', filename, - '-ac', '1', # Mono audio - '-b:a', '500', # 500k bitrate - '-acodec', 'libopus', - output_filename - ] - - try: - logger.info(f"Running FFmpeg command: {' '.join(ffmpeg_command)}") - process = await asyncio.create_subprocess_exec( - *ffmpeg_command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - - stdout, stderr = await process.communicate() - - if process.returncode != 0: - error_msg = stderr.decode() - logger.error(f"FFmpeg error ({process.returncode}): {error_msg}") - await ctx.respond("eror kompresujync pilk", ephemeral=True) - return None - - if not os.path.exists(output_filename): - logger.error(f"Output file was not created: {output_filename}") - await ctx.respond("eror kompresujync pilk - output file not created", ephemeral=True) - return None - - logger.info(f"Successfully compressed file: {output_filename}") - return output_filename - - except Exception as e: - logger.error(f"Error compressing file: {str(e)}", exc_info=True) - await ctx.respond(f"eror kompresujync pilk: {str(e)}", ephemeral=True) - return None - - @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) - """ - # Check if user is in a voice channel - 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) - if delete_after_play 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)}") - return - - is_url = source.startswith(('http://', 'https://')) - - if not is_url: - # Handle local file - 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): - error_msg = f"File not found: {source}" - logger.error(error_msg) - await ctx.respond("Nie znaleziono pliku do odtworzenia", ephemeral=True) - return - - filename = os.path.basename(source) - display_name = os.path.splitext(filename)[0] - else: - # Handle URL - logger.info(f"Attempting to play stream from URL: {source}") - display_name = source # Use URL as display name for streams - - # Get or create voice client - vc = ctx.voice_client - target_channel = ctx.author.voice.channel - - 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}") - - # Stop any currently playing audio - if vc.is_playing() or vc.is_paused(): - vc.stop() - - # Play the audio - 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' - } - source = discord.FFmpegPCMAudio(source, **ffmpeg_options) - else: - # For local files, use default options - 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)}") - - vc.play(source) - logger.info(f"Playing: {display_name}") - await ctx.respond(f"grom: {display_name}") - - except Exception as e: - logger.error(f"Error in play: {str(e)}", exc_info=True) - await ctx.respond(f"eror: {str(e)}", ephemeral=True) - - # -- LOCAL PLAY -- - - @classmethod - async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False): - """Search for audio files with fuzzy matching and optional autocomplete support. - - Args: - query: Search query string (can be None for autocomplete) - max_results: Maximum number of results to return - autocomplete: If True, returns results even with empty query - - Returns: - List of matching file paths with scores - """ - if not query and not autocomplete: - return [] - - query = (query or '').lower().strip() - logger.info(f"Starting local search for query: '{query}'") - - try: - with open('settings.json') as f: - data = json.load(f) - search_paths = data.get('audio_paths', []) - logger.info(f"Searching in paths: {search_paths}") - - matches = [] - file_count = 0 - - for path in search_paths: - path = Path(path) - if not path.exists(): - logger.warning(f"Path does not exist: {path}") - continue - - logger.debug(f"Searching in directory: {path}") - - # Walk through all subdirectories - for file_path in path.rglob('*'): - if not file_path.is_file(): - continue - - # Check if file has a valid extension - if file_path.suffix.lower() not in {'.mp4', '.mp3', '.wav', '.opus', '.m4a'}: - continue - - file_count += 1 - file_name = file_path.name.lower() - - # For autocomplete, return all files if query is empty - if not query and autocomplete: - matches.append((0, str(file_path.absolute()))) - # For regular search, calculate fuzzy match score - elif query and len(query) >= 2: - # First do a simple substring check for debugging - if query in file_name: - logger.debug(f"Found direct match: {file_name} (contains '{query}')") - - score = Utils.combined_score(query, file_name) - logger.debug(f"File: {file_name} - Score: {score}") - - if score >= 30: # Minimum score threshold - matches.append((score, str(file_path.absolute()))) - - # Early return if we have enough high-scoring matches - if len(matches) >= max_results * 5: # Get more candidates for sorting - break - - if len(matches) >= max_results * 5: # Limit memory usage - break - - logger.info(f"Scanned {file_count} files total") - - # Sort by score (descending) and take top results - matches.sort(reverse=True, key=lambda x: x[0]) - result = [path for score, path in matches[:max_results]] - - logger.info(f"Search complete. Found {len(result)} matches out of {file_count} files.") - if not result and query: - logger.warning(f"No matches found for query: '{query}'. Try a different search term.") - return result - - except Exception as e: - logger.error(f"Error during local search: {str(e)}", exc_info=True) - return [] - - @classmethod - async def search_local_autocomplete(cls, query: str = None, max_results: int = 10): - """Wrapper for autocomplete searches that returns more results and allows empty query.""" - # For autocomplete, we want to show more options and allow empty queries - if not query: - return [] - - return await cls.search_local(query, max_results=max_results, autocomplete=True) - - - diff --git a/cogs/music/search_utils.py b/cogs/music/search_utils.py new file mode 100644 index 0000000..50d229c --- /dev/null +++ b/cogs/music/search_utils.py @@ -0,0 +1,286 @@ +# 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 json +import logging +import re +from pathlib import Path +import aiofiles +from rapidfuzz import fuzz + +logger = logging.getLogger(__name__) + +class SearchUtils: + """Utilities for searching and scoring audio files.""" + + @staticmethod + def normalize_name(p: Path, keep_ext=False): + """Preprocess filenames: lower, replace separators, strip extensions optionally.""" + name = str(p.name if keep_ext else p.stem) + return name.lower().replace('_', ' ').replace('-', ' ') + + @staticmethod + def length_penalty(query, candidate): + """Penalize tiny filenames when query is long; returns multiplier [0.0,1.0].""" + ql = len(query) + cl = len(candidate) + if ql == 0: + return 1.0 + ratio = cl / ql + # Prefer candidate lengths near query length; gentle falloff otherwise + return math.exp(-0.5 * ((math.log(max(ratio, 1e-6)))**2)) + + @staticmethod + def combined_score(query, candidate): + """Calculate a combined fuzzy matching score using multiple strategies. + + Args: + query: The search query + candidate: The candidate string to score against + + Returns: + float: Score between 0-100, where higher means better match + """ + if not query or not candidate: + return 0.0 + + query = query.lower().strip() + candidate = candidate.lower() + + # Check for direct matches first + direct_match_score = SearchUtils._check_direct_matches(query, candidate) + if direct_match_score is not None: + return direct_match_score + + # Handle short queries specially + short_query_score = SearchUtils._handle_short_queries(query, candidate) + if short_query_score is not None: + return short_query_score + + # For longer queries, use complex scoring + return SearchUtils._calculate_complex_score(query, candidate) + + @staticmethod + def _check_direct_matches(query: str, candidate: str) -> float: + """Check for direct substring matches.""" + if query in candidate: + base_score = 100 + if re.search(rf'\b{re.escape(query)}\b', candidate): + base_score = 110 + return min(100, base_score) + return 0 + + @staticmethod + def _handle_short_queries(query: str, candidate: str) -> float: + """Handle scoring for short queries (2-3 chars).""" + if len(query) <= 3: + if re.search(rf'\b{re.escape(query)}', candidate): + return 90 + if query in candidate: + return 80 + return 0 + + @staticmethod + def _calculate_complex_score(query: str, candidate: str) -> float: + """Calculate complex score for longer queries.""" + query_words = {word for word in re.split(r'\W+', query) if len(word) > 1} + if not query_words: + return 0.0 + + candidate_words = set(re.split(r'\W+', candidate)) + word_ratio = SearchUtils._calculate_word_ratio(query_words, candidate_words) + + if word_ratio == 0: + return 0.0 + + # Calculate various similarity metrics + similarity_scores = SearchUtils._get_similarity_scores(query, candidate) + word_scores = SearchUtils._get_word_scores(query, candidate, query_words, candidate_words) + + # Calculate final weighted score + final_score = SearchUtils._calculate_weighted_score(similarity_scores, word_scores, word_ratio) + + # Apply penalties and boosts + final_score = SearchUtils._apply_adjustments(final_score, query, candidate, word_ratio) + + return min(100, max(0, int(final_score))) + + @staticmethod + def _calculate_word_ratio(query_words: set, candidate_words: set) -> float: + """Calculate the ratio of matching words.""" + matching_words = sum(1 for word in query_words if word in candidate_words) + return matching_words / len(query_words) + + @staticmethod + def _get_similarity_scores(query: str, candidate: str) -> dict: + """Get various fuzzy similarity scores.""" + return { + 'ratio': fuzz.ratio(query, candidate), + 'partial': fuzz.partial_ratio(query, candidate), + 'token_set': fuzz.token_set_ratio(query, candidate) + } + + @staticmethod + def _get_word_scores(query: str, candidate: str, query_words: set, candidate_words: set) -> dict: + """Calculate word-based scores.""" + query_tokens = set(query.split()) + candidate_tokens = set(candidate.split()) + + word_intersection = len(query_tokens & candidate_tokens) + word_union = len(query_tokens | candidate_tokens) + jaccard_sim = word_intersection / word_union if word_union > 0 else 0 + + prefix_bonus = SearchUtils._calculate_prefix_bonus(query_words, candidate_words) + + return { + 'jaccard': jaccard_sim, + 'prefix': min(prefix_bonus, 100) + } + + @staticmethod + def _calculate_prefix_bonus(query_words: set, candidate_words: set) -> float: + """Calculate prefix matching bonus.""" + prefix_bonus = 0 + for q_word in query_words: + for c_word in candidate_words: + if c_word.startswith(q_word): + prefix_bonus += (len(q_word) / max(len(c_word), 1)) * 20 + return prefix_bonus + + @staticmethod + def _calculate_weighted_score(similarity_scores: dict, word_scores: dict, word_ratio: float) -> float: + """Calculate the final weighted score.""" + return ( + 0.40 * similarity_scores['token_set'] + + 0.25 * similarity_scores['ratio'] + + 0.15 * similarity_scores['partial'] + + 0.15 * (word_scores['jaccard'] * 100) + + 0.05 * word_scores['prefix'] + ) + + @staticmethod + def _apply_adjustments(score: float, query: str, candidate: str, word_ratio: float) -> float: + """Apply length penalty and word ratio boost.""" + import math + + # Apply length penalty + score *= min(1.0, 0.9 + (0.1 * SearchUtils.length_penalty(query, candidate))) + + # Apply word ratio boost + score = score * (0.7 + (0.3 * word_ratio)) + + return score + +class AudioSearcher: + """Handles searching for audio files in configured paths.""" + + @staticmethod + def _is_valid_audio_file(file_path: Path) -> bool: + """Check if file is a valid audio file.""" + valid_extensions = {'.mp4', '.mp3', '.wav', '.opus', '.m4a'} + return file_path.is_file() and file_path.suffix.lower() in valid_extensions + + @staticmethod + def _calculate_match_score(query: str, file_name: str, autocomplete: bool) -> tuple[int, str]: + """Calculate match score for a file.""" + if not query and autocomplete: + return 0, file_name + + if query and len(query) >= 2: + if query in file_name: + logger.debug(f"Found direct match: {file_name} (contains '{query}')") + + score = SearchUtils.combined_score(query, file_name) + logger.debug(f"File: {file_name} - Score: {score}") + + if score >= 30: + return score, file_name + + return -1, file_name + + @staticmethod + async def _search_single_path(path: Path, query: str, max_results: int, autocomplete: bool) -> list[tuple[int, str]]: + """Search for audio files in a single path.""" + matches = [] + + logger.debug(f"Searching in directory: {path}") + + for file_path in path.rglob('*'): + if not AudioSearcher._is_valid_audio_file(file_path): + continue + + file_name = file_path.name.lower() + score, _ = AudioSearcher._calculate_match_score(query, file_name, autocomplete) + + if score >= 0: + matches.append((score, str(file_path.absolute()))) + + if len(matches) >= max_results * 5: + break + + return matches + + @classmethod + async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False): + """Search for audio files with fuzzy matching and optional autocomplete support. + + Args: + query: Search query string (can be None for autocomplete) + max_results: Maximum number of results to return + autocomplete: If True, returns results even with empty query + + Returns: + List of matching file paths with scores + """ + if not query and not autocomplete: + return [] + + query = (query or '').lower().strip() + logger.info(f"Starting local search for query: '{query}'") + + try: + async with aiofiles.open('settings.json') as f: + data = json.load(f) + search_paths = data.get('audio_paths', []) + logger.info(f"Searching in paths: {search_paths}") + + matches = [] + file_count = 0 + + for path_str in search_paths: + path = Path(path_str) + if not path.exists(): + logger.warning(f"Path does not exist: {path}") + continue + + path_matches = await cls._search_single_path(path, query, max_results, autocomplete) + matches.extend(path_matches) + file_count += len(path_matches) + + if len(matches) >= max_results * 5: + break + + logger.info(f"Scanned {file_count} files total") + + matches.sort(reverse=True, key=lambda x: x[0]) + result = [path for score, path in matches[:max_results]] + + logger.info(f"Search complete. Found {len(result)} matches out of {file_count} files.") + if not result and query: + logger.warning(f"No matches found for query: '{query}'. Try a different search term.") + return result + + except Exception as e: + logger.error(f"Error during local search: {str(e)}", exc_info=True) + return [] + + @classmethod + async def search_local_autocomplete(cls, query: str = None, max_results: int = 10): + """Wrapper for autocomplete searches that returns more results and allows empty query.""" + # For autocomplete, we want to show more options and allow empty queries + if not query: + return [] + + return await cls.search_local(query, max_results=max_results, autocomplete=True) -- cgit v1.2.3