mirror of
https://github.com/stoatchat/python-client-sdk.git
synced 2026-07-22 02:25:22 -04:00
add basic help command framework
This commit is contained in:
@@ -5,4 +5,5 @@ from .context import *
|
||||
from .converters import *
|
||||
from .errors import *
|
||||
from .group import *
|
||||
from .help import *
|
||||
from .cog import *
|
||||
|
||||
@@ -2,11 +2,14 @@ from __future__ import annotations
|
||||
|
||||
import traceback
|
||||
import sys
|
||||
from typing import Any, Union, Protocol, runtime_checkable
|
||||
from typing import Any, Union, Protocol, runtime_checkable, Optional, TYPE_CHECKING
|
||||
from importlib import import_module
|
||||
|
||||
import revolt
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .help import HelpCommand
|
||||
|
||||
from .command import Command
|
||||
from .context import Context
|
||||
from .errors import CheckError, CommandNotFound, MissingSetup
|
||||
@@ -44,7 +47,9 @@ class CommandsClient(revolt.Client, metaclass=CommandsMeta):
|
||||
|
||||
_commands: list[Command]
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
def __init__(self, *args, help_command: Optional[HelpCommand] = None, **kwargs):
|
||||
from .help import DefaultHelpCommand, help_command_impl
|
||||
|
||||
self.all_commands: dict[str, Command] = {}
|
||||
self.cogs: dict[str, Cog] = {}
|
||||
self.extensions: dict[str, ExtensionProtocol] = {}
|
||||
@@ -55,6 +60,12 @@ class CommandsClient(revolt.Client, metaclass=CommandsMeta):
|
||||
for alias in command.aliases:
|
||||
self.all_commands[alias] = command
|
||||
|
||||
if help_command is None:
|
||||
help_command = DefaultHelpCommand()
|
||||
|
||||
self.help_command = help_command
|
||||
self.add_command(help_command_impl)
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
@property
|
||||
|
||||
@@ -10,11 +10,13 @@ import revolt
|
||||
from revolt.utils import copy_doc, maybe_coroutine
|
||||
|
||||
from .errors import InvalidLiteralArgument, UnionConverterError
|
||||
from .utils import evaluate_parameters
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .checks import Check
|
||||
from .context import Context
|
||||
from .checks import Check
|
||||
from .group import Group
|
||||
from .cog import Cog
|
||||
|
||||
__all__ = (
|
||||
"Command",
|
||||
@@ -37,18 +39,25 @@ class Command:
|
||||
The aliases of the command
|
||||
parent: Optional[:class:`Group`]
|
||||
The parent of the command if this command is a subcommand
|
||||
cog: Optional[:class:`Cog`]
|
||||
The cog the command is apart of.
|
||||
usage: Optional[:class:`str`]
|
||||
The usage string for the command
|
||||
"""
|
||||
__slots__ = ("callback", "name", "aliases", "signature", "checks", "parent", "_error_handler", "cog")
|
||||
__slots__ = ("callback", "name", "aliases", "signature", "checks", "parent", "_error_handler", "cog", "description", "usage", "parameters")
|
||||
|
||||
def __init__(self, callback: Callable[..., Coroutine[Any, Any, Any]], name: str, aliases: list[str]):
|
||||
def __init__(self, callback: Callable[..., Coroutine[Any, Any, Any]], name: str, aliases: list[str], usage: Optional[str] = None):
|
||||
self.callback = callback
|
||||
self.name = name
|
||||
self.aliases = aliases
|
||||
self.usage = usage
|
||||
self.signature = inspect.signature(self.callback)
|
||||
self.parameters = evaluate_parameters(self.signature.parameters.values(), getattr(callback, "__globals__", {}))
|
||||
self.checks: list[Check] = getattr(callback, "_checks", [])
|
||||
self.parent: Optional[Group] = None
|
||||
self.cog = None
|
||||
self.cog: Optional[Cog] = None
|
||||
self._error_handler: Callable[[Any, Context, Exception], Coroutine[Any, Any, Any]] = type(self)._default_error_handler
|
||||
self.description = callback.__doc__
|
||||
|
||||
async def invoke(self, context: Context, *args, **kwargs) -> Any:
|
||||
"""Runs the command and calls the error handler if the command errors.
|
||||
@@ -128,14 +137,14 @@ class Command:
|
||||
else:
|
||||
raise InvalidLiteralArgument(arg)
|
||||
else:
|
||||
return await maybe_coroutine(cast(Callable[[Any, Context], Any], annotation), arg, context)
|
||||
return await maybe_coroutine(annotation, arg, context)
|
||||
else:
|
||||
return arg
|
||||
|
||||
async def parse_arguments(self, context: Context):
|
||||
for name, parameter in list(self.signature.parameters.items())[2:]:
|
||||
for parameter in self.parameters[2:]:
|
||||
if parameter.kind == parameter.KEYWORD_ONLY:
|
||||
context.kwargs[name] = await self.convert_argument(context.view.get_rest(), parameter.annotation, context)
|
||||
context.kwargs[parameter.name] = await self.convert_argument(context.view.get_rest(), parameter.annotation, context)
|
||||
|
||||
elif parameter.kind == parameter.VAR_POSITIONAL:
|
||||
with suppress(StopIteration):
|
||||
@@ -146,9 +155,47 @@ class Command:
|
||||
context.args.append(await self.convert_argument(context.view.get_next_word(), parameter.annotation, context))
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<Command name=\"{self.name}\">"
|
||||
return f"<{self.__class__.__name__} name=\"{self.name}\">"
|
||||
|
||||
def command(*, name: Optional[str] = None, aliases: Optional[list[str]] = None, cls: type[Command] = Command):
|
||||
@property
|
||||
def short_description(self) -> Optional[str]:
|
||||
"""Returns the first line of the description or None if there is no description."""
|
||||
if self.description:
|
||||
return self.description.split("\n")[0]
|
||||
|
||||
def get_usage(self) -> str:
|
||||
"""Returns the usage string for the command."""
|
||||
if self.usage:
|
||||
return self.usage
|
||||
|
||||
parents = []
|
||||
|
||||
if self.parent:
|
||||
parent = self.parent
|
||||
|
||||
while parent:
|
||||
parents.append(parent.name)
|
||||
parent = parent.parent
|
||||
|
||||
parameters = []
|
||||
|
||||
for parameter in self.parameters[2:]:
|
||||
if parameter.kind == parameter.POSITIONAL_OR_KEYWORD:
|
||||
if parameter.default is not parameter.empty:
|
||||
parameters.append(f"[{parameter.name}]")
|
||||
else:
|
||||
parameters.append(f"<{parameter.name}>")
|
||||
elif parameter.kind == parameter.KEYWORD_ONLY:
|
||||
if parameter.default is not parameter.empty:
|
||||
parameters.append(f"[{parameter.name}]")
|
||||
else:
|
||||
parameters.append(f"<{parameter.name}...>")
|
||||
elif parameter.kind == parameter.VAR_POSITIONAL:
|
||||
parameters.append(f"[{parameter.name}...]")
|
||||
|
||||
return f"{' '.join(parents[::-1])} {self.name} {' '.join(parameters)}"
|
||||
|
||||
def command(*, name: Optional[str] = None, aliases: Optional[list[str]] = None, cls: type[Command] = Command, usage: Optional[str] = None):
|
||||
"""A decorator that turns a function into a :class:`Command`.
|
||||
|
||||
Parameters
|
||||
@@ -166,6 +213,6 @@ def command(*, name: Optional[str] = None, aliases: Optional[list[str]] = None,
|
||||
A function that takes the command callback and returns a :class:`Command`
|
||||
"""
|
||||
def inner(func: Callable[..., Coroutine[Any, Any, Any]]):
|
||||
return cls(func, name or func.__name__, aliases or [])
|
||||
return cls(func, name or func.__name__, aliases or [], usage)
|
||||
|
||||
return inner
|
||||
|
||||
@@ -85,7 +85,6 @@ class Context(revolt.Messageable):
|
||||
await command.parse_arguments(self)
|
||||
return await command.invoke(self, *self.args, **self.kwargs)
|
||||
|
||||
|
||||
async def can_run(self, command: Optional[Command] = None) -> bool:
|
||||
"""Runs all of the commands checks, and returns true if all of them pass"""
|
||||
command = command or self.command
|
||||
|
||||
Executable
+184
@@ -0,0 +1,184 @@
|
||||
from __future__ import annotations
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from typing import TYPE_CHECKING, TypedDict, Union, Optional
|
||||
from typing_extensions import NotRequired
|
||||
from itertools import groupby
|
||||
|
||||
from .utils import evaluate_parameters
|
||||
from .command import Command, command
|
||||
from .client import CommandsClient
|
||||
from .context import Context
|
||||
from .group import Group
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .cog import Cog
|
||||
from revolt import SendableEmbed, File, MessageReply, Message, Messageable
|
||||
|
||||
__all__ = ("MessagePayload", "HelpCommand", "DefaultHelpCommand", "help_command_impl")
|
||||
|
||||
class MessagePayload(TypedDict):
|
||||
content: str
|
||||
embed: NotRequired[SendableEmbed]
|
||||
embeds: NotRequired[list[SendableEmbed]]
|
||||
attachments: NotRequired[list[File]]
|
||||
replies: NotRequired[list[MessageReply]]
|
||||
|
||||
|
||||
class HelpCommand(ABC):
|
||||
@abstractmethod
|
||||
async def create_bot_help(self, context: Context, commands: dict[Optional[Cog], list[Command]]) -> Union[str, SendableEmbed, MessagePayload]:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
async def create_command_help(self, context: Context, command: Command) -> Union[str, SendableEmbed, MessagePayload]:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
async def create_group_help(self, context: Context, group: Group) -> Union[str, SendableEmbed, MessagePayload]:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
async def create_cog_help(self, context: Context, cog: Cog) -> Union[str, SendableEmbed, MessagePayload]:
|
||||
raise NotImplementedError
|
||||
|
||||
async def send_help_command(self, context: Context, message_payload: MessagePayload) -> Message:
|
||||
return await context.send(**message_payload)
|
||||
|
||||
async def filter_commands(self, context: Context, commands: list[Command]) -> list[Command]:
|
||||
filtered: list[Command] = []
|
||||
|
||||
for command in commands:
|
||||
try:
|
||||
if await context.can_run(command):
|
||||
filtered.append(command)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return filtered
|
||||
|
||||
async def group_commands(self, context: Context, commands: list[Command]) -> dict[Optional[Cog], list[Command]]:
|
||||
return {cog_name: list(commands) for cog_name, commands in groupby(commands, lambda command: command.cog)}
|
||||
|
||||
async def handle_message(self, context: Context, message: Message):
|
||||
pass
|
||||
|
||||
async def get_channel(self, context: Context) -> Messageable:
|
||||
return context
|
||||
|
||||
@abstractmethod
|
||||
async def handle_no_command_found(self, context: Context, name: str):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
async def handle_no_cog_found(self, context: Context, name: str):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class DefaultHelpCommand(HelpCommand):
|
||||
def __init__(self, default_cog_name: str = "No Cog"):
|
||||
self.default_cog_name = default_cog_name
|
||||
|
||||
async def create_bot_help(self, context: Context, commands: dict[Optional[Cog], list[Command]]) -> Union[str, SendableEmbed, MessagePayload]:
|
||||
lines = ["```"]
|
||||
|
||||
for cog, cog_commands in commands.items():
|
||||
cog_lines = []
|
||||
cog_lines.append(f"{cog.qualified_name if cog else self.default_cog_name}:")
|
||||
|
||||
for command in cog_commands:
|
||||
cog_lines.append(f" {command.name} - {command.short_description or 'No description'}")
|
||||
|
||||
lines.append("\n".join(cog_lines))
|
||||
|
||||
lines.append("```")
|
||||
return "\n".join(lines)
|
||||
|
||||
async def create_cog_help(self, context: Context, cog: Cog) -> Union[str, SendableEmbed, MessagePayload]:
|
||||
lines = ["```"]
|
||||
|
||||
lines.append(f"{cog.qualified_name}:")
|
||||
|
||||
for command in cog.commands:
|
||||
lines.append(f" {command.name} - {command.short_description or 'No description'}")
|
||||
|
||||
lines.append("```")
|
||||
return "\n".join(lines)
|
||||
|
||||
async def create_command_help(self, context: Context, command: Command) -> Union[str, SendableEmbed, MessagePayload]:
|
||||
lines = ["```"]
|
||||
|
||||
lines.append(f"{command.name}:")
|
||||
lines.append(f" Usage: {command.get_usage()}")
|
||||
|
||||
if command.aliases:
|
||||
lines.append(f" Aliases: {', '.join(command.aliases)}")
|
||||
|
||||
|
||||
if command.description:
|
||||
lines.append(command.description)
|
||||
|
||||
lines.append("```")
|
||||
return "\n".join(lines)
|
||||
|
||||
async def create_group_help(self, context: Context, group: Group) -> Union[str, SendableEmbed, MessagePayload]:
|
||||
lines = ["```"]
|
||||
|
||||
lines.append(f"{group.name}:")
|
||||
lines.append(f" Usage: {group.get_usage()}")
|
||||
|
||||
if group.aliases:
|
||||
lines.append(f" Aliases: {', '.join(group.aliases)}")
|
||||
|
||||
if group.description:
|
||||
lines.append(group.description)
|
||||
|
||||
for command in group.commands:
|
||||
lines.append(f" {command.name} - {command.short_description or 'No description'}")
|
||||
|
||||
lines.append("```")
|
||||
return "\n".join(lines)
|
||||
|
||||
async def handle_no_command_found(self, context: Context, name: str):
|
||||
channel = await self.get_channel(context)
|
||||
await channel.send(f"Command `{name}` not found.")
|
||||
|
||||
async def handle_no_cog_found(self, context: Context, name: str):
|
||||
channel = await self.get_channel(context)
|
||||
await channel.send(f"Cog `{name}` not found.")
|
||||
|
||||
|
||||
@command(name="help")
|
||||
async def help_command_impl(self: CommandsClient, context: Context, *arguments: str):
|
||||
filtered_commands = await context.client.help_command.filter_commands(context, self.commands)
|
||||
commands = await self.help_command.group_commands(context, filtered_commands)
|
||||
|
||||
if not arguments:
|
||||
payload = await self.help_command.create_bot_help(context, commands)
|
||||
else:
|
||||
command_name = arguments[0]
|
||||
|
||||
try:
|
||||
command = self.get_command(command_name)
|
||||
except KeyError:
|
||||
cog = self.cogs.get(command_name)
|
||||
if cog:
|
||||
payload = await self.help_command.create_cog_help(context, cog)
|
||||
else:
|
||||
return await self.help_command.handle_no_command_found(context, command_name)
|
||||
else:
|
||||
if isinstance(command, Group):
|
||||
payload = await self.help_command.create_group_help(context, command)
|
||||
else:
|
||||
payload = await self.help_command.create_command_help(context, command)
|
||||
|
||||
msg_payload: MessagePayload
|
||||
|
||||
if isinstance(payload, str):
|
||||
msg_payload = {"content": payload}
|
||||
elif isinstance(payload, SendableEmbed):
|
||||
msg_payload = {"embed": payload, "content": " "}
|
||||
else:
|
||||
msg_payload = payload
|
||||
|
||||
await self.help_command.send_help_command(context, msg_payload)
|
||||
Executable
+16
@@ -0,0 +1,16 @@
|
||||
from inspect import Parameter
|
||||
from typing import Iterable, Any
|
||||
|
||||
__all__ = ("evaluate_parameters",)
|
||||
|
||||
def evaluate_parameters(parameters: Iterable[Parameter], globals: dict[str, Any]) -> list[Parameter]:
|
||||
new_parameters = []
|
||||
|
||||
for parameter in parameters:
|
||||
if parameter.annotation is not parameter.empty:
|
||||
if isinstance(parameter.annotation, str):
|
||||
parameter = parameter.replace(annotation=eval(parameter.annotation, globals))
|
||||
|
||||
new_parameters.append(parameter)
|
||||
|
||||
return new_parameters
|
||||
Reference in New Issue
Block a user