From 87a55bf43e61af57c6c88f1be57e2e0f2036d869 Mon Sep 17 00:00:00 2001 From: salmonik Date: Fri, 26 Sep 2025 09:39:11 +0200 Subject: add local music playing --- cogs/music/play_utils.py | 141 +++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 131 insertions(+), 10 deletions(-) (limited to 'cogs/music/play_utils.py') diff --git a/cogs/music/play_utils.py b/cogs/music/play_utils.py index b837ba0..bc20d7b 100644 --- a/cogs/music/play_utils.py +++ b/cogs/music/play_utils.py @@ -1,4 +1,7 @@ +import json import logging + +import mutagen from youtubesearchpython import VideosSearch import yt_dlp import discord @@ -7,11 +10,29 @@ 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 - async def download(self, ctx, url): + @staticmethod + async def download(ctx, url): # Create temp directory if it doesn't exist temp_dir = os.path.join('cogs', 'temp', 'music') @@ -42,7 +63,8 @@ class PlayUtils: logger.error(f"Error downloading video: {e}") await ctx.respond("eror pobirajync wideło", ephemeral=True) - async def get_url(self, ctx, query): + @staticmethod + async def get_url(ctx, query): search = VideosSearch(query, limit=1) if search.result()["result"]: title = search.result()["result"][0]["title"] @@ -54,7 +76,8 @@ class PlayUtils: await ctx.respond("nic ni ma", ephemeral=True) return None - async def compress(self, ctx, filename): + @staticmethod + async def compress(ctx, filename): if not filename: await ctx.respond("dzie wideło", ephemeral=True) return @@ -76,12 +99,29 @@ class PlayUtils: logger.error(f"Error compressing file: {e}") await ctx.respond("eror kompresujync pilk", ephemeral=True) - async def play(self, ctx, filename, title): + @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 @@ -89,29 +129,110 @@ class PlayUtils: 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}") - if not filename or not os.path.exists(filename): - await ctx.respond("Nie znaleziono pliku do odtworzenia", ephemeral=True) - logger.warning(f"No file to play, filename: {filename}") - return - # Stop any currently playing audio if vc.is_playing() or vc.is_paused(): vc.stop() # Play the audio - source = discord.FFmpegPCMAudio(source=filename) + 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 [] + + + -- cgit v1.2.3