286 lines
No EOL
12 KiB
Python
286 lines
No EOL
12 KiB
Python
import asyncio
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import discord
|
|
from redbot.core import Config, commands
|
|
from redbot.core.bot import Red
|
|
from redbot.core.utils.chat_formatting import box, humanize_list
|
|
from redbot.core.utils.predicates import MessagePredicate
|
|
|
|
log = logging.getLogger("red.valerie.extendedaudio")
|
|
|
|
SUPPORTED_VARIABLES = {
|
|
"title": "Song title",
|
|
"requester": "User who requested the song",
|
|
"duration": "Duration of the song"
|
|
}
|
|
|
|
class ExtendedAudio(commands.Cog):
|
|
"""Extends Red's core Audio cog with additional features"""
|
|
|
|
def __init__(self, bot: Red):
|
|
self.bot = bot
|
|
self.config = Config.get_conf(self, identifier=117117117)
|
|
default_guild = {
|
|
"channel_status": True, # Whether to update voice channel name with current song
|
|
"status_format": "🎵 {title}", # Format for the channel name
|
|
"original_name": None, # Store the original channel name
|
|
}
|
|
self.config.register_guild(**default_guild)
|
|
self.task = self.bot.loop.create_task(self.initialize())
|
|
self.audio = None
|
|
|
|
async def initialize(self):
|
|
"""Initialize the cog"""
|
|
await self.bot.wait_until_ready()
|
|
try:
|
|
self.audio = self.bot.get_cog("Audio")
|
|
if not self.audio:
|
|
log.error("Audio cog not loaded! Some features may not work.")
|
|
except Exception as e:
|
|
log.error(f"Failed to initialize ExtendedAudio: {e}")
|
|
|
|
def cog_unload(self):
|
|
"""Cleanup when cog unloads"""
|
|
if self.task:
|
|
self.task.cancel()
|
|
# Reset all modified channel names
|
|
asyncio.create_task(self._cleanup_on_unload())
|
|
|
|
async def _cleanup_on_unload(self):
|
|
"""Reset all modified channel names when cog unloads"""
|
|
for guild in self.bot.guilds:
|
|
try:
|
|
original_name = await self.config.guild(guild).original_name()
|
|
if original_name and guild.voice_client and guild.voice_client.channel:
|
|
await guild.voice_client.channel.edit(name=original_name)
|
|
await self.config.guild(guild).original_name.set(None)
|
|
except:
|
|
continue
|
|
|
|
async def cog_check(self, ctx: commands.Context):
|
|
"""Check if Audio cog is loaded"""
|
|
if not self.audio:
|
|
self.audio = self.bot.get_cog("Audio")
|
|
if not self.audio:
|
|
await ctx.send("The Audio cog is not loaded. Please load it first with `[p]load audio`")
|
|
return False
|
|
return True
|
|
|
|
@commands.group(aliases=["eaudioset"])
|
|
@commands.guild_only()
|
|
@commands.admin_or_permissions(manage_guild=True)
|
|
async def extendedaudioset(self, ctx: commands.Context):
|
|
"""Configure ExtendedAudio settings"""
|
|
if not ctx.invoked_subcommand:
|
|
guild = ctx.guild
|
|
conf = await self.config.guild(guild).all()
|
|
enabled = conf["channel_status"]
|
|
format_str = conf["status_format"]
|
|
msg = (
|
|
f"**Current Settings**\n"
|
|
f"Channel Status: {'Enabled' if enabled else 'Disabled'}\n"
|
|
f"Status Format: {format_str}\n\n"
|
|
f"Available Variables: {humanize_list([f'{{' + k + '}}' for k in SUPPORTED_VARIABLES.keys()])}\n"
|
|
f"Use `{ctx.prefix}extendedaudioset statusformat` to change the format."
|
|
)
|
|
await ctx.send(box(msg))
|
|
|
|
@extendedaudioset.command(name="channelstatus")
|
|
async def set_channel_status(self, ctx: commands.Context, enabled: bool):
|
|
"""Toggle voice channel status updates
|
|
|
|
This will update the voice channel name with the currently playing song
|
|
|
|
Example: [p]extendedaudioset channelstatus true
|
|
"""
|
|
guild = ctx.guild
|
|
|
|
# Check if bot has required permissions
|
|
if not guild.me.guild_permissions.manage_channels:
|
|
await ctx.send("I need the `Manage Channels` permission to update channel names.")
|
|
return
|
|
|
|
# Store original channel name if enabling and not stored
|
|
if enabled:
|
|
voice_client = guild.voice_client
|
|
if voice_client and voice_client.channel:
|
|
current_name = voice_client.channel.name
|
|
if not await self.config.guild(guild).original_name():
|
|
await self.config.guild(guild).original_name.set(current_name)
|
|
|
|
await self.config.guild(guild).channel_status.set(enabled)
|
|
state = "enabled" if enabled else "disabled"
|
|
await ctx.send(f"Voice channel status updates {state}.")
|
|
|
|
# Reset channel name if disabling
|
|
if not enabled:
|
|
try:
|
|
voice_client = guild.voice_client
|
|
if voice_client and voice_client.channel:
|
|
original_name = await self.config.guild(guild).original_name()
|
|
if original_name:
|
|
await voice_client.channel.edit(name=original_name)
|
|
await self.config.guild(guild).original_name.set(None)
|
|
except discord.Forbidden:
|
|
await ctx.send("I don't have permission to edit the channel name.")
|
|
except Exception as e:
|
|
log.error(f"Error resetting channel name: {e}")
|
|
|
|
@extendedaudioset.command(name="statusformat")
|
|
async def set_status_format(self, ctx: commands.Context, *, format_str: str):
|
|
"""Set the format for voice channel status updates
|
|
|
|
Available variables:
|
|
{title} - Song title
|
|
{requester} - User who requested the song
|
|
{duration} - Duration of the song
|
|
|
|
Example: [p]extendedaudioset statusformat 🎵 {title}
|
|
"""
|
|
# Validate format string
|
|
try:
|
|
# Test with sample data
|
|
test_data = {
|
|
"title": "Test Song",
|
|
"requester": "Test User",
|
|
"duration": "3:45"
|
|
}
|
|
format_str.format(**test_data)
|
|
except KeyError as e:
|
|
invalid_var = str(e).strip("'")
|
|
supported = humanize_list([f"{{" + k + "}}" for k in SUPPORTED_VARIABLES.keys()])
|
|
await ctx.send(
|
|
f"Invalid variable `{invalid_var}` in format string.\n"
|
|
f"Supported variables are: {supported}"
|
|
)
|
|
return
|
|
except Exception as e:
|
|
await ctx.send(f"Invalid format string: {e}")
|
|
return
|
|
|
|
await self.config.guild(ctx.guild).status_format.set(format_str)
|
|
example = format_str.format(title="Never Gonna Give You Up", requester="Rick Astley", duration="3:32")
|
|
await ctx.send(
|
|
f"Status format set! Example:\n"
|
|
f"{box(example)}"
|
|
)
|
|
|
|
@commands.Cog.listener()
|
|
async def on_red_audio_track_start(self, guild: discord.Guild, track: dict, requester: discord.Member):
|
|
"""Update voice channel name when a new track starts"""
|
|
if not guild or not track:
|
|
return
|
|
|
|
try:
|
|
# Check if channel status updates are enabled
|
|
if not await self.config.guild(guild).channel_status():
|
|
return
|
|
|
|
# Get the voice channel
|
|
voice_client = guild.voice_client
|
|
if not voice_client or not voice_client.channel:
|
|
return
|
|
|
|
channel = voice_client.channel
|
|
|
|
# Store original name if not stored
|
|
if not await self.config.guild(guild).original_name():
|
|
await self.config.guild(guild).original_name.set(channel.name)
|
|
|
|
# Get the format and create the new name
|
|
format_str = await self.config.guild(guild).status_format()
|
|
|
|
# Prepare variables
|
|
duration = track.length if hasattr(track, "length") else 0
|
|
minutes = duration // 60000 # Convert milliseconds to minutes
|
|
seconds = (duration % 60000) // 1000 # Convert remaining milliseconds to seconds
|
|
duration_str = f"{minutes}:{seconds:02d}"
|
|
|
|
new_name = format_str.format(
|
|
title=track.title if hasattr(track, "title") else "Unknown",
|
|
requester=requester.display_name,
|
|
duration=duration_str
|
|
)
|
|
|
|
# Ensure name length is valid (Discord limit is 100 characters)
|
|
if len(new_name) > 100:
|
|
new_name = new_name[:97] + "..."
|
|
|
|
# Update channel name if different and we have permissions
|
|
if channel.name != new_name and guild.me.guild_permissions.manage_channels:
|
|
await channel.edit(name=new_name)
|
|
|
|
except discord.Forbidden:
|
|
log.warning(f"Missing permissions to edit channel name in {guild.name}")
|
|
except Exception as e:
|
|
log.error(f"Error updating channel status in {guild.name}: {e}")
|
|
|
|
@commands.Cog.listener()
|
|
async def on_red_audio_queue_end(self, guild: discord.Guild, *args, **kwargs):
|
|
"""Reset voice channel name when queue ends"""
|
|
if not guild:
|
|
return
|
|
|
|
try:
|
|
# Check if channel status updates are enabled
|
|
if not await self.config.guild(guild).channel_status():
|
|
return
|
|
|
|
# Get the voice channel
|
|
voice_client = guild.voice_client
|
|
if not voice_client or not voice_client.channel:
|
|
return
|
|
|
|
channel = voice_client.channel
|
|
|
|
# Reset to original name if stored and we have permissions
|
|
original_name = await self.config.guild(guild).original_name()
|
|
if original_name and guild.me.guild_permissions.manage_channels:
|
|
await channel.edit(name=original_name)
|
|
|
|
except discord.Forbidden:
|
|
log.warning(f"Missing permissions to edit channel name in {guild.name}")
|
|
except Exception as e:
|
|
log.error(f"Error resetting channel status in {guild.name}: {e}")
|
|
|
|
@commands.Cog.listener()
|
|
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState):
|
|
"""Reset channel name when bot leaves voice channel"""
|
|
if not member.guild:
|
|
return
|
|
|
|
# Check if this is the bot disconnecting
|
|
if member.id == self.bot.user.id and before.channel and not after.channel:
|
|
try:
|
|
# Reset channel name if we have the original stored and permissions
|
|
guild = member.guild
|
|
if not guild.me.guild_permissions.manage_channels:
|
|
return
|
|
|
|
original_name = await self.config.guild(guild).original_name()
|
|
if original_name and before.channel:
|
|
await before.channel.edit(name=original_name)
|
|
await self.config.guild(guild).original_name.set(None)
|
|
except Exception as e:
|
|
log.error(f"Error resetting channel name on disconnect: {e}")
|
|
|
|
@commands.Cog.listener()
|
|
async def on_red_audio_player_auto_disconnect(self, guild: discord.Guild):
|
|
"""Handle auto-disconnect cleanup"""
|
|
try:
|
|
if guild.me.guild_permissions.manage_channels:
|
|
original_name = await self.config.guild(guild).original_name()
|
|
if original_name:
|
|
voice_client = guild.voice_client
|
|
if voice_client and voice_client.channel:
|
|
await voice_client.channel.edit(name=original_name)
|
|
await self.config.guild(guild).original_name.set(None)
|
|
except Exception as e:
|
|
log.error(f"Error handling auto-disconnect cleanup: {e}")
|
|
|
|
@commands.Cog.listener()
|
|
async def on_red_audio_player_auto_pause(self, guild: discord.Guild):
|
|
"""Handle auto-pause"""
|
|
# We don't change the channel name on pause to maintain consistency |