niku/TuneBot
1
0
Fork 0
mirror of https://github.com/Matthww/TuneBot.git synced 2026-09-21 19:07:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrated settings cog

Browse Source
This commit is contained in:
niku
2022-11-12 22:43:03 +01:00
parent c80d17ca6f
commit 89c2e12489
9 changed files with 201 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot.py
+11 -1
View File
@@ -12,7 +12,6 @@ import lavalink
from aioredis import Redis
from aioredis.client import Redis
from discord import ActivityType
from discord import app_commands
from discord import Message
from discord.colour import Color
from discord.ext import commands
@@ -23,6 +22,8 @@ from discord.ext.commands.errors import ExtensionNotFound
from discord.ext.commands.errors import NoEntryPointError
from context import CustomContext
from tunebot.redis import RedisPlaylistSource
from tunebot.redis import RedisAutoJoin
from tunebot.redis import GlobalRedisAutoJoin
from tunebot.redis import GlobalRedisPlaylist
from tunebot.redis import GlobalRedisPlaylistSource
@@ -30,9 +31,12 @@ from utils.assets import process_colours
from utils.log import logger
if TYPE_CHECKING:
from tunebot import AutoJoin
from tunebot import PlaylistSource
from tunebot import GlobalPlaylist
from tunebot import GlobalPlaylistSource
from tunebot import GlobalAutoJoin
from tunebot.context import ContextLike
config_path = "config.json"
@@ -136,6 +140,12 @@ class TuneBot(commands.Bot):
async def get_context(self, message: Message, *, cls=CustomContext):
return await super().get_context(message, cls=cls)
def autojoin_context(self, ctx: "ContextLike") -> "AutoJoin":
return RedisAutoJoin(ctx)
def playlist_source_context(self, ctx: "ContextLike") -> "PlaylistSource":
return RedisPlaylistSource(ctx)
@tasks.loop(seconds=30)
async def update_status(self):
await self.wait_until_ready()
cogs/music/__init__.py
+1 -1
View File
@@ -54,7 +54,7 @@ class MusicCog(BaseCog, name="Music"):
embed = helper.create_track_embed(event.player.current)
await channel.send(embed=embed)
elif isinstance(event, lavalink.events.TrackEndEvent):
await self.fill_player_queue(event.player, 1)
await helper.fill_player_queue(self.bot, event.player, 1)
async def setup(bot: "TuneBot"):
cogs/settings/__init__.py
+5 -151
View File
@@ -1,163 +1,17 @@
from typing import Any
from discord.ext import commands
from discord.message import Message
from bot import TuneBot
from context import CustomContext
from utils.classes import BaseCog
from utils.decorators import source_manager_only
from utils.EmbedGenerator import EmbedGenerator
from cogs.settings.interactions import SettingCommands
from cogs.settings.interactions import SourceCommands
class SettingsCog(BaseCog, name="Settings"):
@commands.group(aliases=["aj"], invoke_without_command=True)
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
async def autojoin(self, ctx: CustomContext):
"""Enable/Disable the bot automatically joining"""
await EmbedGenerator.Message(
ctx,
"Autojoin",
f"Usage:\n\n`{ctx.prefix}autojoin enable`\n`{ctx.prefix}autojoin disable`",
)
@autojoin.command(name="enable", aliases=["set"])
@commands.has_permissions(manage_channels=True)
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
async def autojoin_set(self, ctx: CustomContext):
"""Enable the bot automatically joining"""
voice_state = ctx.author.voice
if not voice_state:
embed = ctx.create_embed()
embed.title = "Please join a voice channel before running this command."
await ctx.send(embed=embed)
return
await ctx.autojoin.update(voice_state.channel.id, ctx.message.channel.id)
embed = ctx.create_embed()
embed.title = f"AutoJoin enabled for #{voice_state.channel.name}"
await ctx.send(embed=embed)
@autojoin.command(name="disable", aliases=["unset"])
@commands.has_permissions(manage_channels=True)
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
async def autojoin_del(self, ctx: CustomContext):
"""Disable the bot automatically joining"""
await ctx.autojoin.disable()
embed = ctx.create_embed()
embed.title = f"AutoJoin disabled"
await ctx.send(embed=embed)
@source_manager_only()
@commands.group(
name="source",
aliases=["src"],
invoke_without_command=True,
hidden=True,
)
async def source(self, ctx: CustomContext):
"""Displays all possible options for the `source` command"""
prefix = self.bot.config["prefixes"][0]
embed = ctx.create_embed()
embed.title = "All options:"
embed.description = f"```{prefix}source list\n{prefix}source add <url>\n{prefix}source remove <url>\n{prefix}source sync```"
await ctx.send(embed=embed)
@source_manager_only()
@source.command(name="remove")
async def source_remove(self, ctx: CustomContext, source_url: str):
"""Removes a source from the bot"""
if await ctx.playlist_source.remove(source_url):
prefix = self.bot.config["prefixes"][0]
embed = ctx.create_embed()
embed.title = "Removed source succesfully"
embed.description = (
f"Please use `{prefix}source sync` to persist these changes."
)
await ctx.send(embed=embed)
return
embed = ctx.create_embed()
embed.title = "Could not remove source, the specified source might not exist"
await ctx.send(embed=embed)
@source_manager_only()
@source.command(name="add")
async def source_add(self, ctx: CustomContext, source_url: str):
"""
Add's a source to the bot
Supported sources: YouTube, SoundCloud, Bandcamp, Vimeo, Twitch and HTTP(S) URL's
"""
embed = ctx.create_embed()
embed.title = "Started processing source"
message = await ctx.send(embed=embed)
query_result: Any = await self.bot.lavalink.get_tracks(source_url)
if query_result["loadType"] == "LOAD_FAILED":
embed = ctx.create_embed()
embed.title = "The specified URL is not a valid source"
embed.description = f"Supported sources: YouTube, SoundCloud, Bandcamp, Vimeo, Twitch and HTTP(S) URL's"
if isinstance(message, Message):
await message.edit(embed=embed)
else:
await ctx.send(embed=embed)
return
await ctx.playlist_source.add(source_url)
track_urls = [str(track["info"]["uri"]) for track in query_result["tracks"]]
await self.bot.global_playlist.add_tracks(track_urls)
embed = ctx.create_embed()
embed.title = "Finished processing source"
embed.description = f"Added {len(track_urls)} tracks"
if isinstance(message, Message):
await message.edit(embed=embed)
else:
await ctx.send(embed=embed)
@source_manager_only()
@source.command(name="list", aliases=["ls"])
async def source_list(self, ctx: CustomContext):
"""Display a list of sources"""
# TODO: Implement pagination for sources
sources = await self.bot.global_playlist_source.fetch_sources()
if len(sources) > 0:
description = "\n".join([f"[{source}]({source})" for source in sources])
else:
description = "This bot has no sources yet"
embed = ctx.create_embed()
embed.title = "All sources:"
embed.description = description
await ctx.send(embed=embed)
@source_manager_only()
@source.command(name="sync")
async def source_sync(self, ctx: CustomContext):
"""Forcefully resyncs all sources"""
failed_sources: list[str] = []
await self.bot.global_playlist.clear()
sources = await self.bot.global_playlist_source.fetch_sources()
for source_url in sources:
query_result: Any = await self.bot.lavalink.get_tracks(source_url)
if query_result["loadType"] == "LOAD_FAILED":
failed_sources.append(source_url)
continue
track_urls = [str(track["info"]["uri"]) for track in query_result["tracks"]]
await self.bot.global_playlist.add_tracks(track_urls)
embed = ctx.create_embed()
embed.title = f"Finished sync ({len(failed_sources)} issues)"
if len(failed_sources) > 0:
embed.description = "\n".join(
[f"[{source_url}]({source_url})" for source_url in failed_sources]
)
await ctx.send(embed=embed)
pass
async def setup(bot: TuneBot):
await bot.add_cog(SettingsCog(bot))
bot.tree.add_command(SettingCommands(bot), override=True)
bot.tree.add_command(SourceCommands(bot), override=True)
cogs/settings/interactions.py
+145
View File
@@ -0,0 +1,145 @@
import typing
from discord import app_commands
from discord import Interaction
from tunebot.context import ContextLike
from utils.embed import create_embed
if typing.TYPE_CHECKING:
from discord import VoiceState
from bot import TuneBot
async def is_source_owner(ctx: Interaction) -> bool:
is_manager = ctx.user.id in ctx.client.config["manager_ids"]
is_owner: bool = await ctx.client.is_owner(ctx.user)
return is_manager or is_owner
class SourceCommands(app_commands.Group):
def __init__(self, bot: "TuneBot"):
super().__init__(name="source", description="Manage the radio sources")
self.bot: "TuneBot" = bot
@app_commands.command(name="list", description="List all radio sources")
@app_commands.check(is_source_owner)
async def list(self, ctx: Interaction):
sources = await self.bot.global_playlist_source.fetch_sources()
if len(sources) > 0:
description = "\n".join([f"[{source}]({source})" for source in sources])
else:
description = "This bot has no sources yet"
embed = create_embed(ctx.user)
embed.title = "All sources:"
embed.description = description
await ctx.response.send_message(embed=embed)
@app_commands.command(name="add", description="Add a new radio source")
@app_commands.describe(url="The URL of the to be added source")
@app_commands.check(is_source_owner)
async def add(self, ctx: Interaction, url: str):
embed = create_embed(ctx.user)
clike = ContextLike.from_discord_interaction(ctx)
playlist_source = self.bot.playlist_source_context(clike)
query_result: typing.Any = await self.bot.lavalink.get_tracks(url)
if query_result["loadType"] == "LOAD_FAILED":
embed = create_embed(ctx.user)
embed.title = "The specified URL is not a valid source"
embed.description = "Supported sources: YouTube, SoundCloud, Bandcamp, Vimeo, Twitch and HTTP(S) URL's"
await ctx.response.send_message(embed=embed)
return
await playlist_source.add(url)
track_urls = [str(track["info"]["uri"]) for track in query_result["tracks"]]
await self.bot.global_playlist.add_tracks(track_urls)
embed.title = "Finished processing source"
embed.description = f"Added {len(track_urls)} tracks"
await ctx.response.send_message(embed=embed)
@app_commands.command(name="remove", description="Remove a radio source")
@app_commands.describe(url="The URL of the to be removed source")
@app_commands.check(is_source_owner)
async def remove(self, ctx: Interaction, url: str):
embed = create_embed(ctx.user)
clike = ContextLike.from_discord_interaction(ctx)
playlist_source = self.bot.playlist_source_context(clike)
if await playlist_source.remove(url):
prefix = self.bot.config["prefixes"][0]
embed.title = "Removed source succesfully"
embed.description = (
f"Please use `{prefix}source sync` to persist these changes."
)
await ctx.response.send_message(embed=embed)
return
embed.title = "Could not remove source, the specified source might not exist"
await ctx.response.send_message(embed=embed)
@app_commands.command(name="sync", description="Synchronize all sources")
@app_commands.check(is_source_owner)
async def sync(self, ctx: Interaction):
embed = create_embed(ctx.user)
failed_sources: list[str] = []
await self.bot.global_playlist.clear()
sources = await self.bot.global_playlist_source.fetch_sources()
for source_url in sources:
query_result: typing.Any = await self.bot.lavalink.get_tracks(source_url)
if query_result["loadType"] == "LOAD_FAILED":
failed_sources.append(source_url)
continue
track_urls = [str(track["info"]["uri"]) for track in query_result["tracks"]]
await self.bot.global_playlist.add_tracks(track_urls)
embed.title = f"Finished sync ({len(failed_sources)} issues)"
if len(failed_sources) > 0:
embed.description = "\n".join(
[f"[{source_url}]({source_url})" for source_url in failed_sources]
)
await ctx.response.send_message(embed=embed)
class SettingCommands(app_commands.Group):
def __init__(self, bot: "TuneBot"):
super().__init__(name="settings", description="Tweak and customize")
self.bot: "TuneBot" = bot
@app_commands.command(
name="autojoin",
description="Let the bot automatically join a channel after occassional maintenance",
)
@app_commands.describe(state="enabled or disabled")
@app_commands.choices(
state=[
app_commands.Choice(name="enable", value=0),
app_commands.Choice(name="disable", value=1),
]
)
@app_commands.check(is_source_owner)
async def autojoin(self, ctx: Interaction, state: app_commands.Choice[int]):
voice_state: "VoiceState" = ctx.user.voice
embed = create_embed(ctx.user)
if not voice_state:
embed.title = "Please join a voice channel before running this command."
await ctx.response.send_message(embed=embed)
return
clike = ContextLike.from_discord_interaction(ctx)
autojoin = self.bot.autojoin_context(clike)
enabled = state.value
if not enabled:
await autojoin.update(voice_state.channel.id, ctx.channel_id)
embed.title = f"AutoJoin enabled for #{voice_state.channel.name}"
await ctx.response.send_message(embed=embed)
return
await autojoin.disable()
embed.title = "AutoJoin disabled"
await ctx.response.send_message(embed=embed)
context.py
+5 -2
View File
@@ -6,6 +6,7 @@ from discord.ext import commands
from discord.ext.commands.errors import CommandInvokeError
from lavalink.models import DefaultPlayer
from tunebot.context import ContextLike
from tunebot.redis import RedisAutoJoin
from tunebot.redis import RedisPlaylistSource
@@ -43,11 +44,13 @@ class CustomContext(commands.Context):
@property
def playlist_source(self) -> "PlaylistSource":
if not hasattr(self, "_playlist_source"):
self._playlist_source = RedisPlaylistSource(self)
ctx = ContextLike.from_discord_context(self)
self._playlist_source = self.bot.playlist_source_context(ctx)
return self._playlist_source
@property
def autojoin(self) -> "AutoJoin":
if not hasattr(self, "_autojoin"):
self._autojoin = RedisAutoJoin(self)
ctx = ContextLike.from_discord_context(self)
self._autojoin = self.bot.autojoin_context(ctx)
return self._autojoin
tunebot/context.py
+28
View File
@@ -0,0 +1,28 @@
import typing
from dataclasses import dataclass
if typing.TYPE_CHECKING:
from aioredis import Redis
from bot import TuneBot
from discord import Interaction
from context import CustomContext
@dataclass
class ContextLike:
redis: "Redis"
bot: "TuneBot"
guild_id: int
@staticmethod
def from_discord_context(ctx: "CustomContext") -> "ContextLike":
return ContextLike(redis=ctx.redis, bot=ctx.bot, guild_id=ctx.guild.id)
@staticmethod
def from_discord_interaction(ctx: "Interaction") -> "ContextLike":
return ContextLike(
redis=ctx.client._redis_client, bot=ctx.client, guild_id=ctx.guild_id
)
__all__ = ("ContextLike",)
tunebot/redis/autojoin.py
+4 -4
View File
@@ -22,20 +22,20 @@ class RedisAutoJoin(RedisContextEntity, AutoJoin):
voice_channel_id (int): [description]
text_channel_id (int): [description]
"""
if not self.ctx.guild:
if not self.ctx.guild_id:
raise Exception("This method can only be invoked inside of a guild.")
value = f"{voice_channel_id}:{text_channel_id}"
await self.redis.hset(self.key("autojoin"), self.ctx.guild.id, value)
await self.redis.hset(self.key("autojoin"), self.ctx.guild_id, value)
async def disable(self):
"""
Removes the AutoJoin configuration of a guild.
"""
if not self.ctx.guild:
if not self.ctx.guild_id:
raise Exception("This method can only be invoked inside of a guild.")
await self.redis.hdel(self.key("autojoin"), self.ctx.guild.id)
await self.redis.hdel(self.key("autojoin"), self.ctx.guild_id)
__all__ = ("GlobalRedisAutoJoin", "RedisAutoJoin")
tunebot/redis/entity.py
+2 -2
View File
@@ -3,7 +3,7 @@ from abc import abstractmethod
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from context import CustomContext
from tunebot.context import ContextLike
from aioredis.client import Redis
@@ -34,7 +34,7 @@ class RedisBotEntity(RedisEntity):
class RedisContextEntity(RedisEntity):
def __init__(self, ctx: "CustomContext") -> None:
def __init__(self, ctx: "ContextLike") -> None:
self.ctx = ctx
super().__init__()
utils/decorators.py
-27
View File
@@ -1,27 +0,0 @@
from typing import Callable
from typing import TYPE_CHECKING
from typing import TypeVar
from discord.ext.commands import check
from discord.ext.commands.errors import NotOwner
if TYPE_CHECKING:
from context import CustomContext
T = TypeVar("T")
def source_manager_only() -> Callable[[T], T]:
"""
A :func:`.check` that checks if the person invoking this command is allowed to modify the radio sources.
"""
async def predicate(ctx: "CustomContext") -> bool:
is_manager = ctx.author.id in ctx.bot.config["manager_ids"]
is_owner = await ctx.bot.is_owner(ctx.author)
if not is_manager and not is_owner:
raise NotOwner("You are not allowed to modify the sources.")
return True
return check(predicate)