-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
242 lines (179 loc) · 7.41 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
import asyncio
import logging
import os
import discord
import yt_dlp
from discord.ext import commands
from dotenv import load_dotenv
# Suppress noise about console usage from errors
yt_dlp.utils.bug_reports_message = lambda: ""
ytdl_format_options = {
"format": "bestaudio/best[abr<=75]", # TODO: Determine if the abr selector works
"outtmpl": "%(extractor)s-%(id)s-%(title)s.%(ext)s",
"restrictfilenames": True,
"noplaylist": True,
"nocheckcertificate": True,
"ignoreerrors": False,
"logtostderr": False,
"quiet": True,
"no_warnings": True,
"default_search": "auto",
"source_address": "0.0.0.0", # bind to ipv4 since ipv6 addresses cause issues sometimes
}
ffmpeg_options = {
"options": "-vn", # TODO: does -bufsize 2 work?
"before_options": "-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5",
}
ytdl = yt_dlp.YoutubeDL(ytdl_format_options)
default_presence_activity = discord.Activity(type=discord.ActivityType.listening, name="!play")
class YTDLSource(discord.PCMVolumeTransformer):
def __init__(self, source, *, data, volume=0.5):
super().__init__(source, volume)
self.data = data
self.title = data.get("title")
self.url = data.get("url")
self.artist = data.get("artist", None)
@classmethod
async def from_url(cls, url, *, loop=None, stream=False):
loop = loop or asyncio.get_event_loop()
data = await loop.run_in_executor(None, lambda: ytdl.extract_info(url, download=not stream))
if "entries" in data:
# take first item from a playlist
data = data["entries"][0]
filename = data["url"] if stream else ytdl.prepare_filename(data)
return cls(discord.FFmpegPCMAudio(filename, **ffmpeg_options), data=data)
class Music(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.queue = []
# @commands.command()
# async def join(self, ctx, *, channel: discord.VoiceChannel = None):
# """Joins a voice channel -- the one you're in, or the one you specify"""
# if ctx.voice_client is not None:
# return await ctx.voice_client.move_to(channel)
# if channel is None:
# if ctx.author.voice:
# await ctx.author.voice.channel.connect()
# else:
# await ctx.send("You are not connected to a voice channel.")
# raise commands.CommandError("Author not connected to a voice channel.")
# else:
# await channel.connect()
@commands.command()
async def play(self, ctx, *, url):
"""Streams music from a url or adds to the queue"""
if ctx.voice_client.is_playing():
self.queue.append(url)
print(f"Added song to queue as we're already playing something, queue={self.queue}")
await ctx.send("Added to queue")
else:
async with ctx.typing():
await self.play_song(ctx, url)
@commands.command(aliases=["next"])
async def skip(self, ctx):
"""Skips the current song"""
print("Skipping by user request")
# Note: this causes the `after` function to be invoked, which is on_finish_streaming
ctx.voice_client.stop()
@commands.command()
async def clear(self, ctx):
"""Clear the queue"""
print("Clearing queue by user request")
self.clear_queue()
await ctx.send("Cleared queue")
def on_finish_streaming(self, ctx, e=None):
"""When we're done with a song, we play the next one, if available"""
if e:
print(f"Player error: {e}")
print(f"Finished streaming, remaining queue: {self.queue}")
if not ctx.voice_client:
print("No more voice client, skipping playing the next song")
return
if self.queue:
url = self.queue.pop(0)
# Execute an async function from this synchronous callback:
asyncio.run_coroutine_threadsafe(self.play_song(ctx, url), self.bot.loop)
else:
# Clear our activity if there is nothing else to play
asyncio.run_coroutine_threadsafe(
self.bot.change_presence(activity=default_presence_activity),
self.bot.loop
)
async def play_song(self, ctx, url):
print(f"Playing song: {url}")
player = await YTDLSource.from_url(url, loop=self.bot.loop, stream=True)
ctx.voice_client.play(player, after=lambda e: self.on_finish_streaming(ctx, e))
title = (
f"{player.title}" if not player.artist else
f"{player.title} by {player.artist}"
)
sanitized_title = discord.utils.escape_markdown(title)
sanitized_title = discord.utils.escape_mentions(sanitized_title)
# TODO: Sanitize links in youtube titles?
# Disable mentions. Even if we escape them, there could still be some input that
# triggers an @everyone due to a bug.
await ctx.send(f"Now playing: {sanitized_title}",
allowed_mentions=discord.AllowedMentions.none())
# We can use the unsanitized title in presence, no pings or markdown syntax work here
await ctx.bot.change_presence(
activity=discord.Activity(type=discord.ActivityType.listening, name=title)
)
def clear_queue(self):
self.queue.clear()
@commands.command(aliases=["leave", "quit", "exit", "end"])
async def stop(self, ctx):
"""Stops and disconnects the bot from voice"""
await self.clear(ctx)
if ctx.voice_client:
await ctx.voice_client.disconnect()
await ctx.bot.change_presence(
activity=default_presence_activity
)
@play.before_invoke
async def ensure_voice(self, ctx):
if ctx.voice_client is None:
if ctx.author.voice:
await ctx.author.voice.channel.connect()
else:
await ctx.send("You are not connected to a voice channel.")
raise commands.CommandError("Author not connected to a voice channel.")
@commands.Cog.listener(name="on_command")
async def log_command(self, ctx):
print(f"Command issued: {ctx.guild.name} > {ctx.author}"
+ f" > {ctx.command} [{ctx.message.content}]")
intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(
command_prefix=commands.when_mentioned_or("!"),
description="Simple music bot with queueing",
intents=intents,
)
@bot.event
async def on_ready():
print(f"Logged in as {bot.user} (ID: {bot.user.id})")
print("------")
await bot.change_presence(
activity=default_presence_activity
)
@bot.event
async def on_message(message: discord.Message):
# Filter out DMs from command processing
if message.guild:
await bot.process_commands(message)
@bot.event
async def on_voice_state_update(member, before, after):
if not bot.voice_clients:
return
print("Voice state update")
for client in bot.voice_clients:
if len(client.channel.members) <= 1:
print("Left alone in voice channel, leaving")
bot.get_cog("Music").clear_queue()
await client.disconnect()
async def main():
load_dotenv()
async with bot:
await bot.add_cog(Music(bot))
await bot.start(os.environ.get("DISCORD_TOKEN"))
discord.utils.setup_logging(level=logging.INFO, root=False)
asyncio.run(main())