From 456842d6f1c36d5a04cb4f8821e641be80e077f9 Mon Sep 17 00:00:00 2001 From: salmonik Date: Mon, 23 Mar 2026 17:43:36 +0100 Subject: play music using the server --- cogs/music/play_commands.py | 166 ++++++++++---------- cogs/utils/file_search.py | 370 +++----------------------------------------- 2 files changed, 104 insertions(+), 432 deletions(-) diff --git a/cogs/music/play_commands.py b/cogs/music/play_commands.py index e3ca4c1..23808d3 100644 --- a/cogs/music/play_commands.py +++ b/cogs/music/play_commands.py @@ -4,8 +4,8 @@ # https://codeberg.org/salmonik/senator-sperma import os +import urllib.parse import discord -import mutagen from typing import List, Tuple, Callable, Awaitable from os import PathLike @@ -37,20 +37,8 @@ class PlayCommands(commands.Cog): 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 + 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 @@ -86,69 +74,46 @@ class PlayCommands(commands.Cog): 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.UniversalSearcher.search_local_autocomplete(query, max_results=25, filetype="Audio") + results = await search_utils.UniversalSearcher.search_copyparty_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) + + 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() - - # 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 + + self.path_mapping[display] = url + + if len(display) > 90: 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: + async 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) + await 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) + await ctx.respond("Możesz podać tylko plik LUB stream, nie oba naraz.", ephemeral=True) return False return True @@ -157,38 +122,75 @@ class PlayCommands(commands.Cog): """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: + async def _handle_compressed_file_play(self, ctx: discord.ApplicationContext, url: 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): + is_url = url.startswith(('http://', 'https://')) + + if is_url: + import aiohttp + local_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] + local_path = os.path.join(self.temp_dir, filename) + os.makedirs(self.temp_dir, exist_ok=True) + + with open(local_path, 'wb') as f: + async for chunk in response.content.iter_chunked(8192): + f.write(chunk) + + compressed_path = await media_processing.AudioCompressor.compress(ctx, local_path) + if not compressed_path: + await ctx.respond("Nie udało się skompresować pliku.", ephemeral=True) + return + 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 + await audio_playback.AudioPlaybackManager.play(ctx, compressed_path, delete_after_play=True) + except Exception as e: + if os.path.exists(compressed_path): + try: + os.remove(compressed_path) + except Exception: + pass + raise e + finally: + if os.path.exists(local_path): + try: + os.remove(local_path) + except Exception: + pass + except Exception as e: + logger.error(f"Error downloading file from URL: {str(e)}", exc_info=True) + await ctx.respond(f"Błąd pobierania pliku: {str(e)}", ephemeral=True) + else: + compressed_path = await media_processing.AudioCompressor.compress(ctx, url) + 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: + if os.path.exists(compressed_path): + try: + os.remove(compressed_path) + except Exception: + pass + 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 + """Handle file playback from copyparty URL.""" + url = self.path_mapping.get(file, file) if compressed: - await self._handle_compressed_file_play(ctx, full_path) + await self._handle_compressed_file_play(ctx, url) else: - await audio_playback.AudioPlaybackManager.play(ctx, full_path) + await audio_playback.AudioPlaybackManager.play(ctx, url) @discord.slash_command(name="grajcuj", description="grajcujmy wszyscy!") async def play( diff --git a/cogs/utils/file_search.py b/cogs/utils/file_search.py index 2685e25..8d3e5d6 100644 --- a/cogs/utils/file_search.py +++ b/cogs/utils/file_search.py @@ -5,368 +5,41 @@ import json import logging -import re -from pathlib import Path -import aiofiles -from rapidfuzz import fuzz import aiohttp import os from dotenv import load_dotenv +from cogs.utils.settings import get_baseurl load_dotenv() 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 UniversalSearcher: - """Handles searching for files in configured paths with support for multiple file types.""" - - FILE_TYPE_EXTENSIONS = { - 'Audio': {'.mp4', '.mp3', '.wav', '.opus', '.m4a', '.flac', '.aac', '.ogg'}, - 'Video': {'.mp4', '.avi', '.mkv', '.mov', '.wmv', '.flv', '.webm', '.m4v', '.3gp', '.gif'}, - 'Image': {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp', '.svg', '.ico'} - } - - @staticmethod - def _is_valid_file(file_path: Path, filetype: str = None) -> bool: - """Check if file is valid for the specified filetype.""" - if not file_path.is_file(): - return False - - if filetype is None: - # Search all file types - all_extensions = set() - for extensions in UniversalSearcher.FILE_TYPE_EXTENSIONS.values(): - all_extensions.update(extensions) - return file_path.suffix.lower() in all_extensions - - filetype = filetype.capitalize() - if filetype not in UniversalSearcher.FILE_TYPE_EXTENSIONS: - return False - - valid_extensions = UniversalSearcher.FILE_TYPE_EXTENSIONS[filetype] - return 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, filetype: str = None) -> list[tuple[int, str]]: - """Search for files in a single path.""" - matches = [] - - logger.debug(f"Searching in directory: {path} (filetype: {filetype or 'All'})") - - for file_path in path.rglob('*'): - if not UniversalSearcher._is_valid_file(file_path, filetype): - continue - - file_name = file_path.name.lower() - score, _ = UniversalSearcher._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 - - @staticmethod - def _get_search_paths(data: dict, filetype: str = None) -> list: - """Get the search paths from settings.""" - return data.get('file_paths', []) - - @classmethod - async def _search_paths(cls, search_paths: list, query: str, max_results: int, autocomplete: bool, filetype: str = None) -> tuple[list, int]: - """Search through the provided paths and return matches and file count.""" - 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, filetype) - matches.extend(path_matches) - file_count += len(path_matches) - - if len(matches) >= max_results * 5: - break - - return matches, file_count - - @classmethod - async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False, filetype: str = None): - """Search for 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 - filetype: Type of files to search for ('Audio', 'Video', 'Image', or None for all) - - 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}' (filetype: {filetype or 'All'})") - - try: - async with aiofiles.open('settings.json') as f: - content = await f.read() - data = json.loads(content) - - search_paths = cls._get_search_paths(data, filetype) - logger.info(f"Searching in paths: {search_paths}") - - matches, file_count = await cls._search_paths(search_paths, query, max_results, autocomplete, filetype) - - 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, filetype: str = None): - """Wrapper for autocomplete searches that returns more results and allows empty query. - - Args: - query: Search query string (can be None for autocomplete) - max_results: Maximum number of results to return - filetype: Type of files to search for ('Audio', 'Video', 'Image', or None for all) - """ - # 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, filetype=filetype) - - @staticmethod - async def _load_baseurl(): - """Load baseurl from settings.json""" - try: - async with aiofiles.open('settings.json') as f: - content = await f.read() - data = json.loads(content) - return data.get('baseurl', 'http://localhost:8080').rstrip('/') - except Exception as e: - logger.error(f"Error loading baseurl: {e}") - return 'http://localhost:8080' - + """Handles searching for files via copyparty API.""" + @staticmethod async def search_copyparty(query: str, max_results: int = 25): """Search for files using copyparty API (jPOST method). - + Args: query: Search query string max_results: Maximum number of results to return - + Returns: List of matching file URLs """ - - if not query: return [] - + query = query.strip() logger.info(f"Starting copyparty search for query: '{query}'") query = f'name like "*{query}*"' - + try: - baseurl = await UniversalSearcher._load_baseurl() + baseurl = get_baseurl() password = os.getenv("COPYPARTY_PASSWORD") - + async with aiohttp.ClientSession() as session: headers = {"Content-Type": "application/json"} if password: @@ -377,20 +50,20 @@ class UniversalSearcher: logger.info(f"Copyparty response: {data}") files = UniversalSearcher._extract_files_from_response(data) results = UniversalSearcher._convert_files_to_urls(files, baseurl, max_results) - + logger.info(f"Copyparty search complete. Found {len(results)} matches.") return results else: logger.error(f"Copyparty API returned status {response.status}") return [] - + except aiohttp.ClientError as e: logger.error(f"HTTP error during copyparty search: {e}") return [] except Exception as e: logger.error(f"Error during copyparty search: {str(e)}", exc_info=True) return [] - + @staticmethod def _extract_files_from_response(data): """Extract files list from copyparty API response.""" @@ -407,26 +80,25 @@ class UniversalSearcher: return data else: return [] - + @staticmethod def _convert_files_to_urls(files, baseurl, max_results): """Convert file info objects to URLs.""" results = [] - + for file_info in files[:max_results]: if isinstance(file_info, dict): file_url = UniversalSearcher._extract_url_from_dict(file_info, baseurl) if file_url: results.append(file_url) elif isinstance(file_info, str): - # If it's just a string path/URL if file_info.startswith('http'): results.append(file_info) else: results.append(f"{baseurl}/{file_info.lstrip('/')}") - + return results - + @staticmethod def _extract_url_from_dict(file_info, baseurl): """Extract URL from file info dictionary.""" @@ -440,21 +112,19 @@ class UniversalSearcher: return f"{baseurl}/{file_info['name'].lstrip('/')}" else: return None - + @staticmethod async def search_copyparty_autocomplete(query: str, max_results: int = 10): """Wrapper for autocomplete searches using copyparty. - + Args: query: Search query string max_results: Maximum number of results to return - + Returns: List of matching file URLs """ if not query: return [] - - return await UniversalSearcher.search_copyparty(query, max_results) - + return await UniversalSearcher.search_copyparty(query, max_results) -- cgit v1.2.3