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/play.py | 219 ----------------------------------------------------- 1 file changed, 219 deletions(-) delete mode 100644 cogs/music/play.py (limited to 'cogs/music/play.py') 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 -- cgit v1.2.3