# 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 logging import os import urllib.parse import asyncio from os import PathLike from typing import Awaitable, Callable, List, Tuple import discord from discord.commands import OptionChoice from discord.ext import commands from cogs.music import audio_playback, media_processing from cogs.utils import file_search as search_utils logger = logging.getLogger(__name__) class PlayCommands(commands.Cog): """Discord slash commands for music playback.""" def __init__(self, bot): from cogs.utils.temp import get_temp_dir self.bot = bot self.temp_dir = get_temp_dir() 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 = candidate.split("/")[-1].split("?")[0] display_name = os.path.splitext(filename)[0] 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 search_utils.UniversalSearcher.search_copyparty_autocomplete_audio( query, max_results=25 ) choices = [] self.path_mapping = {} # Reset the mapping for each new query for url in results: filename = url.split("/")[-1] filename = filename.split("?")[0] # Remove query params if any filename = urllib.parse.unquote(filename) # Decode URL encoding display = os.path.splitext(filename)[0] # Remove extension display = display.strip() if len(display) > 100: display = display[:99] + "…" self.path_mapping[display] = url choices.append(OptionChoice(display, display)) if len(choices) >= 25: break return choices except Exception as e: logger.error(f"Error in autocomplete: {str(e)}", exc_info=True) return [] async def _handle_compressed_file_play( self, ctx: discord.ApplicationContext, url: str ) -> None: """Download from copyparty, compress, then play locally.""" import aiohttp local_path = None compressed_path = None try: async with aiohttp.ClientSession() as session: async with session.get(url) as response: if response.status != 200: await ctx.respond("Nie udało się pobrać pliku.", ephemeral=True) return filename = url.split("/")[-1].split("?")[0] filename = urllib.parse.unquote(filename) local_path = os.path.join(self.temp_dir, filename) os.makedirs(self.temp_dir, exist_ok=True) loop = asyncio.get_event_loop() with open(local_path, "wb") as f: async for chunk in response.content.iter_chunked(8192): await loop.run_in_executor(None, f.write, chunk) compressed_path = await media_processing.AudioCompressor.compress( ctx, local_path ) if not compressed_path: return await audio_playback.AudioPlaybackManager.play( ctx, compressed_path, delete_after_play=True ) except Exception as e: logger.error(f"Error in compressed playback: {str(e)}", exc_info=True) await ctx.respond(f"Błąd: {str(e)}", ephemeral=True) finally: if local_path and os.path.exists(local_path): try: os.remove(local_path) except Exception: pass async def _handle_local_file_play( self, ctx: discord.ApplicationContext, file: str, compressed: bool ) -> None: """Handle file playback from copyparty URL.""" url = self.path_mapping.get(file, file) # Look up URL from display name if compressed: await self._handle_compressed_file_play(ctx, url) else: await audio_playback.AudioPlaybackManager.play(ctx, url) @discord.slash_command(name="grajcuj", description="grajcujmy wszyscy!") async def play( self, ctx: discord.ApplicationContext, melodia: str = discord.Option( str, "odtwórz plik z serwera bądź strumienia", autocomplete=_search_autocomplete, required=False, ), file: discord.Attachment = discord.Option( discord.Attachment, "plik do zagrania", required=False ), compressed: bool = discord.Option( bool, "Czy skompresować dźwięk?", default=False ), ): """Odtwarza wybrany plik dźwiękowy lub stream internetowy.""" await ctx.defer() if melodia and file: await ctx.respond( "Możesz podać plik z serwera LUB upload, nie oba naraz.", ephemeral=True ) return if not melodia and not file: await ctx.respond("Musisz podać plik do odtworzenia.", ephemeral=True) return if file: file_path: PathLike[str] | str = os.path.join(self.temp_dir, file.filename) await file.save(file_path) await audio_playback.AudioPlaybackManager.play(ctx, file_path) else: try: await self._handle_local_file_play(ctx, melodia, compressed) except Exception as e: logger.error(f"Error playing: {str(e)}", exc_info=True) await ctx.respond( f"Wystąpił błąd podczas odtworzenia: {str(e)}", ephemeral=True ) def setup(bot): bot.add_cog(PlayCommands(bot))