path: root/cogs/music/audio_playback.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 asyncio
import logging
import os
import urllib.parse

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:
            # Check if vc is valid and actually connected
            if vc is not None and vc.is_connected():
                # Already connected - check if need to move
                if 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

            # Not connected - connect now
            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}")
            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 and extract clean filename
            before_options = "-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5"
            audio_source = discord.FFmpegPCMAudio(source, before_options=before_options)
            # Extract clean filename for URL (remove extension and decode URL encoding)
            filename = os.path.basename(source.split("?")[0])  # Remove query parameters
            filename = urllib.parse.unquote(filename)  # Decode URL encoding
            clean_filename = os.path.splitext(filename)[0]  # Remove file extension
            return audio_source, clean_filename
        else:
            # For local files, use default options
            audio_source = discord.FFmpegPCMAudio(source=source)

            # Try to get metadata for local files
            try:
                metadata = mutagen.open(source)  # type: ignore[call-overload]
                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 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, interrupt=True):
        """Execute the actual audio playback.

        Args:
            ctx: Discord context
            source: Audio source path or URL
            display_name: Name to display
            is_url: Whether source is a URL
            interrupt: If True, stop any currently playing audio. If False, skip if audio is playing.
        """
        target_channel = await AudioPlaybackManager.validate_voice_channel(ctx)
        if not target_channel:
            return False

        # Get existing voice client from bot or connect
        guild = ctx.guild
        vc = discord.utils.get(ctx.bot.voice_clients, guild=guild)

        try:
            # Handle various connection states
            if vc is None:
                logger.info("No existing voice client, connecting...")
                try:
                    vc = await target_channel.connect()
                except discord.ClientException:
                    vc = discord.utils.get(ctx.bot.voice_clients, guild=guild)
            elif not vc.is_connected():
                logger.info("Voice client exists but disconnected, reconnecting...")
                try:
                    await vc.disconnect()
                except Exception:
                    pass
                try:
                    vc = await target_channel.connect()
                except discord.ClientException:
                    vc = discord.utils.get(ctx.bot.voice_clients, guild=guild)
            elif vc.channel != target_channel:
                logger.info(f"Moving to voice channel: {target_channel.name}")
                await vc.move_to(target_channel)

            # Stop any currently playing audio (if interrupt is True)
            if interrupt and (vc.is_playing() or vc.is_paused()):  # type: ignore[union-attr]
                vc.stop()  # type: ignore[union-attr]
            elif not interrupt and vc.is_playing():  # type: ignore[union-attr]
                logger.info("Audio already playing and interrupt=False, skipping playback")
                return False

            # Prepare audio source
            (
                audio_source,
                final_display_name,
            ) = await AudioPlaybackManager.prepare_audio_source(
                source, is_url, display_name
            )

            # Track playback state
            playback_state = {"error": None, "finished": False}

            def after_playing(error):
                if error:
                    playback_state["error"] = str(error)
                    logger.error(f"Playback error: {error}")
                playback_state["finished"] = True

            # Play the audio
            vc.play(audio_source, after=after_playing)
            logger.info(f"Playing: {final_display_name}")
            await ctx.followup.send(f"grom: {final_display_name}")

            # Wait for FFmpeg to attempt connection - errors usually happen within 5 seconds
            await asyncio.sleep(5)

            # Check if playback failed to start (not just finished quickly)
            if playback_state["error"]:
                logger.error(f"Playback failed: {playback_state['error']}")
                await ctx.followup.send(
                    "eror: Timeout - nie udało się połączyć ze streamem", ephemeral=True
                )
                return False

            # If playback finished within 5s, it was a short file - not an error
            if playback_state["finished"]:
                logger.info(f"Playback completed quickly: {final_display_name}")
                return True

            if not vc.is_playing():
                logger.error(f"Stream failed - not playing after 5s: {source}")
                await ctx.followup.send(
                    "eror: Timeout - nie udało się połączyć ze streamem", ephemeral=True
                )
                return False

            return True

        except Exception as e:
            logger.error(f"Error in play: {str(e)}", exc_info=True)
            await ctx.followup.send(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, interrupt=True):
        """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)
            interrupt: If True, stop any currently playing audio. If False, skip if audio is playing.
        """
        # 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, interrupt
        )

        # Cleanup if needed
        if not success:
            await AudioPlaybackManager.cleanup_if_needed(
                prepared_source, delete_after_play, is_url
            )