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
|
# 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 os
import logging
from pathlib import Path
import discord
import mutagen
logger = logging.getLogger(__name__)
class AudioPlaybackManager:
"""Handles audio playback, voice connections, and source preparation."""
@staticmethod
async def cleanup_temp_file(source):
"""Clean up temporary file if it exists."""
if source 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)}")
@staticmethod
async def validate_voice_channel(ctx):
"""Validate user is in a voice channel. Returns channel or None."""
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 None
return ctx.author.voice.channel
@staticmethod
async def prepare_file_source(source):
"""Prepare and validate local file source. Returns (display_name, absolute_path) or (None, None)."""
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):
logger.error(f"File not found: {source}")
return None, None
filename = os.path.basename(source)
display_name = os.path.splitext(filename)[0]
return display_name, source
@staticmethod
async def connect_to_voice_channel(ctx, vc, target_channel):
"""Connect or move to voice channel. Returns voice client."""
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}")
return vc
except Exception as e:
logger.error(f"Error connecting to voice channel: {str(e)}")
raise
@staticmethod
async def prepare_audio_source(source, is_url, display_name):
"""Prepare audio source for playback. Returns (source, final_display_name)."""
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'
}
audio_source = discord.FFmpegPCMAudio(source, **ffmpeg_options)
return audio_source, display_name
else:
# For local files, use default options
audio_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)}")
return audio_source, display_name
@staticmethod
async def cleanup_if_needed(source, delete_after_play, is_url):
"""Clean up temporary file if needed."""
if delete_after_play and not is_url:
await AudioPlaybackManager.cleanup_temp_file(source)
@staticmethod
async def prepare_source(ctx, source):
"""Prepare source and get display name. Returns (source, display_name, is_url) or (None, None, None)."""
is_url = source.startswith(('http://', 'https://'))
if is_url:
logger.info(f"Attempting to play stream from URL: {source}")
return source, source, is_url
display_name, absolute_source = await AudioPlaybackManager.prepare_file_source(source)
if not display_name:
return None, None, None
return absolute_source, display_name, is_url
@staticmethod
async def execute_playback(ctx, source, display_name, is_url):
"""Execute the actual audio playback."""
vc = ctx.voice_client
target_channel = await AudioPlaybackManager.validate_voice_channel(ctx)
if not target_channel:
return False
try:
vc = await AudioPlaybackManager.connect_to_voice_channel(ctx, vc, target_channel)
# Stop any currently playing audio
if vc.is_playing() or vc.is_paused():
vc.stop()
# Prepare audio source
audio_source, final_display_name = await AudioPlaybackManager.prepare_audio_source(source, is_url, display_name)
# Play the audio
vc.play(audio_source)
logger.info(f"Playing: {final_display_name}")
await ctx.respond(f"grom: {final_display_name}")
return True
except Exception as e:
logger.error(f"Error in play: {str(e)}", exc_info=True)
await ctx.respond(f"eror: {str(e)}", ephemeral=True)
return False
@staticmethod
async def validate_and_prepare_source(ctx, source):
"""Validate and prepare source for playback. Returns (source, display_name, is_url) or (None, None, None)."""
prepared_source, display_name, is_url = await AudioPlaybackManager.prepare_source(ctx, source)
if not prepared_source:
return None, None, None
return prepared_source, display_name, is_url
@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)
"""
# Validate and prepare source
prepared_source, display_name, is_url = await AudioPlaybackManager.validate_and_prepare_source(ctx, source)
if not prepared_source:
await AudioPlaybackManager.cleanup_if_needed(source, delete_after_play, False)
return
# Execute playback
success = await AudioPlaybackManager.execute_playback(ctx, prepared_source, display_name, is_url)
# Cleanup if needed
if not success:
await AudioPlaybackManager.cleanup_if_needed(prepared_source, delete_after_play, is_url)
|
