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
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")
                        try:
                            await vc.disconnect()
                        except Exception:
                            pass
                        # Forcefully clear stale internal state so the fresh
                        # connect doesn't hang for 30s waiting for Discord.
                        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
                            vc = None
                else:
                    logger.info("Voice client exists but disconnected, cleaning up stale entry")
                    try:
                        await vc.disconnect()
                    except Exception:
                        pass
                    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