1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
|
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 []
|
