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): """Search for audio files that contain the query in their name (case-insensitive). Args: query: Search query string max_results: Maximum number of results to return Returns: List of matching file paths """ 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 = [] q = query.lower() 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 # Check if query is in filename (case-insensitive) if q in file_path.name.lower(): matches.append(str(file_path.absolute())) if len(matches) >= max_results: logger.info(f"Found {len(matches)} matches, stopping search") return matches logger.info(f"Search complete. Found {len(matches)} matches.") return matches except Exception as e: logger.error(f"Error during local search: {str(e)}", exc_info=True) return []