import json import logging import math import os import asyncio 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.""" # Use several scorers and combine with weights r_ratio = fuzz.ratio(query, candidate) # overall edit similarity r_partial = fuzz.partial_ratio(query, candidate) # substring allowance r_token_set = fuzz.token_set_ratio(query, candidate) # word overlap ignoring extras # Calculate prefix score using partial_ratio of the start of the strings min_len = min(len(query), len(candidate)) if min_len > 0: r_prefix = fuzz.ratio(query, candidate[:min_len]) # prefix matching else: r_prefix = 0 # Weighted sum (tune weights as needed) score = (0.35 * r_token_set + # Increased weight for token set 0.30 * r_partial + # Slightly increased weight for partial matches 0.25 * r_ratio + # Standard ratio 0.10 * r_prefix) # Reduced weight for prefix matches score *= Utils.length_penalty(query, candidate) # apply length penalty return min(100, score) # Ensure score doesn't exceed 100 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 ffmpeg_command = f"ffmpeg -i {filename} -ac 1 -b:a 500 -acodec libopus {filename}" try: process = await asyncio.create_subprocess_shell( ffmpeg_command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode != 0: logger.error(f"FFmpeg error: {stderr.decode()}") await ctx.respond("eror kompresujync pilk", ephemeral=True) else: logger.info(f"Compressed file: {filename}") except Exception as e: logger.error(f"Error compressing file: {e}") await ctx.respond("eror kompresujync pilk", ephemeral=True) @staticmethod async def play(ctx, filepath): # 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) return logger.info(f"Attempting to play: {filepath}") # Get absolute path and verify file exists if not os.path.isabs(filepath): filepath = os.path.abspath(filepath) logger.debug(f"Converted to absolute path: {filepath}") if not os.path.exists(filepath): error_msg = f"File not found: {filepath}" logger.error(error_msg) await ctx.respond("Nie znaleziono pliku do odtworzenia", ephemeral=True) return filename = os.path.basename(filepath) logger.info(f"Playing file: {filename} (from: {filepath})") # 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 source = discord.FFmpegPCMAudio(source=filepath) vc.play(source) try: metadata = mutagen.File(filepath) title = metadata.get('title', None) logger.debug(f"Read metadata - title: {title}") except Exception as e: logger.warning(f"Could not read metadata from {filepath}: {str(e)}") title = None if title is None: title = os.path.splitext(os.path.basename(filepath))[0] logger.debug(f"Using filename as title: {title}") logger.info(f"Playing file: {title}") await ctx.respond(f"grom: {title}") 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 -- @staticmethod async def search_local(query, max_results=5, min_score=30): """Search for audio files using fuzzy matching on both filename and path. Args: query: Search query string max_results: Maximum number of results to return min_score: Minimum score threshold for matches (0-100) Returns: List of matching file paths, sorted by relevance """ 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}") candidates = [] files_checked = 0 q = query.lower().replace('_', ' ').replace('-', ' ') 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 files_checked += 1 # Check if file has a valid extension if file_path.suffix.lower() not in {'.mp4', '.mp3', '.wav', '.opus', '.m4a'}: continue # Get relative path from search root rel_path = file_path.relative_to(path).parent # Create searchable strings filename = Utils.normalize_name(file_path, keep_ext=False) full_path_str = str(rel_path / file_path.stem).lower() # Calculate scores score_filename = Utils.combined_score(q, filename) score_path = Utils.combined_score(q, full_path_str) # Use the better score score = max(score_filename, score_path) # Apply bonuses depth_bonus = max(0, 1.0 - (len(rel_path.parts) / 10.0)) # Prefer shallower paths ext_bonus = 1.1 if file_path.suffix.lower() in q.split('.')[-1].lower() else 1.0 final_score = score * (0.9 + depth_bonus * 0.1) * ext_bonus if final_score >= min_score: candidates.append((final_score, str(file_path.absolute()))) logger.debug(f"Match: {file_path} (score: {final_score:.1f})") logger.info(f"Search complete. Checked {files_checked} files, found {len(candidates)} matches.") if not candidates: logger.warning("No matching audio files found.") return [] # Sort by score (descending) and get top results candidates.sort(key=lambda x: x[0], reverse=True) top_matches = candidates[:max_results] logger.info(f"Top {len(top_matches)} matches:") for i, (score, path) in enumerate(top_matches, 1): logger.info(f" {i}. {path} (score: {score:.1f})") return [c[1] for c in top_matches] except Exception as e: logger.error(f"Error during local search: {str(e)}", exc_info=True) return []