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

Merge branch 'dev' into fix_wlinfo

Browse Source
This commit is contained in:
matthew
2021-11-06 21:38:29 +01:00
committed by GitHub
parent bc94157a0c a8db9e70df
commit a3d9d71af4
18 changed files with 506 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cogs/information.py
+21 -25
View File
@@ -1,24 +1,25 @@
import datetime
import time
from typing import Optional
import discord
from discord.ext import tasks, commands
import time
from discord.ext.commands import Context
import humanize
import datetime
import lavalink
from bot import ChristmasBot
from discord.ext import commands
from discord.ext import tasks
from discord.ext.commands import Context
from bot import TuneBot
from context import CustomContext
from utils.database import AutoJoin
from utils.EmbedGenerator import EmbedGenerator
from utils.classes import BaseCog
from utils.paginator import HelpPaginator
class InformationCog(commands.Cog, name="Information"):
def __init__(self, bot: ChristmasBot):
self.bot = bot
class InformationCog(BaseCog, name="Information"):
@commands.command(name="ping", aliases=["pong"])
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
@commands.command(description="PONG!", aliases=["pong"])
async def ping(self, ctx: Context):
"""Test the latency"""
avatar = ctx.author.avatar.with_static_format("jpeg")
@@ -39,24 +40,18 @@ class InformationCog(commands.Cog, name="Information"):
embed.set_footer(text=f"Requested by: {ctx.author}", icon_url=f"{avatar}")
await msg.edit(embed=embed)
@commands.command(
name="invite", description="Gets the invite link!", slash_commands=True
)
@commands.command(name="invite")
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
async def invite(self, ctx: Context):
"""Gets the invite link!"""
"""Gets the invite link"""
await EmbedGenerator.Message(
ctx, "Add our bot to your server:", self.bot.invite_link
)
@commands.command(
name="wlinfo",
description="Retrieve various Node/Server/Player information.",
slash_commands=True,
)
@commands.command(name="wlinfo")
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
async def wlinfo(self, ctx: Context):
"""Retrieve various Node/Server/Player information."""
async def wlinfo(self, ctx: CustomContext):
"""Retrieve various node/server/player information"""
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
nodes = self.bot.lavalink.node_manager.available_nodes
@@ -77,8 +72,9 @@ class InformationCog(commands.Cog, name="Information"):
#f"Server Uptime: `{datetime.timedelta(milliseconds=node.stats.uptime)}`"
)
await ctx.send(fmt)
AutoJoin.get_channels()
@commands.command(name="help", aliases=["about", "info"], slash_command=True)
@commands.command(name="help", aliases=["about", "info"])
@commands.cooldown(1, 1, commands.BucketType.user)
async def about(
self,
@@ -87,7 +83,7 @@ class InformationCog(commands.Cog, name="Information"):
description="Show help for a command or category"
),
):
"""ChristmasBot command list"""
"""Retrieve a list of possible commands"""
if command:
entity = self.bot.get_cog(command) or self.bot.get_command(command)
@@ -127,6 +123,6 @@ class InformationCog(commands.Cog, name="Information"):
await ctx.send(embed=embed)
def setup(bot: ChristmasBot):
def setup(bot: TuneBot):
bot.remove_command("help")
bot.add_cog(InformationCog(bot))
cogs/music.py
+94 -58
View File
@@ -1,21 +1,27 @@
import asyncio
import datetime
import re
from typing import Optional
from aioredis.client import Redis
import discord
from discord.channel import TextChannel
from discord.ext.commands.context import Context
from discord.ext.commands.errors import CommandError
import lavalink
from discord.ext import commands
from lavalink.models import AudioTrack, DefaultPlayer
from bot import ChristmasBot
from utils.EmbedGenerator import EmbedGenerator
from utils.database import AutoJoin
from context import CustomContext
from discord import Embed
from discord.channel import TextChannel
from discord.ext import commands
from discord.ext.commands.context import Context
from lavalink.models import AudioTrack
from lavalink.models import DefaultPlayer
from bot import TuneBot
from utils.classes import BaseCog
from context import CustomContext
from utils.database import AutoJoin
from utils.database import Playlist
from utils.exceptions import EmbeddedCommandException
from utils.EmbedGenerator import EmbedGenerator
url_rx = re.compile(r"https?://(?:www\.)?.+")
@@ -69,10 +75,7 @@ class LavalinkVoiceClient(discord.VoiceClient):
self.cleanup()
class Music(commands.Cog):
def __init__(self, bot: ChristmasBot):
self.bot = bot
class Music(BaseCog):
@commands.Cog.listener()
async def on_ready(self):
if not hasattr(
@@ -96,7 +99,7 @@ class Music(commands.Cog):
for guild_id, (voicechannel_id, textchannel_id) in redis_result.items():
player = self.bot.lavalink.player_manager.create(guild_id)
player.store('channel', textchannel_id)
player.store("channel", textchannel_id)
voice_channel = await self.bot.fetch_channel(voicechannel_id)
await voice_channel.connect(cls=LavalinkVoiceClient)
if not player.is_playing:
@@ -109,16 +112,10 @@ class Music(commands.Cog):
await textchannel.send("Automatically joined the voice channel")
async def fill_player_queue(self, player: DefaultPlayer, buffer: Optional[int] = 1):
pipeline = self.bot._redis_client.pipeline()
for _ in range(buffer):
pipeline.randomkey()
queries = await pipeline.execute()
print(queries)
queries = await Playlist.random(self.bot._redis_client, buffer)
# Get the results for the query from Lavalink.
for query in queries:
result = await player.node.get_tracks(query)
print(result)
if not result or not result["tracks"]:
continue
@@ -127,6 +124,26 @@ class Music(commands.Cog):
)
player.add(requester=self.bot.user.id, track=track)
async def create_track_embed(self, track: AudioTrack) -> Embed:
embed_color = self.bot.colors["embed"]
embed = discord.Embed(
title=f"Now playing...",
colour=embed_color,
)
embed.description = f"[{track.title}]({track.uri})"
embed.set_thumbnail(
url=f"https://i3.ytimg.com/vi/{track.identifier}/mqdefault.jpg"
)
try:
duration = str(datetime.timedelta(milliseconds=int(track.duration)))
except OverflowError:
duration = "🔴 LIVE"
embed.add_field(name="Duration", value=duration)
embed.add_field(name="Author", value=track.author)
return embed
def cog_unload(self):
"""Cog unload handler. This removes any event hooks that were registered."""
self.bot.lavalink._event_hooks.clear()
@@ -147,13 +164,15 @@ class Music(commands.Cog):
return guild_check
async def cog_command_error(self, ctx, error):
async def cog_command_error(self, ctx: CustomContext, error: CommandError):
if isinstance(error, commands.CommandInvokeError):
await ctx.send(error.original)
# The above handles errors thrown in this cog and shows them to the user.
# This shouldn't be a problem as the only errors thrown in this cog are from `ensure_voice`
# which contain a reason string, such as "Join a voicechannel" etc. You can modify the above
# if you want to do things differently.
elif isinstance(error, EmbeddedCommandException):
await error.send(ctx)
async def ensure_voice(self, ctx):
"""This check ensures that the bot and command author are in the same voicechannel."""
@@ -178,7 +197,14 @@ class Music(commands.Cog):
if not player.is_connected:
if not should_connect:
raise commands.CommandInvokeError("Not connected.")
bot_name = self.bot.config["info"]["name"]
embed = await EmbedGenerator.Message(
ctx,
f"{bot_name} is not connected",
"However you can start playing music using `/connect`",
no_send=True,
)
raise EmbeddedCommandException(embed)
permissions = ctx.author.voice.channel.permissions_for(ctx.me)
@@ -206,68 +232,78 @@ class Music(commands.Cog):
elif isinstance(event, lavalink.events.TrackStartEvent):
channel_id = int(event.player.fetch("channel"))
channel: TextChannel = self.bot.get_channel(channel_id)
color = self.bot.colors["embed"]
current_track: AudioTrack = event.player.current
embed = Embed(
title="Now playing:",
description=f"[{current_track.title}]({current_track.uri})",
color=color,
)
embed = await self.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)
@commands.command(name="connect", aliases=["p", "play", "join"])
async def play(self, ctx: CustomContext):
"""Starts playing Christmas bangers"""
"""Start the radio"""
# Get the player for this guild from cache.
player: DefaultPlayer = self.bot.lavalink.player_manager.get(ctx.guild.id)
await self.fill_player_queue(player, self.bot.config["queue_buffer_size"])
if ctx.player.is_connected:
await ctx.send("Already connected")
return
if not player.is_playing:
await player.play()
await self.fill_player_queue(
ctx.player, self.bot.config["queue_buffer_size"] + 1
)
await ctx.send("Started playing")
if not ctx.player.is_playing:
await ctx.player.play()
await EmbedGenerator.Title(ctx, "*⃣ | Connected.")
return
@commands.command(name="skip", aliases=["next"])
async def skip(self, ctx: Context):
"""I heard this song way too often"""
player: DefaultPlayer = self.bot.lavalink.player_manager.get(ctx.guild.id)
await player.skip()
async def skip(self, ctx: CustomContext):
"""Skip the current song"""
await ctx.player.skip()
await ctx.send("Skipped current song")
@commands.command(name="queue")
async def queue(self, ctx: Context):
"""Ghetto queue"""
player: DefaultPlayer = self.bot.lavalink.player_manager.get(ctx.guild.id)
await EmbedGenerator.Message(ctx, "Queue:", player.queue)
async def queue(self, ctx: CustomContext):
"""Display the current radio queue"""
embed_color = self.bot.colors["embed"]
embed = Embed(title="Coming Up...", colour=embed_color)
if len(ctx.player.queue) > 0:
embed.description = "\n".join(
f"{index}. [{track.title}]({track.uri})"
for index, track in enumerate(ctx.player.queue, 1)
)
else:
embed.description = "We are still determining a playlist"
await ctx.send(embed=embed)
@commands.command(name="disconnect", aliases=["dc", "stop"])
async def disconnect(self, ctx: Context):
"""Disconnects ChristmasBot"""
player: DefaultPlayer = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not player.is_connected:
return await EmbedGenerator.Title(ctx, "Not connected.")
async def disconnect(self, ctx: CustomContext):
"""Disconnects the radio from the channel"""
if not ctx.author.voice or (
player.is_connected
and ctx.author.voice.channel.id != int(player.channel_id)
ctx.player.is_connected
and ctx.author.voice.channel.id != int(ctx.player.channel_id)
):
# Abuse prevention. Users not in voice channels, or not in the same voice channel as the bot
# may not disconnect the bot.
return await EmbedGenerator.Title(ctx, "You're not in my voicechannel!")
await EmbedGenerator.Title(ctx, "You're not in my voicechannel!")
return
# Clear the queue to ensure old tracks don't start playing
# when someone else queues something.
player.queue.clear()
ctx.player.queue.clear()
# Stop the current track so Lavalink consumes less resources.
await player.stop()
await ctx.player.stop()
# Disconnect from the voice channel.
await ctx.voice_client.disconnect(force=True)
await EmbedGenerator.Title(ctx, "*⃣ | Disconnected.")
@commands.command(name="now")
async def now_playing(self, ctx: CustomContext):
"""Displays information about the currently played track"""
embed = await self.create_track_embed(ctx.player.current)
await ctx.send(embed=embed)
def setup(bot: ChristmasBot):
def setup(bot: TuneBot):
bot.add_cog(Music(bot))
cogs/owner.py
+12 -12
View File
@@ -1,24 +1,24 @@
import discord
from discord.ext import commands
import textwrap
import io
import traceback
import asyncio
import io
import textwrap
import time
import traceback
from asyncio.subprocess import PIPE
from contextlib import redirect_stdout
from io import BytesIO
from platform import python_version
from contextlib import redirect_stdout
import discord
from discord.ext import commands
from discord.ext.commands.context import Context
from bot import ChristmasBot
from bot import TuneBot
from utils.classes import BaseCog
class OwnerCog(commands.Cog):
def __init__(self, bot: ChristmasBot):
self.bot = bot
class OwnerCog(BaseCog):
def __init__(self, bot: TuneBot):
super().__init__(bot)
self._last_result = None
@staticmethod
@@ -209,5 +209,5 @@ class OwnerCog(commands.Cog):
await message.edit(content=f"```fix\n{content}\n```")
def setup(bot: ChristmasBot):
def setup(bot: TuneBot):
bot.add_cog(OwnerCog(bot))
cogs/settings.py
+15 -9
View File
@@ -1,18 +1,21 @@
from context import CustomContext
from discord.ext import commands
from utils.EmbedGenerator import EmbedGenerator
from utils.classes import BaseCog
from utils.database import AutoJoin
from bot import ChristmasBot
from bot import TuneBot
from discord.ext.commands import Context
from bot import TuneBot
from context import CustomContext
from utils.database import AutoJoin
from utils.EmbedGenerator import EmbedGenerator
class SettingsCog(commands.Cog, name="Settings"):
def __init__(self, bot: ChristmasBot):
self.bot = bot
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",
@@ -23,19 +26,22 @@ class SettingsCog(commands.Cog, name="Settings"):
@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"""
voicechannel_id = ctx.author.voice.channel.id
textchannel_id = ctx.message.channel.id
await AutoJoin.update_channel(ctx.get_redis(), ctx.guild.id, voicechannel_id, textchannel_id)
await AutoJoin.update_channel(
ctx.redis, ctx.guild.id, voicechannel_id, textchannel_id
)
await EmbedGenerator.Message(ctx, "Autojoin", "`enabled`")
@autojoin.command(name="disable")
@commands.has_permissions(manage_channels=True)
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
async def autojoin_del(self, ctx: CustomContext):
vc = ctx.author.voice.channel
await AutoJoin.del_channel(ctx.get_redis(), ctx.guild.id)
"""Disable the bot automatically joining"""
await AutoJoin.del_channel(ctx.redis, ctx.guild.id)
await EmbedGenerator.Message(ctx, "Autojoin", "`disabled`")
def setup(bot: ChristmasBot):
def setup(bot: TuneBot):
bot.add_cog(SettingsCog(bot))