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