346 lines
11 KiB
Python
346 lines
11 KiB
Python
from discord.ext import commands
|
|
from discord.ext.commands.context import Context
|
|
|
|
import cogs.music.util as util
|
|
import cogs.music.queue as queue
|
|
import cogs.music.translate as translate
|
|
|
|
|
|
from cogs.music.help import music_help
|
|
|
|
import spotipy
|
|
from spotipy.oauth2 import SpotifyClientCredentials
|
|
|
|
|
|
# Fix this pls
|
|
|
|
import json
|
|
#from .. import config
|
|
# Read data from JSON file in ./data/config.json
|
|
def read_data():
|
|
with open("./data/config.json", "r") as file:
|
|
return json.load(file)
|
|
|
|
raise Exception("Could not load config data")
|
|
|
|
|
|
def get_spotify_creds():
|
|
data = read_data()
|
|
data = data.get("spotify")
|
|
|
|
SCID = data.get("SCID")
|
|
secret = data.get("SECRET")
|
|
|
|
return SCID, secret
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class music(commands.Cog):
|
|
def __init__(self, client):
|
|
self.client = client
|
|
|
|
self.name = "🎶 Music"
|
|
self.emoji = "🎶"
|
|
|
|
help_command = music_help()
|
|
help_command.cog = self
|
|
self.help_command = help_command
|
|
|
|
SCID, secret = get_spotify_creds()
|
|
# Authentication - without user
|
|
client_credentials_manager = SpotifyClientCredentials(client_id=SCID,
|
|
client_secret=secret)
|
|
|
|
self.sp = spotipy.Spotify(client_credentials_manager=client_credentials_manager)
|
|
|
|
queue.initialize_tables()
|
|
|
|
|
|
@commands.command(
|
|
help="Connects to your current voice channel",
|
|
aliases=['connect'])
|
|
async def join(self, ctx: Context):
|
|
await util.join_vc(ctx)
|
|
await ctx.message.add_reaction('👍')
|
|
|
|
|
|
@commands.command(
|
|
help="Leaves the voice chat if the bot is present",
|
|
aliases=['disconnect'])
|
|
async def leave(self, ctx: Context):
|
|
await util.leave_vc(ctx)
|
|
await ctx.message.add_reaction('👍')
|
|
|
|
@commands.command(
|
|
help="Queues a song into the bot",
|
|
aliases=['p'])
|
|
async def play(self, ctx: Context, *, url=None):
|
|
if url is None:
|
|
raise commands.CommandError("Must provide a link or search query")
|
|
elif ctx.guild is None:
|
|
raise commands.CommandError("Command must be issued in a server")
|
|
|
|
server = ctx.guild.id
|
|
|
|
await util.join_vc(ctx)
|
|
await ctx.message.add_reaction('👍')
|
|
|
|
msg = await ctx.send("Fetching song(s)...")
|
|
async with ctx.typing():
|
|
#TODO potentially save requests before getting stream link
|
|
# Grab video details such as title thumbnail duration
|
|
audio = await translate.main(url, self.sp)
|
|
|
|
await msg.delete()
|
|
|
|
if len(audio) == 0:
|
|
await ctx.message.add_reaction('🚫')
|
|
await ctx.send("Failed to find song!")
|
|
return
|
|
|
|
|
|
#TODO make sure user isn't queuing in dm for some stupid reason
|
|
|
|
# Setup first song's position
|
|
audio[0]['position'] = await queue.add_song(
|
|
server,
|
|
audio[0],
|
|
ctx.author.display_name)
|
|
|
|
# Add any other songs
|
|
for song in audio[1:]:
|
|
await queue.add_song(
|
|
server,
|
|
song,
|
|
ctx.author.display_name)
|
|
|
|
await util.queue_message(ctx, audio[0])
|
|
|
|
|
|
if await queue.is_server_playing(server):
|
|
return
|
|
|
|
await queue.update_server(server, True)
|
|
await queue.play(ctx)
|
|
|
|
|
|
@commands.command(
|
|
help="Display the current music queue",
|
|
aliases=['q', 'songs'])
|
|
async def queue(self, ctx: Context):
|
|
|
|
server = ctx.guild
|
|
|
|
# Perform usual checks
|
|
if server is None:
|
|
raise commands.CommandError("Command must be issued in a server")
|
|
|
|
|
|
# Grab all songs from this server
|
|
n, songs = await queue.grab_songs(server.id)
|
|
|
|
# Check once more
|
|
if len(songs) == 0 and await queue.is_server_playing(ctx.guild.id) == False:
|
|
await ctx.send("🚫 This server has no queue currently. Start the party by queuing up a song!")
|
|
return
|
|
|
|
# Display songs
|
|
await util.display_server_queue(ctx, songs, n)
|
|
|
|
|
|
@commands.command(
|
|
help="Skips the current song that is playing, can include number to skip more songs",
|
|
aliases=['s'])
|
|
async def skip(self, ctx: Context, n='1'):
|
|
server = ctx.guild
|
|
|
|
if server is None:
|
|
raise commands.CommandError("Command must be issued in a server")
|
|
|
|
if ctx.voice_client is None:
|
|
raise commands.CommandError("I'm not in a voice channel")
|
|
|
|
if not n.isdigit():
|
|
raise commands.CommandError("Please enter a number to skip")
|
|
n = int(n)
|
|
|
|
if n <= 0:
|
|
raise commands.CommandError("Please enter a positive number")
|
|
|
|
# Skip specificed number of songs
|
|
for _ in range(n-1):
|
|
await queue.pop(server.id, True)
|
|
|
|
# Safe to ignore error for now
|
|
ctx.voice_client.stop()
|
|
|
|
|
|
@commands.command(
|
|
help="Toggle loop mode: off -> song -> queue -> off",
|
|
aliases=['l', 'repeat'])
|
|
async def loop(self, ctx: Context, mode: str = None):
|
|
"""Toggle between loop modes or set a specific mode"""
|
|
server = ctx.guild
|
|
|
|
if server is None:
|
|
raise commands.CommandError("Command must be issued in a server")
|
|
|
|
current_mode = await queue.get_loop_mode(server.id)
|
|
|
|
# If no mode specified, cycle through modes
|
|
if mode is None:
|
|
if current_mode == 'off':
|
|
new_mode = 'song'
|
|
elif current_mode == 'song':
|
|
new_mode = 'queue'
|
|
else:
|
|
new_mode = 'off'
|
|
else:
|
|
# Set specific mode
|
|
mode = mode.lower()
|
|
if mode not in ['off', 'song', 'queue']:
|
|
raise commands.CommandError("Loop mode must be: off, song, or queue")
|
|
new_mode = mode
|
|
|
|
await queue.set_loop_mode(server.id, new_mode)
|
|
|
|
# Response messages
|
|
emojis = {'off': '⏹️', 'song': '🔂', 'queue': '🔁'}
|
|
messages = {
|
|
'off': 'Loop disabled',
|
|
'song': 'Looping current song 🔂',
|
|
'queue': 'Looping entire queue 🔁'
|
|
}
|
|
|
|
await ctx.send(f"{emojis[new_mode]} {messages[new_mode]}")
|
|
|
|
|
|
@commands.command(
|
|
help="Shuffle the queue randomly",
|
|
aliases=['mix', 'randomize'])
|
|
async def shuffle(self, ctx: Context):
|
|
"""Shuffle all songs in the queue"""
|
|
server = ctx.guild
|
|
|
|
if server is None:
|
|
raise commands.CommandError("Command must be issued in a server")
|
|
|
|
success = await queue.shuffle_queue(server.id)
|
|
|
|
if not success:
|
|
await ctx.send("🚫 Not enough songs in the queue to shuffle!")
|
|
else:
|
|
await ctx.message.add_reaction('🔀')
|
|
await ctx.send("🔀 Queue shuffled!")
|
|
|
|
|
|
@commands.command(
|
|
help="Set playback volume (0-200%)",
|
|
aliases=['vol', 'v'])
|
|
async def volume(self, ctx: Context, vol: str = None):
|
|
"""Set or display the current volume"""
|
|
server = ctx.guild
|
|
|
|
if server is None:
|
|
raise commands.CommandError("Command must be issued in a server")
|
|
|
|
if vol is None:
|
|
# Display current volume
|
|
current_vol = await queue.get_volume(server.id)
|
|
await ctx.send(f"🔊 Current volume: {current_vol}%")
|
|
return
|
|
|
|
# Set volume
|
|
if not vol.isdigit():
|
|
raise commands.CommandError("Volume must be a number (0-200)")
|
|
|
|
vol = int(vol)
|
|
if vol < 0 or vol > 200:
|
|
raise commands.CommandError("Volume must be between 0 and 200")
|
|
|
|
new_vol = await queue.set_volume(server.id, vol)
|
|
|
|
# Update the current playing song's volume if something is playing
|
|
if ctx.voice_client and ctx.voice_client.source:
|
|
ctx.voice_client.source.volume = new_vol / 100.0
|
|
|
|
# Pick an emoji based on volume
|
|
if new_vol == 0:
|
|
emoji = '🔇'
|
|
elif new_vol < 50:
|
|
emoji = '🔉'
|
|
elif new_vol < 100:
|
|
emoji = '🔊'
|
|
else:
|
|
emoji = '📢'
|
|
|
|
await ctx.send(f"{emoji} Volume set to {new_vol}%")
|
|
|
|
|
|
@commands.command(
|
|
help="Apply audio effects to playback",
|
|
aliases=['fx', 'filter'])
|
|
async def effect(self, ctx: Context, effect_name: str = None):
|
|
"""Apply or list audio effects"""
|
|
server = ctx.guild
|
|
|
|
if server is None:
|
|
raise commands.CommandError("Command must be issued in a server")
|
|
|
|
# If no effect specified, show current effect and list options
|
|
if effect_name is None:
|
|
current = await queue.get_effect(server.id)
|
|
emoji = queue.get_effect_emoji(current)
|
|
desc = queue.get_effect_description(current)
|
|
|
|
effects_list = '\n'.join([
|
|
f"`{e}` - {queue.get_effect_description(e)}"
|
|
for e in queue.list_all_effects()[:7] # Show first 7
|
|
])
|
|
more_effects = '\n'.join([
|
|
f"`{e}` - {queue.get_effect_description(e)}"
|
|
for e in queue.list_all_effects()[7:] # Show rest
|
|
])
|
|
|
|
await ctx.send(
|
|
f"{emoji} **Current effect:** {current} - {desc}\n\n"
|
|
f"**Available effects:**\n{effects_list}\n\n"
|
|
f"**More effects:**\n{more_effects}\n\n"
|
|
f"Use `=effect <name>` to apply an effect!"
|
|
)
|
|
return
|
|
|
|
# Set effect
|
|
effect_name = effect_name.lower()
|
|
|
|
if effect_name not in queue.list_all_effects():
|
|
raise commands.CommandError(
|
|
f"Unknown effect! Use `=effect` to see available effects."
|
|
)
|
|
|
|
await queue.set_effect(server.id, effect_name)
|
|
|
|
emoji = queue.get_effect_emoji(effect_name)
|
|
desc = queue.get_effect_description(effect_name)
|
|
|
|
# Special warning for earrape
|
|
if effect_name == 'earrape':
|
|
await ctx.send(
|
|
f"⚠️ **EARRAPE MODE ACTIVATED** ⚠️\n"
|
|
f"RIP your eardrums. Effect will apply to next song.\n"
|
|
f"Use `=effect none` to disable."
|
|
)
|
|
else:
|
|
await ctx.send(
|
|
f"{emoji} Effect set to **{effect_name}**\n"
|
|
f"{desc}\n"
|
|
f"Effect will apply to next song!"
|
|
)
|
|
|
|
# If something is currently playing, notify about skip
|
|
if ctx.voice_client and ctx.voice_client.is_playing():
|
|
await ctx.send("💡 Tip: Use `=skip` to apply effect to current song immediately!")
|