Files
TuneBot/cogs/music/helper.py
T

64 lines
1.9 KiB
Python

import datetime
import typing
import discord
import lavalink
from bot import colors
from cogs.music.voice_client import LavalinkVoiceClient
from utils.log import logger
if typing.TYPE_CHECKING:
from lavalink import DefaultPlayer
from lavalink_player import CustomPlayer
from bot import TuneBot
def get_player(bot: "TuneBot", guild_id: int) -> "CustomPlayer":
if player := bot.lavalink.player_manager.get(guild_id=guild_id):
return player
return bot.lavalink.player_manager.create(guild_id)
async def fill_player_queue(
bot: "TuneBot", player: "DefaultPlayer", amount: typing.Optional[int] = 1
):
queries = await bot.global_playlist.pick_random(amount)
logger.info(f"Got queries for {player.guild_id} {queries=} {amount=}")
for query in queries:
result = await player.node.get_tracks(query)
if not result or not result["tracks"]:
continue
track = lavalink.models.AudioTrack(
result["tracks"][0], bot.user.id, recommended=False
)
player.add(requester=bot.user.id, track=track)
async def ensure_voice(
permissions: discord.Permissions,
player: "DefaultPlayer",
author: discord.Member,
voice_client: typing.Any,
channel_id: int,
):
"""This ensures that the bot and command author are in the same voicechannel."""
if not author.voice or not author.voice.channel:
raise Exception("Join a voicechannel first.")
if not voice_client:
if not permissions.connect or not permissions.speak: # Check user limit too?
raise Exception("I need to be able to join your channel and speak")
player.store("channel", channel_id)
await author.voice.channel.connect(cls=LavalinkVoiceClient)
else:
if voice_client.channel.id != author.voice.channel.id:
raise Exception("You need to be in my voicechannel")
__all__ = ("fill_player_queue", "get_player", "ensure_voice")