From cd8ed8bc5793985df2c1692072a932ceee517b95 Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 8 Jul 2026 02:13:35 +0200 Subject: OPEN THE RUST GATES Signed-off-by: salmonik --- cogs/music/play_commands.py | 320 -------------------------------------------- 1 file changed, 320 deletions(-) delete mode 100644 cogs/music/play_commands.py (limited to 'cogs/music/play_commands.py') diff --git a/cogs/music/play_commands.py b/cogs/music/play_commands.py deleted file mode 100644 index c6a55b0..0000000 --- a/cogs/music/play_commands.py +++ /dev/null @@ -1,320 +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 logging -import os -import urllib.parse -import asyncio -from os import PathLike -from typing import Awaitable, Callable, List, Tuple -import re -import requests -from yt_dlp import YoutubeDL - -import discord -from discord.commands import OptionChoice -from discord.ext import commands - -from cogs.music import media_processing -from cogs.utils.file_search import UniversalSearcher -from cogs.utils.settings import get_uploadurl -from cogs.utils.url_helper import clean_url_for_ytdlp -from cogs.voice import get_voice_manager - -logger = logging.getLogger(__name__) - -EXTERNAL_URL_PATTERN = re.compile( - r"^https://(?=.*(?:youtube|youtu\.be|soundcloud|bandcamp))" -) - - -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.""" - if ctx.value.lower().startswith(("http://", "https://")): - return [] - - query = ctx.value.lower() - try: - results = await 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 get_voice_manager(self.bot).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_external_url_play( - self, ctx: discord.ApplicationContext, url: str - ) -> None: - """Download from external URL, upload to copyparty, then play.""" - local_path = None - try: - ytdl_opts = { - "outtmpl": os.path.join(self.temp_dir, "%(title)s.%(ext)s"), - "postprocessors": [ - { - "key": "FFmpegExtractAudio", - "preferredcodec": "mp3", - "preferredquality": "192", - } - ], - "force_ipv4": True, - } - - loop = asyncio.get_event_loop() - url = clean_url_for_ytdlp(url) - with YoutubeDL(ytdl_opts) as ydl: - info_dict = await loop.run_in_executor( - None, lambda: ydl.extract_info(url, download=True) - ) - title = info_dict.get("title", "nieznany") - base_filename = os.path.splitext(ydl.prepare_filename(info_dict))[0] - local_path = f"{base_filename}.mp3" - - if not os.path.exists(local_path): - await ctx.respond("Nie udało się pobrać pliku.", ephemeral=True) - return - - upload_task = asyncio.create_task(self._upload_to_copyparty_async(local_path)) - - await get_voice_manager(self.bot).play( - ctx, local_path, delete_after_play=True - ) - - try: - await asyncio.wait_for(upload_task, timeout=30.0) - except asyncio.TimeoutError: - logger.warning("Upload timed out") - except Exception as e: - logger.error(f"Error downloading external URL: {str(e)}", exc_info=True) - if local_path and os.path.exists(local_path): - try: - os.remove(local_path) - except Exception: - pass - await ctx.respond(f"Błąd pobierania: {str(e)}", ephemeral=True) - - async def _upload_to_copyparty_async(self, file_path: str) -> None: - """Upload file to copyparty asynchronously (fire and forget).""" - try: - logger.info(f"Starting async upload: {file_path}") - baseurl = get_uploadurl() - password = os.getenv("COPYPARTY_PASSWORD", "") - - logger.info(f"Upload - baseurl: {baseurl}, password set: {bool(password)}") - - if baseurl == "http://localhost:3939": - logger.warning("COPYPARTY_URL_UPLOAD not set! Using default") - - with open(file_path, "rb") as f: - content = f.read() - - files = {"f": (os.path.basename(file_path), content, "application/octet-stream")} - headers = {"Cookie": f"cppwd={password}"} - params = {"j": "1", "ck": "b2"} - - logger.info(f"Posting to {baseurl}") - response = requests.post(baseurl, files=files, headers=headers, params=params) - text = response.text - logger.info(f"Upload response: {response.status_code} - {text[:200] if text else 'empty'}") - - if response.status_code in [200, 201]: - result = response.json() - logger.info(f"Uploaded to copyparty: {result.get('fileurl')}") - except Exception as e: - logger.error(f"Async upload failed: {e}", exc_info=True) - - 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 get_voice_manager(self.bot).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 get_voice_manager(self.bot).play(ctx, file_path) - else: - try: - if EXTERNAL_URL_PATTERN.match(melodia.lower()): - await self._handle_external_url_play(ctx, melodia) - else: - 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)) -- cgit v1.2.3