aboutsummaryrefslogtreecommitdiffstats
path: root/cogs/music/play_utils.py
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--cogs/music/play_utils.py426
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 @@
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
generated by cgit v1.2.3 (git 2.46.0) at 2026-11-08 17:39:59 +0000