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)