[Discord Bot Python] 디스코드 유튜브 음악 재생 봇 - 플레이리스트

2023.05.18 - [Discord/Discord Bot Python] - Discord Bot 만들기 - 유튜브 음악 재생 봇\

Discord Bot 만들기 - 유튜브 음악 재생 봇

이 페이지에서 설명하는 코드는 위 "디스코드 유튜브 음악 재생 봇" 페이지를 바탕으로 작성되었습니다.

 

전체 코드에 대한 기본 설명은 위 페이지를 참조해주시기 바랍니다.

 

---

이번 포스트에서는 기존 유튜브 음악 재생 봇에 플레이리스트 기능을 추가하여 사용자가 play 명령어를 입력할 때마다 음악 대기열(플레이리스트)을 생성하는 기능을 작성해보록 한다.

 

설치에 필요한 라이브러리

import asyncio
 
import discord
import yt_dlp as youtube_dl
 
from discord.ext import commands
from dico_token import token

 

기본적으로 사용되는 라이브러리는 위와 같다.

 

기존 유튜브 음악 재생 봇 포스트를 작성하면서 dico_token 모듈에 대한 질문이 많은데, 해당 모듈은 시크릿 키(토큰)를 따로 파일로 저장하여 사용한 것으로 자세한 내용은 아래에서 한번 더 설명하도록 하겠다.

 

 

코드 설명 

기존에 작성했던 코드에 기능만 추가한 것이므로 Music 클래스 부분에 추가한 기능 위주로 설명한다.

 

# 음악 재생 클래스. 커맨드 포함.
class Music(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.queue = []
        self.current_player = None
        self.max_queue_size = 10  # 대기열 최대 길이 설정

 

 

기본적으로 단순 플레이리스트는 큐(Queue)를 사용한다.

 

먼저 추가한 곡이 가장 먼저 재생되어야 하기 때문에 선입선출 구조를 가지게 되므로, 음악 대기열을 생성하기 위한 queue라는 배열을 하나 만들어 준다.

 

current_player = None

플레이리스트에서 현재 재생중인 곡과 다음에 재생할 곡을 구별해 주기 위해 미리 초기화.

 

max_queue_size = 10

이 부분은 플레이리스트의 최대 사이즈를 결정하기 위해 작성.

초기값을 10으로 설정했는데, 플레이리스트에 더 많은 곡을 추가하고 싶으면 뒤에 정수를 수정해주기만 하면 된다.

 

 

async def play_next(self, ctx):
        """다음 곡을 자동으로 재생합니다"""
        if len(self.queue) > 0:  # 대기열이 비어있지 않은지 확인
            await self.skip(ctx)
    
@commands.command(aliases=["다음"])
async def skip(self, ctx):
    """대기열에서 다음 곡을 재생합니다"""
    if not ctx.voice_client:
        await ctx.send("봇이 음성 채널에 연결되어 있지 않습니다.")
        return

    if not self.queue:  # 대기열이 비어있는지 확인
        await ctx.send("다음 재생할 곡이 대기열에 없습니다.\n음악을 계속 재생하시려면 음악을 추가해주세요.")
        return

    # 현재 재생중인 음악 중지
    if ctx.voice_client.is_playing():
        ctx.voice_client.stop()

    try: 
        current_url, current_title = self.queue.pop(0)  # 바로 대기열에서 제거

        async with ctx.typing():
            self.current_player = await YTDLSource.from_url(current_url, loop=self.bot.loop, stream=True)

            ctx.voice_client.play(self.current_player, after=lambda _: asyncio.run_coroutine_threadsafe(self.play_next(ctx), self.bot.loop))

            await ctx.send(f'지금 재생 중: {self.current_player.title}')

    except Exception as e:
        await ctx.send(f"재생 중 오류가 발생했습니다: {str(e)}")
        print(f"재생 오류: {e}")
        # 오류 발생 시 대기열에 기존 재생 곡 추가
        if current_url and current_title:
            self.queue.insert(0, (current_url, current_title))

 

skip

(명령어: !skip, !다음)

해당 부분은 다음 곡을 재생할 수 있도록 처리해주는 코드.

!skip 또는 !다음을 입력하면 현재 재생 중인 곡을 중단하고 재생중인 곡을 알려주며, 대기열에 있는 다음 곡을 재생한다.

 

만약 재생 중 오류가 발생하면 기존 재생중이던 곡을 대기열의 첫번째에 추가하여 플레이리스트를 복구한다.

 

 

 

@commands.command(aliases=["재생"])
async def play(self, ctx, *, url):
    """URL에서 음악을 재생하고 대기열에 추가합니다"""

    try:
        if not ctx.author.voice:
            await ctx.send("음성 채널에 먼저 입장해주세요.")
            return

        if not ctx.voice_client:
            await ctx.author.voice.channel.connect()

        if len(self.queue) >= self.max_queue_size:
            await ctx.send(f"대기열이 가득 찼습니다. 최대 {self.max_queue_size}곡까지만 추가할 수 있습니다.")
            return


        # 추가하려는 곡의 정보를 미리 가져옴
        async with ctx.typing():
            player = await YTDLSource.from_url(url, loop=self.bot.loop, stream=True)
            self.queue.append((url, player.title))
            if not ctx.voice_client.is_playing():
                await self.skip(ctx)
            else:
                queue_info = f'대기열에 "{player.title}" 노래가 추가되었습니다.\n현재 대기열 ({len(self.queue)}곡):\n'
                for i, (_, title) in enumerate(self.queue, 1):
                    queue_info += f"{i}. {title}\n"
                await ctx.send(queue_info)

    except Exception as e:
        await ctx.send(f"음악을 추가하는 중 오류가 발생했습니다: {str(e)}")
        print(f"재생 오류: {e}")

 

기존 재생 코드에서 대기열(플레이리스트)와 관련한 동작 코드를 추가해주어야 한다.

 

만약 기존에 설정한 대기열 최대 크기를 초과하면 대기열이 가득 찼다는 메시지와 함께 음악을 추가하지 못하도록 설정.

 

!play '유튜브 링크' 입력 시, 자동으로 대기열에 추가하도록 설정하였으며, 추가한 곡의 정보와 대기열에 현재 몇 곡이 추가되어있는지 알려준다.

 

 

@commands.command(aliases=["q","플레이리스트","대기열"])
async def queue(self, ctx):
    """현재 대기열을 보여줍니다"""
    if len(self.queue) == 0:
        await ctx.send("대기열이 비어있습니다.")
        return

    queue_list = f"현재 대기열 ({len(self.queue)}/{self.max_queue_size}곡):\n"
    for i, (_, title) in enumerate(self.queue, 1):
        queue_list += f"{i}. {title}\n"
    await ctx.send(queue_list)

 

queue

(명령어: !queue, !q, !플레이리스트, !대기열)

 !q, !플레이리스트, !대기열 명령어 모두 동작하므로 성향에 맞게 골라쓰면 된다.

해당 명령어 입력 시 대기열에 있는 곡 목록을 출력한다.

 

 

@commands.command()
async def now(self, ctx):
    """현재 재생중인 음악의 제목을 보여줍니다"""
    if not ctx.voice_client or not ctx.voice_client.is_playing():
        await ctx.send("현재 재생 중인 음악이 없습니다.")
        return

    if self.current_player:
        await ctx.send(f"현재 재생 중: {self.current_player.title}")
    else:
        await ctx.send("현재 재생 중인 음악 정보를 가져올 수 없습니다.")

@commands.command(aliases=["삭제", "제거"])
async def remove(self, ctx, index: int):
    """대기열에서 특정 곡을 삭제합니다"""
    if len(self.queue) == 0:
        await ctx.send("대기열이 비어있습니다.")
        return

    if not 1 <= index <= len(self.queue):
        await ctx.send(f"올바른 번호를 입력해주세요. (1 ~ 대기열 길이`({len(self.queue)})`)")
        return

    removed_title = self.queue.pop(index-1)
    await ctx.send(f"대기열에서 {index}번 곡 '{removed_title}'이(가) 제거되었습니다.")

 

now

(명령어: !now)

 

현재 재생중인 음악의 제목 정보를 출력한다.

url 정보도 queue에 추가했기 때문에 가져올 수 있지만 여기서 따로 추가하진 않았다.(필요하면 가져오면 된다.)

 

 

remove

(명령어: !remove, !삭제, !제거)

 

대기열에서 특정 인덱스에 해당하는 곡을 삭제한다.

만약 대기열에 있는 여러 곡 중 삭제하고 싶은 곡이 있다면 !remove "삭제할 곡 번호" 명령어를 입력하면 된다.

ex) !remove 1 -> 1번 곡을 삭제한다

 

 

music 부분 전체 코드

class Music(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.queue = []
        self.current_player = None
        self.max_queue_size = 10  # 대기열 최대 길이 설정

    @commands.command(aliases=["입장"])
    async def join(self, ctx):
        """음성 채널에 입장합니다"""
        if not ctx.author.voice:
            await ctx.send("음성 채널에 먼저 입장해주세요.")
            return
            
        channel = ctx.author.voice.channel
        
        if ctx.voice_client is not None:
            return await ctx.voice_client.move_to(channel)

        await channel.connect()
        await ctx.send(f"{channel} 채널에 입장했습니다.")
    
    async def play_next(self, ctx):
        """다음 곡을 자동으로 재생합니다"""
        if len(self.queue) > 0:  # 대기열이 비어있지 않은지 확인
            await self.skip(ctx)
    
    @commands.command(aliases=["다음"])
    async def skip(self, ctx):
        """대기열에서 다음 곡을 재생합니다"""
        if not ctx.voice_client:
            await ctx.send("봇이 음성 채널에 연결되어 있지 않습니다.")
            return
            
        if not self.queue:  # 대기열이 비어있는지 확인
            await ctx.send("다음 재생할 곡이 대기열에 없습니다.\n음악을 계속 재생하시려면 음악을 추가해주세요.")
            return
        
        # 현재 재생중인 음악 중지
        if ctx.voice_client.is_playing():
            ctx.voice_client.stop()
        
        try: 
            current_url, current_title = self.queue.pop(0)  # 바로 대기열에서 제거
            
            async with ctx.typing():
                self.current_player = await YTDLSource.from_url(current_url, loop=self.bot.loop, stream=True)
                
                ctx.voice_client.play(self.current_player, after=lambda _: asyncio.run_coroutine_threadsafe(self.play_next(ctx), self.bot.loop))
                
                await ctx.send(f'지금 재생 중: {self.current_player.title}')
                    
        except Exception as e:
            await ctx.send(f"재생 중 오류가 발생했습니다: {str(e)}")
            print(f"재생 오류: {e}")
            # 오류 발생 시 대기열에 기존 재생 곡 추가
            if current_url and current_title:
                self.queue.insert(0, (current_url, current_title))
   
    @commands.command(aliases=["재생"])
    async def play(self, ctx, *, url):
        """URL에서 음악을 재생하고 대기열에 추가합니다"""
        
        try:
            if not ctx.author.voice:
                await ctx.send("음성 채널에 먼저 입장해주세요.")
                return

            if not ctx.voice_client:
                await ctx.author.voice.channel.connect()
                
            if len(self.queue) >= self.max_queue_size:
                await ctx.send(f"대기열이 가득 찼습니다. 최대 {self.max_queue_size}곡까지만 추가할 수 있습니다.")
                return
            

            # 추가하려는 곡의 정보를 미리 가져옴
            async with ctx.typing():
                player = await YTDLSource.from_url(url, loop=self.bot.loop, stream=True)
                self.queue.append((url, player.title))
                if not ctx.voice_client.is_playing():
                    await self.skip(ctx)
                else:
                    queue_info = f'대기열에 "{player.title}" 노래가 추가되었습니다.\n현재 대기열 ({len(self.queue)}곡):\n'
                    for i, (_, title) in enumerate(self.queue, 1):
                        queue_info += f"{i}. {title}\n"
                    await ctx.send(queue_info)
                    
        except Exception as e:
            await ctx.send(f"음악을 추가하는 중 오류가 발생했습니다: {str(e)}")
            print(f"재생 오류: {e}")

    @commands.command(aliases=["음량","볼륨","소리"])
    async def volume(self, ctx, volume: int):
        """플레이어의 볼륨을 조절합니다"""

        if ctx.voice_client is None:
            return await ctx.send("음성 채널에 연결되어 있지 않습니다.")

        if not 0 <= volume <= 100:
            return await ctx.send("볼륨은 0에서 100 사이의 값이어야 합니다.")

        ctx.voice_client.source.volume = volume / 100
        await ctx.send(f"볼륨이 {volume}%로 변경되었습니다")

    @commands.command()
    async def stop(self, ctx):
        """재생을 멈추고 음성 채널에서 나갑니다"""
        if not ctx.voice_client:
            await ctx.send("봇이 음성 채널에 연결되어 있지 않습니다.")
            return
            
        self.queue.clear()
        if ctx.voice_client.is_playing():
            ctx.voice_client.stop()
        await ctx.voice_client.disconnect()
        await ctx.send("재생을 멈추고 채널에서 나갔습니다.")
        
    @commands.command()
    async def pause(self, ctx):
        """음악을 일시정지합니다"""
        if not ctx.voice_client:
            await ctx.send("봇이 음성 채널에 연결되어 있지 않습니다.")
            return

        if ctx.voice_client.is_paused() or not ctx.voice_client.is_playing():
            await ctx.send("음악이 이미 일시 정지 중이거나 재생 중이지 않습니다.")
            return
            
        ctx.voice_client.pause()
        await ctx.send("음악이 일시정지되었습니다.")
            
    @commands.command()
    async def resume(self, ctx):
        """일시정지된 음악을 다시 재생합니다"""
        if not ctx.voice_client:
            await ctx.send("봇이 음성 채널에 연결되어 있지 않습니다.")
            return

        if ctx.voice_client.is_playing() or not ctx.voice_client.is_paused():
            await ctx.send("음악이 이미 재생 중이거나 재생할 음악이 존재하지 않습니다.")
            return
            
        ctx.voice_client.resume()
        await ctx.send("음악이 다시 재생됩니다.")

    @commands.command(aliases=["q","플레이리스트","대기열"])
    async def queue(self, ctx):
        """현재 대기열을 보여줍니다"""
        if len(self.queue) == 0:
            await ctx.send("대기열이 비어있습니다.")
            return
            
        queue_list = f"현재 대기열 ({len(self.queue)}/{self.max_queue_size}곡):\n"
        for i, (_, title) in enumerate(self.queue, 1):
            queue_list += f"{i}. {title}\n"
        await ctx.send(queue_list)

    @commands.command()
    async def now(self, ctx):
        """현재 재생중인 음악의 제목을 보여줍니다"""
        if not ctx.voice_client or not ctx.voice_client.is_playing():
            await ctx.send("현재 재생 중인 음악이 없습니다.")
            return
            
        if self.current_player:
            await ctx.send(f"현재 재생 중: {self.current_player.title}")
        else:
            await ctx.send("현재 재생 중인 음악 정보를 가져올 수 없습니다.")

    @commands.command(aliases=["삭제", "제거"])
    async def remove(self, ctx, index: int):
        """대기열에서 특정 곡을 삭제합니다"""
        if len(self.queue) == 0:
            await ctx.send("대기열이 비어있습니다.")
            return
            
        if not 1 <= index <= len(self.queue):
            await ctx.send(f"올바른 번호를 입력해주세요. (1 ~ 대기열 길이`({len(self.queue)})`)")
            return
            
        _, removed_title = self.queue.pop(index-1)
        await ctx.send(f"대기열에서 {index}번 곡 '{removed_title}'이(가) 제거되었습니다.")

    @play.before_invoke
    @skip.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("음성 채널에 먼저 입장해주세요.")
                raise commands.CommandError("사용자가 음성 채널에 연결되어 있지 않습니다.")

 

해당 코드는 Music 클래스의 코드만 작성했기 때문에 전체 동작 코드는

상단에 링크한 "유튜브 음악 재생 봇" 페이지를 참조하면 된다.

 

전체 동작 코드에 여기서 작성한 Music 클래스를 붙여넣거나 사용한다.

 

동작 방식은 기존 작성했던 명령어와 동일하다.

 

명령어: !play '유튜브 링크' 

ex) !play https://www.youtube.com/~

 

 

실행 결과 이미지

 

대기열이 비어있을 경우 비어있다는 메시지, 대기열이 존재할 경우 현재 대기열 곡 목록 수와 제목, 아티스트 정보를 표시한다.

 

 

 

제거하고 싶은 곡의 번호(인덱스)를 입력하면 대기열에서 삭제된다.

 

 

 

!skip 명령어 입력 시 현재 재생중인 곡을 종료하고 바로 다음 곡을 재생한다.

 

---

 

오류 Q&A

** 해당 코드에서 사용한 Python 버전은 3.12.5 입니다.

Q: 봇을 실행시키고 명령어는 동작하는데, 봇이 아무 반응이 없어요(소리가 나오지 않아요)

A: 
1. 디스코드 개발자 포탈의 OAuth2 설정에서 bot의 권한이 Admin이거나, bot 권한 중 음성과 text 권한을 허용해주셔야 합니다.
2. 디스코드 개발자 포탈의 Privileged Gateway Intents 설정의 3가지 권한이 모두 허용되어 있어야 합니다.
3. 디스코드의 사용자 설정 - 음성 및 오디오 - 출력 장치의 설정이 사용자가 설정한 스피커로 설정되어 있는지 확인해주셔야 합니다.
4. IDE(VSCode) 터미널에서 오류 메시지가 출력되지는 않았는지 확인하셔야 합니다.

5. 디스코드 봇이 들어오고 초록색으로 표시되는 경우, !volume 50 명령어를 입력하셔서 볼륨 설정이 잘못되지 않았는지 확인하셔야 합니다.

 

위 방법 모두 동작하지 않는다면 python 버전을 최신(3.12) 버전으로 재설치 후 라이브러리를 pip install 해주시기 바랍니다.

pip install discord[voice]
pip install discord.py
pip install yt-dlp
pip install PyNaCl

 

파이썬 재설치 후 ffmpeg를 파이썬이 설치된 경로에 넣어주시기 바랍니다.

 

참조: VSCode에서 파이썬 버전 설정 방법

1. VSCode 우측 하단 파이썬 버전 클릭

 

2. 파이썬 버전을 3.12로 설정

 

Python 3.9 버전은 실행 불가 확인. 최신 버전의 파이썬 사용을 권장합니다. 

 

Q: TypeError가 발생해요.

A: 오류 메시지가 출력되는 부분에서 잘못 작성한 코드 또는 전체 코드를 빠뜨린 부분이 없는지 확인 부탁드립니다.

 

Q: dico_token이라는 파일이 없다고 떠요

A: 

1. dico_token.py라는 이름으로 파일을 만들어줍니다.(파일 이름은 사용자 마음대로 지정하셔도 됩니다.)

 

2. 디스코드 개발자 포털 - Settings - Bot - Reset Token 버튼을 클릭하여 토큰을 발급받아 복사합니다.

 

3. 발급받아 복사한 토큰을 아까 만들어 두었던 파일에 붙여넣기 합니다.

 

 

기타 오류 사항은 아래 페이지를 확인해주시기 바랍니다.

 

https://kante-kante.tistory.com/38

디스코드 봇 개발 시 Error 관련