aboutsummaryrefslogtreecommitdiffstats
path: root/cogs/voice.py
blob: a224336ad975a6871e061671a61bb6843c9886a9 (plain) (blame)
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
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
# 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 discord
import logging
import asyncio
import os
import urllib.parse
from dataclasses import dataclass, field
from typing import Any

import mutagen

logger = logging.getLogger(__name__)


_voice_managers: dict[int, "VoiceManager"] = {}


def get_voice_manager(bot) -> "VoiceManager":
    """Get or create VoiceManager for a specific bot instance."""
    bot_id = bot.user.id if bot.user else id(bot)
    if bot_id not in _voice_managers:
        _voice_managers[bot_id] = VoiceManager(bot)
    return _voice_managers[bot_id]


def is_voice_client_alive(voice_client, guild):
    """Check if a voice client connection is healthy."""
    if voice_client is None:
        return False
    if not voice_client.is_connected():
        return False
    if voice_client.channel is None:
        return False
    if guild.get_channel(voice_client.channel.id) is None:
        return False

    try:
        ws = voice_client._connection
        s = ws.socket
        if s is None or s.fileno() == -1:
            return False
    except Exception:
        return False

    return True


@dataclass
class VoiceManager:
    bot: Any
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
    _connecting: set[int] = field(default_factory=set, init=False)

    # --- Connection -----------------------------------------------------------

    async def _ensure_voice_client(self, channel):
        """Unified connect/move/reconnect logic.

        Returns a healthy VoiceClient connected to *channel*, or None on failure.
        """
        guild = channel.guild
        guild_id = guild.id

        async with self._lock:
            if guild_id in self._connecting:
                logger.info(f"Connection already in progress for guild {guild_id}, skipping")
                return None

            vc = discord.utils.get(self.bot.voice_clients, guild=guild)

            # --- resolve existing client --------------------------------------
            if vc is not None:
                if vc.is_connected():
                    if vc.channel == channel:
                        if is_voice_client_alive(vc, guild):
                            logger.info(f"Already connected to {channel}")
                            return vc
                        logger.warning(f"Stale connection to {channel}, reconnecting")
                        # Disconnect resets the state machine to 'disconnected'
                        # but keeps the VoiceClient alive.  Reusing it avoids a
                        # race where Discord sends VOICE_SERVER_UPDATE between
                        # the destroy and the fresh-connect below, causing the
                        # new client to timeout waiting for the event (since the
                        # old one already consumed it).
                        try:
                            await vc.disconnect()
                        except Exception:
                            pass
                        try:
                            await vc.connect(
                                reconnect=False, timeout=10.0, self_deaf=False, self_mute=False
                            )
                            logger.info(f"Reconnected to voice channel: {channel}")
                            return vc
                        except Exception as e:
                            logger.error(f"Failed to reconnect stale voice client: {e}")
                            try:
                                channel._state._remove_voice_client(guild.id)
                            except Exception:
                                pass
                            vc = None
                    else:
                        try:
                            await vc.move_to(channel)
                            logger.info(f"Moved to voice channel: {channel}")
                            return vc
                        except Exception as e:
                            logger.error(f"Failed to move voice client: {e}")
                            try:
                                await vc.disconnect()
                            except Exception:
                                pass
                            try:
                                await vc.connect(
                                    reconnect=False, timeout=10.0, self_deaf=False, self_mute=False
                                )
                                logger.info(f"Reconnected after failed move: {channel}")
                                return vc
                            except Exception:
                                pass
                            try:
                                channel._state._remove_voice_client(guild.id)
                            except Exception:
                                pass
                            vc = None
                else:
                    logger.info("Voice client exists but disconnected, reconnecting")
                    try:
                        await vc.disconnect()
                    except Exception:
                        pass
                    try:
                        await vc.connect(
                            reconnect=False, timeout=10.0, self_deaf=False, self_mute=False
                        )
                        logger.info(f"Reconnected disconnected client: {channel}")
                        return vc
                    except Exception as e:
                        logger.error(f"Failed to reconnect disconnected client: {e}")
                        try:
                            channel._state._remove_voice_client(guild.id)
                        except Exception:
                            pass
                        vc = None

            # --- create fresh connection --------------------------------------
            self._connecting.add(guild_id)
            try:
                await asyncio.sleep(0.3)
                try:
                    vc = await channel.connect(timeout=10.0, reconnect=False)
                    logger.info(f"Connected to voice channel: {channel}")
                    return vc
                except discord.ClientException as e:
                    if "Already connected" in str(e):
                        logger.warning(f"Stale voice state detected, forcing reconnect: {e}")
                        vc = discord.utils.get(self.bot.voice_clients, guild=guild)
                        if vc:
                            try:
                                await vc.disconnect()
                            except Exception:
                                pass
                        else:
                            try:
                                channel._state._remove_voice_client(guild.id)
                            except Exception:
                                pass
                        await asyncio.sleep(0.3)
                        try:
                            vc = await channel.connect(timeout=10.0, reconnect=False)
                            logger.info(f"Reconnected to voice channel: {channel}")
                            return vc
                        except Exception as e2:
                            logger.error(f"Failed to reconnect after stale cleanup: {e2}")
                            return None
                    logger.error(f"Failed to connect to voice channel: {e}")
                    return None
                except Exception as e:
                    logger.error(f"Failed to connect to voice channel: {e}")
                    return None
            finally:
                self._connecting.discard(guild_id)

    async def connect(self, channel):
        """Join a voice channel (public alias)."""
        return await self._ensure_voice_client(channel)

    async def join_channel(self, channel):
        """Backward-compatible alias for connect()."""
        return await self.connect(channel)

    async def join_by_channel_id(self, channel_id):
        """Join a voice channel by channel ID."""
        channel = self.bot.get_channel(channel_id)
        if not channel:
            logger.error(f"Channel {channel_id} not found")
            return None
        if not isinstance(channel, discord.VoiceChannel):
            logger.error(f"Channel {channel_id} is not a voice channel")
            return None
        return await self.connect(channel)

    async def get_client(self, guild):
        """Get existing voice client for a guild."""
        return discord.utils.get(self.bot.voice_clients, guild=guild)

    async def get_voice_client(self, guild):
        """Backward-compatible alias for get_client()."""
        return await self.get_client(guild)

    async def get_status(self, guild):
        """Get voice client status.

        Returns:
            tuple: (voice_client, is_connected, is_playing) or (None, False, False)
        """
        vc = await self.get_client(guild)
        if vc is None:
            return None, False, False
        return vc, vc.is_connected(), vc.is_playing()

    async def get_voice_status(self, guild):
        """Backward-compatible alias for get_status()."""
        return await self.get_status(guild)

    async def disconnect(self, guild):
        """Stop playback and disconnect from voice."""
        vc = await self.get_client(guild)
        if vc:
            vc.stop()
            logger.info("Disconnecting from voice")
            await vc.disconnect()
            logger.info(f"Disconnected from guild {guild}")

    async def stop(self, guild):
        """Backward-compatible alias for disconnect()."""
        await self.disconnect(guild)

    # --- Playback ------------------------------------------------------------

    async def _play_await(self, vc, audio_source):
        """Play an audio source and wait for completion using an asyncio.Future.

        Returns True on success, False on error.  No polling, no magic sleeps.
        """
        loop = asyncio.get_event_loop()
        future = loop.create_future()

        def after(error):
            if not future.done():
                if error:
                    future.set_exception(Exception(str(error)))
                else:
                    future.set_result(True)

        vc.play(audio_source, after=after)

        try:
            await future
            return True
        except Exception as e:
            logger.error(f"Playback error: {e}")
            return False

    async def play_source(self, channel, audio_source, interrupt=True, volume=1.0):
        """Connect to *channel* and play a pre-made audio source.

        Returns True on success, False on failure.
        """
        vc = await self.connect(channel)
        if not vc:
            logger.error("play_source: could not connect to voice channel")
            return False

        if interrupt and (vc.is_playing() or vc.is_paused()):
            vc.stop()

        if abs(volume - 1.0) > 1e-9:
            audio_source = discord.PCMVolumeTransformer(audio_source)
            audio_source.volume = volume

        logger.info("Playing via play_source")
        return await self._play_await(vc, audio_source)

    # --- Source preparation (folded from AudioPlaybackManager) ---------------

    @staticmethod
    async def validate_voice_channel(ctx):
        """Validate user is in a voice channel. Returns the 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):
        """Validate a local file and return (display_name, absolute_path) or (None, None)."""
        logger.info(f"Attempting to play local file: {source}")

        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 prepare_audio_source(source, is_url, display_name):
        """Prepare FFmpegPCMAudio from a URL or local file.

        Returns (audio_source, final_display_name).
        """
        if is_url:
            before_options = "-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5"
            audio_source = discord.FFmpegPCMAudio(source, before_options=before_options)
            filename = os.path.basename(source.split("?")[0])
            filename = urllib.parse.unquote(filename)
            clean_filename = os.path.splitext(filename)[0]
            return audio_source, clean_filename
        else:
            audio_source = discord.FFmpegPCMAudio(source=source)
            try:
                metadata = mutagen.open(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}")
                else:
                    display_name = os.path.splitext(os.path.basename(source))[0]
            except Exception as e:
                logger.warning(f"Could not read metadata from {source}: {str(e)}")
                display_name = os.path.splitext(os.path.basename(source))[0]

            return audio_source, display_name

    @staticmethod
    async def prepare_source(source):
        """Determine whether *source* is a URL or local file.

        Returns (source_path, display_name, is_url) or (None, None, None) on failure.
        """
        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 VoiceManager.prepare_file_source(source)
        if not display_name:
            return None, None, None

        return absolute_source, display_name, is_url

    @staticmethod
    async def cleanup_temp_file(source):
        """Remove a 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 cleanup_if_needed(source, delete_after_play, is_url):
        """Conditionally clean up a file after playback."""
        if delete_after_play and not is_url:
            await VoiceManager.cleanup_temp_file(source)

    # --- High-level play (full flow) -----------------------------------------

    async def play(self, ctx, source, delete_after_play=False, interrupt=True):
        """High-level playback: validate context, prepare source, connect, play, cleanup.

        Sends ``grom: {name}`` immediately on start, and an error follow-up
        if playback fails.

        Args:
            ctx: Discord application context (used for channel resolution & error messages).
            source: Local file path or URL to stream.
            delete_after_play: Remove local file after playback (no-op for URLs).
            interrupt: Stop currently playing audio before starting.

        Returns:
            bool: True if playback completed successfully, False otherwise.
        """
        prepared_source, display_name, is_url = await VoiceManager.prepare_source(source)
        if not prepared_source:
            await VoiceManager.cleanup_if_needed(source, delete_after_play, False)
            return False

        target_channel = await VoiceManager.validate_voice_channel(ctx)
        if not target_channel:
            await VoiceManager.cleanup_if_needed(prepared_source, delete_after_play, is_url)
            return False

        vc = await self.connect(target_channel)
        if not vc:
            await VoiceManager.cleanup_if_needed(prepared_source, delete_after_play, is_url)
            return False

        if interrupt and (vc.is_playing() or vc.is_paused()):
            vc.stop()

        audio_source, final_display_name = await VoiceManager.prepare_audio_source(
            prepared_source, is_url, display_name
        )

        try:
            await ctx.followup.send(f"grom: {final_display_name}")
        except Exception as e:
            logger.error(e)

        success = await self._play_await(vc, audio_source)

        if success:
            logger.info(f"Playback completed: {final_display_name}")
        else:
            logger.error(f"Playback failed: {prepared_source}")
            try:
                await ctx.followup.send(
                    "eror: Timeout - nie udało się połączyć ze streamem", ephemeral=True
                )
            except Exception:
                pass
            await VoiceManager.cleanup_if_needed(prepared_source, delete_after_play, is_url)

        return success