# 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