ekzyis
/
musicube
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Run pre-commit on all files

This commit is contained in:
ekzyis 2022-04-15 01:22:45 +02:00
parent 33f471ee9c
commit bfc593f39b
11 changed files with 83 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.dockerignore
View File

@ -1,4 +1,4 @@
**
!src/
!*cookies.txt
!requirements.txt
!requirements.txt

2
.env.template
View File

@ -2,4 +2,4 @@ DISCORD_BOT_TOKEN=
DISCORD_CMD_PREFIX=
DISCORD_LOG_CHANNEL_ID=
DISCORD_LOG_CHANNEL_CLEAR_ON_STARTUP=False
YOUTUBE_COOKIES=youtube.com_cookies.txt
YOUTUBE_COOKIES=youtube.com_cookies.txt

2
.gitignore vendored
View File

@ -3,4 +3,4 @@ __pycache__
.env
docker.env
.coverage
*cookies.txt
*cookies.txt

2
.pylintrc
View File

@ -21,4 +21,4 @@ indent-string=' '
max-line-length=160
[SIMILARITIES]
ignore-comments = no
ignore-comments = no

2
pyproject.toml
View File

@ -1,4 +1,4 @@
[tool.autopep8]
indent-size=2
max-line-length=160
ignore = ["E402"]
ignore = ["E402"]

58
src/bot.py
View File

@ -1,18 +1,18 @@
import asyncio
import logging
import os
import sys
import asyncio
from dataclasses import dataclass
from distutils.util import strtobool
import logging
import discord
import youtube_dl
from discord.ext import commands, tasks
from dotenv import load_dotenv
import youtube_dl
from error import ErrorHandler
from message import NowPlayingMessage, QueuedMessage, ErrorMessage
from log import create_logger, DiscordLogger
from log import DiscordLogger, create_logger
from message import ErrorMessage, NowPlayingMessage, QueuedMessage
load_dotenv()
@ -55,10 +55,10 @@ class Music(commands.Cog):
# pylint: disable=no-member
self._handle_playback.start()
self.logger = logger if logger else create_logger("bot")
if log_channel_id := os.environ.get("DISCORD_LOG_CHANNEL_ID", None):
clear_on_startup = os.environ.get("DISCORD_LOG_CHANNEL_CLEAR_ON_STARTUP", False)
if clear_on_startup == "":
self.logger = logger if logger else create_logger('bot')
if log_channel_id := os.environ.get('DISCORD_LOG_CHANNEL_ID', None):
clear_on_startup = os.environ.get('DISCORD_LOG_CHANNEL_CLEAR_ON_STARTUP', False)
if clear_on_startup == '':
clear_on_startup = False
if isinstance(clear_on_startup, str):
clear_on_startup = bool(strtobool(clear_on_startup))
@ -70,7 +70,7 @@ class Music(commands.Cog):
def _next(self):
"""Trigger playback of next song."""
self.logger.info("Song finished. Triggering playback of next song")
self.logger.info('Song finished. Triggering playback of next song')
self._queue.task_done()
self._current_skip_message = None
if self._queue_lock.locked():
@ -80,11 +80,11 @@ class Music(commands.Cog):
async def _handle_playback(self):
while True:
try:
self.logger.info("Waiting for queue lock to acquire ...")
self.logger.info('Waiting for queue lock to acquire ...')
await self._queue_lock.acquire()
self.logger.info("Queue lock acquired! Waiting for song queue to return a song ...")
self.logger.info('Queue lock acquired! Waiting for song queue to return a song ...')
ctx, song = await self._queue.get()
self.logger.info("Queue returned a song!")
self.logger.info('Queue returned a song!')
if ctx.voice_client is None:
# Bot is no longer in a voice channel.
# This could be the case because a stop command was issued.
@ -96,7 +96,7 @@ class Music(commands.Cog):
def after(err):
if err:
self.logger.error("Player error: %s", err)
self.logger.error('Player error: %s', err)
self._next()
self.logger.info('Now playing song "%s"', song.title)
ctx.voice_client.play(audio, after=after)
@ -105,14 +105,14 @@ class Music(commands.Cog):
await self._add_skip_button(msg)
# pylint: disable=broad-except
except Exception as err:
self.logger.error("Error during playback: %s", err)
self.logger.error('Error during playback: %s', err)
if ctx:
embed = ErrorMessage(str(err))
await ctx.send(embed=embed)
self._next()
async def _add_skip_button(self, msg):
await msg.add_reaction("⏭️")
await msg.add_reaction('⏭️')
self._current_skip_message = msg
@commands.Cog.listener()
@ -120,7 +120,7 @@ class Music(commands.Cog):
if not user.bot \
and self._current_skip_message \
and reaction.message.id == self._current_skip_message.id \
and reaction.emoji == "⏭️":
and reaction.emoji == '⏭️':
voice_client = reaction.message.guild.voice_client
self._skip(voice_client)
@ -148,10 +148,10 @@ class Music(commands.Cog):
def _skip(self, voice_client):
"""Skip to next song."""
if voice_client is None or not voice_client.is_playing():
raise commands.CommandError("No song playing")
raise commands.CommandError('No song playing')
# This skips to next song because the bot does not differentiate between
# a song stopping because it is finished or because it was manually stopped.
self.logger.info("Skipping song")
self.logger.info('Skipping song')
voice_client.stop()
@commands.command()
@ -161,37 +161,37 @@ class Music(commands.Cog):
@commands.command()
async def stop(self, ctx):
self.logger.info("Stopping playback")
self.logger.info('Stopping playback')
await ctx.voice_client.disconnect()
@play.before_invoke
async def ensure_voice(self, ctx):
if ctx.voice_client is None:
if ctx.author.voice:
self.logger.info("Connecting to voice channel ...")
self.logger.info('Connecting to voice channel ...')
await ctx.author.voice.channel.connect()
self.logger.info("Connected")
self.logger.info('Connected')
else:
raise commands.CommandError("Author not connected to a voice channel")
raise commands.CommandError('Author not connected to a voice channel')
if __name__ == "__main__":
prefix = os.environ.get("DISCORD_CMD_PREFIX", "!")
logger = create_logger("bot")
if __name__ == '__main__':
prefix = os.environ.get('DISCORD_CMD_PREFIX', '!')
logger = create_logger('bot')
bot = commands.Bot(command_prefix=commands.when_mentioned_or(prefix),
description='Relatively simple music bot example')
@bot.event
async def on_ready():
logger.info("Logged in as %s (%s)", bot.user, bot.user.id)
logger.info('Logged in as %s (%s)', bot.user, bot.user.id)
logger.info('------')
bot.add_cog(Music(bot, logger=logger))
bot.add_cog(ErrorHandler(bot, logger=logger))
token = os.environ.get("DISCORD_BOT_TOKEN", None)
token = os.environ.get('DISCORD_BOT_TOKEN', None)
if not token:
logger.warning("Discord bot token not found")
logger.warning('Discord bot token not found')
sys.exit(1)
bot.run(token)

8
src/error.py
View File

@ -20,16 +20,16 @@ class ErrorHandler(commands.Cog):
self.logger.error(error)
return
if isinstance(error, commands.MissingPermissions):
message = "You are missing the required permissions to run this command!"
message = 'You are missing the required permissions to run this command!'
elif isinstance(error, commands.UserInputError):
message = "Something about your input was wrong, please check your input and try again!"
message = 'Something about your input was wrong, please check your input and try again!'
elif isinstance(error, commands.CommandError):
message = str(error)
else:
message = "Oh no! Something went wrong while running the command!"
message = 'Oh no! Something went wrong while running the command!'
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
message = ansi_escape.sub("", message)
message = ansi_escape.sub('', message)
self.logger.error('Error during command "%s": %s', command_name, message)
embed = ErrorMessage(message, command_name=command_name)

9
src/log.py
View File

@ -1,14 +1,13 @@
import asyncio
import logging
from logging import LogRecord
import sys
import os
import sys
from datetime import datetime
from logging import LogRecord
from discord import TextChannel
from discord.ext.commands import Bot as DiscordBot
__LOG_FORMAT__ = '%(asctime)s %(name)-5s %(levelname)-8s %(message)s'
__DATE_FORMAT__ = '%Y-%m-%d %H:%M:%S'
@ -21,8 +20,8 @@ def create_logger(name: str) -> logging.Logger:
stream_handler = logging.StreamHandler(stream=sys.stdout)
COMMIT = os.getenv('GIT_COMMIT') or 'no-commit'
timestamp = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
log_path = f"{COMMIT}_{timestamp}.log"
timestamp = datetime.now().strftime('%Y-%m-%dT%H:%M:%S')
log_path = f'{COMMIT}_{timestamp}.log'
file_handler = logging.FileHandler(log_path)

8
src/message.py
View File

@ -5,7 +5,7 @@ import discord
class BotMessage(discord.Embed):
def __init__(self, **kwargs):
title = kwargs.pop("title", None)[:256]
title = kwargs.pop('title', None)[:256]
if title is not None:
# Max embed title length is 256
title = title[:256]
@ -21,7 +21,7 @@ class ErrorMessage(BotMessage):
if command_name:
title = f'Error during command "{command_name}"'
description = None
if match := re.search(r"(?P<error>\w+Error): ?(ERROR: ?)?(?P<message>.*): ?Traceback", message):
if match := re.search(r'(?P<error>\w+Error): ?(ERROR: ?)?(?P<message>.*): ?Traceback', message):
description = f"{match.group('error')}: {match.group('message')}"
super().__init__(
title=title,
@ -33,7 +33,7 @@ class ErrorMessage(BotMessage):
class NowPlayingMessage(BotMessage):
def __init__(self, title, url):
super().__init__(
title=f"Now playing: {title}",
title=f'Now playing: {title}',
description=url,
color=discord.Color.green()
)
@ -42,7 +42,7 @@ class NowPlayingMessage(BotMessage):
class QueuedMessage(BotMessage):
def __init__(self, title, url):
super().__init__(
title=f"Queued: {title}",
title=f'Queued: {title}',
description=url,
color=discord.Color.blue()
)

8
test/conftest.py
View File

@ -1,9 +1,9 @@
import asyncio
import glob
import os
import sys
from pathlib import Path
from unittest.mock import Mock
import glob
import os
import pytest
from pytest_mock import MockerFixture
@ -15,7 +15,7 @@ sys.path.insert(0, SRC_PATH)
from bot import Music
@pytest.fixture(scope="session", autouse=True)
@pytest.fixture(scope='session', autouse=True)
def global_teardown():
yield
logs = glob.glob('*.log')
@ -40,7 +40,7 @@ def mbot(bot):
@pytest.fixture
def ctx(mocker: MockerFixture):
ctx_mock = mocker.patch('discord.ext.commands.Context', autospec=True)
ctx_mock.voice_client.stop = lambda: ctx_mock.voice_client.play.call_args.kwargs["after"](None)
ctx_mock.voice_client.stop = lambda: ctx_mock.voice_client.play.call_args.kwargs['after'](None)
return ctx_mock

66
test/test_bot.py
View File

@ -1,5 +1,5 @@
import asyncio
from unittest.mock import Mock, AsyncMock, patch
from unittest.mock import AsyncMock, Mock, patch
import pytest
from discord.ext import commands
@ -11,7 +11,7 @@ async def test_bot_ensure_voice(mbot, ctx):
ctx.voice_client = None
ctx.author.voice = AsyncMock()
await mbot.ensure_voice(ctx)
assert ctx.author.voice.channel.connect.call_count == 1, "Did not connect to voice channel of author"
assert ctx.author.voice.channel.connect.call_count == 1, 'Did not connect to voice channel of author'
ctx.reset_mock(return_value=True)
# TEST: Error if author not inside a channel
@ -22,7 +22,7 @@ async def test_bot_ensure_voice(mbot, ctx):
def mock_ytdl_extract_info(ytdl, url, title):
ytdl.extract_info.return_value = {"entries": [{"url": url, "title": title}]}
ytdl.extract_info.return_value = {'entries': [{'url': url, 'title': title}]}
def mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio):
@ -38,66 +38,66 @@ async def test_bot_playback(mbot, ctx):
# TEST: First song queued is immediately played
ctx.voice_client.is_playing.return_value = False
url = "https://www.youtube.com/watch?v=Wr9LZ1hAFpQ"
title = "In Flames - Deliver Us (Official Video)"
url = 'https://www.youtube.com/watch?v=Wr9LZ1hAFpQ'
title = 'In Flames - Deliver Us (Official Video)'
mock_ytdl_extract_info(ytdl, url, title)
deliver_us_audio = mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio)
query = 'in flames deliver us'
# pylint: disable=too-many-function-args
await mbot.play(mbot, ctx, query=query)
assert \
ytdl.extract_info.call_args.args == (query,) and ytdl.extract_info.call_args.kwargs == {"download": False}, \
f"ytdl.extract_info was not called with {query}, {{ download: False }}"
ytdl.extract_info.call_args.args == (query,) and ytdl.extract_info.call_args.kwargs == {'download': False}, \
f'ytdl.extract_info was not called with {query}, {{ download: False }}'
assert \
ffmpeg_pcm_audio.call_args is None, \
f"FFmpegPCMAudio was immediately called with {url} instead of being queued"
f'FFmpegPCMAudio was immediately called with {url} instead of being queued'
assert \
ctx.voice_client.play.call_args is None, \
"Did immediately playback audio instead of being queued"
'Did immediately playback audio instead of being queued'
assert \
ctx.send.call_args is None, \
"Did immediately send 'Now playing:' message of being queued"
await asyncio.sleep(0)
assert \
ffmpeg_pcm_audio.call_args.args == (url,), \
f"FFmpegPCMAudio was not called with {url}"
f'FFmpegPCMAudio was not called with {url}'
assert \
ctx.voice_client.play.call_args.args == (deliver_us_audio,), \
"Did not playback correct audio"
embed = ctx.send.call_args.kwargs["embed"]
assert embed.title == f"Now playing: {title}", "Did not send 'Now playing:' message"
'Did not playback correct audio'
embed = ctx.send.call_args.kwargs['embed']
assert embed.title == f'Now playing: {title}', "Did not send 'Now playing:' message"
# TEST: Following songs are put inside a queue
ctx.voice_client.is_playing.return_value = True
url = "https://www.youtube.com/watch?v=pMDcYX2wRSg"
title = "Three Days Grace - Time of Dying (lyrics)"
url = 'https://www.youtube.com/watch?v=pMDcYX2wRSg'
title = 'Three Days Grace - Time of Dying (lyrics)'
mock_ytdl_extract_info(ytdl, url, title)
time_of_dying_audio = mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio)
# pylint: disable=too-many-function-args
query = "three days grace time of dying"
query = 'three days grace time of dying'
await mbot.play(mbot, ctx, query=query)
assert \
ytdl.extract_info.call_args.args == (query,) and ytdl.extract_info.call_args.kwargs == {"download": False}, \
f"ytdl.extract_info was not called with {query}, {{ download: False }}"
ytdl.extract_info.call_args.args == (query,) and ytdl.extract_info.call_args.kwargs == {'download': False}, \
f'ytdl.extract_info was not called with {query}, {{ download: False }}'
assert \
not ffmpeg_pcm_audio.call_args.args == (url,), \
f"FFmpegPCMAudio was immediately called with {url} instead of being queued"
f'FFmpegPCMAudio was immediately called with {url} instead of being queued'
assert \
not ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \
"Did immediately playback audio instead of being queued"
embed = ctx.send.call_args.kwargs["embed"]
assert embed.title == f"Queued: {title}", "Did not send 'Queued:' message"
'Did immediately playback audio instead of being queued'
embed = ctx.send.call_args.kwargs['embed']
assert embed.title == f'Queued: {title}', "Did not send 'Queued:' message"
await asyncio.sleep(0)
# Assert that there is still no playback because previous song is not finished yet
assert \
not ffmpeg_pcm_audio.call_args.args == (url,), \
f"FFmpegPCMAudio was called with {url} before previous song finished"
f'FFmpegPCMAudio was called with {url} before previous song finished'
# Execute callback for song finish event
ctx.voice_client.play.call_args.kwargs["after"](None)
ctx.voice_client.play.call_args.kwargs['after'](None)
await asyncio.sleep(0)
assert \
ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \
"Did not queue next song"
'Did not queue next song'
@pytest.mark.asyncio
@ -109,8 +109,8 @@ async def test_bot_skip(mbot, ctx):
# Queue first song
ctx.voice_client.is_playing.return_value = False
url = "https://www.youtube.com/watch?v=Wr9LZ1hAFpQ"
title = "In Flames - Deliver Us (Official Video)"
url = 'https://www.youtube.com/watch?v=Wr9LZ1hAFpQ'
title = 'In Flames - Deliver Us (Official Video)'
mock_ytdl_extract_info(ytdl, url, title)
deliver_us_audio = mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio)
query = 'in flames deliver us'
@ -119,28 +119,28 @@ async def test_bot_skip(mbot, ctx):
await asyncio.sleep(0)
assert \
ctx.voice_client.play.call_args.args == (deliver_us_audio,), \
"Did not playback correct audio"
'Did not playback correct audio'
# Queue second song
ctx.voice_client.is_playing.return_value = True
url = "https://www.youtube.com/watch?v=pMDcYX2wRSg"
title = "Three Days Grace - Time of Dying (lyrics)"
url = 'https://www.youtube.com/watch?v=pMDcYX2wRSg'
title = 'Three Days Grace - Time of Dying (lyrics)'
mock_ytdl_extract_info(ytdl, url, title)
time_of_dying_audio = mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio)
# pylint: disable=too-many-function-args
query = "three days grace time of dying"
query = 'three days grace time of dying'
await mbot.play(mbot, ctx, query=query)
await asyncio.sleep(0)
assert \
not ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \
"Did immediately playback audio instead of being queued"
'Did immediately playback audio instead of being queued'
# Now skip first song
await mbot.skip(mbot, ctx)
await asyncio.sleep(0)
assert \
ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \
"Did not skip song"
'Did not skip song'
# TEST: Error if no song playing
ctx.voice_client.is_playing.return_value = False