class searchError(Exception):
pass
class VoiceStateError(Exception):
pass
class Song(disnake.PCMVolumeTransformer):
def search_tracks(ctx, query):
stats = True
while stats:
try:
searcher = client.search(str(query))
if not searcher.tracks: raise searchError("Not searched tracks")
# print(searcher.tracks['results'][0])
track_object = searcher.tracks['results'][0]
stats = False
text = None
lyrics = track_object.get_supplement()
if lyrics.lyrics: text = lyrics.lyrics.full_lyrics
with sqlite3.connect("database.db") as db:
cursor = db.cursor()
cursor.execute("INSERT INTO songs(name, requester, author, id, albumid, lyrics, guild) VALUES(?, ?, ?, ?, ?, ?, ?)", (track_object['title'],ctx.author.id,track_object['artists'][0]['name'],track_object['id'],track_object['albums'][0]["id"],text,ctx.guild.id,))
print("Success queued")
return {
"id": track_object["id"],
"title": track_object["title"],
"artist": track_object["artists"][0]["name"],
"albumid": track_object["albums"][0]["id"],
"lyrics": text
}
except NetworkError:
pass
def download_track(ctx, track):
stats = True
while stats:
try:
client.tracks(track['id'])[0].download(f"songs/{ctx.guild.id}.mp3")
stats = False
except NetworkError:
pass
# def check_playing(ctx):
# voice_state = ctx.guild.voice_client
# if not voice_state: return raise VoiceStateError("Not voice_state")
# if not voice_state.is_playing(): return raise VoiceStateError("Not playing")
async def join_channel(ctx):
voice = ctx.author.voice
if not voice: return "notChannel"
if voice:
await voice.channel.connect()
return None
async def leave_channel(ctx):
voice_state = ctx.guild.voice_client
if not voice_state:
return "notState"
if voice_state.is_connected():
await voice_state.disconnect()
return None
else:
return "notState"
async def stop(ctx):
voice_state = ctx.guild.voice_client
if not voice_state:
return "notState"
if voice_state.is_connected():
#await voice_state.stop()
await Song.leave_channel(ctx)
return None
else:
return "notState"
async def my_after(ctx):
#os.remove(f"songs/{ctx.guild.id}.mp3")
voice_state = ctx.guild.voice_client
if not voice_state.is_playing:
names = []
requesters = []
artists = []
ids = []
albumids = []
positions = []
guilds = []
lyrics = []
with sqlite3.connect("database.db") as db:
c = db.cursor()
for n, r, a, i, ai, ly, g, p in c.execute("SELECT * FROM songs WHERE guild = ?", (ctx.guild.id,)):
positions.append(p)
names.append(n)
requesters.append(r)
artists.append(a)
guilds.append(g)
ids.append(i)
albumids.append(ai)
if n and not ly:
lyrics.append(None)
else:
lyrics.append(ly)
c.execute("DELETE FROM songs WHERE guild = ? AND position = ?", (guilds[0], positions[0],))
print("Success removed")
if positions[1]:
Song.download_track(ctx, {"id": ids[1]})
await Song.play(ctx)
async def play_next(ctx):
names = []
requesters = []
artists = []
ids = []
albumids = []
positions = []
guilds = []
lyrics = []
with sqlite3.connect("database.db") as db:
c = db.cursor()
for n, r, a, i, ai, ly, g, p in c.execute("SELECT * FROM songs WHERE guild = ?", (ctx.guild.id,)):
positions.append(p)
names.append(n)
requesters.append(r)
artists.append(a)
guilds.append(g)
ids.append(i)
albumids.append(ai)
if n and not ly:
lyrics.append(None)
else:
lyrics.append(ly)
c.execute("DELETE FROM songs WHERE guild = ? AND position = ?", (guilds[0], positions[0],))
print("Success removed")
if positions[1]:
Song.download_track(ctx, {"id": ids[1], "title": names[1], "artist": artists[1], "albumid": albumids[1], "lyrics": lyrics[1]})
await Song.play(ctx)
async def play(ctx):
voice_state = ctx.guild.voice_client
if not voice_state:
error = await Song.join_channel(ctx)
if error:
return error
voice_state = ctx.guild.voice_client
if voice_state.is_playing():
return "alreadyPlay"
voice_state.play(disnake.FFmpegPCMAudio(f"songs/{ctx.guild.id}.mp3"), after=await Song.my_after(ctx))
return None
def add_queue(ctx, track):
with sqlite3.connect("database.db") as db:
cursor = db.cursor()
cursor.execute("INSERT INTO songs(name, requester, author, id, albumid, lyrics, guild) VALUES(?, ?, ?, ?, ?, ?, ?)", (track['title'],ctx.author.id,track['artist'],track['id'],track['albumid'],track['lyrics'],ctx.guild.id,))
print("Success queued")
async def skip(ctx):
voice_state = ctx.guild.voice_client
if voice_state.is_playing():
try:
await voice_state.stop()
except:
pass
await Song.play_next(ctx)
return None
else:
return "notState"
def lyrics(ctx):
names = []
requesters = []
artists = []
ids = []
albumids = []
lyrics = []
with sqlite3.connect("database.db") as db:
c = db.cursor()
for n, r, a, i, ai, ly, g, p in c.execute("SELECT * FROM songs WHERE guild = ?", (ctx.guild.id,)):
names.append(n)
requesters.append(r)
artists.append(a)
ids.append(i)
albumids.append(ai)
if n and not ly:
lyrics.append(None)
else:
lyrics.append(ly)
return {
"name": names[0],
"lyrics": lyrics[0]
}
def now_playing(ctx):
names = []
requesters = []
artists = []
ids = []
albumids = []
positions = []
with sqlite3.connect("database.db") as db:
c = db.cursor()
for n, r, a, i, ai, ly, g, p in c.execute("SELECT * FROM songs WHERE guild = ?", (ctx.guild.id,)):
names.append(n)
requesters.append(r)
artists.append(a)
ids.append(i)
albumids.append(ai)
positions.append(p)
return {
"name": names[0],
"requester": requesters[0],
"artist": artists[0],
"pos": positions[0],
"uri": f"https://music.yandex.ru/album/{ids[0]}/track/{albumids[0]}"
}
