refactor: minor fixes everywhere

.envrc

@@ -1,6 +0,0 @@
-dotenv
-
-export VIRTUAL_ENV=."venv"
-layout python
-
-use flake

.gitignore

@@ -1,4 +1,2 @@
-.direnv
 .env
-.venv
 __pycache__

Dockerfile

@@ -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", "-m", "errornocord.main"]

README.md

@@ -1,3 +1,3 @@
 # ErrorNoCord
 
-Hot-reloadable Discord music bot
+Discord music bot for testing purposes, with live reloading support

arguments.py

@@ -2,15 +2,13 @@ import argparse
 import contextlib
 import io
 
-from . import utils
+import utils
 
 
 class ArgumentParser:
     def __init__(self, command, description):
         self.parser = argparse.ArgumentParser(
-            command,
+            command, description=description, exit_on_error=False
-            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:
+    except ValueError:
-        raise argparse.ArgumentTypeError("value is not a valid integer") from e
+        raise argparse.ArgumentTypeError("value is not a valid integer")
 
-    if lower <= value <= upper:
+    if min <= value <= max:
         return value
-
+    else:
-    raise argparse.ArgumentTypeError(f"value is not in range {lower}-{upper}")
+        raise argparse.ArgumentTypeError(f"value is not in range {min}-{max}")

commands/__init__.py

@@ -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("_")})

commands/bot.py

@@ -0,0 +1,38 @@
+import time
+
+import arguments
+import commands
+import utils
+from state import start_time
+
+
+async def help(message):
+    await utils.reply(
+        message,
+        ", ".join(
+            [f"`{command.value}`" for command in commands.Command.__members__.values()]
+        ),
+    )
+
+
+async def uptime(message):
+    tokens = commands.tokenize(message.content)
+    parser = arguments.ArgumentParser(
+        tokens[0],
+        "print bot uptime",
+    )
+    parser.add_argument(
+        "-s",
+        "--since",
+        action="store_true",
+        help="bot up since",
+    )
+    if not (args := await parser.parse_args(message, tokens)):
+        return
+
+    if args.since:
+        await utils.reply(message, f"{round(start_time)}")
+    else:
+        await utils.reply(
+            message, f"up {utils.format_duration(int(time.time() - start_time))}"
+        )

commands/tools.py

@@ -0,0 +1,104 @@
+import re
+
+import arguments
+import commands
+import utils
+
+
+async def clear(message):
+    tokens = commands.tokenize(message.content)
+    parser = arguments.ArgumentParser(
+        tokens[0],
+        "bulk delete messages in the current channel matching certain criteria",
+    )
+    parser.add_argument(
+        "count",
+        type=lambda c: arguments.range_type(c, min=1, max=1000),
+        help="amount of messages to delete",
+    )
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument(
+        "-r",
+        "--regex",
+        required=False,
+        help="delete messages with content matching this regex",
+    )
+    group.add_argument(
+        "-c",
+        "--contains",
+        required=False,
+        help="delete messages with content containing this substring",
+    )
+    parser.add_argument(
+        "-i",
+        "--case-insensitive",
+        action="store_true",
+        help="ignore case sensitivity when deleting messages",
+    )
+    parser.add_argument(
+        "-a",
+        "--author-id",
+        type=int,
+        action="append",
+        help="delete messages whose author matches this id",
+    )
+    parser.add_argument(
+        "-o",
+        "--oldest-first",
+        action="store_true",
+        help="delete oldest messages first",
+    )
+    parser.add_argument(
+        "-R",
+        "--reactions",
+        action="store_true",
+        help="delete messages with reactions",
+    )
+    parser.add_argument(
+        "-d",
+        "--delete-command",
+        action="store_true",
+        help="delete the command message as well",
+    )
+    if not (args := await parser.parse_args(message, tokens)):
+        return
+
+    if args.delete_command:
+        try:
+            await message.delete()
+        except Exception:
+            pass
+
+    regex = None
+    if r := args.regex:
+        regex = re.compile(r, re.IGNORECASE if args.case_insensitive else 0)
+
+    def check(m):
+        c = []
+        if regex:
+            c.append(regex.search(m.content))
+        if s := args.contains:
+            if args.case_insensitive:
+                c.append(s.lower() in m.content.lower())
+            else:
+                c.append(s in m.content)
+        if i := args.author_id:
+            c.append(m.author.id in i)
+        if args.reactions:
+            c.append(len(m.reactions) > 0)
+        return all(c)
+
+    messages = len(
+        await message.channel.purge(
+            limit=args.count, check=check, oldest_first=args.oldest_first
+        )
+    )
+
+    if not args.delete_command:
+        try:
+            await utils.reply(
+                message,
+                f"purged **{messages}/{args.count} {'message' if args.count == 1 else 'messages'}**",
+            )
+        except Exception:
+            pass

commands/utils.py

@@ -1,20 +1,15 @@
-from enum import Enum
+import enum
-from functools import lru_cache
 
-from .. import constants
+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():
+    if token.lower() == "r":
-        case "r":
         return [Command.RELOAD]
-        case "s":
-            return [Command.SKIP]
-        case "c":
-            return [Command.CURRENT]
 
     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) -> list[str]:
-def tokenize(string: str, remove_prefix: bool = True) -> list[str]:
     tokens = []
     token = ""
     in_quotes = False
     quote_char = None
     escape = False
 
-    if remove_prefix:
+    for char in string[len(constants.PREFIX) :]:
-        string = string[len(constants.PREFIX) :]
-
-    for char in string:
         if escape:
             token += char
             escape = False

commands/voice.py

@@ -0,0 +1,374 @@
+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):
+    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:
+        if queued in players[messages[0].guild.id].queue:
+            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

constants.py

@@ -0,0 +1,40 @@
+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",
+    "format": "bestaudio/best",
+    "ignoreerrors": False,
+    "logtostderr": False,
+    "no_warnings": True,
+    "noplaylist": True,
+    "outtmpl": "%(extractor)s-%(id)s-%(title)s.%(ext)s",
+    "quiet": True,
+    "restrictfilenames": True,
+    "socket_timeout": 15,
+    "source_address": "0.0.0.0",
+}
+
+
+SECRETS = {
+    "TOKEN": os.getenv("BOT_TOKEN"),
+}

core.py

@@ -0,0 +1,161 @@
+import asyncio
+import contextlib
+import importlib
+import inspect
+import io
+import textwrap
+import time
+import traceback
+
+import disnake
+import disnake_paginator
+
+import commands
+import constants
+import core
+import utils
+from state import client, command_locks, idle_tracker
+
+
+async def on_message(message, edited=False):
+    if not message.content.startswith(constants.PREFIX) or message.author.bot:
+        return
+
+    tokens = commands.tokenize(message.content)
+    if not tokens:
+        return
+    matched = commands.match_token(tokens[0])
+    if not matched:
+        return
+
+    idle_tracker["last_used"] = time.time()
+    if idle_tracker["is_idle"]:
+        idle_tracker["is_idle"] = False
+        await client.change_presence(status=disnake.Status.online)
+
+    if len(matched) > 1:
+        await utils.reply(
+            message,
+            f"ambiguous command, could be {' or '.join([f'`{match.value}`' for match in matched])}",
+        )
+        return
+
+    if message.guild.id not in command_locks:
+        command_locks[message.guild.id] = asyncio.Lock()
+
+    C = commands.Command
+    try:
+        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 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:
+                    with contextlib.redirect_stdout(stdout):
+                        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(
+                                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:
+                    output = output.replace("`", "\\`")
+                    await disnake_paginator.ButtonPaginator(
+                        prefix="```\n",
+                        suffix="```",
+                        invalid_user_function=utils.invalid_user_handler,
+                        color=constants.EMBED_COLOR,
+                        segments=disnake_paginator.split(output),
+                    ).start(
+                        disnake_paginator.wrappers.MessageInteractionWrapper(message)
+                    )
+                elif len(output.strip()) == 0:
+                    await utils.add_check_reaction(message)
+                else:
+                    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:
+                async with command_locks[message.guild.id]:
+                    await commands.voice.queue_or_play(message, edited)
+            case C.SKIP:
+                async with command_locks[message.guild.id]:
+                    await commands.voice.skip(message)
+            case C.RESUME:
+                await commands.voice.resume(message)
+            case C.PAUSE:
+                await commands.voice.pause(message)
+            case C.VOLUME:
+                await commands.voice.volume(message)
+            case C.HELP:
+                await commands.bot.help(message)
+            case C.UPTIME:
+                await commands.bot.uptime(message)
+            case C.PLAYING:
+                await commands.voice.playing(message)
+    except Exception as e:
+        await utils.reply(
+            message,
+            f"exception occurred while processing command: ```\n{''.join(traceback.format_exception(e)).replace('`', '\\`')}```",
+        )
+
+
+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]
+
+    c = None
+    if is_empty(before.channel):
+        c = before.channel
+    elif is_empty(after.channel):
+        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 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)
+
+    importlib.reload(module)
+
+    if "__reload_module__" in dir(module):
+        module.__reload_module__()

default.nix

@@ -1,59 +0,0 @@
-{
-  lib,
-  python3Packages,
-  self,
-  ...
-}:
-let
-  disnake = python3Packages.disnake.overrideAttrs (old: {
-    src = self.pins.disnake;
-
-    propagatedBuildInputs =
-      with python3Packages;
-      old.propagatedBuildInputs
-      ++ [
-        typing-extensions
-        versioningit
-      ];
-
-    nativeBuildInputs = old.nativeBuildInputs ++ [ python3Packages.hatchling ];
-  });
-
-  disnake_paginator = python3Packages.buildPythonPackage {
-    pname = "disnake-paginator";
-    version = "1.0.8";
-
-    src = self.pins.disnake-paginator;
-
-    pyproject = true;
-    build-system = [ python3Packages.setuptools ];
-
-    propagatedBuildInputs = [
-      disnake
-    ];
-
-    doCheck = false;
-  };
-in
-python3Packages.buildPythonApplication {
-  pname = "errornocord";
-  version = "0.1.0";
-
-  src = lib.cleanSource ./.;
-
-  pyproject = true;
-  build-system = [ python3Packages.setuptools ];
-
-  propagatedBuildInputs = with python3Packages; [
-    aiohttp
-    audioop-lts
-    disnake
-    disnake_paginator
-    psutil
-    typing-extensions
-    youtube-transcript-api
-    yt-dlp
-  ];
-
-  doCheck = false;
-}

errornocord/audio/__init__.py

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

errornocord/audio/discord.py

@@ -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)

errornocord/audio/queue.py

@@ -1,111 +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__()

errornocord/audio/utils.py

@@ -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)}"

errornocord/audio/youtubedl.py

@@ -1,75 +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)

errornocord/commands/bot.py

@@ -1,102 +0,0 @@
-import os
-import threading
-import time
-
-import disnake
-import psutil
-from yt_dlp import version
-
-from .. import arguments, commands
-from ..constants import EMBED_COLOR
-from ..state import client, start_time
-from ..utils import format_duration, reply, surround
-
-
-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"),
-    )
-    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):
-    tokens = commands.tokenize(message.content)
-    parser = arguments.ArgumentParser(
-        tokens[0],
-        "print bot uptime",
-    )
-    parser.add_argument(
-        "-s",
-        "--since",
-        action="store_true",
-        help="bot up since",
-    )
-    if not (args := await parser.parse_args(message, tokens)):
-        return
-
-    if args.since:
-        await 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()],
-        ),
-    )

errornocord/commands/tools.py

@@ -1,277 +0,0 @@
-import re
-
-import aiohttp
-import disnake
-
-from .. import arguments, commands, 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",
-    )
-    parser.add_argument(
-        "count",
-        type=lambda c: arguments.range_type(c, lower=1, upper=1000),
-        help="amount of messages to delete",
-    )
-    group = parser.add_mutually_exclusive_group()
-    group.add_argument(
-        "-r",
-        "--regex",
-        required=False,
-        help="delete messages with content matching this regex",
-    )
-    group.add_argument(
-        "-c",
-        "--contains",
-        required=False,
-        help="delete messages with content containing this substring",
-    )
-    parser.add_argument(
-        "-i",
-        "--case-insensitive",
-        action="store_true",
-        help="ignore case sensitivity when deleting messages",
-    )
-    parser.add_argument(
-        "-a",
-        "--author-id",
-        type=int,
-        action="append",
-        help="delete messages whose author matches this id",
-    )
-    parser.add_argument(
-        "-o",
-        "--oldest-first",
-        action="store_true",
-        help="delete oldest messages first",
-    )
-    parser.add_argument(
-        "-R",
-        "--reactions",
-        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
-
-    if args.delete_command:
-        try:
-            await message.delete()
-        except Exception:
-            pass
-
-    regex = None
-    if r := args.regex:
-        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))
-        if s := args.contains:
-            if args.case_insensitive:
-                c.append(s.lower() in m.content.lower())
-            else:
-                c.append(s in m.content)
-        if i := args.author_id:
-            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,
-        ),
-    )
-
-    if not args.delete_command:
-        try:
-            await utils.reply(
-                message,
-                f"purged **{messages}/{args.count} {'message' if args.count == 1 else 'messages'}**",
-            )
-        except Exception:
-            pass

errornocord/commands/voice/__init__.py

@@ -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",
-]

errornocord/commands/voice/channel.py

@@ -1,27 +0,0 @@
-from ... import utils
-from ...state import players
-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
-
-    if message.guild.id in players:
-        del players[message.guild.id]
-    await message.guild.voice_client.disconnect()
-
-    await utils.add_check_reaction(message)

errornocord/commands/voice/playback.py

@@ -1,166 +0,0 @@
-import disnake_paginator
-
-from ... import arguments, commands, sponsorblock, 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)

errornocord/commands/voice/queue.py

@@ -1,268 +0,0 @@
-import itertools
-
-import disnake
-import disnake_paginator
-
-from ... import arguments, audio, commands, 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 players[message.guild.id] and not players[message.guild.id].queue:
-        del players[message.guild.id]
-        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)

errornocord/commands/voice/sponsorblock.py

@@ -1,44 +0,0 @@
-import disnake
-
-from ... import audio, sponsorblock, 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,
-        ),
-    )

errornocord/commands/voice/utils.py

@@ -1,72 +0,0 @@
-from logging import error
-
-import disnake
-
-from ... 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

errornocord/constants.py

@@ -1,112 +0,0 @@
-import os
-
-YTDL_OPTIONS = {
-    "color": "never",
-    "default_search": "auto",
-    "format": "bestaudio/best",
-    "ignoreerrors": False,
-    "logtostderr": False,
-    "no_warnings": True,
-    "noplaylist": True,
-    "outtmpl": "%(extractor)s-%(id)s-%(title)s.%(ext)s",
-    "quiet": True,
-    "restrictfilenames": True,
-    "socket_timeout": 15,
-    "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 = [
-    "errornocord.arguments",
-    "errornocord.audio",
-    "errornocord.audio.discord",
-    "errornocord.audio.queue",
-    "errornocord.audio.utils",
-    "errornocord.audio.youtubedl",
-    "errornocord.commands",
-    "errornocord.commands.bot",
-    "errornocord.commands.tools",
-    "errornocord.commands.utils",
-    "errornocord.commands.voice",
-    "errornocord.commands.voice.channel",
-    "errornocord.commands.voice.playback",
-    "errornocord.commands.voice.playing",
-    "errornocord.commands.voice.queue",
-    "errornocord.commands.voice.sponsorblock",
-    "errornocord.commands.voice.utils",
-    "errornocord.constants",
-    "errornocord.core",
-    "errornocord.events",
-    "errornocord.extra",
-    "errornocord.fun",
-    "errornocord.sponsorblock",
-    "errornocord.tasks",
-    "errornocord.utils",
-    "errornocord.utils.common",
-    "errornocord.utils.discord",
-    "errornocord.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"),
-}

errornocord/core.py

@@ -1,198 +0,0 @@
-import asyncio
-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
-
-from . import commands, 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, players
-
-
-async def on_message(message, edited=False):
-    if not message.content.startswith(PREFIX) or message.author.bot:
-        return
-
-    tokens = commands.tokenize(message.content)
-    if not tokens:
-        return
-    matched = commands.match_token(tokens[0])
-    if not matched:
-        return
-
-    idle_tracker["last_used"] = time.time()
-    if idle_tracker["is_idle"]:
-        idle_tracker["is_idle"] = False
-        await client.change_presence(status=disnake.Status.online)
-
-    if len(matched) > 1:
-        await utils.reply(
-            message,
-            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()
-
-    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",
-                )
-
-                await utils.add_check_reaction(message)
-
-            case C.EXECUTE if message.author.id in OWNERS:
-                code = message.content[len(tokens[0]) + 1 :].strip().strip("`")
-                for replacement in ["python", "py"]:
-                    if code.startswith(replacement):
-                        code = code[len(replacement) :]
-
-                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())
-                            await globals()["run_code"]()
-                        else:
-                            dictionary = dict(locals(), **globals())
-                            exec(wrapped_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:
-                    output = output.replace("`", "\\`")
-                    await disnake_paginator.ButtonPaginator(
-                        prefix="```\n",
-                        suffix="```",
-                        invalid_user_function=utils.invalid_user_handler,
-                        color=EMBED_COLOR,
-                        segments=disnake_paginator.split(output),
-                    ).start(utils.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 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)
-            case C.SKIP:
-                await commands.voice.skip(message)
-            case C.RESUME:
-                await commands.voice.resume(message)
-            case C.PAUSE:
-                await commands.voice.pause(message)
-            case C.VOLUME:
-                await commands.voice.volume(message)
-            case C.HELP:
-                await commands.bot.help(message)
-            case C.UPTIME:
-                await commands.bot.uptime(message)
-            case C.PLAYING | C.CURRENT:
-                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(member, before, after):
-    def is_alone(channel):
-        return [m.id for m in (channel.members if channel else [])] == [client.user.id]
-
-    if member.id == client.user.id and is_alone(after.channel):
-        if before.channel.guild.id in players:
-            del players[before.channel.guild.id]
-        await after.channel.guild.voice_client.disconnect()
-        return
-
-    if is_alone(before.channel):
-        if before.channel.guild.id in players:
-            del players[before.channel.guild.id]
-        await before.channel.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
-        ),
-        vars(module).values(),
-    ):
-        rreload(reloaded_modules, submodule)
-
-    importlib.reload(module)
-
-    if "__reload_module__" in dir(module):
-        module.__reload_module__()
-
-
-def reload(*_):
-    reloaded_modules = set()
-    rreload(reloaded_modules, __import__("errornocord.core"))
-    rreload(reloaded_modules, __import__("errornocord.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)

errornocord/extra.py

@@ -1,98 +0,0 @@
-import asyncio
-import string
-
-import disnake
-from youtube_transcript_api._api import YouTubeTranscriptApi
-
-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 = YouTubeTranscriptApi().list(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.snippets:
-        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)

errornocord/fun.py

@@ -1,13 +0,0 @@
-import random
-
-from . 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

errornocord/main.py

@@ -1,24 +0,0 @@
-import logging
-
-from . import constants, events
-from .state import client
-
-
-def 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"])
-
-
-if __name__ == "__main__":
-    main()

errornocord/sponsorblock.py

@@ -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]

errornocord/state.py

@@ -1,20 +0,0 @@
-import time
-
-import disnake
-
-from .utils import LimitedSizeDict
-
-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 = []

errornocord/tasks.py

@@ -1,33 +0,0 @@
-import asyncio
-import time
-from logging import debug, error
-
-import disnake
-
-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():
-            if len(player.queue) == 0 and not player.current:
-                targets.append(guild_id)
-        for target in targets:
-            del players[target]
-        if len(targets):
-            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}")

errornocord/tests/__init__.py

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

errornocord/tests/test_filter_secrets.py

@@ -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}),
-        )

errornocord/tests/test_format_duration.py

@@ -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")

errornocord/utils/__init__.py

@@ -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",
-]

errornocord/utils/common.py

@@ -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)

errornocord/utils/discord.py

@@ -1,115 +0,0 @@
-import ctypes
-import time
-from logging import debug, error
-
-import disnake
-
-from .. 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():
-    path = ctypes.util._findLib_ld("opus")
-    try:
-        disnake.opus.load_opus(path)
-        debug(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)

events.py

@@ -1,12 +1,13 @@
 import asyncio
 import threading
-from logging import debug, info, warning
 
-from . import commands, core, fun, tasks
+import commands
-from .state import client
+import core
+import tasks
+from state import client
 
 
-def prepare():
+async def on_ready():
     threading.Thread(
         name="cleanup",
         target=asyncio.run_coroutine_threadsafe,
@@ -17,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):
@@ -37,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():
+for k, v in client.get_listeners().items():
-    info(f"logged in as {client.user}")
+    for f in v:
-
+        client.remove_listener(f, k)
-
-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)
 
 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")

flake.lock

@@ -1,61 +0,0 @@
-{
-  "nodes": {
-    "flake-parts": {
-      "inputs": {
-        "nixpkgs-lib": "nixpkgs-lib"
-      },
-      "locked": {
-        "lastModified": 1772408722,
-        "narHash": "sha256-rHuJtdcOjK7rAHpHphUb1iCvgkU3GpfvicLMwwnfMT0=",
-        "owner": "hercules-ci",
-        "repo": "flake-parts",
-        "rev": "f20dc5d9b8027381c474144ecabc9034d6a839a3",
-        "type": "github"
-      },
-      "original": {
-        "owner": "hercules-ci",
-        "repo": "flake-parts",
-        "type": "github"
-      }
-    },
-    "nixpkgs": {
-      "locked": {
-        "lastModified": 1773821835,
-        "narHash": "sha256-TJ3lSQtW0E2JrznGVm8hOQGVpXjJyXY2guAxku2O9A4=",
-        "owner": "NixOS",
-        "repo": "nixpkgs",
-        "rev": "b40629efe5d6ec48dd1efba650c797ddbd39ace0",
-        "type": "github"
-      },
-      "original": {
-        "owner": "NixOS",
-        "ref": "nixos-unstable",
-        "repo": "nixpkgs",
-        "type": "github"
-      }
-    },
-    "nixpkgs-lib": {
-      "locked": {
-        "lastModified": 1772328832,
-        "narHash": "sha256-e+/T/pmEkLP6BHhYjx6GmwP5ivonQQn0bJdH9YrRB+Q=",
-        "owner": "nix-community",
-        "repo": "nixpkgs.lib",
-        "rev": "c185c7a5e5dd8f9add5b2f8ebeff00888b070742",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-community",
-        "repo": "nixpkgs.lib",
-        "type": "github"
-      }
-    },
-    "root": {
-      "inputs": {
-        "flake-parts": "flake-parts",
-        "nixpkgs": "nixpkgs"
-      }
-    }
-  },
-  "root": "root",
-  "version": 7
-}

flake.nix

@@ -1,42 +0,0 @@
-{
-  inputs = {
-    flake-parts.url = "github:hercules-ci/flake-parts";
-
-    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
-  };
-
-  outputs =
-    {
-      flake-parts,
-      self,
-      ...
-    }@inputs:
-    flake-parts.lib.mkFlake { inherit inputs; } {
-      systems = [
-        "aarch64-linux"
-        "x86_64-linux"
-      ];
-
-      perSystem =
-        { pkgs, self', ... }:
-        {
-          devShells.default = pkgs.mkShell {
-            name = "errornocord";
-
-            buildInputs = with pkgs; [
-              ffmpeg
-              self'.packages.default
-            ];
-          };
-
-          packages = rec {
-            errornocord = pkgs.callPackage ./. { inherit self; };
-            default = errornocord;
-          };
-        };
-
-      flake.pins = import ./npins;
-    };
-
-  description = "Hot-reloadable Discord music bot";
-}

main.py

@@ -0,0 +1,15 @@
+import time
+
+import constants
+import events
+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()
+
+
+client.run(constants.SECRETS["TOKEN"])

npins/default.nix

@@ -1,249 +0,0 @@
-/*
-  This file is provided under the MIT licence:
-
-  Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
-
-  The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
-
-  THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
-*/
-# Generated by npins. Do not modify; will be overwritten regularly
-let
-  # Backwards-compatibly make something that previously didn't take any arguments take some
-  # The function must return an attrset, and will unfortunately be eagerly evaluated
-  # Same thing, but it catches eval errors on the default argument so that one may still call it with other arguments
-  mkFunctor =
-    fn:
-    let
-      e = builtins.tryEval (fn { });
-    in
-    (if e.success then e.value else { error = fn { }; }) // { __functor = _self: fn; };
-
-  # https://github.com/NixOS/nixpkgs/blob/0258808f5744ca980b9a1f24fe0b1e6f0fecee9c/lib/lists.nix#L295
-  range =
-    first: last: if first > last then [ ] else builtins.genList (n: first + n) (last - first + 1);
-
-  # https://github.com/NixOS/nixpkgs/blob/0258808f5744ca980b9a1f24fe0b1e6f0fecee9c/lib/strings.nix#L257
-  stringToCharacters = s: map (p: builtins.substring p 1 s) (range 0 (builtins.stringLength s - 1));
-
-  # https://github.com/NixOS/nixpkgs/blob/0258808f5744ca980b9a1f24fe0b1e6f0fecee9c/lib/strings.nix#L269
-  stringAsChars = f: s: concatStrings (map f (stringToCharacters s));
-  concatStrings = builtins.concatStringsSep "";
-
-  # If the environment variable NPINS_OVERRIDE_${name} is set, then use
-  # the path directly as opposed to the fetched source.
-  # (Taken from Niv for compatibility)
-  mayOverride =
-    name: path:
-    let
-      envVarName = "NPINS_OVERRIDE_${saneName}";
-      saneName = stringAsChars (c: if (builtins.match "[a-zA-Z0-9]" c) == null then "_" else c) name;
-      ersatz = builtins.getEnv envVarName;
-    in
-    if ersatz == "" then
-      path
-    else
-      # this turns the string into an actual Nix path (for both absolute and
-      # relative paths)
-      builtins.trace "Overriding path of \"${name}\" with \"${ersatz}\" due to set \"${envVarName}\"" (
-        if builtins.substring 0 1 ersatz == "/" then
-          /. + ersatz
-        else
-          /. + builtins.getEnv "PWD" + "/${ersatz}"
-      );
-
-  mkSource =
-    name: spec:
-    {
-      pkgs ? null,
-    }:
-    assert spec ? type;
-    let
-      # Unify across builtin and pkgs fetchers.
-      # `fetchGit` requires a wrapper because of slight API differences.
-      fetchers =
-        if pkgs == null then
-          {
-            inherit (builtins) fetchTarball fetchurl;
-            # For some fucking reason, fetchGit has a different signature than the other builtin fetchers …
-            fetchGit = args: (builtins.fetchGit args).outPath;
-          }
-        else
-          {
-            fetchTarball =
-              {
-                url,
-                sha256,
-              }:
-              pkgs.fetchzip {
-                inherit url sha256;
-                extension = "tar";
-              };
-            inherit (pkgs) fetchurl;
-            fetchGit =
-              {
-                url,
-                submodules,
-                rev,
-                name,
-                narHash,
-              }:
-              pkgs.fetchgit {
-                inherit url rev name;
-                fetchSubmodules = submodules;
-                hash = narHash;
-              };
-          };
-
-      # Dispatch to the correct code path based on the type
-      path =
-        if spec.type == "Git" then
-          mkGitSource fetchers spec
-        else if spec.type == "GitRelease" then
-          mkGitSource fetchers spec
-        else if spec.type == "PyPi" then
-          mkPyPiSource fetchers spec
-        else if spec.type == "Channel" then
-          mkChannelSource fetchers spec
-        else if spec.type == "Tarball" then
-          mkTarballSource fetchers spec
-        else if spec.type == "Container" then
-          mkContainerSource pkgs spec
-        else
-          builtins.throw "Unknown source type ${spec.type}";
-    in
-    spec // { outPath = mayOverride name path; };
-
-  mkGitSource =
-    {
-      fetchTarball,
-      fetchGit,
-      ...
-    }:
-    {
-      repository,
-      revision,
-      url ? null,
-      submodules,
-      hash,
-      ...
-    }:
-    assert repository ? type;
-    # At the moment, either it is a plain git repository (which has an url), or it is a GitHub/GitLab repository
-    # In the latter case, there we will always be an url to the tarball
-    if url != null && !submodules then
-      fetchTarball {
-        inherit url;
-        sha256 = hash;
-      }
-    else
-      let
-        url =
-          if repository.type == "Git" then
-            repository.url
-          else if repository.type == "GitHub" then
-            "https://github.com/${repository.owner}/${repository.repo}.git"
-          else if repository.type == "GitLab" then
-            "${repository.server}/${repository.repo_path}.git"
-          else if repository.type == "Forgejo" then
-            "${repository.server}/${repository.owner}/${repository.repo}.git"
-          else
-            throw "Unrecognized repository type ${repository.type}";
-        urlToName =
-          url: rev:
-          let
-            matched = builtins.match "^.*/([^/]*)(\\.git)?$" url;
-
-            short = builtins.substring 0 7 rev;
-
-            appendShort = if (builtins.match "[a-f0-9]*" rev) != null then "-${short}" else "";
-          in
-          "${if matched == null then "source" else builtins.head matched}${appendShort}";
-        name = urlToName url revision;
-      in
-      fetchGit {
-        rev = revision;
-        narHash = hash;
-
-        inherit name submodules url;
-      };
-
-  mkPyPiSource =
-    { fetchurl, ... }:
-    {
-      url,
-      hash,
-      ...
-    }:
-    fetchurl {
-      inherit url;
-      sha256 = hash;
-    };
-
-  mkChannelSource =
-    { fetchTarball, ... }:
-    {
-      url,
-      hash,
-      ...
-    }:
-    fetchTarball {
-      inherit url;
-      sha256 = hash;
-    };
-
-  mkTarballSource =
-    { fetchTarball, ... }:
-    {
-      url,
-      locked_url ? url,
-      hash,
-      ...
-    }:
-    fetchTarball {
-      url = locked_url;
-      sha256 = hash;
-    };
-
-  mkContainerSource =
-    pkgs:
-    {
-      image_name,
-      image_tag,
-      image_digest,
-      ...
-    }:
-    if pkgs == null then
-      builtins.throw "container sources require passing in a Nixpkgs value: https://github.com/andir/npins/blob/master/README.md#using-the-nixpkgs-fetchers"
-    else
-      pkgs.dockerTools.pullImage {
-        imageName = image_name;
-        imageDigest = image_digest;
-        finalImageTag = image_tag;
-      };
-in
-mkFunctor (
-  {
-    input ? ./sources.json,
-  }:
-  let
-    data =
-      if builtins.isPath input then
-        # while `readFile` will throw an error anyways if the path doesn't exist,
-        # we still need to check beforehand because *our* error can be caught but not the one from the builtin
-        # *piegames sighs*
-        if builtins.pathExists input then
-          builtins.fromJSON (builtins.readFile input)
-        else
-          throw "Input path ${toString input} does not exist"
-      else if builtins.isAttrs input then
-        input
-      else
-        throw "Unsupported input type ${builtins.typeOf input}, must be a path or an attrset";
-    version = data.version;
-  in
-  if version == 7 then
-    builtins.mapAttrs (name: spec: mkFunctor (mkSource name spec)) data.pins
-  else
-    throw "Unsupported format version ${toString version} in sources.json. Try running `npins upgrade`"
-)

npins/sources.json

@@ -1,26 +0,0 @@
-{
-  "pins": {
-    "disnake": {
-      "type": "Git",
-      "repository": {
-        "type": "GitHub",
-        "owner": "DisnakeDev",
-        "repo": "disnake"
-      },
-      "branch": "master",
-      "submodules": false,
-      "revision": "79afc2d8673299f43636730d4a1518e7901f95fa",
-      "url": "https://github.com/DisnakeDev/disnake/archive/79afc2d8673299f43636730d4a1518e7901f95fa.tar.gz",
-      "hash": "sha256-7gsHFnqTyANdV+eosLI4MlzsMyebLdYZShEVObJPhl8="
-    },
-    "disnake-paginator": {
-      "type": "PyPi",
-      "name": "disnake-paginator",
-      "version_upper_bound": null,
-      "version": "1.0.8",
-      "url": "https://files.pythonhosted.org/packages/8c/db/3a86b247c7653a3f1676d16a316daf400edcf76295098ab60544f2a8f8b7/disnake_paginator-1.0.8.tar.gz",
-      "hash": "sha256-Sn0qbKNkJIdX8GCRRRPxwWutBtG9c1Zt7JnQAsl5I4s="
-    }
-  },
-  "version": 7
-}

pyproject.toml

@@ -1,13 +0,0 @@
-[project]
-name = "errornocord"
-version = "0.1.0"
-
-[project.scripts]
-errornocord = "errornocord.main:main"
-
-[tool.setuptools.packages.find]
-where = ["."]
-include = ["errornocord*"]
-
-[tool.basedpyright]
-typeCheckingMode = "basic"

requirements.txt

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

state.py

@@ -0,0 +1,32 @@
+import collections
+import time
+
+import disnake
+
+
+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_locks = LimitedSizeDict()
+idle_tracker = {"is_idle": False, "last_used": time.time()}
+message_responses = LimitedSizeDict()
+players = {}
+start_time = time.time()

tasks.py

@@ -0,0 +1,25 @@
+import asyncio
+import time
+
+import disnake
+
+from state import client, idle_tracker, players
+
+
+async def cleanup():
+    while True:
+        await asyncio.sleep(3600)
+
+        targets = []
+        for id, player in players:
+            if len(player.queue) == 0:
+                targets.append(id)
+        for target in targets:
+            del players[target]
+
+        if (
+            not idle_tracker["is_idle"]
+            and time.time() - idle_tracker["last_used"] >= 3600
+        ):
+            await client.change_presence(status=disnake.Status.idle)
+            idle_tracker["is_idle"] = True

utils.py

@@ -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

youtubedl.py

@@ -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)