ErrorNoInternet/ErrorNoCord
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: ErrorNoInternet:main
Branches Tags
ErrorNoInternet:main
..
compare: ErrorNoInternet:25d0519b621eadf37a37bff147bc02de4f65015b
Branches Tags
ErrorNoInternet:main

2 Commits
main ... 25d0519b62

Author SHA1 Message Date
ErrorNoInternet
25d0519b62 fix(commands/voice): return early if no player 2025-01-06 00:37:14 -05:00
ErrorNoInternet
3f36823b73 refactor: minor fixes everywhere 2025-01-06 00:27:35 -05:00
37 changed files with 712 additions and 1878 deletions
Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1,3 +1,2 @@
.env
.venv
__pycache__

10
Dockerfile
View File

@@ -1,10 +0,0 @@
FROM python:3.13-alpine
RUN apk --no-cache add ffmpeg gcc linux-headers musl-dev opus python3-dev
WORKDIR /bot
COPY . .
RUN pip install -r requirements.txt
CMD ["python", "main.py"]

19
arguments.py
View File

@@ -8,9 +8,7 @@ import utils
class ArgumentParser:
def __init__(self, command, description):
self.parser = argparse.ArgumentParser(
command,
description=description,
exit_on_error=False,
command, description=description, exit_on_error=False
)
def print_help(self):
@@ -28,20 +26,21 @@ class ArgumentParser:
async def parse_args(self, message, tokens) -> argparse.Namespace | None:
try:
with contextlib.redirect_stdout(io.StringIO()):
return self.parser.parse_args(tokens[1:])
args = self.parser.parse_args(tokens[1:])
return args
except SystemExit:
await utils.reply(message, f"```\n{self.print_help()}```")
except Exception as e:
await utils.reply(message, f"`{e}`")
def range_type(string: str, lower=0, upper=100) -> int:
def range_type(string, min=0, max=100):
try:
value = int(string)
except ValueError as e:
raise argparse.ArgumentTypeError("value is not a valid integer") from e
except ValueError:
raise argparse.ArgumentTypeError("value is not a valid integer")
if lower <= value <= upper:
if min <= value <= max:
return value
raise argparse.ArgumentTypeError(f"value is not in range {lower}-{upper}")
else:
raise argparse.ArgumentTypeError(f"value is not in range {min}-{max}")

8
audio/__init__.py
View File

@@ -1,8 +0,0 @@
from . import discord, queue, utils, youtubedl
__all__ = [
"discord",
"queue",
"utils",
"youtubedl",
]

39
audio/discord.py
View File

@@ -1,39 +0,0 @@
import audioop
import disnake
class TrackedAudioSource(disnake.AudioSource):
def __init__(self, source):
self._source = source
self.read_count = 0
def read(self) -> bytes:
data = self._source.read()
if data:
self.read_count += 1
return data
def fast_forward(self, seconds: int):
for _ in range(int(seconds / 0.02)):
self.read()
@property
def progress(self) -> float:
return self.read_count * 0.02
class PCMVolumeTransformer(disnake.AudioSource):
def __init__(self, original: TrackedAudioSource, volume: float = 1.0) -> None:
if original.is_opus():
raise disnake.ClientException("AudioSource must not be Opus encoded")
self.original = original
self.volume = volume
def cleanup(self) -> None:
self.original.cleanup()
def read(self) -> bytes:
ret = self.original.read()
return audioop.mul(ret, 2, self.volume)

112
audio/queue.py
View File

@@ -1,112 +0,0 @@
import collections
from dataclasses import dataclass
from typing import ClassVar, Optional
import disnake
from constants import BAR_LENGTH, EMBED_COLOR
from .utils import format_duration
from .youtubedl import YTDLSource
@dataclass
class Song:
player: YTDLSource
trigger_message: disnake.Message
def format(self, show_queuer=False, hide_preview=False, multiline=False) -> str:
title = f"[`{self.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''})"
duration = (
format_duration(self.player.duration) if self.player.duration else "stream"
)
if multiline:
queue_time = (
self.trigger_message.edited_at or self.trigger_message.created_at
)
return f"{title}\n**duration:** {duration}" + (
f", **queued by:** <@{self.trigger_message.author.id}> <t:{round(queue_time.timestamp())}:R>"
if show_queuer
else ""
)
return f"{title} [**{duration}**]" + (
f" (<@{self.trigger_message.author.id}>)" if show_queuer else ""
)
def embed(self, is_paused=False):
progress = 0
if self.player.duration:
progress = self.player.original.progress / self.player.duration
embed = disnake.Embed(
color=EMBED_COLOR,
title=self.player.title,
url=self.player.original_url,
description=(
f"{'⏸️ ' if is_paused else ''}"
f"`[{'#' * int(progress * BAR_LENGTH)}{'-' * int((1 - progress) * BAR_LENGTH)}]` "
+ (
f"**{format_duration(int(self.player.original.progress))}** / **{format_duration(self.player.duration)}** (**{round(progress * 100)}%**)"
if self.player.duration
else "[**stream**]"
)
),
timestamp=self.trigger_message.edited_at or self.trigger_message.created_at,
)
uploader_value = None
if self.player.uploader_url:
if self.player.uploader:
uploader_value = f"[{self.player.uploader}]({self.player.uploader_url})"
else:
uploader_value = self.player.uploader_url
elif self.player.uploader:
uploader_value = self.player.uploader
if uploader_value:
embed.add_field(name="Uploader", value=uploader_value)
if self.player.like_count:
embed.add_field(name="Likes", value=f"{self.player.like_count:,}")
if self.player.view_count:
embed.add_field(name="Views", value=f"{self.player.view_count:,}")
if self.player.timestamp:
embed.add_field(name="Published", value=f"<t:{int(self.player.timestamp)}>")
if self.player.volume:
embed.add_field(name="Volume", value=f"{int(self.player.volume * 100)}%")
if self.player.thumbnail_url:
embed.set_image(self.player.thumbnail_url)
embed.set_footer(
text=f"Queued by {self.trigger_message.author.name}",
icon_url=(
self.trigger_message.author.avatar.url
if self.trigger_message.author.avatar
else None
),
)
return embed
def __str__(self):
return self.__repr__()
@dataclass
class Player:
queue: ClassVar = collections.deque()
current: Optional[Song] = None
def queue_pop(self):
popped = self.queue.popleft()
self.current = popped
return popped
def queue_push(self, item):
self.queue.append(item)
def queue_push_front(self, item):
self.queue.appendleft(item)
def __str__(self):
return self.__repr__()

7
audio/utils.py
View File

@@ -1,7 +0,0 @@
def format_duration(duration: int | float) -> str:
hours, duration = divmod(int(duration), 3600)
minutes, duration = divmod(duration, 60)
segments = [hours, minutes, duration]
if len(segments) == 3 and segments[0] == 0:
del segments[0]
return f"{':'.join(f'{s:0>2}' for s in segments)}"

76
audio/youtubedl.py
View File

@@ -1,76 +0,0 @@
import asyncio
from typing import Any, Optional
import disnake
import yt_dlp
from constants import YTDL_OPTIONS
from .discord import PCMVolumeTransformer, TrackedAudioSource
ytdl = yt_dlp.YoutubeDL(YTDL_OPTIONS)
class YTDLSource(PCMVolumeTransformer):
def __init__(
self,
source: TrackedAudioSource,
*,
data: dict[str, Any],
volume: float = 0.5,
):
super().__init__(source, volume)
self.description = data.get("description")
self.duration = data.get("duration")
self.id = data.get("id")
self.like_count = data.get("like_count")
self.original_url = data.get("original_url")
self.thumbnail_url = data.get("thumbnail")
self.timestamp = data.get("timestamp")
self.title = data.get("title")
self.uploader = data.get("uploader")
self.uploader_url = data.get("uploader_url")
self.view_count = data.get("view_count")
@classmethod
async def from_url(
cls,
url,
*,
loop: Optional[asyncio.AbstractEventLoop] = None,
stream: bool = False,
):
loop = loop or asyncio.get_event_loop()
data: Any = await loop.run_in_executor(
None,
lambda: ytdl.extract_info(url, download=not stream),
)
if "entries" in data:
if not data["entries"]:
raise Exception("no results found!")
data = data["entries"][0]
if "url" not in data:
raise Exception("no url returned!")
return cls(
TrackedAudioSource(
disnake.FFmpegPCMAudio(
data["url"] if stream else ytdl.prepare_filename(data),
before_options="-vn -reconnect 1",
),
),
data=data,
)
def __repr__(self):
return f"<YTDLSource title={self.title} original_url={self.original_url} duration={self.duration}>"
def __str__(self):
return self.__repr__()
def __reload_module__():
global ytdl
ytdl = yt_dlp.YoutubeDL(YTDL_OPTIONS)

10
commands/__init__.py
View File

@@ -3,11 +3,15 @@ from .utils import Command, match, match_token, tokenize
__all__ = [
"bot",
"tools",
"utils",
"voice",
"Command",
"match",
"match_token",
"tokenize",
"tools",
"utils",
"voice",
]
def __reload_module__():
globals().update({k: v for k, v in vars(utils).items() if not k.startswith("_")})

89
commands/bot.py
View File

@@ -1,65 +1,18 @@
import os
import threading
import time
import disnake
import psutil
from yt_dlp import version
import arguments
import commands
from constants import EMBED_COLOR
from state import client, start_time
from utils import format_duration, reply, surround
import utils
from state import start_time
async def status(message):
member_count = 0
channel_count = 0
for guild in client.guilds:
member_count += len(guild.members)
channel_count += len(guild.channels)
process = psutil.Process(os.getpid())
memory_usage = process.memory_info().rss / 1048576
embed = disnake.Embed(color=EMBED_COLOR)
embed.add_field(
name="Latency",
value=surround(f"{round(client.latency * 1000, 1)} ms"),
async def help(message):
await utils.reply(
message,
", ".join(
[f"`{command.value}`" for command in commands.Command.__members__.values()]
),
)
embed.add_field(
name="Memory",
value=surround(f"{round(memory_usage, 1)} MiB"),
)
embed.add_field(
name="Threads",
value=surround(threading.active_count()),
)
embed.add_field(
name="Guilds",
value=surround(len(client.guilds)),
)
embed.add_field(
name="Members",
value=surround(member_count),
)
embed.add_field(
name="Channels",
value=surround(channel_count),
)
embed.add_field(
name="Disnake",
value=surround(disnake.__version__),
)
embed.add_field(
name="yt-dlp",
value=surround(version.__version__),
)
embed.add_field(
name="Uptime",
value=surround(format_duration(int(time.time() - start_time), short=True)),
)
await reply(message, embed=embed)
async def uptime(message):
@@ -78,26 +31,8 @@ async def uptime(message):
return
if args.since:
await reply(message, f"{round(start_time)}")
await utils.reply(message, f"{round(start_time)}")
else:
await reply(message, f"up {format_duration(int(time.time() - start_time))}")
async def ping(message):
await reply(
message,
embed=disnake.Embed(
title="Pong :ping_pong:",
description=f"Latency: **{round(client.latency * 1000, 1)} ms**",
color=EMBED_COLOR,
),
)
async def help(message):
await reply(
message,
", ".join(
[f"`{command.value}`" for command in commands.Command.__members__.values()],
),
)
await utils.reply(
message, f"up {utils.format_duration(int(time.time() - start_time))}"
)

183
commands/tools.py
View File

@@ -1,175 +1,19 @@
import re
import aiohttp
import disnake
import arguments
import commands
import utils
from constants import APPLICATION_FLAGS, BADGE_EMOJIS, EMBED_COLOR, PUBLIC_FLAGS
from state import client
async def lookup(message):
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(
tokens[0],
"look up a discord user or application by ID",
)
parser.add_argument(
"-a",
"--application",
action="store_true",
help="look up applications instead of users",
)
parser.add_argument(
"id",
type=int,
help="the ID to perform a search for",
)
if not (args := await parser.parse_args(message, tokens)):
return
if args.application:
session = aiohttp.ClientSession()
response = await (
await session.get(f"https://discord.com/api/v9/applications/{args.id}/rpc")
).json()
if "code" in response.keys():
await utils.reply(message, "application not found!")
return
embed = disnake.Embed(description=response["description"], color=EMBED_COLOR)
embed.set_thumbnail(
url=f"https://cdn.discordapp.com/app-icons/{response['id']}/{response['icon']}.webp",
)
embed.add_field(name="Application Name", value=response["name"])
embed.add_field(name="Application ID", value="`" + response["id"] + "`")
embed.add_field(
name="Public Bot",
value=f"{'`' + str(response['bot_public']) + '`' if 'bot_public' in response else 'No bot'}",
)
embed.add_field(name="Public Flags", value="`" + str(response["flags"]) + "`")
embed.add_field(
name="Terms of Service",
value=(
"None"
if "terms_of_service_url" not in response.keys()
else f"[Link]({response['terms_of_service_url']})"
),
)
embed.add_field(
name="Privacy Policy",
value=(
"None"
if "privacy_policy_url" not in response.keys()
else f"[Link]({response['privacy_policy_url']})"
),
)
embed.add_field(
name="Creation Time",
value=f"<t:{utils.snowflake_timestamp(int(response['id']))}:R>",
)
embed.add_field(
name="Default Invite URL",
value=(
"None"
if "install_params" not in response.keys()
else f"[Link](https://discord.com/oauth2/authorize?client_id={response['id']}&permissions={response['install_params']['permissions']}&scope={'%20'.join(response['install_params']['scopes'])})"
),
)
embed.add_field(
name="Custom Invite URL",
value=(
"None"
if "custom_install_url" not in response.keys()
else f"[Link]({response['custom_install_url']})"
),
)
bot_intents = []
for application_flag, intent_name in APPLICATION_FLAGS.items():
if response["flags"] & application_flag == application_flag:
if intent_name.replace(" (unverified)", "") not in bot_intents:
bot_intents.append(intent_name)
embed.add_field(
name="Application Flags",
value=", ".join(bot_intents) if bot_intents else "None",
)
bot_tags = ""
if "tags" in response.keys():
for tag in response["tags"]:
bot_tags += tag + ", "
embed.add_field(
name="Tags",
value="None" if bot_tags == "" else bot_tags[:-2],
inline=False,
)
else:
try:
user = await client.fetch_user(args.id)
except Exception:
await utils.reply(message, "user not found!")
return
badges = ""
for flag, flag_name in PUBLIC_FLAGS.items():
if user.public_flags.value & flag == flag:
if flag_name != "None":
try:
badges += BADGE_EMOJIS[PUBLIC_FLAGS[flag]]
except Exception as e:
raise Exception(
f"unable to find badge: {PUBLIC_FLAGS[flag]}"
) from e
user_object = await client.fetch_user(user.id)
accent_color = 0x000000
if user_object.accent_color is not None:
accent_color = user_object.accent_color
embed = disnake.Embed(color=accent_color)
embed.add_field(
name="User ID",
value=f"`{user.id}`",
)
embed.add_field(
name="Discriminator",
value=f"`{user.name}#{user.discriminator}`",
)
embed.add_field(
name="Creation Time",
value=f"<t:{utils.snowflake_timestamp(int(user.id))}:R>",
)
embed.add_field(
name="Public Flags",
value=f"`{user.public_flags.value}` {badges}",
)
embed.add_field(
name="Bot User",
value=f"`{user.bot}`",
)
embed.add_field(
name="System User",
value=f"`{user.system}`",
)
embed.set_thumbnail(url=user.avatar if user.avatar else user.default_avatar)
if user_object.banner:
embed.set_image(url=user_object.banner)
await utils.reply(message, embed=embed)
async def clear(message):
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(
tokens[0],
"bulk delete messages in the current channel matching specified criteria",
"bulk delete messages in the current channel matching certain criteria",
)
parser.add_argument(
"count",
type=lambda c: arguments.range_type(c, lower=1, upper=1000),
type=lambda c: arguments.range_type(c, min=1, max=1000),
help="amount of messages to delete",
)
group = parser.add_mutually_exclusive_group()
@@ -210,25 +54,12 @@ async def clear(message):
action="store_true",
help="delete messages with reactions",
)
parser.add_argument(
"-A",
"--attachments",
action="store_true",
help="delete messages with attachments",
)
parser.add_argument(
"-d",
"--delete-command",
action="store_true",
help="delete the command message as well",
)
parser.add_argument(
"-I",
"--ignore-ids",
type=int,
action="append",
help="ignore messages with this id",
)
if not (args := await parser.parse_args(message, tokens)):
return
@@ -243,8 +74,6 @@ async def clear(message):
regex = re.compile(r, re.IGNORECASE if args.case_insensitive else 0)
def check(m):
if (ids := args.ignore_ids) and m.id in ids:
return False
c = []
if regex:
c.append(regex.search(m.content))
@@ -257,16 +86,12 @@ async def clear(message):
c.append(m.author.id in i)
if args.reactions:
c.append(len(m.reactions) > 0)
if args.attachments:
c.append(len(m.attachments) > 0)
return all(c)
messages = len(
await message.channel.purge(
limit=args.count,
check=check,
oldest_first=args.oldest_first,
),
limit=args.count, check=check, oldest_first=args.oldest_first
)
)
if not args.delete_command:

34
commands/utils.py
View File

@@ -1,20 +1,15 @@
from enum import Enum
from functools import lru_cache
import enum
import constants
class Command(Enum):
class Command(enum.Enum):
CLEAR = "clear"
CURRENT = "current"
EXECUTE = "execute"
FAST_FORWARD = "ff"
HELP = "help"
JOIN = "join"
LEAVE = "leave"
LOOKUP = "lookup"
PAUSE = "pause"
PING = "ping"
PLAY = "play"
PLAYING = "playing"
PURGE = "purge"
@@ -22,27 +17,19 @@ class Command(Enum):
RELOAD = "reload"
RESUME = "resume"
SKIP = "skip"
SPONSORBLOCK = "sponsorblock"
STATUS = "status"
UPTIME = "uptime"
VOLUME = "volume"
@lru_cache
def match_token(token: str) -> list[Command]:
match token.lower():
case "r":
return [Command.RELOAD]
case "s":
return [Command.SKIP]
case "c":
return [Command.CURRENT]
if token.lower() == "r":
return [Command.RELOAD]
if exact_match := list(
filter(
lambda command: command.value == token.lower(),
Command.__members__.values(),
),
)
):
return exact_match
@@ -50,28 +37,23 @@ def match_token(token: str) -> list[Command]:
filter(
lambda command: command.value.startswith(token.lower()),
Command.__members__.values(),
),
)
)
@lru_cache
def match(command: str) -> list[Command] | None:
if tokens := tokenize(command):
return match_token(tokens[0])
@lru_cache
def tokenize(string: str, remove_prefix: bool = True) -> list[str]:
def tokenize(string: str) -> list[str]:
tokens = []
token = ""
in_quotes = False
quote_char = None
escape = False
if remove_prefix:
string = string[len(constants.PREFIX) :]
for char in string:
for char in string[len(constants.PREFIX) :]:
if escape:
token += char
escape = False

379
commands/voice.py Normal file
View File

@@ -0,0 +1,379 @@
import itertools
import disnake
import disnake_paginator
import arguments
import commands
import constants
import utils
import youtubedl
from state import client, players
async def queue_or_play(message, edited=False):
await ensure_joined(message)
if not command_allowed(message):
return
if message.guild.id not in players:
players[message.guild.id] = youtubedl.QueuedPlayer()
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(
tokens[0], "queue a song, list the queue, or resume playback"
)
parser.add_argument("query", nargs="?", help="yt-dlp URL or query to get song")
parser.add_argument(
"-v",
"--volume",
default=50,
type=lambda v: arguments.range_type(v, min=0, max=150),
help="the volume level (0 - 150) for the specified song",
)
parser.add_argument(
"-i",
"--remove-index",
type=int,
help="remove a queued song by index",
)
parser.add_argument(
"-m",
"--remove-multiple",
action="store_true",
help="continue removing queued after finding a match",
)
parser.add_argument(
"-c",
"--clear",
action="store_true",
help="remove all queued songs",
)
parser.add_argument(
"--now",
action="store_true",
help="play the specified song immediately",
)
parser.add_argument(
"--next",
action="store_true",
help="play the specified song next",
)
parser.add_argument(
"-t",
"--remove-title",
help="remove queued songs by title",
)
parser.add_argument(
"-q",
"--remove-queuer",
type=int,
help="remove queued songs by queuer",
)
if not (args := await parser.parse_args(message, tokens)):
return
if edited:
found = None
for queued in players[message.guild.id].queue:
if queued.trigger_message.id == message.id:
found = queued
break
if found:
players[message.guild.id].queue.remove(found)
if args.clear:
players[message.guild.id].queue.clear()
await utils.add_check_reaction(message)
return
elif i := args.remove_index:
if i <= 0 or i > len(players[message.guild.id].queue):
await utils.reply(message, "invalid index!")
return
queued = players[message.guild.id].queue[i - 1]
del players[message.guild.id].queue[i - 1]
await utils.reply(message, f"**X** {queued.format()}")
elif args.remove_title or args.remove_queuer:
targets = []
for queued in players[message.guild.id].queue:
if t := args.remove_title:
if t in queued.player.title:
targets.append(queued)
continue
if q := args.remove_queuer:
if q == queued.trigger_message.author.id:
targets.append(queued)
if not args.remove_multiple:
targets = targets[:1]
for target in targets:
players[message.guild.id].queue.remove(target)
await utils.reply(
message,
f"removed **{len(targets)}** queued {'song' if len(targets) == 1 else 'songs'}",
)
elif query := args.query:
if (
not message.channel.permissions_for(message.author).manage_channels
and len(
list(
filter(
lambda queued: queued.trigger_message.author.id
== message.author.id,
players[message.guild.id].queue,
)
)
)
>= 5
and not len(message.guild.voice_client.channel.members) == 2
):
await utils.reply(
message,
"you can only queue **5 items** without the manage channels permission!",
)
return
try:
async with message.channel.typing():
player = await youtubedl.YTDLSource.from_url(
query, loop=client.loop, stream=True
)
player.volume = float(args.volume) / 100.0
except Exception as e:
await utils.reply(
message, f"**failed to queue:** `{e}`", suppress_embeds=True
)
return
queued = youtubedl.QueuedSong(player, message)
if args.now or args.next:
players[message.guild.id].queue_add_front(queued)
else:
players[message.guild.id].queue_add(queued)
if (
not message.guild.voice_client.is_playing()
and not message.guild.voice_client.is_paused()
):
play_next(message)
elif args.now:
message.guild.voice_client.stop()
else:
await utils.reply(
message,
f"**{len(players[message.guild.id].queue)}.** {queued.format()}",
)
else:
if tokens[0].lower() == "play":
await resume(message)
else:
if players[message.guild.id].queue:
formatted_duration = utils.format_duration(
sum(
[
queued.player.duration if queued.player.duration else 0
for queued in players[message.guild.id].queue
]
)
)
def embed(description):
e = disnake.Embed(
description=description,
color=constants.EMBED_COLOR,
)
if formatted_duration:
e.set_footer(text=f"{formatted_duration} long")
return e
await disnake_paginator.ButtonPaginator(
invalid_user_function=utils.invalid_user_handler,
color=constants.EMBED_COLOR,
segments=list(
map(
embed,
[
"\n\n".join(
[
f"**{i + 1}.** {queued.format(show_queuer=True, hide_preview=True, multiline=True)}"
for i, queued in batch
]
)
for batch in itertools.batched(
enumerate(players[message.guild.id].queue), 10
)
],
)
),
).start(disnake_paginator.wrappers.MessageInteractionWrapper(message))
else:
await utils.reply(
message,
"nothing is queued!",
)
async def playing(message):
if not command_allowed(message):
return
if message.guild.voice_client.source:
await utils.reply(
message,
f"{'(paused) ' if message.guild.voice_client.is_paused() else ''} {players[message.guild.id].current.format(show_queuer=True)}",
)
else:
await utils.reply(
message,
"nothing is playing!",
)
async def skip(message):
if not command_allowed(message):
return
if not players[message.guild.id].queue:
message.guild.voice_client.stop()
await utils.reply(
message,
"the queue is empty now!",
)
else:
message.guild.voice_client.stop()
await utils.add_check_reaction(message)
if (
not message.guild.voice_client.is_playing()
and not message.guild.voice_client.is_paused()
):
play_next(message)
async def join(message):
if message.guild.voice_client:
return await message.guild.voice_client.move_to(message.channel)
await message.channel.connect()
await utils.add_check_reaction(message)
async def leave(message):
if not command_allowed(message):
return
await message.guild.voice_client.disconnect()
await utils.add_check_reaction(message)
async def resume(message):
if not command_allowed(message):
return
if message.guild.voice_client.is_paused():
message.guild.voice_client.resume()
await utils.add_check_reaction(message)
else:
await utils.reply(
message,
"nothing is paused!",
)
async def pause(message):
if not command_allowed(message):
return
if message.guild.voice_client.is_playing():
message.guild.voice_client.pause()
await utils.add_check_reaction(message)
else:
await utils.reply(
message,
"nothing is playing!",
)
async def volume(message):
if not command_allowed(message):
return
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(tokens[0], "set the current volume level")
parser.add_argument(
"volume",
nargs="?",
type=lambda v: arguments.range_type(v, min=0, max=150),
help="the volume level (0 - 150)",
)
if not (args := await parser.parse_args(message, tokens)):
return
if not message.guild.voice_client.source:
await utils.reply(message, "nothing is playing!")
return
if args.volume is None:
await utils.reply(
message,
f"{int(message.guild.voice_client.source.volume * 100)}",
)
else:
message.guild.voice_client.source.volume = float(args.volume) / 100.0
await utils.add_check_reaction(message)
def delete_queued(messages):
if messages[0].guild.id not in players:
return
if len(players[messages[0].guild.id].queue) == 0:
return
found = []
for message in messages:
for queued in players[message.guild.id].queue:
if queued.trigger_message.id == message.id:
found.append(queued)
for queued in found:
players[messages[0].guild.id].queue.remove(queued)
def play_after_callback(e, message, once):
if e:
print(f"player error: {e}")
if not once:
play_next(message)
def play_next(message, once=False):
message.guild.voice_client.stop()
if players[message.guild.id].queue:
queued = players[message.guild.id].queue_pop()
try:
message.guild.voice_client.play(
queued.player, after=lambda e: play_after_callback(e, message, once)
)
except Exception as e:
client.loop.create_task(
utils.channel_send(message, f"error while trying to play: `{e}`")
)
return
client.loop.create_task(
utils.channel_send(message, f"**0.** {queued.format(show_queuer=True)}")
)
async def ensure_joined(message):
if message.guild.voice_client is None:
if message.author.voice:
await message.author.voice.channel.connect()
else:
await utils.reply(message, "You are not connected to a voice channel.")
def command_allowed(message):
if not message.author.voice or not message.guild.voice_client:
return False
return message.author.voice.channel.id == message.guild.voice_client.channel.id

20
commands/voice/__init__.py
View File

@@ -1,20 +0,0 @@
from .channel import join, leave
from .playback import fast_forward, pause, playing, resume, volume
from .queue import queue_or_play, skip
from .sponsorblock import sponsorblock_command
from .utils import remove_queued
__all__ = [
"fast_forward",
"join",
"leave",
"pause",
"playing",
"queue_or_play",
"remove_queued",
"resume",
"skip",
"skip",
"sponsorblock_command",
"volume",
]

24
commands/voice/channel.py
View File

@@ -1,24 +0,0 @@
import utils
from .utils import command_allowed
async def join(message):
if message.author.voice:
if message.guild.voice_client:
await message.guild.voice_client.move_to(message.channel)
else:
await message.author.voice.channel.connect()
else:
await utils.reply(message, "you are not connected to a voice channel!")
return
await utils.add_check_reaction(message)
async def leave(message):
if not command_allowed(message):
return
await message.guild.voice_client.disconnect()
await utils.add_check_reaction(message)

168
commands/voice/playback.py
View File

@@ -1,168 +0,0 @@
import disnake_paginator
import arguments
import commands
import sponsorblock
import utils
from constants import EMBED_COLOR
from state import players
from .utils import command_allowed
async def playing(message):
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(
tokens[0],
"get information about the currently playing song",
)
parser.add_argument(
"-d",
"--description",
action="store_true",
help="get the description",
)
if not (args := await parser.parse_args(message, tokens)):
return
if not command_allowed(message, immutable=True):
return
if source := message.guild.voice_client.source:
if args.description:
if description := source.description:
paginator = disnake_paginator.ButtonPaginator(
invalid_user_function=utils.invalid_user_handler,
color=EMBED_COLOR,
title=source.title,
segments=disnake_paginator.split(description),
)
for embed in paginator.embeds:
embed.url = source.original_url
await paginator.start(utils.MessageInteractionWrapper(message))
else:
await utils.reply(
message,
source.description or "no description found!",
)
return
await utils.reply(
message,
embed=players[message.guild.id].current.embed(
is_paused=message.guild.voice_client.is_paused(),
),
)
else:
await utils.reply(
message,
"nothing is playing!",
)
async def resume(message):
if not command_allowed(message):
return
if message.guild.voice_client.is_paused():
message.guild.voice_client.resume()
await utils.add_check_reaction(message)
else:
await utils.reply(
message,
"nothing is paused!",
)
async def pause(message):
if not command_allowed(message):
return
if message.guild.voice_client.is_playing():
message.guild.voice_client.pause()
await utils.add_check_reaction(message)
else:
await utils.reply(
message,
"nothing is playing!",
)
async def fast_forward(message):
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(tokens[0], "skip the current sponsorblock segment")
parser.add_argument(
"-s",
"--seconds",
nargs="?",
type=lambda v: arguments.range_type(v, lower=0, upper=300),
help="the number of seconds to fast forward instead",
)
if not (args := await parser.parse_args(message, tokens)):
return
if not command_allowed(message):
return
if not message.guild.voice_client.source:
await utils.reply(message, "nothing is playing!")
return
seconds = args.seconds
if not seconds:
video = await sponsorblock.get_segments(
players[message.guild.id].current.player.id,
)
if not video:
await utils.reply(
message,
"no sponsorblock segments were found for this video!",
)
return
progress = message.guild.voice_client.source.original.progress
for segment in video["segments"]:
begin, end = map(float, segment["segment"])
if progress >= begin and progress < end:
seconds = end - message.guild.voice_client.source.original.progress
if not seconds:
await utils.reply(message, "no sponsorblock segment is currently playing!")
return
message.guild.voice_client.pause()
message.guild.voice_client.source.original.fast_forward(seconds)
message.guild.voice_client.resume()
await utils.add_check_reaction(message)
async def volume(message):
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(tokens[0], "get or set the current volume level")
parser.add_argument(
"volume",
nargs="?",
type=lambda v: arguments.range_type(v, lower=0, upper=150),
help="the volume level (0 - 150)",
)
if not (args := await parser.parse_args(message, tokens)):
return
if not command_allowed(message, immutable=True):
return
if not message.guild.voice_client.source:
await utils.reply(message, "nothing is playing!")
return
if args.volume is None:
await utils.reply(
message,
f"{int(message.guild.voice_client.source.volume * 100)}",
)
else:
if not command_allowed(message):
return
message.guild.voice_client.source.volume = float(args.volume) / 100.0
await utils.add_check_reaction(message)

270
commands/voice/queue.py
View File

@@ -1,270 +0,0 @@
import itertools
import disnake
import disnake_paginator
import arguments
import audio
import commands
import utils
from constants import EMBED_COLOR
from state import client, players, trusted_users
from .playback import resume
from .utils import command_allowed, ensure_joined, play_next
async def queue_or_play(message, edited=False):
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(
tokens[0],
"queue a song, list the queue, or resume playback",
)
parser.add_argument("query", nargs="*", help="yt-dlp URL or query to get song")
parser.add_argument(
"-v",
"--volume",
default=50,
type=lambda v: arguments.range_type(v, lower=0, upper=150),
help="the volume level (0 - 150) for the specified song",
)
parser.add_argument(
"-i",
"--remove-index",
type=int,
nargs="*",
help="remove queued songs by index",
)
parser.add_argument(
"-m",
"--match-multiple",
action="store_true",
help="continue removing queued after finding a match",
)
parser.add_argument(
"-c",
"--clear",
action="store_true",
help="remove all queued songs",
)
parser.add_argument(
"--now",
action="store_true",
help="play the specified song immediately",
)
parser.add_argument(
"--next",
action="store_true",
help="play the specified song next",
)
parser.add_argument(
"-t",
"--remove-title",
help="remove queued songs by title",
)
parser.add_argument(
"-q",
"--remove-queuer",
type=int,
help="remove queued songs by queuer",
)
if not (args := await parser.parse_args(message, tokens)):
return
await ensure_joined(message)
if len(tokens) == 1 and tokens[0].lower() != "play":
if not command_allowed(message, immutable=True):
return
elif not command_allowed(message):
return
if message.guild.id not in players:
players[message.guild.id] = audio.queue.Player()
if edited:
found = next(
filter(
lambda queued: queued.trigger_message.id == message.id,
players[message.guild.id].queue,
),
None,
)
if found:
players[message.guild.id].queue.remove(found)
if args.clear:
players[message.guild.id].queue.clear()
await utils.add_check_reaction(message)
elif indices := args.remove_index:
targets = []
for i in indices:
if i <= 0 or i > len(players[message.guild.id].queue):
await utils.reply(message, f"invalid index `{i}`!")
return
targets.append(players[message.guild.id].queue[i - 1])
for target in targets:
if target in players[message.guild.id].queue:
players[message.guild.id].queue.remove(target)
if len(targets) == 1:
await utils.reply(message, f"**removed** {targets[0].format()}")
else:
await utils.reply(
message,
f"removed **{len(targets)}** queued {'song' if len(targets) == 1 else 'songs'}",
)
elif args.remove_title or args.remove_queuer:
targets = set()
for queued in players[message.guild.id].queue:
if t := args.remove_title:
if t in queued.player.title:
targets.add(queued)
if q := args.remove_queuer:
if q == queued.trigger_message.author.id:
targets.add(queued)
targets = list(targets)
if not args.match_multiple:
targets = targets[:1]
for target in targets:
players[message.guild.id].queue.remove(target)
await utils.reply(
message,
f"removed **{len(targets)}** queued {'song' if len(targets) == 1 else 'songs'}",
)
elif query := args.query:
if (
not message.channel.permissions_for(message.author).manage_channels
and len(
list(
filter(
lambda queued: queued.trigger_message.author.id
== message.author.id,
players[message.guild.id].queue,
),
),
)
>= 5
and not len(message.guild.voice_client.channel.members) == 2
and message.author.id not in trusted_users
):
await utils.reply(
message,
"you can only queue **5 items** without the manage channels permission!",
)
return
try:
async with message.channel.typing():
player = await audio.youtubedl.YTDLSource.from_url(
" ".join(query),
loop=client.loop,
stream=True,
)
player.volume = float(args.volume) / 100.0
except Exception as e:
await utils.reply(message, f"failed to queue: `{e}`")
return
queued = audio.queue.Song(player, message)
if args.now or args.next:
players[message.guild.id].queue_push_front(queued)
else:
players[message.guild.id].queue_push(queued)
if not message.guild.voice_client:
await utils.reply(message, "unexpected disconnect from voice channel!")
return
elif not message.guild.voice_client.source:
play_next(message, first=True)
elif args.now:
message.guild.voice_client.stop()
else:
await utils.reply(
message,
f"**{1 if args.next else len(players[message.guild.id].queue)}.** {queued.format()}",
)
utils.cooldown(message, 2)
elif tokens[0].lower() == "play":
await resume(message)
else:
if players[message.guild.id].queue:
formatted_duration = utils.format_duration(
sum(
[
queued.player.duration if queued.player.duration else 0
for queued in players[message.guild.id].queue
],
),
natural=True,
)
def embed(description):
e = disnake.Embed(
description=description,
color=EMBED_COLOR,
)
if formatted_duration and len(players[message.guild.id].queue) > 1:
e.set_footer(text=f"{formatted_duration} in total")
return e
await disnake_paginator.ButtonPaginator(
invalid_user_function=utils.invalid_user_handler,
color=EMBED_COLOR,
segments=list(
map(
embed,
[
"\n\n".join(
[
f"**{i + 1}.** {queued.format(show_queuer=True, hide_preview=True, multiline=True)}"
for i, queued in batch
],
)
for batch in itertools.batched(
enumerate(players[message.guild.id].queue),
10,
)
],
),
),
).start(utils.MessageInteractionWrapper(message))
else:
await utils.reply(
message,
"nothing is queued!",
)
async def skip(message):
tokens = commands.tokenize(message.content)
parser = arguments.ArgumentParser(tokens[0], "skip the song currently playing")
parser.add_argument(
"-n",
"--next",
action="store_true",
help="skip the next song",
)
if not (args := await parser.parse_args(message, tokens)):
return
if not command_allowed(message):
return
if not players[message.guild.id].queue:
message.guild.voice_client.stop()
await utils.reply(
message,
"the queue is empty now!",
)
elif args.next:
next = players[message.guild.id].queue.pop()
await utils.reply(message, f"**skipped** {next.format()}")
else:
message.guild.voice_client.stop()
await utils.add_check_reaction(message)
if not message.guild.voice_client.source:
play_next(message)

47
commands/voice/sponsorblock.py
View File

@@ -1,47 +0,0 @@
import disnake
import audio
import sponsorblock
import utils
from constants import EMBED_COLOR, SPONSORBLOCK_CATEGORY_NAMES
from state import players
from .utils import command_allowed
async def sponsorblock_command(message):
if not command_allowed(message, immutable=True):
return
if not message.guild.voice_client.source:
await utils.reply(message, "nothing is playing!")
return
progress = message.guild.voice_client.source.original.progress
video = await sponsorblock.get_segments(players[message.guild.id].current.player.id)
if not video:
await utils.reply(
message,
"no sponsorblock segments were found for this video!",
)
return
text = []
for segment in video["segments"]:
begin, end = map(int, segment["segment"])
if (category := segment["category"]) in SPONSORBLOCK_CATEGORY_NAMES:
category = SPONSORBLOCK_CATEGORY_NAMES[category]
current = "**" if progress >= begin and progress < end else ""
text.append(
f"{current}`{audio.utils.format_duration(begin)}` - `{audio.utils.format_duration(end)}`: {category}{current}",
)
await utils.reply(
message,
embed=disnake.Embed(
title="Sponsorblock segments",
description="\n".join(text),
color=EMBED_COLOR,
),
)

72
commands/voice/utils.py
View File

@@ -1,72 +0,0 @@
from logging import error
import disnake
import utils
from state import client, players
def play_after_callback(e, message, once):
if e:
error(f"player error: {e}")
if not once:
play_next(message)
def play_next(message, once=False, first=False):
if not message.guild.voice_client:
return
message.guild.voice_client.stop()
if not disnake.opus.is_loaded():
utils.load_opus()
if message.guild.id in players and players[message.guild.id].queue:
queued = players[message.guild.id].queue_pop()
message.guild.voice_client.play(
queued.player,
after=lambda e: play_after_callback(e, message, once),
)
embed = queued.embed()
if first and len(players[message.guild.id].queue) == 0:
client.loop.create_task(utils.reply(message, embed=embed))
else:
client.loop.create_task(utils.channel_send(message, embed=embed))
def remove_queued(messages):
if messages[0].guild.id not in players:
return
if len(players[messages[0].guild.id].queue) == 0:
return
found = []
for message in messages:
for queued in players[message.guild.id].queue:
if queued.trigger_message.id == message.id:
found.append(queued)
for queued in found:
players[messages[0].guild.id].queue.remove(queued)
async def ensure_joined(message):
if message.guild.voice_client is None:
if message.author.voice:
await message.author.voice.channel.connect()
else:
await utils.reply(message, "you are not connected to a voice channel!")
def command_allowed(message, immutable=False):
if not message.guild.voice_client:
return False
if immutable:
return True
if not message.author.voice:
return False
return message.author.voice.channel.id == message.guild.voice_client.channel.id

110
constants.py
View File

@@ -1,5 +1,24 @@
import os
EMBED_COLOR = 0xFF6600
OWNERS = [531392146767347712]
PREFIX = "%"
RELOADABLE_MODULES = [
"arguments",
"commands",
"commands.bot",
"commands.tools",
"commands.utils",
"commands.voice",
"constants",
"core",
"events",
"tasks",
"utils",
"voice",
"youtubedl",
]
YTDL_OPTIONS = {
"color": "never",
"default_search": "auto",
@@ -15,97 +34,6 @@ YTDL_OPTIONS = {
"source_address": "0.0.0.0",
}
BAR_LENGTH = 35
EMBED_COLOR = 0xFF6600
OWNERS = [531392146767347712]
PREFIX = "%"
SPONSORBLOCK_CATEGORY_NAMES = {
"music_offtopic": "non-music",
"selfpromo": "self promotion",
"sponsor": "sponsored",
}
REACTIONS = {
"cat": ["🐈"],
"dog": ["🐕"],
"gn": ["💤", "😪", "😴", "🛌"],
"pizza": ["🍕"],
}
RELOADABLE_MODULES = [
"arguments",
"audio",
"audio.discord",
"audio.queue",
"audio.utils",
"audio.youtubedl",
"commands",
"commands.bot",
"commands.tools",
"commands.utils",
"commands.voice",
"commands.voice.channel",
"commands.voice.playback",
"commands.voice.playing",
"commands.voice.queue",
"commands.voice.sponsorblock",
"commands.voice.utils",
"constants",
"core",
"events",
"extra",
"fun",
"sponsorblock",
"tasks",
"utils",
"utils.common",
"utils.discord",
"voice",
"yt_dlp",
"yt_dlp.version",
]
PUBLIC_FLAGS = {
1 << 0: "Discord Employee",
1 << 1: "Discord Partner",
1 << 2: "HypeSquad Events",
1 << 3: "Bug Hunter Level 1",
1 << 6: "HypeSquad Bravery",
1 << 7: "HypeSquad Brilliance",
1 << 8: "HypeSquad Balance",
1 << 9: "Early Supporter",
1 << 10: "Team User",
1 << 14: "Bug Hunter Level 2",
1 << 16: "Verified Bot",
1 << 17: "Verified Bot Developer",
1 << 18: "Discord Certified Moderator",
1 << 19: "HTTP Interactions Only",
1 << 22: "Active Developer",
}
BADGE_EMOJIS = {
"Discord Employee": "<:DiscordStaff:879666899980546068>",
"Discord Partner": "<:DiscordPartner:879668340434534400>",
"HypeSquad Events": "<:HypeSquadEvents:879666970310606848>",
"Bug Hunter Level 1": "<:BugHunter1:879666851448234014>",
"HypeSquad Bravery": "<:HypeSquadBravery:879666945153175612>",
"HypeSquad Brilliance": "<:HypeSquadBrilliance:879666956884643861>",
"HypeSquad Balance": "<:HypeSquadBalance:879666934717771786>",
"Early Supporter": "<:EarlySupporter:879666916493496400>",
"Team User": "<:TeamUser:890866907996127305>",
"Bug Hunter Level 2": "<:BugHunter2:879666866971357224>",
"Verified Bot": "<:VerifiedBot:879670687554498591>",
"Verified Bot Developer": "<:VerifiedBotDeveloper:879669786550890507>",
"Discord Certified Moderator": "<:DiscordModerator:879666882976837654>",
"HTTP Interactions Only": "<:HTTPInteractionsOnly:1047141867806015559>",
"Active Developer": "<:ActiveDeveloper:1047141451244523592>",
}
APPLICATION_FLAGS = {
1 << 12: "Presence Intent",
1 << 13: "Presence Intent (unverified)",
1 << 14: "Guild Members Intent",
1 << 15: "Guild Members Intent (unverified)",
1 << 16: "Unusual Growth (verification suspended)",
1 << 18: "Message Content Intent",
1 << 19: "Message Content Intent (unverified)",
1 << 23: "Suports Application Commands",
}
SECRETS = {
"TOKEN": os.getenv("BOT_TOKEN"),

129
core.py
View File

@@ -3,24 +3,21 @@ import contextlib
import importlib
import inspect
import io
import signal
import textwrap
import time
import traceback
from logging import debug
import disnake
import disnake_paginator
import commands
import constants
import utils
from commands import Command as C
from constants import EMBED_COLOR, OWNERS, PREFIX, RELOADABLE_MODULES
from state import client, command_cooldowns, command_locks, idle_tracker
from state import client, command_locks, idle_tracker
async def on_message(message, edited=False):
if not message.content.startswith(PREFIX) or message.author.bot:
if not message.content.startswith(constants.PREFIX) or message.author.bot:
return
tokens = commands.tokenize(message.content)
@@ -41,56 +38,50 @@ async def on_message(message, edited=False):
f"ambiguous command, could be {' or '.join([f'`{match.value}`' for match in matched])}",
)
return
matched = matched[0]
if (message.guild.id, message.author.id) not in command_locks:
command_locks[(message.guild.id, message.author.id)] = asyncio.Lock()
await command_locks[(message.guild.id, message.author.id)].acquire()
if message.guild.id not in command_locks:
command_locks[message.guild.id] = asyncio.Lock()
C = commands.Command
try:
if (cooldowns := command_cooldowns.get(message.author.id)) and not edited:
if (end_time := cooldowns.get(matched)) and (
remaining_time := round(end_time - time.time()) > 0
):
await utils.reply(
message,
f"please wait **{utils.format_duration(remaining_time, natural=True)}** before using this command again!",
)
return
match matched:
case C.RELOAD if message.author.id in OWNERS:
start = time.time()
reloaded_modules = reload()
end = time.time()
debug(
f"reloaded {len(reloaded_modules)} modules in {round(end - start, 2)}s",
)
match matched[0]:
case C.RELOAD if message.author.id in constants.OWNERS:
reloaded_modules = set()
for module in filter(
lambda v: inspect.ismodule(v)
and v.__name__ in constants.RELOADABLE_MODULES,
globals().values(),
):
core.rreload(reloaded_modules, module)
await utils.add_check_reaction(message)
case C.EXECUTE if message.author.id in OWNERS:
case C.EXECUTE if message.author.id in constants.OWNERS:
code = message.content[len(tokens[0]) + 1 :].strip().strip("`")
for replacement in ["python", "py"]:
if code.startswith(replacement):
code = code[len(replacement) :]
stdout = io.StringIO()
try:
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
wrapped_code = (
f"async def run_code():\n{textwrap.indent(code, ' ')}"
)
if "# globals" in code:
exec(wrapped_code, globals())
if "#globals" in code:
exec(
f"async def run_code():\n{textwrap.indent(code, ' ')}",
globals(),
)
await globals()["run_code"]()
else:
dictionary = dict(locals(), **globals())
exec(wrapped_code, dictionary, dictionary)
exec(
f"async def run_code():\n{textwrap.indent(code, ' ')}",
dictionary,
dictionary,
)
await dictionary["run_code"]()
output = stdout.getvalue()
except Exception as e:
output = "`" + str(e) + "`"
output = utils.filter_secrets(output)
if len(output) > 2000:
@@ -99,24 +90,27 @@ async def on_message(message, edited=False):
prefix="```\n",
suffix="```",
invalid_user_function=utils.invalid_user_handler,
color=EMBED_COLOR,
color=constants.EMBED_COLOR,
segments=disnake_paginator.split(output),
).start(utils.MessageInteractionWrapper(message))
).start(
disnake_paginator.wrappers.MessageInteractionWrapper(message)
)
elif len(output.strip()) == 0:
await utils.add_check_reaction(message)
else:
await utils.reply(message, output)
case C.CLEAR | C.PURGE if message.author.id in OWNERS:
await utils.channel_send(message, output)
case C.CLEAR | C.PURGE if message.author.id in constants.OWNERS:
await commands.tools.clear(message)
case C.JOIN:
await commands.voice.join(message)
case C.LEAVE:
await commands.voice.leave(message)
case C.QUEUE | C.PLAY:
await commands.voice.queue_or_play(message, edited)
async with command_locks[message.guild.id]:
await commands.voice.queue_or_play(message, edited)
case C.SKIP:
await commands.voice.skip(message)
async with command_locks[message.guild.id]:
await commands.voice.skip(message)
case C.RESUME:
await commands.voice.resume(message)
case C.PAUSE:
@@ -127,49 +121,35 @@ async def on_message(message, edited=False):
await commands.bot.help(message)
case C.UPTIME:
await commands.bot.uptime(message)
case C.PLAYING | C.CURRENT:
case C.PLAYING:
await commands.voice.playing(message)
case C.FAST_FORWARD:
await commands.voice.fast_forward(message)
case C.STATUS:
await commands.bot.status(message)
case C.PING:
await commands.bot.ping(message)
case C.LOOKUP:
await commands.tools.lookup(message)
case C.SPONSORBLOCK:
await commands.voice.sponsorblock_command(message)
except Exception as e:
await utils.reply(
message,
f"exception occurred while processing command: ```\n{''.join(traceback.format_exception(e)).replace('`', '\\`')}```",
)
raise e
finally:
command_locks[(message.guild.id, message.author.id)].release()
async def on_voice_state_update(_, before, after):
def is_empty(channel):
return [m.id for m in (channel.members if channel else [])] == [client.user.id]
channel = None
c = None
if is_empty(before.channel):
channel = before.channel
c = before.channel
elif is_empty(after.channel):
channel = after.channel
if channel:
await channel.guild.voice_client.disconnect()
c = after.channel
if c:
await c.guild.voice_client.disconnect()
def rreload(reloaded_modules, module):
reloaded_modules.add(module.__name__)
for submodule in filter(
lambda sm: inspect.ismodule(sm)
and sm.__name__ in RELOADABLE_MODULES
and sm.__name__ not in reloaded_modules,
lambda v: inspect.ismodule(v)
and v.__name__ in constants.RELOADABLE_MODULES
and v.__name__ not in reloaded_modules,
vars(module).values(),
):
rreload(reloaded_modules, submodule)
@@ -178,18 +158,3 @@ def rreload(reloaded_modules, module):
if "__reload_module__" in dir(module):
module.__reload_module__()
def reload(*_):
reloaded_modules = set()
rreload(reloaded_modules, __import__("core"))
rreload(reloaded_modules, __import__("extra"))
for module in filter(
lambda v: inspect.ismodule(v) and v.__name__ in RELOADABLE_MODULES,
globals().values(),
):
rreload(reloaded_modules, module)
return reloaded_modules
signal.signal(signal.SIGUSR1, reload)

42
events.py
View File

@@ -1,15 +1,13 @@
import asyncio
import threading
from logging import debug, info, warning
import commands
import core
import fun
import tasks
from state import client
def prepare():
async def on_ready():
threading.Thread(
name="cleanup",
target=asyncio.run_coroutine_threadsafe,
@@ -20,17 +18,8 @@ def prepare():
).start()
async def on_bulk_message_delete(messages):
commands.voice.remove_queued(messages)
async def on_message(message):
await core.on_message(message)
await fun.on_message(message)
async def on_message_delete(message):
commands.voice.remove_queued([message])
async def on_message_edit(before, after):
@@ -40,31 +29,24 @@ async def on_message_edit(before, after):
await core.on_message(after, edited=True)
async def on_message_delete(message):
commands.voice.delete_queued([message])
async def on_bulk_message_delete(messages):
commands.voice.delete_queued(messages)
async def on_voice_state_update(member, before, after):
await core.on_voice_state_update(member, before, after)
async def on_ready():
info(f"logged in as {client.user}")
async def on_connect():
debug("connected to the gateway!")
async def on_disconnect():
warning("disconnected from the gateway!")
for event_type, handlers in client.get_listeners().items():
for handler in handlers:
client.remove_listener(handler, event_type)
for k, v in client.get_listeners().items():
for f in v:
client.remove_listener(f, k)
client.add_listener(on_bulk_message_delete, "on_bulk_message_delete")
client.add_listener(on_connect, "on_connect")
client.add_listener(on_disconnect, "on_disconnect")
client.add_listener(on_message, "on_message")
client.add_listener(on_message_delete, "on_message_delete")
client.add_listener(on_message_edit, "on_message_edit")
client.add_listener(on_ready, "on_ready")
client.add_listener(on_voice_state_update, "on_voice_state_update")

101
extra.py
View File

@@ -1,101 +0,0 @@
import asyncio
import string
import disnake
import youtube_transcript_api
from state import client, kill, players
async def transcript(
message,
languages=["en"],
max_messages=6,
min_messages=3,
upper=True,
):
initial_id = message.guild.voice_client.source.id
transcript_list = youtube_transcript_api.YouTubeTranscriptApi.list_transcripts(
initial_id,
)
try:
transcript = transcript_list.find_manually_created_transcript(languages).fetch()
except Exception:
transcript = transcript_list.find_generated_transcript(languages).fetch()
await message.channel.send("(autogenerated)")
messages = []
for line in transcript:
if (
players[message.guild.id].current.player.original.progress
>= line["start"] + line["duration"]
):
continue
while (
players[message.guild.id].current.player.original.progress < line["start"]
):
await asyncio.sleep(0.2)
messages.insert(
0,
await message.channel.send(line["text"].upper() if upper else line["text"]),
)
if len(messages) > max_messages:
try:
count = min(min_messages, len(messages))
if count == 1:
await messages.pop().delete()
else:
await message.channel.delete_messages(
[messages.pop() for _ in range(count)],
)
except Exception:
pass
if (message.guild.voice_client.source.id != initial_id) or kill["transcript"]:
kill["transcript"] = False
break
def messages_per_second(limit=500):
oldest = 2**64
newest = 0
guilds = set()
members = set()
cached_messages = list(client.cached_messages)[-limit:]
for message in cached_messages:
if message.guild:
guilds.add(message.guild.id)
members.add(message.author.id)
t = message.created_at.timestamp()
if t < oldest:
oldest = t
elif t > newest:
newest = t
average = round(len(cached_messages) / (newest - oldest), 1)
if average == 1.0:
average = 1
print(
f"I am receiving **{average} {'message' if average == 1 else 'messages'} per second** "
f"from **{len(members)} {'member' if len(members) == 1 else 'members'}** across **{len(guilds)} {'guild' if len(guilds) == 1 else 'guilds'}**",
)
async def auto_count(channel_id: int):
if (channel := await client.fetch_channel(channel_id)) and isinstance(
channel,
disnake.TextChannel,
):
last_message = (await channel.history(limit=1).flatten())[0]
try:
result = str(
int("".join(filter(lambda d: d in string.digits, last_message.content)))
+ 1,
)
except Exception:
result = "where number"
await channel.send(result)

13
fun.py
View File

@@ -1,13 +0,0 @@
import random
import commands
from constants import REACTIONS
async def on_message(message):
if random.random() < 0.01:
tokens = commands.tokenize(message.content, remove_prefix=False)
for keyword, options in REACTIONS.items():
if keyword in tokens:
await message.add_reaction(random.choice(options))
break

24
main.py
View File

@@ -1,20 +1,16 @@
import logging
import time
import constants
import events
from state import client
from state import client, start_time
@client.event
async def on_ready():
print(f"logged in as {client.user} in {round(time.time() - start_time, 1)}s")
await events.on_ready()
if __name__ == "__main__":
logging.basicConfig(
format=(
"%(asctime)s %(levelname)s %(name)s:%(module)s %(message)s"
if __debug__
else "%(asctime)s %(levelname)s %(message)s"
),
datefmt="%Y-%m-%d %T",
level=logging.DEBUG if __debug__ else logging.INFO,
)
logging.getLogger("disnake").setLevel(logging.WARNING)
events.prepare()
client.run(constants.SECRETS["TOKEN"])

5
requirements.txt
View File

@@ -1,8 +1,5 @@
aiohttp
audioop-lts
disnake
disnake_paginator
psutil
PyNaCl
youtube_transcript_api
yt-dlp[default] @ https://github.com/yt-dlp/yt-dlp/archive/master.tar.gz
yt-dlp

37
sponsorblock.py
View File

@@ -1,37 +0,0 @@
import hashlib
import json
import aiohttp
from state import sponsorblock_cache
categories = json.dumps(
[
"interaction",
"intro",
"music_offtopic",
"outro",
"preview",
"selfpromo",
"sponsor",
],
)
async def get_segments(video_id: str):
if video_id in sponsorblock_cache:
return sponsorblock_cache[video_id]
hash_prefix = hashlib.sha256(video_id.encode()).hexdigest()[:4]
session = aiohttp.ClientSession()
response = await session.get(
f"https://sponsor.ajay.app/api/skipSegments/{hash_prefix}",
params={"categories": categories},
)
if response.status == 200 and (
results := list(
filter(lambda v: video_id == v["videoID"], await response.json()),
)
):
sponsorblock_cache[video_id] = results[0]
return results[0]

22
state.py
View File

@@ -1,20 +1,32 @@
import collections
import time
import disnake
from utils import LimitedSizeDict
class LimitedSizeDict(collections.OrderedDict):
def __init__(self, *args, **kwds):
self.size_limit = kwds.pop("size_limit", 1000)
super().__init__(*args, **kwds)
self._check_size_limit()
def __setitem__(self, key, value):
super().__setitem__(key, value)
self._check_size_limit()
def _check_size_limit(self):
if self.size_limit is not None:
while len(self) > self.size_limit:
self.popitem(last=False)
intents = disnake.Intents.default()
intents.message_content = True
intents.members = True
client = disnake.Client(intents=intents)
command_cooldowns = LimitedSizeDict()
command_locks = LimitedSizeDict()
idle_tracker = {"is_idle": False, "last_used": time.time()}
kill = {"transcript": False}
message_responses = LimitedSizeDict()
players = {}
sponsorblock_cache = LimitedSizeDict()
start_time = time.time()
trusted_users = []

15
tasks.py
View File

@@ -1,6 +1,5 @@
import asyncio
import time
from logging import debug, error
import disnake
@@ -8,25 +7,19 @@ from state import client, idle_tracker, players
async def cleanup():
debug("spawned cleanup thread")
while True:
await asyncio.sleep(3600)
targets = []
for guild_id, player in players.items():
for id, player in players:
if len(player.queue) == 0:
targets.append(guild_id)
targets.append(id)
for target in targets:
del players[target]
debug(f"cleanup thread removed {len(targets)} empty players")
if (
not idle_tracker["is_idle"]
and time.time() - idle_tracker["last_used"] >= 3600
):
try:
await client.change_presence(status=disnake.Status.idle)
idle_tracker["is_idle"] = True
except Exception as e:
error(f"failed to change status to idle: {e}")
await client.change_presence(status=disnake.Status.idle)
idle_tracker["is_idle"] = True

3
tests/__init__.py
View File

@@ -1,3 +0,0 @@
from . import test_filter_secrets, test_format_duration
__all__ = ["test_filter_secrets", "test_format_duration"]

21
tests/test_filter_secrets.py
View File

@@ -1,21 +0,0 @@
import unittest
import utils
class TestFilterSecrets(unittest.TestCase):
def test_filter_secrets(self):
secret = "PLACEHOLDER_TOKEN"
self.assertFalse(
secret in utils.filter_secrets(f"HELLO{secret}WORLD", {"TOKEN": secret}),
)
self.assertFalse(secret in utils.filter_secrets(secret, {"TOKEN": secret}))
self.assertFalse(
secret in utils.filter_secrets(f"123{secret}", {"TOKEN": secret}),
)
self.assertFalse(
secret in utils.filter_secrets(f"{secret}{secret}", {"TOKEN": secret}),
)
self.assertFalse(
secret in utils.filter_secrets(f"{secret}@#(*&*$)", {"TOKEN": secret}),
)

107
tests/test_format_duration.py
View File

@@ -1,107 +0,0 @@
import unittest
import audio
import utils
class TestFormatDuration(unittest.TestCase):
def test_audio(self):
def f(s):
return audio.utils.format_duration(s)
self.assertEqual(f(0), "00:00")
self.assertEqual(f(0.5), "00:00")
self.assertEqual(f(60.5), "01:00")
self.assertEqual(f(1), "00:01")
self.assertEqual(f(60), "01:00")
self.assertEqual(f(60 + 30), "01:30")
self.assertEqual(f(60 * 60), "01:00:00")
self.assertEqual(f(60 * 60 + 30), "01:00:30")
def test_utils(self):
def f(s):
return utils.format_duration(s)
self.assertEqual(f(0), "")
self.assertEqual(f(60 * 60 * 24 * 7), "1 week")
self.assertEqual(f(60 * 60 * 24 * 21), "3 weeks")
self.assertEqual(
f((60 * 60 * 24 * 21) - 1),
"2 weeks, 6 days, 23 hours, 59 minutes, 59 seconds",
)
self.assertEqual(f(60), "1 minute")
self.assertEqual(f(60 * 2), "2 minutes")
self.assertEqual(f(60 * 59), "59 minutes")
self.assertEqual(f(60 * 60), "1 hour")
self.assertEqual(f(60 * 60 * 2), "2 hours")
self.assertEqual(f(1), "1 second")
self.assertEqual(f(60 + 5), "1 minute, 5 seconds")
self.assertEqual(f(60 * 60 + 30), "1 hour, 30 seconds")
self.assertEqual(f(60 * 60 + 60 + 30), "1 hour, 1 minute, 30 seconds")
self.assertEqual(f(60 * 60 * 24 * 7 + 30), "1 week, 30 seconds")
def test_utils_natural(self):
def f(s):
return utils.format_duration(s, natural=True)
self.assertEqual(f(0), "")
self.assertEqual(f(60 * 60 * 24 * 7), "1 week")
self.assertEqual(f(60 * 60 * 24 * 21), "3 weeks")
self.assertEqual(
f((60 * 60 * 24 * 21) - 1),
"2 weeks, 6 days, 23 hours, 59 minutes and 59 seconds",
)
self.assertEqual(f(60), "1 minute")
self.assertEqual(f(60 * 2), "2 minutes")
self.assertEqual(f(60 * 59), "59 minutes")
self.assertEqual(f(60 * 60), "1 hour")
self.assertEqual(f(60 * 60 * 2), "2 hours")
self.assertEqual(f(1), "1 second")
self.assertEqual(f(60 + 5), "1 minute and 5 seconds")
self.assertEqual(f(60 * 60 + 30), "1 hour and 30 seconds")
self.assertEqual(f(60 * 60 + 60 + 30), "1 hour, 1 minute and 30 seconds")
self.assertEqual(f(60 * 60 * 24 * 7 + 30), "1 week and 30 seconds")
def test_utils_short(self):
def f(s):
return utils.format_duration(s, short=True)
self.assertEqual(f(0), "")
self.assertEqual(f(60 * 60 * 24 * 7), "1w")
self.assertEqual(f(60 * 60 * 24 * 21), "3w")
self.assertEqual(
f((60 * 60 * 24 * 21) - 1),
"2w 6d 23h 59m 59s",
)
self.assertEqual(f(60), "1m")
self.assertEqual(f(60 * 2), "2m")
self.assertEqual(f(60 * 59), "59m")
self.assertEqual(f(60 * 60), "1h")
self.assertEqual(f(60 * 60 * 2), "2h")
self.assertEqual(f(1), "1s")
self.assertEqual(f(60 + 5), "1m 5s")
self.assertEqual(f(60 * 60 + 30), "1h 30s")
self.assertEqual(f(60 * 60 + 60 + 30), "1h 1m 30s")
self.assertEqual(f(60 * 60 * 24 * 7 + 30), "1w 30s")
def test_utils_natural_short(self):
def f(s):
return utils.format_duration(s, natural=True, short=True)
self.assertEqual(f(0), "")
self.assertEqual(f(60 * 60 * 24 * 7), "1w")
self.assertEqual(f(60 * 60 * 24 * 21), "3w")
self.assertEqual(
f((60 * 60 * 24 * 21) - 1),
"2w 6d 23h 59m and 59s",
)
self.assertEqual(f(60), "1m")
self.assertEqual(f(60 * 2), "2m")
self.assertEqual(f(60 * 59), "59m")
self.assertEqual(f(60 * 60), "1h")
self.assertEqual(f(60 * 60 * 2), "2h")
self.assertEqual(f(1), "1s")
self.assertEqual(f(60 + 5), "1m and 5s")
self.assertEqual(f(60 * 60 + 30), "1h and 30s")
self.assertEqual(f(60 * 60 + 60 + 30), "1h 1m and 30s")
self.assertEqual(f(60 * 60 * 24 * 7 + 30), "1w and 30s")

74
utils.py Normal file
View File

@@ -0,0 +1,74 @@
import disnake
import constants
from state import message_responses
def format_duration(duration: int):
def format_plural(noun, count):
return noun if count == 1 else noun + "s"
segments = []
weeks, duration = divmod(duration, 604800)
if weeks > 0:
segments.append(f"{weeks} {format_plural('week', weeks)}")
days, duration = divmod(duration, 86400)
if days > 0:
segments.append(f"{days} {format_plural('day', days)}")
hours, duration = divmod(duration, 3600)
if hours > 0:
segments.append(f"{hours} {format_plural('hour', hours)}")
minutes, duration = divmod(duration, 60)
if minutes > 0:
segments.append(f"{minutes} {format_plural('minute', minutes)}")
if duration > 0:
segments.append(f"{duration} {format_plural('second', duration)}")
return ", ".join(segments)
async def add_check_reaction(message):
await message.add_reaction("✅")
async def reply(message, *args, **kwargs):
if message.id in message_responses:
await message_responses[message.id].edit(
*args, **kwargs, allowed_mentions=disnake.AllowedMentions.none()
)
else:
response = await message.reply(
*args, **kwargs, allowed_mentions=disnake.AllowedMentions.none()
)
message_responses[message.id] = response
async def channel_send(message, *args, **kwargs):
if message.id in message_responses:
await message_responses[message.id].edit(
*args, **kwargs, allowed_mentions=disnake.AllowedMentions.none()
)
else:
response = await message.channel.send(
*args, **kwargs, allowed_mentions=disnake.AllowedMentions.none()
)
message_responses[message.id] = response
async def invalid_user_handler(interaction):
await interaction.response.send_message(
"You are not the intended receiver of this message!", ephemeral=True
)
def filter_secrets(text: str) -> str:
for secret_name, secret in constants.SECRETS.items():
if not secret:
continue
text = text.replace(secret, f"<{secret_name}>")
return text

28
utils/__init__.py
View File

@@ -1,28 +0,0 @@
from .common import LimitedSizeDict, filter_secrets, format_duration, surround
from .discord import (
ChannelResponseWrapper,
MessageInteractionWrapper,
add_check_reaction,
channel_send,
cooldown,
invalid_user_handler,
load_opus,
reply,
snowflake_timestamp,
)
__all__ = [
"add_check_reaction",
"channel_send",
"ChannelResponseWrapper",
"cooldown",
"filter_secrets",
"format_duration",
"invalid_user_handler",
"LimitedSizeDict",
"load_opus",
"MessageInteractionWrapper",
"reply",
"snowflake_timestamp",
"surround",
]

64
utils/common.py
View File

@@ -1,64 +0,0 @@
from collections import OrderedDict
from constants import SECRETS
def surround(inner: str, outer="```") -> str:
return outer + str(inner) + outer
def format_duration(duration: int, natural: bool = False, short: bool = False) -> str:
def format_plural(noun, count):
if short:
return noun[0]
return " " + (noun if count == 1 else noun + "s")
segments = []
weeks, duration = divmod(duration, 604800)
if weeks > 0:
segments.append(f"{weeks}{format_plural('week', weeks)}")
days, duration = divmod(duration, 86400)
if days > 0:
segments.append(f"{days}{format_plural('day', days)}")
hours, duration = divmod(duration, 3600)
if hours > 0:
segments.append(f"{hours}{format_plural('hour', hours)}")
minutes, duration = divmod(duration, 60)
if minutes > 0:
segments.append(f"{minutes}{format_plural('minute', minutes)}")
if duration > 0:
segments.append(f"{duration}{format_plural('second', duration)}")
separator = " " if short else ", "
if not natural or len(segments) <= 1:
return separator.join(segments)
return separator.join(segments[:-1]) + f" and {segments[-1]}"
def filter_secrets(text: str, secrets=SECRETS) -> str:
for secret_name, secret in secrets.items():
if not secret:
continue
text = text.replace(secret, f"<{secret_name}>")
return text
class LimitedSizeDict(OrderedDict):
def __init__(self, *args, **kwargs):
self.size_limit = kwargs.pop("size_limit", 100)
super().__init__(*args, **kwargs)
self._check_size_limit()
def __setitem__(self, key, value):
super().__setitem__(key, value)
self._check_size_limit()
def _check_size_limit(self):
if self.size_limit is not None:
while len(self) > self.size_limit:
self.popitem(last=False)

118
utils/discord.py
View File

@@ -1,118 +0,0 @@
import time
from logging import error, info
from pathlib import Path
import disnake
import commands
from constants import OWNERS
from state import command_cooldowns, message_responses
def cooldown(message, cooldown_time: int):
if message.author.id in OWNERS:
return
possible_commands = commands.match(message.content)
if not possible_commands or len(possible_commands) > 1:
return
command = possible_commands[0]
end_time = time.time() + cooldown_time
if message.author.id in command_cooldowns:
command_cooldowns[message.author.id][command] = end_time
else:
command_cooldowns[message.author.id] = {command: end_time}
async def reply(message, *args, **kwargs):
if message.id in message_responses:
if len(args) == 0:
kwargs["content"] = None
elif len(kwargs) == 0:
kwargs["embeds"] = []
try:
await message_responses[message.id].edit(
*args,
**kwargs,
allowed_mentions=disnake.AllowedMentions.none(),
)
return
except Exception:
pass
try:
response = await message.reply(
*args,
**kwargs,
allowed_mentions=disnake.AllowedMentions.none(),
)
except Exception:
response = await channel_send(message, *args, **kwargs)
message_responses[message.id] = response
return message_responses[message.id]
async def channel_send(message, *args, **kwargs):
await message.channel.send(
*args,
**kwargs,
allowed_mentions=disnake.AllowedMentions.none(),
)
def load_opus():
for path in filter(
lambda p: Path(p).exists(),
["/usr/lib64/libopus.so.0", "/usr/lib/libopus.so.0"],
):
try:
disnake.opus.load_opus(path)
info(f"successfully loaded opus from {path}")
return
except Exception as e:
error(f"failed to load opus from {path}: {e}")
raise Exception("could not locate working opus library")
def snowflake_timestamp(snowflake) -> int:
return round(((snowflake >> 22) + 1420070400000) / 1000)
async def add_check_reaction(message):
await message.add_reaction("✅")
async def invalid_user_handler(interaction):
await interaction.response.send_message(
"you are not the intended receiver of this message!",
ephemeral=True,
)
class ChannelResponseWrapper:
def __init__(self, message):
self.message = message
self.sent_message = None
async def send_message(self, **kwargs):
kwargs.pop("ephemeral", None)
self.sent_message = await reply(self.message, **kwargs)
async def edit_message(self, content=None, embed=None, view=None):
if self.sent_message:
content = content or self.sent_message.content
if not embed and len(self.sent_message.embeds) > 0:
embed = self.sent_message.embeds[0]
await self.sent_message.edit(content=content, embed=embed, view=view)
class MessageInteractionWrapper:
def __init__(self, message):
self.message = message
self.author = message.author
self.response = ChannelResponseWrapper(message)
async def edit_original_message(self, content=None, embed=None, view=None):
await self.response.edit_message(content=content, embed=embed, view=view)

109
youtubedl.py Normal file
View File

@@ -0,0 +1,109 @@
import asyncio
import collections
from dataclasses import dataclass
from typing import Any, Optional
import disnake
import yt_dlp
import constants
ytdl = yt_dlp.YoutubeDL(constants.YTDL_OPTIONS)
class YTDLSource(disnake.PCMVolumeTransformer):
def __init__(
self, source: disnake.AudioSource, *, data: dict[str, Any], volume: float = 0.5
):
super().__init__(source, volume)
self.title = data.get("title")
self.original_url = data.get("original_url")
self.duration = data.get("duration")
@classmethod
async def from_url(
cls,
url,
*,
loop: Optional[asyncio.AbstractEventLoop] = None,
stream: bool = False,
):
loop = loop or asyncio.get_event_loop()
data: Any = await loop.run_in_executor(
None, lambda: ytdl.extract_info(url, download=not stream)
)
if "entries" in data:
data = data["entries"][0]
return cls(
disnake.FFmpegPCMAudio(
data["url"] if stream else ytdl.prepare_filename(data),
before_options="-vn -reconnect 1",
),
data=data,
)
def __repr__(self):
return f"<YTDLSource title={self.title} original_url=<{self.original_url}> duration={self.duration}>"
def __str__(self):
return self.__repr__()
@dataclass
class QueuedSong:
player: YTDLSource
trigger_message: disnake.Message
def format(self, show_queuer=False, hide_preview=False, multiline=False) -> str:
if multiline:
return (
f"[`{self.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''})\n**duration:** {self.format_duration(self.player.duration) if self.player.duration else '[live]'}"
+ (
f", **queuer:** <@{self.trigger_message.author.id}>"
if show_queuer
else ""
)
)
else:
return (
f"[`{self.player.title}`]({'<' if hide_preview else ''}{self.player.original_url}{'>' if hide_preview else ''}) **[{self.format_duration(self.player.duration) if self.player.duration else 'live'}]**"
+ (f" (<@{self.trigger_message.author.id}>)" if show_queuer else "")
)
def format_duration(self, duration: int) -> str:
hours, duration = divmod(duration, 3600)
minutes, duration = divmod(duration, 60)
segments = [hours, minutes, duration]
if len(segments) == 3 and segments[0] == 0:
del segments[0]
return f"{':'.join(f'{s:0>2}' for s in segments)}"
def __str__(self):
return self.__repr__()
@dataclass
class QueuedPlayer:
queue = collections.deque()
current: Optional[QueuedSong] = None
def queue_pop(self):
popped = self.queue.popleft()
self.current = popped
return popped
def queue_add(self, item):
self.queue.append(item)
def queue_add_front(self, item):
self.queue.appendleft(item)
def __str__(self):
return self.__repr__()
def __reload_module__():
global ytdl
ytdl = yt_dlp.YoutubeDL(constants.YTDL_OPTIONS)