# 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 os import logging 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: # Connect to voice channel if not connected if vc is None: 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}") # Move to user's channel if in a different one elif 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 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 ffmpeg_options = { 'options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5', 'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5' } audio_source = discord.FFmpegPCMAudio(source, **ffmpeg_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.File(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 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): """Execute the actual audio playback.""" vc = ctx.voice_client target_channel = await AudioPlaybackManager.validate_voice_channel(ctx) if not target_channel: return False try: vc = await AudioPlaybackManager.connect_to_voice_channel(ctx, vc, target_channel) # Stop any currently playing audio if vc.is_playing() or vc.is_paused(): vc.stop() # Prepare audio source audio_source, final_display_name = await AudioPlaybackManager.prepare_audio_source(source, is_url, display_name) # Play the audio vc.play(audio_source) logger.info(f"Playing: {final_display_name}") await ctx.respond(f"grom: {final_display_name}") return True except Exception as e: logger.error(f"Error in play: {str(e)}", exc_info=True) await ctx.respond(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): """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) """ # 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) # Cleanup if needed if not success: await AudioPlaybackManager.cleanup_if_needed(prepared_source, delete_after_play, is_url)