path: root/cogs/voice.py

# 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
from discord.ext import commands
import logging
import asyncio
from dataclasses import dataclass, field
from typing import Any

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.

    Verifies the client is connected, the channel still exists,
    and the underlying voice socket is still open.
    """
    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  # type: ignore[no-any-import]
    _voice_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
    _connecting: set[int] = field(default_factory=set, init=False)

    async def join_channel(self, channel):
        """Join a voice channel by channel object."""
        async with self._voice_lock:
            guild = channel.guild
            guild_id = guild.id

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

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

            if voice_client and voice_client.is_connected():
                if voice_client.channel == channel:
                    if is_voice_client_alive(voice_client, guild):
                        logger.info(f"Already connected to {channel}")
                        return voice_client
                    logger.warning(f"Stale connection to {channel}, reconnecting")
                    try:
                        await voice_client.disconnect()
                    except Exception:
                        pass
                else:
                    try:
                        await voice_client.move_to(channel)
                        logger.info(f"Moved to voice channel: {channel}")
                        return voice_client
                    except Exception as e:
                        logger.error(f"Failed to move voice client: {e}")
                        logger.info("Disconnecting after failed move attempt")
                        try:
                            await voice_client.disconnect()
                        except Exception:
                            pass
            elif voice_client and not voice_client.is_connected():
                logger.info("Voice client exists but disconnected, cleaning up stale entry")
                try:
                    await voice_client.disconnect()
                except Exception:
                    pass
                voice_client = None

            self._connecting.add(guild_id)
            try:
                await asyncio.sleep(0.3)
                try:
                    voice_client = await channel.connect(timeout=30.0, reconnect=False)
                    logger.info(f"Connected to voice channel: {channel}")
                    return voice_client
                except discord.ClientException as e:
                    if "Already connected" in str(e):
                        logger.warning(f"Stale voice state detected, forcing disconnect and reconnect: {e}")
                        voice_client = discord.utils.get(self.bot.voice_clients, guild=guild)
                        if voice_client:
                            try:
                                await voice_client.disconnect()
                            except Exception:
                                pass
                        else:
                            try:
                                channel._state._remove_voice_client(guild.id)
                            except Exception:
                                pass
                        await asyncio.sleep(0.3)
                        try:
                            voice_client = await channel.connect(timeout=30.0, reconnect=False)
                            logger.info(f"Reconnected to voice channel: {channel}")
                            return voice_client
                        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 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.join_channel(channel)

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

    async def get_voice_status(self, guild):
        """Get voice client status including connection and playback state.

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

        is_connected = voice_client.is_connected()
        is_playing = voice_client.is_playing() if is_connected else False

        return voice_client, is_connected, is_playing

    async def play_audio(self, channel, audio_source, volume=1.0):
        """Play audio in a voice channel."""
        voice_client = await self.join_channel(channel)
        if not voice_client:
            return False

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

        try:
            voice_client.play(audio_source)
            while voice_client.is_playing():
                await asyncio.sleep(0.1)
            return True
        except Exception as e:
            logger.error(f"Error playing audio: {e}")
            return False

    async def stop(self, guild):
        """Stop playing and disconnect from voice."""
        voice_client = await self.get_voice_client(guild)
        if voice_client:
            voice_client.stop()
            logger.info("stop() called - disconnecting from voice")
            await voice_client.disconnect()
            logger.info(f"Stopped and disconnected from guild {guild}")


class Voice(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self._voice_lock = asyncio.Lock()

    async def _validate_user_voice_state(self, ctx):
        if not ctx.author.voice or not ctx.author.voice.channel:
            await ctx.send("You need to be in a voice channel first!")
            return False
        return True

    async def _create_voice_client(self, ctx):
        async with self._voice_lock:
            try:
                await asyncio.sleep(0.5)
                voice_client = await ctx.author.voice.channel.connect(timeout=30.0, reconnect=False)
                logger.info(f"Connected to voice channel: {ctx.author.voice.channel}")
                return voice_client
            except asyncio.TimeoutError:
                await ctx.send("Connection to voice channel timed out.")
                return None
            except discord.ClientException as e:
                logger.error(f"Failed to connect to voice channel: {e}")
                await ctx.send("Failed to connect to voice channel. Please try again.", ephemeral=True)
                return None

    async def _move_voice_client(self, voice_client, ctx):
        try:
            await voice_client.move_to(ctx.author.voice.channel)
            logger.info(f"Moved to voice channel: {ctx.author.voice.channel}")
            return voice_client
        except asyncio.TimeoutError:
            await ctx.send("Connection to voice channel timed out.")
            return None
        except discord.ClientException as e:
            logger.error(f"Failed to move to voice channel: {e}")
            await ctx.send("Failed to move to your voice channel. Please try again.", ephemeral=True)
            return None

    async def _verify_connection(self, voice_client):
        if not voice_client.is_connected():
            try:
                await voice_client.connect(reconnect=False, timeout=10.0)
                return voice_client
            except Exception as e:
                logger.error(f"Failed to reconnect to voice: {e}")
                return None
        return voice_client

    async def ensure_voice_client(self, ctx):
        if not await self._validate_user_voice_state(ctx):
            return None

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

        if voice_client and not voice_client.is_connected():
            try:
                logger.info("Voice client exists but not connected - disconnecting stale client")
                await voice_client.disconnect()
            except Exception:
                pass
            voice_client = None

        if not voice_client or not voice_client.is_connected():
            if not await self._validate_user_voice_state(ctx):
                return None
            return await self._create_voice_client(ctx)

        if voice_client.channel != ctx.author.voice.channel:
            return await self._move_voice_client(voice_client, ctx)

        return await self._verify_connection(voice_client)

    async def connect(self, ctx):
        voice_client = await self.ensure_voice_client(ctx)
        if not voice_client:
            return None
        return voice_client


def setup(bot):
    bot.add_cog(Voice(bot))