doot-doot/cogs/youtube.py
# This cog handles support of YT links and playthrough of them using lavalink
# While this is not directly music bot, it implements feature request from Github.
# Code uses config for easy selfhosting setup, while this makes .add_node look disgusting it makes it easier for end user to host on their own
# imports
import math
import re
import json
import discord
import lavalink
from discord.ext import commands
url_regex = re.compile('https?:\\/\\/(?:www\\.)?.+')
# config initiation
def getConfig(path):
configFile = open(path, "r")
return json.loads(configFile.read())
config_ll = getConfig("config.json")
# cog initiation
class Youtube(commands.Cog):
def __init__(self, bot):
self.bot = bot
if not hasattr(bot, 'lavalink'): # This ensures the client isn't overwritten during cog reloads.
bot.lavalink = lavalink.Client(bot.user.id)
bot.lavalink.add_node(config_ll['ll_ip'], config_ll['ll_port'], config_ll['ll_password'], config_ll['ll_region'], config_ll['ll_node_name']) # Host, Port, Password, Region, Name
bot.add_listener(bot.lavalink.voice_update_handler, 'on_socket_response')
bot.lavalink.add_event_hook(self.track_hook)
def cog_unload(self):
self.bot.lavalink._event_hooks.clear()
async def cog_before_invoke(self, ctx):
check_if_in_guild = ctx.guild is not None
# Check if user sent command in Guild
if check_if_in_guild:
await self.ensure_voice(ctx)
# Verify that bot and end user share mutal voice channel.
return check_if_in_guild
async def cog_command_error(self, ctx, error):
if isinstance(error, commands.CommandInvokeError):
await ctx.send(error.original)
# informs user about any errors such as "need to be in voice cahannel"
async def track_hook(self, event):
if isinstance(event, lavalink.events.QueueEndEvent):
guild_id = int(event.player.guild_id)
await self.connect_to(guild_id, None)
# Disconnect from the channel after queue finishes.
async def connect_to(self, guild_id: int, channel_id: str):
""" Connects to the given voicechannel ID. A channel_id of `None` means disconnect. """
ws = self.bot._connection._get_websocket(guild_id)
await ws.voice_state(str(guild_id), channel_id)
# play YT video from either URL or search querry
@commands.command(aliases=['p'])
async def play(self, ctx, *, query: str):
""" Searches and plays a song from youtube"""
player = self.bot.lavalink.players.get(ctx.guild.id)
query = query.strip('<>')
if not url_regex.match(query):
query = f'ytsearch:{query}'
results = await player.node.get_tracks(query)
if not results or not results['tracks']:
return await ctx.send('Nothing found!')
embed = discord.Embed(color=discord.Color.blurple())
if results['loadType'] == 'PLAYLIST_LOADED':
tracks = results['tracks']
for track in tracks:
player.add(requester=ctx.author.id, track=track)
embed.title = 'Playlist Enqueued!'
embed.description = f'{results["playlistInfo"]["name"]} - {len(tracks)} tracks'
else:
track = results['tracks'][0]
embed.title = 'Track Enqueued'
embed.description = f'[{track["info"]["title"]}]({track["info"]["uri"]})'
player.add(requester=ctx.author.id, track=track)
await ctx.send(embed=embed)
if not player.is_playing:
await player.play()
@commands.command(aliases=['np', 'n', 'playing'])
async def now(self, ctx):
""" Shows currently playing song. """
player = self.bot.lavalink.players.get(ctx.guild.id)
if not player.current:
return await ctx.send('Nothing playing.')
position = lavalink.utils.format_time(player.position)
if player.current.stream:
duration = '🔴 LIVE'
else:
duration = lavalink.utils.format_time(player.current.duration)
song = f'**[{player.current.title}]({player.current.uri})**\n({position}/{duration})'
embed = discord.Embed(color=discord.Color.blurple(),
title='Now Playing', description=song)
await ctx.send(embed=embed)
# allows to manually disconnect bot from channel clearing it queue
@commands.command(aliases=['dc'])
async def disconnect(self, ctx):
""" Disconnects the player from the voice channel and clears its queue. """
player = self.bot.lavalink.players.get(ctx.guild.id)
if not player.is_connected:
return await ctx.send('Not connected.')
if not ctx.author.voice or (player.is_connected and ctx.author.voice.channel.id != int(player.channel_id)):
return await ctx.send('You\'re not in my voicechannel!')
player.queue.clear()
await player.stop()
await self.connect_to(ctx.guild.id, None)
await ctx.send('*⃣ | Disconnected.')
async def ensure_voice(self, ctx):
""" This check ensures that the bot and command author are in the same voicechannel. """
player = self.bot.lavalink.players.create(ctx.guild.id, endpoint=str(ctx.guild.region))
# Create returns a player if one exists, otherwise creates.
should_connect = ctx.command.name in ('play') # Add commands that require joining voice to work.
if not ctx.author.voice or not ctx.author.voice.channel:
raise commands.CommandInvokeError('You need to be in Voice Channel to use this command.')
if not player.is_connected:
if not should_connect:
raise commands.CommandInvokeError('Not connected.')
permissions = ctx.author.voice.channel.permissions_for(ctx.me)
if not permissions.connect or not permissions.speak: # Check user limit too?
raise commands.CommandInvokeError('I need the `CONNECT` and `SPEAK` permissions.')
player.store('channel', ctx.channel.id)
await self.connect_to(ctx.guild.id, str(ctx.author.voice.channel.id))
else:
if int(player.channel_id) != ctx.author.voice.channel.id:
raise commands.CommandInvokeError('You need to be in my voice channel.')
def setup(bot):
bot.add_cog(Youtube(bot))