refactor music

1 files changed, 0 insertions, 426 deletions

diff --git a/cogs/music/play_utils.py b/cogs/music/play_utils.py
deleted file mode 100644
index ed98395..0000000
--- a/cogs/music/play_utils.py
+++ /dev/null
@@ -1,426 +0,0 @@
-# This file is part of Senator Sperma
-# Licensed under the GPL-3.0 License
-# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
-# https://codeberg.org/salmonik/senator-sperma
-
-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)
-
-
-