aboutsummaryrefslogtreecommitdiffstats
path: root/cogs/music
diff options
context:
space:
mode:
Diffstat (limited to 'cogs/music')
-rw-r--r--cogs/music/audio_playback.py179
-rw-r--r--cogs/music/media_processing.py124
-rw-r--r--cogs/music/play_commands.py (renamed from cogs/music/play.py)145
-rw-r--r--cogs/music/play_utils.py426
-rw-r--r--cogs/music/search_utils.py286
5 files changed, 670 insertions, 490 deletions
diff --git a/cogs/music/audio_playback.py b/cogs/music/audio_playback.py
new file mode 100644
index 0000000..3d4f453
--- /dev/null
+++ b/cogs/music/audio_playback.py
@@ -0,0 +1,179 @@
1# This file is part of Senator Sperma
2# Licensed under the GPL-3.0 License
3# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
4# https://codeberg.org/salmonik/senator-sperma
5
6import os
7import logging
8from pathlib import Path
9
10import discord
11import mutagen
12
13logger = logging.getLogger(__name__)
14
15class AudioPlaybackManager:
16 """Handles audio playback, voice connections, and source preparation."""
17
18 @staticmethod
19 async def cleanup_temp_file(source):
20 """Clean up temporary file if it exists."""
21 if source and os.path.exists(source):
22 try:
23 os.remove(source)
24 logger.info(f"Cleaned up temporary file: {source}")
25 except Exception as e:
26 logger.error(f"Error cleaning up file {source}: {str(e)}")
27
28 @staticmethod
29 async def validate_voice_channel(ctx):
30 """Validate user is in a voice channel. Returns channel or None."""
31 if not ctx.author.voice or not ctx.author.voice.channel:
32 await ctx.respond("Musisz być na kanale głosowym, żebym mógł dołączyć!", ephemeral=True)
33 return None
34 return ctx.author.voice.channel
35
36 @staticmethod
37 async def prepare_file_source(source):
38 """Prepare and validate local file source. Returns (display_name, absolute_path) or (None, None)."""
39 logger.info(f"Attempting to play local file: {source}")
40
41 # Get absolute path and verify file exists
42 if not os.path.isabs(source):
43 source = os.path.abspath(source)
44 logger.debug(f"Converted to absolute path: {source}")
45
46 if not os.path.exists(source):
47 logger.error(f"File not found: {source}")
48 return None, None
49
50 filename = os.path.basename(source)
51 display_name = os.path.splitext(filename)[0]
52 return display_name, source
53
54 @staticmethod
55 async def connect_to_voice_channel(ctx, vc, target_channel):
56 """Connect or move to voice channel. Returns voice client."""
57 try:
58 # Connect to voice channel if not connected
59 if vc is None:
60 logger.info(f"Connecting to voice channel: {target_channel.name}")
61 vc = await target_channel.connect()
62 logger.info(f"Connected to voice channel: {target_channel.name}")
63 # Move to user's channel if in a different one
64 elif vc.channel != target_channel:
65 logger.info(f"Moving to voice channel: {target_channel.name}")
66 await vc.move_to(target_channel)
67 logger.info(f"Moved to voice channel: {target_channel.name}")
68 return vc
69 except Exception as e:
70 logger.error(f"Error connecting to voice channel: {str(e)}")
71 raise
72
73 @staticmethod
74 async def prepare_audio_source(source, is_url, display_name):
75 """Prepare audio source for playback. Returns (source, final_display_name)."""
76 if is_url:
77 # For URLs, use FFmpeg with the direct URL
78 ffmpeg_options = {
79 'options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5',
80 'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5'
81 }
82 audio_source = discord.FFmpegPCMAudio(source, **ffmpeg_options)
83 return audio_source, display_name
84 else:
85 # For local files, use default options
86 audio_source = discord.FFmpegPCMAudio(source=source)
87
88 # Try to get metadata for local files
89 try:
90 metadata = mutagen.File(source)
91 if metadata:
92 title = metadata.get('title', [None])[0] if hasattr(metadata, 'get') else None
93 if title:
94 display_name = title
95 logger.debug(f"Read metadata - title: {title}")
96 except Exception as e:
97 logger.warning(f"Could not read metadata from {source}: {str(e)}")
98
99 return audio_source, display_name
100
101 @staticmethod
102 async def cleanup_if_needed(source, delete_after_play, is_url):
103 """Clean up temporary file if needed."""
104 if delete_after_play and not is_url:
105 await AudioPlaybackManager.cleanup_temp_file(source)
106
107 @staticmethod
108 async def prepare_source(ctx, source):
109 """Prepare source and get display name. Returns (source, display_name, is_url) or (None, None, None)."""
110 is_url = source.startswith(('http://', 'https://'))
111
112 if is_url:
113 logger.info(f"Attempting to play stream from URL: {source}")
114 return source, source, is_url
115
116 display_name, absolute_source = await AudioPlaybackManager.prepare_file_source(source)
117 if not display_name:
118 return None, None, None
119
120 return absolute_source, display_name, is_url
121
122 @staticmethod
123 async def execute_playback(ctx, source, display_name, is_url):
124 """Execute the actual audio playback."""
125 vc = ctx.voice_client
126 target_channel = await AudioPlaybackManager.validate_voice_channel(ctx)
127 if not target_channel:
128 return False
129
130 try:
131 vc = await AudioPlaybackManager.connect_to_voice_channel(ctx, vc, target_channel)
132
133 # Stop any currently playing audio
134 if vc.is_playing() or vc.is_paused():
135 vc.stop()
136
137 # Prepare audio source
138 audio_source, final_display_name = await AudioPlaybackManager.prepare_audio_source(source, is_url, display_name)
139
140 # Play the audio
141 vc.play(audio_source)
142 logger.info(f"Playing: {final_display_name}")
143 await ctx.respond(f"grom: {final_display_name}")
144 return True
145
146 except Exception as e:
147 logger.error(f"Error in play: {str(e)}", exc_info=True)
148 await ctx.respond(f"eror: {str(e)}", ephemeral=True)
149 return False
150
151 @staticmethod
152 async def validate_and_prepare_source(ctx, source):
153 """Validate and prepare source for playback. Returns (source, display_name, is_url) or (None, None, None)."""
154 prepared_source, display_name, is_url = await AudioPlaybackManager.prepare_source(ctx, source)
155 if not prepared_source:
156 return None, None, None
157 return prepared_source, display_name, is_url
158
159 @staticmethod
160 async def play(ctx, source, delete_after_play=False):
161 """Play audio from a local file or URL.
162
163 Args:
164 ctx: Discord context
165 source: Either a local file path or a URL to stream from
166 delete_after_play: If True, delete the file after playing (only for local files)
167 """
168 # Validate and prepare source
169 prepared_source, display_name, is_url = await AudioPlaybackManager.validate_and_prepare_source(ctx, source)
170 if not prepared_source:
171 await AudioPlaybackManager.cleanup_if_needed(source, delete_after_play, False)
172 return
173
174 # Execute playback
175 success = await AudioPlaybackManager.execute_playback(ctx, prepared_source, display_name, is_url)
176
177 # Cleanup if needed
178 if not success:
179 await AudioPlaybackManager.cleanup_if_needed(prepared_source, delete_after_play, is_url)
diff --git a/cogs/music/media_processing.py b/cogs/music/media_processing.py
new file mode 100644
index 0000000..430b51f
--- /dev/null
+++ b/cogs/music/media_processing.py
@@ -0,0 +1,124 @@
1# This file is part of Senator Sperma
2# Licensed under the GPL-3.0 License
3# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
4# https://codeberg.org/salmonik/senator-sperma
5
6import os
7import asyncio
8import logging
9from pathlib import Path
10
11import yt_dlp
12from youtubesearchpython import VideosSearch
13
14logger = logging.getLogger(__name__)
15
16class MediaDownloader:
17 """Handles downloading media from various sources."""
18
19 @staticmethod
20 async def download(ctx, url):
21 """Download audio from YouTube URL."""
22 # Create temp directory if it doesn't exist
23 temp_dir = os.path.join('cogs', 'temp', 'music')
24 os.makedirs(temp_dir, exist_ok=True)
25
26 ytdl_opts = {
27 "outtmpl": os.path.join(temp_dir, "%(title)s.%(ext)s"),
28 "postprocessors": [
29 {
30 "key": "FFmpegExtractAudio",
31 "preferredcodec": "mp3",
32 "preferredquality": "192",
33 }
34 ],
35 "force_ipv4": True,
36 }
37
38 try:
39 with yt_dlp.YoutubeDL(ytdl_opts) as ydl:
40 info_dict = ydl.extract_info(url, download=True)
41 # Get the base filename without extension
42 base_filename = os.path.splitext(ydl.prepare_filename(info_dict))[0]
43 # Add .mp3 extension since we're extracting audio
44 filename = f"{base_filename}.mp3"
45 logger.info(f"Downloaded file: {filename}")
46 return filename
47 except Exception as e:
48 logger.error(f"Error downloading video: {e}")
49 await ctx.respond("eror pobirajync wideło", ephemeral=True)
50
51 @staticmethod
52 async def get_url(ctx, query):
53 """Search YouTube and get URL for a query."""
54 search = VideosSearch(query, limit=1)
55 if search.result()["result"]:
56 title = search.result()["result"][0]["title"]
57 url = search.result()["result"][0]["link"]
58 logger.info(f"Found video: {title}")
59 return url, title
60 else:
61 logger.warning("No results found for the query.")
62 await ctx.respond("nic ni ma", ephemeral=True)
63 return None
64
65class AudioCompressor:
66 """Handles audio compression using FFmpeg."""
67
68 @staticmethod
69 async def compress(ctx, filename):
70 """Compress audio file to opus format."""
71 if not filename:
72 await ctx.respond("dzie wideło", ephemeral=True)
73 return None
74
75 # Create temp directory if it doesn't exist
76 temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'temp', 'compressed')
77 os.makedirs(temp_dir, exist_ok=True)
78
79 # Create a temporary output filename in the temp directory
80 base_name = os.path.basename(filename)
81 output_filename = os.path.join(temp_dir, f"{os.path.splitext(base_name)[0]}_compressed.opus")
82
83 # Ensure the output directory exists
84 os.makedirs(os.path.dirname(output_filename) or '.', exist_ok=True)
85
86 # Use proper quoting for filenames with spaces
87 ffmpeg_command = [
88 'ffmpeg',
89 '-y', # Overwrite output file if it exists
90 '-i', filename,
91 '-ac', '1', # Mono audio
92 '-b:a', '500', # 500k bitrate
93 '-acodec', 'libopus',
94 output_filename
95 ]
96
97 try:
98 logger.info(f"Running FFmpeg command: {' '.join(ffmpeg_command)}")
99 process = await asyncio.create_subprocess_exec(
100 *ffmpeg_command,
101 stdout=asyncio.subprocess.PIPE,
102 stderr=asyncio.subprocess.PIPE
103 )
104
105 _, stderr = await process.communicate()
106
107 if process.returncode != 0:
108 error_msg = stderr.decode()
109 logger.error(f"FFmpeg error ({process.returncode}): {error_msg}")
110 await ctx.respond("eror kompresujync pilk", ephemeral=True)
111 return None
112
113 if not os.path.exists(output_filename):
114 logger.error(f"Output file was not created: {output_filename}")
115 await ctx.respond("eror kompresujync pilk - output file not created", ephemeral=True)
116 return None
117
118 logger.info(f"Successfully compressed file: {output_filename}")
119 return output_filename
120
121 except Exception as e:
122 logger.error(f"Error compressing file: {str(e)}", exc_info=True)
123 await ctx.respond(f"eror kompresujync pilk: {str(e)}", ephemeral=True)
124 return None
diff --git a/cogs/music/play.py b/cogs/music/play_commands.py
index c7dd9f1..b25ce84 100644
--- a/cogs/music/play.py
+++ b/cogs/music/play_commands.py
@@ -9,14 +9,16 @@ import mutagen
9from typing import List, Tuple, Callable, Awaitable 9from typing import List, Tuple, Callable, Awaitable
10from os import PathLike 10from os import PathLike
11 11
12from cogs.music import play_utils 12from cogs.music import search_utils, audio_playback, media_processing
13from discord.ext import commands 13from discord.ext import commands
14from discord.commands import OptionChoice 14from discord.commands import OptionChoice
15import logging 15import logging
16 16
17logger = logging.getLogger(__name__) 17logger = logging.getLogger(__name__)
18 18
19class Play(commands.Cog): 19class PlayCommands(commands.Cog):
20 """Discord slash commands for music playback."""
21
20 def __init__(self, bot): 22 def __init__(self, bot):
21 self.bot = bot 23 self.bot = bot
22 self.temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "temp") 24 self.temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "temp")
@@ -83,38 +85,38 @@ class Play(commands.Cog):
83 85
84 await ctx.respond(embed=embed, view=view, ephemeral=True) 86 await ctx.respond(embed=embed, view=view, ephemeral=True)
85 87
88 def _get_display_name_from_metadata(self, path: str, filename: str) -> str:
89 """Extract display name from audio file metadata."""
90 try:
91 if os.path.splitext(path)[1].lower() in ['.mp3', '.wav', '.opus', '.m4a']:
92 metadata = mutagen.File(path, easy=True)
93
94 title = metadata.get('title', [filename])[0]
95 artist = metadata.get('artist', [''])[0]
96 album_artist = metadata.get('albumartist', [artist])[0]
97
98 if album_artist and title:
99 return f"{album_artist} - {title}"
100 elif artist and title:
101 return f"{artist} - {title}"
102 elif title:
103 return title
104 except Exception as e:
105 logger.debug(f"Could not read metadata for {filename}: {str(e)}")
106
107 return filename
108
86 async def _search_autocomplete(self, ctx: discord.AutocompleteContext) -> List[OptionChoice]: 109 async def _search_autocomplete(self, ctx: discord.AutocompleteContext) -> List[OptionChoice]:
87 """Autocomplete handler for search queries.""" 110 """Autocomplete handler for search queries."""
88 query = ctx.value.lower() 111 query = ctx.value.lower()
89 try: 112 try:
90 results = await play_utils.PlayUtils.search_local_autocomplete(query, max_results=25) 113 results = await search_utils.AudioSearcher.search_local_autocomplete(query, max_results=25)
91 choices = [] 114 choices = []
92 self.path_mapping = {} # Reset the mapping for each new query 115 self.path_mapping = {} # Reset the mapping for each new query
93 116
94 for path in results: 117 for path in results:
95 filename = os.path.splitext(os.path.basename(path))[0] # Remove extension 118 filename = os.path.splitext(os.path.basename(path))[0] # Remove extension
96 display = filename # Default display name 119 display = self._get_display_name_from_metadata(path, filename)
97
98 # Try to get metadata for display
99 try:
100 if os.path.splitext(path)[1].lower() in ['.mp3', '.wav', '.opus', '.m4a']:
101 metadata = mutagen.File(path, easy=True)
102
103 # Get metadata fields, falling back to filename if not available
104 title = metadata.get('title', [filename])[0]
105 artist = metadata.get('artist', [''])[0]
106 album_artist = metadata.get('albumartist', [artist])[0]
107
108 # Build display name based on available metadata
109 if album_artist and title:
110 display = f"{album_artist} - {title}"
111 elif artist and title:
112 display = f"{artist} - {title}"
113 elif title:
114 display = title
115
116 except Exception as e:
117 logger.debug(f"Could not read metadata for {filename}: {str(e)}")
118 120
119 # Clean up the display name (remove any extra whitespace or brackets) 121 # Clean up the display name (remove any extra whitespace or brackets)
120 display = display.strip() 122 display = display.strip()
@@ -138,6 +140,55 @@ class Play(commands.Cog):
138 logger.error(f"Error in autocomplete: {str(e)}", exc_info=True) 140 logger.error(f"Error in autocomplete: {str(e)}", exc_info=True)
139 return [] 141 return []
140 142
143 def _validate_play_inputs(self, file: str, stream: str, ctx: discord.ApplicationContext) -> bool:
144 """Validate input parameters for play command."""
145 if not file and not stream:
146 ctx.respond("Musisz podać plik do odtworzenia lub adres streamu.", ephemeral=True)
147 return False
148
149 if file and stream:
150 ctx.respond("Możesz podać tylko plik LUB stream, nie oba naraz.", ephemeral=True)
151 return False
152
153 return True
154
155 async def _handle_stream_play(self, ctx: discord.ApplicationContext, stream: str) -> None:
156 """Handle internet stream playback."""
157 await audio_playback.AudioPlaybackManager.play(ctx, stream)
158
159 async def _handle_compressed_file_play(self, ctx: discord.ApplicationContext, full_path: str) -> None:
160 """Handle compressed file playback with cleanup."""
161 compressed_path = await media_processing.AudioCompressor.compress(ctx, full_path)
162 if not compressed_path:
163 await ctx.respond("Nie udało się skompresować pliku.", ephemeral=True)
164 return
165
166 try:
167 await audio_playback.AudioPlaybackManager.play(ctx, compressed_path, delete_after_play=True)
168 except Exception as e:
169 # Clean up the compressed file if play fails
170 if os.path.exists(compressed_path):
171 try:
172 os.remove(compressed_path)
173 logger.info(f"Cleaned up temporary file after play error: {compressed_path}")
174 except Exception as cleanup_error:
175 logger.error(f"Error cleaning up file {compressed_path}: {str(cleanup_error)}")
176 raise e
177
178 async def _handle_local_file_play(self, ctx: discord.ApplicationContext, file: str, compressed: bool) -> None:
179 """Handle local file playback."""
180 full_path = self.path_mapping.get(file, file)
181
182 if not os.path.exists(full_path):
183 logger.warning(f"File not found: {full_path} (display name: {file})")
184 await ctx.respond("Wybrany plik nie istnieje.", ephemeral=True)
185 return
186
187 if compressed:
188 await self._handle_compressed_file_play(ctx, full_path)
189 else:
190 await audio_playback.AudioPlaybackManager.play(ctx, full_path)
191
141 @discord.slash_command(name="grajcuj", description="grajcujmy wszyscy!") 192 @discord.slash_command(name="grajcuj", description="grajcujmy wszyscy!")
142 async def play( 193 async def play(
143 self, 194 self,
@@ -162,48 +213,14 @@ class Play(commands.Cog):
162 """Odtwarza wybrany plik dźwiękowy lub stream internetowy.""" 213 """Odtwarza wybrany plik dźwiękowy lub stream internetowy."""
163 await ctx.defer() 214 await ctx.defer()
164 215
165 # Check if either file or stream is provided 216 if not self._validate_play_inputs(file, stream, ctx):
166 if not file and not stream:
167 await ctx.respond("Musisz podać plik do odtworzenia lub adres streamu.", ephemeral=True)
168 return
169
170 if file and stream:
171 await ctx.respond("Możesz podać tylko plik LUB stream, nie oba naraz.", ephemeral=True)
172 return 217 return
173 218
174 try: 219 try:
175 if stream: 220 if stream:
176 # Handle internet stream 221 await self._handle_stream_play(ctx, stream)
177 await play_utils.PlayUtils(self.bot).play(ctx, stream)
178 else: 222 else:
179 # Handle local file 223 await self._handle_local_file_play(ctx, file, compressed)
180 full_path = self.path_mapping.get(file, file)
181
182 if not os.path.exists(full_path):
183 logger.warning(f"File not found: {full_path} (display name: {file})")
184 await ctx.respond("Wybrany plik nie istnieje.", ephemeral=True)
185 return
186
187 if compressed:
188 # Compress the file before playing
189 compressed_path = await play_utils.PlayUtils(self.bot).compress(ctx, full_path)
190 if not compressed_path:
191 await ctx.respond("Nie udało się skompresować pliku.", ephemeral=True)
192 return
193
194 try:
195 await play_utils.PlayUtils(self.bot).play(ctx, compressed_path, delete_after_play=True)
196 except Exception as e:
197 # Clean up the compressed file if play fails
198 if os.path.exists(compressed_path):
199 try:
200 os.remove(compressed_path)
201 logger.info(f"Cleaned up temporary file after play error: {compressed_path}")
202 except Exception as cleanup_error:
203 logger.error(f"Error cleaning up file {compressed_path}: {str(cleanup_error)}")
204 raise e
205 else:
206 await play_utils.PlayUtils(self.bot).play(ctx, full_path)
207 except Exception as e: 224 except Exception as e:
208 logger.error(f"Error playing: {str(e)}", exc_info=True) 225 logger.error(f"Error playing: {str(e)}", exc_info=True)
209 await ctx.respond(f"Wystąpił błąd podczas odtwarzania: {str(e)}", ephemeral=True) 226 await ctx.respond(f"Wystąpił błąd podczas odtwarzania: {str(e)}", ephemeral=True)
@@ -213,7 +230,7 @@ class Play(commands.Cog):
213 await ctx.defer() 230 await ctx.defer()
214 file_path: PathLike[str] | str = os.path.join(self.temp_dir, attachment.filename) 231 file_path: PathLike[str] | str = os.path.join(self.temp_dir, attachment.filename)
215 await attachment.save(file_path) 232 await attachment.save(file_path)
216 await play_utils.PlayUtils(self.bot).play(ctx, file_path) 233 await audio_playback.AudioPlaybackManager.play(ctx, file_path)
217 234
218def setup(bot): 235def setup(bot):
219 bot.add_cog(Play(bot)) \ No newline at end of file 236 bot.add_cog(PlayCommands(bot))
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 @@
1# This file is part of Senator Sperma
2# Licensed under the GPL-3.0 License
3# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
4# https://codeberg.org/salmonik/senator-sperma
5
6import json
7import logging
8import math
9import os
10import asyncio
11import re
12from pathlib import Path
13
14import discord
15import mutagen
16import yt_dlp
17from rapidfuzz import fuzz
18from youtubesearchpython import VideosSearch
19
20logger = logging.getLogger(__name__)
21
22class Utils:
23 @staticmethod
24 def normalize_name(p: Path, keep_ext=False):
25 """Preprocess filenames: lower, replace separators, strip extensions optionally."""
26 name = str(p.name if keep_ext else p.stem)
27 return name.lower().replace('_', ' ').replace('-', ' ')
28
29 @staticmethod
30 def length_penalty(query, candidate):
31 """Penalize tiny filenames when query is long; returns multiplier [0.0,1.0]."""
32 ql = len(query)
33 cl = len(candidate)
34 if ql == 0:
35 return 1.0
36 ratio = cl / ql
37 # Prefer candidate lengths near query length; gentle falloff otherwise
38 return math.exp(-0.5 * ((math.log(max(ratio, 1e-6)))**2))
39
40 @staticmethod
41 def combined_score(query, candidate):
42 """Calculate a combined fuzzy matching score using multiple strategies.
43
44 Args:
45 query: The search query
46 candidate: The candidate string to score against
47
48 Returns:
49 float: Score between 0-100, where higher means better match
50 """
51 if not query or not candidate:
52 return 0.0
53
54 query = query.lower().strip()
55 candidate = candidate.lower()
56
57 # First check for direct substring match (fast path)
58 if query in candidate:
59 # Base score of 100 for exact match, with bonus for being at word boundary
60 base_score = 100
61 # Check if match is at word boundary
62 if re.search(rf'\b{re.escape(query)}\b', candidate):
63 base_score = 110 # Slight bonus for word boundary match
64 return min(100, base_score)
65
66 # For very short queries (2-3 chars), be more lenient
67 if len(query) <= 3:
68 # Check for partial matches at word boundaries
69 if re.search(rf'\b{re.escape(query)}', candidate):
70 return 90 # High score for short query matching word start
71 if query in candidate:
72 return 80 # Still good for any substring match with short query
73
74 # For longer queries, check word presence
75 query_words = set(word for word in re.split(r'\W+', query) if len(word) > 1) # Reduced min word length to 1
76 if query_words:
77 candidate_words = set(re.split(r'\W+', candidate))
78 # Check if all query words are in candidate (with some leniency)
79 matching_words = sum(1 for word in query_words if word in candidate_words)
80 word_ratio = matching_words / len(query_words)
81
82 # If no words match at all, return 0
83 if matching_words == 0:
84 return 0.0
85
86 # Calculate different similarity metrics
87 r_ratio = fuzz.ratio(query, candidate) # Overall edit similarity
88 r_partial = fuzz.partial_ratio(query, candidate) # Best substring match
89 r_token_set = fuzz.token_set_ratio(query, candidate) # Word overlap
90
91 # Calculate word-based scores
92 query_tokens = set(query.split())
93 candidate_tokens = set(candidate.split())
94 word_intersection = len(query_tokens & candidate_tokens)
95 word_union = len(query_tokens | candidate_tokens)
96
97 # Jaccard similarity for word overlap
98 jaccard_sim = word_intersection / word_union if word_union > 0 else 0
99
100 # Calculate prefix score (bonus for matching start of words)
101 prefix_bonus = 0
102 for q_word in query_words:
103 for c_word in candidate_words:
104 if c_word.startswith(q_word):
105 prefix_bonus += (len(q_word) / max(len(c_word), 1)) * 20 # Increased weight
106
107 # Weighted sum with adjusted weights
108 score = (
109 0.40 * r_token_set + # Increased weight for token set
110 0.25 * r_ratio + # Overall similarity
111 0.15 * r_partial + # Substring matches
112 0.15 * (jaccard_sim * 100) + # Word overlap ratio
113 0.05 * min(prefix_bonus, 100) # Prefix bonus (capped at 100)
114 )
115
116 # Apply length penalty (less severe)
117 score *= min(1.0, 0.9 + (0.1 * Utils.length_penalty(query, candidate)))
118
119 # Boost score based on word ratio
120 score = score * (0.7 + (0.3 * word_ratio))
121
122 score = int(score)
123
124 # Ensure score is within bounds
125 return min(100, max(0, score))
126
127 return 0.0 # No match found
128
129class PlayUtils:
130 def __init__(self, bot):
131 self.bot = bot
132
133 @staticmethod
134 async def download(ctx, url):
135
136 # Create temp directory if it doesn't exist
137 temp_dir = os.path.join('cogs', 'temp', 'music')
138 os.makedirs(temp_dir, exist_ok=True)
139
140 ytdl_opts = {
141 "outtmpl": os.path.join(temp_dir, "%(title)s.%(ext)s"),
142 "postprocessors": [
143 {
144 "key": "FFmpegExtractAudio",
145 "preferredcodec": "mp3",
146 "preferredquality": "192",
147 }
148 ],
149 "force_ipv4": True,
150 }
151
152 try:
153 with yt_dlp.YoutubeDL(ytdl_opts) as ydl:
154 info_dict = ydl.extract_info(url, download=True)
155 # Get the base filename without extension
156 base_filename = os.path.splitext(ydl.prepare_filename(info_dict))[0]
157 # Add .mp3 extension since we're extracting audio
158 filename = f"{base_filename}.mp3"
159 logger.info(f"Downloaded file: {filename}")
160 return filename
161 except Exception as e:
162 logger.error(f"Error downloading video: {e}")
163 await ctx.respond("eror pobirajync wideło", ephemeral=True)
164
165 @staticmethod
166 async def get_url(ctx, query):
167 search = VideosSearch(query, limit=1)
168 if search.result()["result"]:
169 title = search.result()["result"][0]["title"]
170 url = search.result()["result"][0]["link"]
171 logger.info(f"Found video: {title}")
172 return url, title
173 else:
174 logger.warning("No results found for the query.")
175 await ctx.respond("nic ni ma", ephemeral=True)
176 return None
177
178 @staticmethod
179 async def compress(ctx, filename):
180 if not filename:
181 await ctx.respond("dzie wideło", ephemeral=True)
182 return None
183
184 # Create temp directory if it doesn't exist
185 temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'temp', 'compressed')
186 os.makedirs(temp_dir, exist_ok=True)
187
188 # Create a temporary output filename in the temp directory
189 base_name = os.path.basename(filename)
190 output_filename = os.path.join(temp_dir, f"{os.path.splitext(base_name)[0]}_compressed.opus")
191
192 # Ensure the output directory exists
193 os.makedirs(os.path.dirname(output_filename) or '.', exist_ok=True)
194
195 # Use proper quoting for filenames with spaces
196 ffmpeg_command = [
197 'ffmpeg',
198 '-y', # Overwrite output file if it exists
199 '-i', filename,
200 '-ac', '1', # Mono audio
201 '-b:a', '500', # 500k bitrate
202 '-acodec', 'libopus',
203 output_filename
204 ]
205
206 try:
207 logger.info(f"Running FFmpeg command: {' '.join(ffmpeg_command)}")
208 process = await asyncio.create_subprocess_exec(
209 *ffmpeg_command,
210 stdout=asyncio.subprocess.PIPE,
211 stderr=asyncio.subprocess.PIPE
212 )
213
214 stdout, stderr = await process.communicate()
215
216 if process.returncode != 0:
217 error_msg = stderr.decode()
218 logger.error(f"FFmpeg error ({process.returncode}): {error_msg}")
219 await ctx.respond("eror kompresujync pilk", ephemeral=True)
220 return None
221
222 if not os.path.exists(output_filename):
223 logger.error(f"Output file was not created: {output_filename}")
224 await ctx.respond("eror kompresujync pilk - output file not created", ephemeral=True)
225 return None
226
227 logger.info(f"Successfully compressed file: {output_filename}")
228 return output_filename
229
230 except Exception as e:
231 logger.error(f"Error compressing file: {str(e)}", exc_info=True)
232 await ctx.respond(f"eror kompresujync pilk: {str(e)}", ephemeral=True)
233 return None
234
235 @staticmethod
236 async def play(ctx, source, delete_after_play=False):
237 """Play audio from a local file or URL.
238
239 Args:
240 ctx: Discord context
241 source: Either a local file path or a URL to stream from
242 delete_after_play: If True, delete the file after playing (only for local files)
243 """
244 # Check if user is in a voice channel
245 if not ctx.author.voice or not ctx.author.voice.channel:
246 await ctx.respond("Musisz być na kanale głosowym, żebym mógł dołączyć!", ephemeral=True)
247 if delete_after_play and os.path.exists(source):
248 try:
249 os.remove(source)
250 logger.info(f"Cleaned up temporary file: {source}")
251 except Exception as e:
252 logger.error(f"Error cleaning up file {source}: {str(e)}")
253 return
254
255 is_url = source.startswith(('http://', 'https://'))
256
257 if not is_url:
258 # Handle local file
259 logger.info(f"Attempting to play local file: {source}")
260
261 # Get absolute path and verify file exists
262 if not os.path.isabs(source):
263 source = os.path.abspath(source)
264 logger.debug(f"Converted to absolute path: {source}")
265
266 if not os.path.exists(source):
267 error_msg = f"File not found: {source}"
268 logger.error(error_msg)
269 await ctx.respond("Nie znaleziono pliku do odtworzenia", ephemeral=True)
270 return
271
272 filename = os.path.basename(source)
273 display_name = os.path.splitext(filename)[0]
274 else:
275 # Handle URL
276 logger.info(f"Attempting to play stream from URL: {source}")
277 display_name = source # Use URL as display name for streams
278
279 # Get or create voice client
280 vc = ctx.voice_client
281 target_channel = ctx.author.voice.channel
282
283 try:
284 # Connect to voice channel if not connected
285 if vc is None:
286 logger.info(f"Connecting to voice channel: {target_channel.name}")
287 vc = await target_channel.connect()
288 logger.info(f"Connected to voice channel: {target_channel.name}")
289 # Move to user's channel if in a different one
290 elif vc.channel != target_channel:
291 logger.info(f"Moving to voice channel: {target_channel.name}")
292 await vc.move_to(target_channel)
293 logger.info(f"Moved to voice channel: {target_channel.name}")
294
295 # Stop any currently playing audio
296 if vc.is_playing() or vc.is_paused():
297 vc.stop()
298
299 # Play the audio
300 if is_url:
301 # For URLs, use FFmpeg with the direct URL
302 ffmpeg_options = {
303 'options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5',
304 'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5'
305 }
306 source = discord.FFmpegPCMAudio(source, **ffmpeg_options)
307 else:
308 # For local files, use default options
309 source = discord.FFmpegPCMAudio(source=source)
310
311 # Try to get metadata for local files
312 try:
313 metadata = mutagen.File(source)
314 if metadata:
315 title = metadata.get('title', [None])[0] if hasattr(metadata, 'get') else None
316 if title:
317 display_name = title
318 logger.debug(f"Read metadata - title: {title}")
319 except Exception as e:
320 logger.warning(f"Could not read metadata from {source}: {str(e)}")
321
322 vc.play(source)
323 logger.info(f"Playing: {display_name}")
324 await ctx.respond(f"grom: {display_name}")
325
326 except Exception as e:
327 logger.error(f"Error in play: {str(e)}", exc_info=True)
328 await ctx.respond(f"eror: {str(e)}", ephemeral=True)
329
330 # -- LOCAL PLAY --
331
332 @classmethod
333 async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False):
334 """Search for audio files with fuzzy matching and optional autocomplete support.
335
336 Args:
337 query: Search query string (can be None for autocomplete)
338 max_results: Maximum number of results to return
339 autocomplete: If True, returns results even with empty query
340
341 Returns:
342 List of matching file paths with scores
343 """
344 if not query and not autocomplete:
345 return []
346
347 query = (query or '').lower().strip()
348 logger.info(f"Starting local search for query: '{query}'")
349
350 try:
351 with open('settings.json') as f:
352 data = json.load(f)
353 search_paths = data.get('audio_paths', [])
354 logger.info(f"Searching in paths: {search_paths}")
355
356 matches = []
357 file_count = 0
358
359 for path in search_paths:
360 path = Path(path)
361 if not path.exists():
362 logger.warning(f"Path does not exist: {path}")
363 continue
364
365 logger.debug(f"Searching in directory: {path}")
366
367 # Walk through all subdirectories
368 for file_path in path.rglob('*'):
369 if not file_path.is_file():
370 continue
371
372 # Check if file has a valid extension
373 if file_path.suffix.lower() not in {'.mp4', '.mp3', '.wav', '.opus', '.m4a'}:
374 continue
375
376 file_count += 1
377 file_name = file_path.name.lower()
378
379 # For autocomplete, return all files if query is empty
380 if not query and autocomplete:
381 matches.append((0, str(file_path.absolute())))
382 # For regular search, calculate fuzzy match score
383 elif query and len(query) >= 2:
384 # First do a simple substring check for debugging
385 if query in file_name:
386 logger.debug(f"Found direct match: {file_name} (contains '{query}')")
387
388 score = Utils.combined_score(query, file_name)
389 logger.debug(f"File: {file_name} - Score: {score}")
390
391 if score >= 30: # Minimum score threshold
392 matches.append((score, str(file_path.absolute())))
393
394 # Early return if we have enough high-scoring matches
395 if len(matches) >= max_results * 5: # Get more candidates for sorting
396 break
397
398 if len(matches) >= max_results * 5: # Limit memory usage
399 break
400
401 logger.info(f"Scanned {file_count} files total")
402
403 # Sort by score (descending) and take top results
404 matches.sort(reverse=True, key=lambda x: x[0])
405 result = [path for score, path in matches[:max_results]]
406
407 logger.info(f"Search complete. Found {len(result)} matches out of {file_count} files.")
408 if not result and query:
409 logger.warning(f"No matches found for query: '{query}'. Try a different search term.")
410 return result
411
412 except Exception as e:
413 logger.error(f"Error during local search: {str(e)}", exc_info=True)
414 return []
415
416 @classmethod
417 async def search_local_autocomplete(cls, query: str = None, max_results: int = 10):
418 """Wrapper for autocomplete searches that returns more results and allows empty query."""
419 # For autocomplete, we want to show more options and allow empty queries
420 if not query:
421 return []
422
423 return await cls.search_local(query, max_results=max_results, autocomplete=True)
424
425
426
diff --git a/cogs/music/search_utils.py b/cogs/music/search_utils.py
new file mode 100644
index 0000000..50d229c
--- /dev/null
+++ b/cogs/music/search_utils.py
@@ -0,0 +1,286 @@
1# This file is part of Senator Sperma
2# Licensed under the GPL-3.0 License
3# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
4# https://codeberg.org/salmonik/senator-sperma
5
6import json
7import logging
8import re
9from pathlib import Path
10import aiofiles
11from rapidfuzz import fuzz
12
13logger = logging.getLogger(__name__)
14
15class SearchUtils:
16 """Utilities for searching and scoring audio files."""
17
18 @staticmethod
19 def normalize_name(p: Path, keep_ext=False):
20 """Preprocess filenames: lower, replace separators, strip extensions optionally."""
21 name = str(p.name if keep_ext else p.stem)
22 return name.lower().replace('_', ' ').replace('-', ' ')
23
24 @staticmethod
25 def length_penalty(query, candidate):
26 """Penalize tiny filenames when query is long; returns multiplier [0.0,1.0]."""
27 ql = len(query)
28 cl = len(candidate)
29 if ql == 0:
30 return 1.0
31 ratio = cl / ql
32 # Prefer candidate lengths near query length; gentle falloff otherwise
33 return math.exp(-0.5 * ((math.log(max(ratio, 1e-6)))**2))
34
35 @staticmethod
36 def combined_score(query, candidate):
37 """Calculate a combined fuzzy matching score using multiple strategies.
38
39 Args:
40 query: The search query
41 candidate: The candidate string to score against
42
43 Returns:
44 float: Score between 0-100, where higher means better match
45 """
46 if not query or not candidate:
47 return 0.0
48
49 query = query.lower().strip()
50 candidate = candidate.lower()
51
52 # Check for direct matches first
53 direct_match_score = SearchUtils._check_direct_matches(query, candidate)
54 if direct_match_score is not None:
55 return direct_match_score
56
57 # Handle short queries specially
58 short_query_score = SearchUtils._handle_short_queries(query, candidate)
59 if short_query_score is not None:
60 return short_query_score
61
62 # For longer queries, use complex scoring
63 return SearchUtils._calculate_complex_score(query, candidate)
64
65 @staticmethod
66 def _check_direct_matches(query: str, candidate: str) -> float:
67 """Check for direct substring matches."""
68 if query in candidate:
69 base_score = 100
70 if re.search(rf'\b{re.escape(query)}\b', candidate):
71 base_score = 110
72 return min(100, base_score)
73 return 0
74
75 @staticmethod
76 def _handle_short_queries(query: str, candidate: str) -> float:
77 """Handle scoring for short queries (2-3 chars)."""
78 if len(query) <= 3:
79 if re.search(rf'\b{re.escape(query)}', candidate):
80 return 90
81 if query in candidate:
82 return 80
83 return 0
84
85 @staticmethod
86 def _calculate_complex_score(query: str, candidate: str) -> float:
87 """Calculate complex score for longer queries."""
88 query_words = {word for word in re.split(r'\W+', query) if len(word) > 1}
89 if not query_words:
90 return 0.0
91
92 candidate_words = set(re.split(r'\W+', candidate))
93 word_ratio = SearchUtils._calculate_word_ratio(query_words, candidate_words)
94
95 if word_ratio == 0:
96 return 0.0
97
98 # Calculate various similarity metrics
99 similarity_scores = SearchUtils._get_similarity_scores(query, candidate)
100 word_scores = SearchUtils._get_word_scores(query, candidate, query_words, candidate_words)
101
102 # Calculate final weighted score
103 final_score = SearchUtils._calculate_weighted_score(similarity_scores, word_scores, word_ratio)
104
105 # Apply penalties and boosts
106 final_score = SearchUtils._apply_adjustments(final_score, query, candidate, word_ratio)
107
108 return min(100, max(0, int(final_score)))
109
110 @staticmethod
111 def _calculate_word_ratio(query_words: set, candidate_words: set) -> float:
112 """Calculate the ratio of matching words."""
113 matching_words = sum(1 for word in query_words if word in candidate_words)
114 return matching_words / len(query_words)
115
116 @staticmethod
117 def _get_similarity_scores(query: str, candidate: str) -> dict:
118 """Get various fuzzy similarity scores."""
119 return {
120 'ratio': fuzz.ratio(query, candidate),
121 'partial': fuzz.partial_ratio(query, candidate),
122 'token_set': fuzz.token_set_ratio(query, candidate)
123 }
124
125 @staticmethod
126 def _get_word_scores(query: str, candidate: str, query_words: set, candidate_words: set) -> dict:
127 """Calculate word-based scores."""
128 query_tokens = set(query.split())
129 candidate_tokens = set(candidate.split())
130
131 word_intersection = len(query_tokens & candidate_tokens)
132 word_union = len(query_tokens | candidate_tokens)
133 jaccard_sim = word_intersection / word_union if word_union > 0 else 0
134
135 prefix_bonus = SearchUtils._calculate_prefix_bonus(query_words, candidate_words)
136
137 return {
138 'jaccard': jaccard_sim,
139 'prefix': min(prefix_bonus, 100)
140 }
141
142 @staticmethod
143 def _calculate_prefix_bonus(query_words: set, candidate_words: set) -> float:
144 """Calculate prefix matching bonus."""
145 prefix_bonus = 0
146 for q_word in query_words:
147 for c_word in candidate_words:
148 if c_word.startswith(q_word):
149 prefix_bonus += (len(q_word) / max(len(c_word), 1)) * 20
150 return prefix_bonus
151
152 @staticmethod
153 def _calculate_weighted_score(similarity_scores: dict, word_scores: dict, word_ratio: float) -> float:
154 """Calculate the final weighted score."""
155 return (
156 0.40 * similarity_scores['token_set'] +
157 0.25 * similarity_scores['ratio'] +
158 0.15 * similarity_scores['partial'] +
159 0.15 * (word_scores['jaccard'] * 100) +
160 0.05 * word_scores['prefix']
161 )
162
163 @staticmethod
164 def _apply_adjustments(score: float, query: str, candidate: str, word_ratio: float) -> float:
165 """Apply length penalty and word ratio boost."""
166 import math
167
168 # Apply length penalty
169 score *= min(1.0, 0.9 + (0.1 * SearchUtils.length_penalty(query, candidate)))
170
171 # Apply word ratio boost
172 score = score * (0.7 + (0.3 * word_ratio))
173
174 return score
175
176class AudioSearcher:
177 """Handles searching for audio files in configured paths."""
178
179 @staticmethod
180 def _is_valid_audio_file(file_path: Path) -> bool:
181 """Check if file is a valid audio file."""
182 valid_extensions = {'.mp4', '.mp3', '.wav', '.opus', '.m4a'}
183 return file_path.is_file() and file_path.suffix.lower() in valid_extensions
184
185 @staticmethod
186 def _calculate_match_score(query: str, file_name: str, autocomplete: bool) -> tuple[int, str]:
187 """Calculate match score for a file."""
188 if not query and autocomplete:
189 return 0, file_name
190
191 if query and len(query) >= 2:
192 if query in file_name:
193 logger.debug(f"Found direct match: {file_name} (contains '{query}')")
194
195 score = SearchUtils.combined_score(query, file_name)
196 logger.debug(f"File: {file_name} - Score: {score}")
197
198 if score >= 30:
199 return score, file_name
200
201 return -1, file_name
202
203 @staticmethod
204 async def _search_single_path(path: Path, query: str, max_results: int, autocomplete: bool) -> list[tuple[int, str]]:
205 """Search for audio files in a single path."""
206 matches = []
207
208 logger.debug(f"Searching in directory: {path}")
209
210 for file_path in path.rglob('*'):
211 if not AudioSearcher._is_valid_audio_file(file_path):
212 continue
213
214 file_name = file_path.name.lower()
215 score, _ = AudioSearcher._calculate_match_score(query, file_name, autocomplete)
216
217 if score >= 0:
218 matches.append((score, str(file_path.absolute())))
219
220 if len(matches) >= max_results * 5:
221 break
222
223 return matches
224
225 @classmethod
226 async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False):
227 """Search for audio files with fuzzy matching and optional autocomplete support.
228
229 Args:
230 query: Search query string (can be None for autocomplete)
231 max_results: Maximum number of results to return
232 autocomplete: If True, returns results even with empty query
233
234 Returns:
235 List of matching file paths with scores
236 """
237 if not query and not autocomplete:
238 return []
239
240 query = (query or '').lower().strip()
241 logger.info(f"Starting local search for query: '{query}'")
242
243 try:
244 async with aiofiles.open('settings.json') as f:
245 data = json.load(f)
246 search_paths = data.get('audio_paths', [])
247 logger.info(f"Searching in paths: {search_paths}")
248
249 matches = []
250 file_count = 0
251
252 for path_str in search_paths:
253 path = Path(path_str)
254 if not path.exists():
255 logger.warning(f"Path does not exist: {path}")
256 continue
257
258 path_matches = await cls._search_single_path(path, query, max_results, autocomplete)
259 matches.extend(path_matches)
260 file_count += len(path_matches)
261
262 if len(matches) >= max_results * 5:
263 break
264
265 logger.info(f"Scanned {file_count} files total")
266
267 matches.sort(reverse=True, key=lambda x: x[0])
268 result = [path for score, path in matches[:max_results]]
269
270 logger.info(f"Search complete. Found {len(result)} matches out of {file_count} files.")
271 if not result and query:
272 logger.warning(f"No matches found for query: '{query}'. Try a different search term.")
273 return result
274
275 except Exception as e:
276 logger.error(f"Error during local search: {str(e)}", exc_info=True)
277 return []
278
279 @classmethod
280 async def search_local_autocomplete(cls, query: str = None, max_results: int = 10):
281 """Wrapper for autocomplete searches that returns more results and allows empty query."""
282 # For autocomplete, we want to show more options and allow empty queries
283 if not query:
284 return []
285
286 return await cls.search_local(query, max_results=max_results, autocomplete=True)
generated by cgit v1.2.3 (git 2.46.0) at 2026-11-08 09:05:13 +0000