import json import logging import mutagen from youtubesearchpython import VideosSearch import yt_dlp import discord import asyncio import os logger = logging.getLogger(__name__) class Utils: @staticmethod def levenshtein_distance(s1, s2): if len(s1) > len(s2): s1, s2 = s2, s1 distances = range(len(s1) + 1) for i2, c2 in enumerate(s2): distances_ = [i2+1] for i1, c1 in enumerate(s1): if c1 == c2: distances_.append(distances[i1]) else: distances_.append( min(distances[i1], distances[i1 + 1], distances_[-1]) + 1) distances = distances_ return distances[-1] 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): logger.info(f"Starting local search for query: '{query}'") try: with open('settings.json') as f: data = json.load(f) paths = data.get('audio_paths', []) logger.info(f"Searching in paths: {paths}") candidates = [] files_checked = 0 for path in paths: if not os.path.exists(path): logger.warning(f"Path does not exist: {path}") continue logger.debug(f"Searching in directory: {path}") # Walk through all subdirectories for root, dirs, files in os.walk(path): logger.debug(f"Scanning directory: {root} (contains {len(files)} files)") for file in files: files_checked += 1 # Check if file has a valid extension if file.lower().endswith(('.mp4', '.mp3', '.wav', '.opus', '.m4a')): # Get relative path from the search path rel_path = os.path.relpath(root, path) # Get filename without extension filename_without_ext = os.path.splitext(file)[0] # Create searchable string with both filename and relative path searchable_name = f"{os.path.join(rel_path, filename_without_ext)}".lower() # Calculate Levenshtein distance for the searchable string distance = Utils.levenshtein_distance(query.lower(), searchable_name) # Also check just the filename for better matching of partial queries filename_distance = Utils.levenshtein_distance(query.lower(), filename_without_ext.lower()) # Use the better (lower) distance score final_distance = min(distance, filename_distance) # Add to candidates with distance and full path file_path = os.path.abspath(os.path.join(root, file)) candidates.append((final_distance, file_path)) logger.debug(f"Found match: {file_path} (distance: {distance})") logger.info(f"Search complete. Checked {files_checked} files, found {len(candidates)} audio files.") if not candidates: logger.warning("No matching audio files found.") return [] # Sort by distance and get top 5 matches sorted_candidates = sorted(candidates, key=lambda x: x[0]) top_matches = sorted_candidates[:5] logger.info(f"Top {len(top_matches)} matches:") for i, (distance, path) in enumerate(top_matches, 1): logger.info(f" {i}. {path} (distance: {distance})") 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 []