ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 19:44:20 +00:00
Code Activity

This might actually work

Browse source
This commit is contained in:
Steffo 2019-08-18 02:53:59 +03:00
parent d2c1f87798
commit d65f677e21
8 changed files with 152 additions and 185 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

102
royalnet/bots/discord.py
View file

@ -2,11 +2,12 @@ import discord
import typing import typing
import logging as _logging import logging as _logging
from .generic import GenericBot from .generic import GenericBot
from ..utils import asyncify, Call, Command, discord_escape from ..utils import *
from ..error import UnregisteredError, NoneFoundError, TooManyFoundError, InvalidConfigError, RoyalnetResponseError from ..error import *
from ..network import RoyalnetConfig, Request, ResponseSuccess, ResponseError from ..network import *
from ..database import DatabaseConfig from ..database import *
from ..audio import playmodes, YtdlDiscord from ..audio import *
from ..commands import *
log = _logging.getLogger(__name__) log = _logging.getLogger(__name__)
@ -24,47 +25,35 @@ class DiscordConfig:
class DiscordBot(GenericBot): class DiscordBot(GenericBot):
"""A bot that connects to `Discord <https://discordapp.com/>`_.""" """A bot that connects to `Discord <https://discordapp.com/>`_."""
interface_name = "discord"
def _init_voice(self): def _init_voice(self):
"""Initialize the variables needed for the connection to voice chat.""" """Initialize the variables needed for the connection to voice chat."""
log.debug(f"Creating music_data dict") log.debug(f"Creating music_data dict")
self.music_data: typing.Dict[discord.Guild, playmodes.PlayMode] = {} self.music_data: typing.Dict[discord.Guild, playmodes.PlayMode] = {}
def _interface_factory(self) -> typing.Type[Call]: def _interface_factory(self) -> typing.Type[CommandInterface]:
log.debug(f"Creating DiscordCall") # noinspection PyPep8Naming
GenericInterface = super()._interface_factory()
# noinspection PyMethodParameters # noinspection PyMethodParameters,PyAbstractClass
class DiscordCall(Call): class DiscordInterface(GenericInterface):
interface_name = self.interface_name name = "discord"
interface_obj = self prefix = "!"
interface_prefix = "!"
alchemy = self.alchemy return DiscordInterface
async def reply(call, text: str): def _data_factory(self) -> typing.Type[CommandData]:
# TODO: don't escape characters inside [c][/c] blocks # noinspection PyMethodParameters,PyAbstractClass
await call.channel.send(discord_escape(text)) class DiscordData(CommandData):
def __init__(data, interface: CommandInterface, message: discord.Message):
data._interface = interface
data.message = message
async def net_request(call, request: Request, destination: str) -> dict: async def reply(data, text: str):
if self.network is None: await data.message.channel.send(discord_escape(text))
raise InvalidConfigError("Royalnet is not enabled on this bot")
response_dict: dict = await self.network.request(request.to_dict(), destination)
if "type" not in response_dict:
raise RoyalnetResponseError("Response is missing a type")
elif response_dict["type"] == "ResponseSuccess":
response: typing.Union[ResponseSuccess, ResponseError] = ResponseSuccess.from_dict(response_dict)
elif response_dict["type"] == "ResponseError":
response = ResponseError.from_dict(response_dict)
else:
raise RoyalnetResponseError("Response type is unknown")
response.raise_on_error()
return response.data
async def get_author(call, error_if_none=False): async def get_author(data, error_if_none=False):
message: discord.Message = call.kwargs["message"] user: discord.Member = data.message.author
user: discord.Member = message.author query = data._interface.session.query(self.master_table)
query = call.session.query(self.master_table)
for link in self.identity_chain: for link in self.identity_chain:
query = query.join(link.mapper.class_) query = query.join(link.mapper.class_)
query = query.filter(self.identity_column == user.id) query = query.filter(self.identity_column == user.id)
@ -73,7 +62,7 @@ class DiscordBot(GenericBot):
raise UnregisteredError("Author is not registered") raise UnregisteredError("Author is not registered")
return result return result
return DiscordCall return DiscordData
def _bot_factory(self) -> typing.Type[discord.Client]: def _bot_factory(self) -> typing.Type[discord.Client]:
"""Create a custom DiscordClient class inheriting from :py:class:`discord.Client`.""" """Create a custom DiscordClient class inheriting from :py:class:`discord.Client`."""
@ -107,20 +96,25 @@ class DiscordBot(GenericBot):
if not text: if not text:
return return
# Skip non-command updates # Skip non-command updates
if not text.startswith(self.command_prefix): if not text.startswith("!"):
return return
# Skip bot messages # Skip bot messages
author: typing.Union[discord.User] = message.author author: typing.Union[discord.User] = message.author
if author.bot: if author.bot:
return return
# Start typing # Find and clean parameters
command_text, *parameters = text.split(" ")
# Don't use a case-sensitive command name
command_name = command_text.lower()
# Find the command
try:
command = self.commands[command_name]
except KeyError:
# Skip the message
return
# Call the command
with message.channel.typing(): with message.channel.typing():
# Find and clean parameters await command.run(CommandArgs(parameters), self._Data(interface=command.interface, message=message))
command_text, *parameters = text.split(" ")
# Don't use a case-sensitive command name
command_name = command_text.lower()
# Call the command
await self.call(command_name, message.channel, parameters, message=message)
async def on_ready(cli): async def on_ready(cli):
log.debug("Connection successful, client is ready") log.debug("Connection successful, client is ready")
@ -146,12 +140,14 @@ class DiscordBot(GenericBot):
def find_channel_by_name(cli, def find_channel_by_name(cli,
name: str, name: str,
guild: typing.Optional[discord.Guild] = None) -> discord.abc.GuildChannel: guild: typing.Optional[discord.Guild] = None) -> discord.abc.GuildChannel:
"""Find the :py:class:`TextChannel`, :py:class:`VoiceChannel` or :py:class:`CategoryChannel` with the specified name. """Find the :py:class:`TextChannel`, :py:class:`VoiceChannel` or :py:class:`CategoryChannel` with the
specified name.
Case-insensitive. Case-insensitive.
Guild is optional, but the method will raise a :py:exc:`TooManyFoundError` if none is specified and there is more than one channel with the same name. Guild is optional, but the method will raise a :py:exc:`TooManyFoundError` if none is specified and
Will also raise a :py:exc:`NoneFoundError` if no channels are found.""" there is more than one channel with the same name. Will also raise a :py:exc:`NoneFoundError` if no
channels are found. """
if guild is not None: if guild is not None:
all_channels = guild.channels all_channels = guild.channels
else: else:
@ -192,16 +188,10 @@ class DiscordBot(GenericBot):
discord_config: DiscordConfig, discord_config: DiscordConfig,
royalnet_config: typing.Optional[RoyalnetConfig] = None, royalnet_config: typing.Optional[RoyalnetConfig] = None,
database_config: typing.Optional[DatabaseConfig] = None, database_config: typing.Optional[DatabaseConfig] = None,
command_prefix: str = "!", commands: typing.List[typing.Type[Command]] = None):
commands: typing.List[typing.Type[Command]] = None,
missing_command: typing.Type[Command] = NullCommand,
error_command: typing.Type[Command] = NullCommand):
super().__init__(royalnet_config=royalnet_config, super().__init__(royalnet_config=royalnet_config,
database_config=database_config, database_config=database_config,
command_prefix=command_prefix, commands=commands)
commands=commands,
missing_command=missing_command,
error_command=error_command)
self._discord_config = discord_config self._discord_config = discord_config
self._init_client() self._init_client()
self._init_voice() self._init_voice()

86
royalnet/bots/generic.py
View file

@ -2,10 +2,10 @@ import sys
import typing import typing
import asyncio import asyncio
import logging import logging
from ..utils import NetworkHandler from ..utils import *
from ..network import RoyalnetLink, Request, Response, ResponseSuccess, ResponseError, RoyalnetConfig from ..network import *
from ..database import Alchemy, DatabaseConfig, relationshiplinkchain from ..database import *
from ..commands import Command, CommandInterface from ..commands import *
from ..error import * from ..error import *
@ -13,32 +13,24 @@ log = logging.getLogger(__name__)
class GenericBot: class GenericBot:
"""A generic bot class, to be used as base for the other more specific classes, such as :ref:`royalnet.bots.TelegramBot` and :ref:`royalnet.bots.DiscordBot`.""" """A generic bot class, to be used as base for the other more specific classes, such as
:ref:`royalnet.bots.TelegramBot` and :ref:`royalnet.bots.DiscordBot`. """
interface_name = NotImplemented interface_name = NotImplemented
def _init_commands(self, def _init_commands(self, commands: typing.List[typing.Type[Command]]) -> None:
command_prefix: str, """Generate the ``commands`` dictionary required to handle incoming messages, and the ``network_handlers``
commands: typing.List[typing.Type[Command]], dictionary required to handle incoming requests. """
missing_command: typing.Type[Command], log.debug(f"Now binding commands")
error_command: typing.Type[Command]) -> None: self._Interface = self._interface_factory()
"""Generate the ``commands`` dictionary required to handle incoming messages, and the ``network_handlers`` dictionary required to handle incoming requests.""" self._Data = self._data_factory()
log.debug(f"Now generating commands") self.commands = {}
self.command_prefix = command_prefix for SelectedCommand in self.commands:
self.commands: typing.Dict[str, typing.Type[Command]] = {} interface = self._Interface()
self.commands[f"{interface.prefix}{SelectedCommand.name}"] = SelectedCommand(interface)
self.network_handlers: typing.Dict[str, typing.Type[NetworkHandler]] = {} self.network_handlers: typing.Dict[str, typing.Type[NetworkHandler]] = {}
for command in commands: log.debug(f"Successfully bound commands")
lower_command_name = command.command_name.lower()
self.commands[f"{command_prefix}{lower_command_name}"] = command
self.missing_command: typing.Type[Command] = missing_command
self.error_command: typing.Type[Command] = error_command
log.debug(f"Successfully generated commands")
def _interface_factory(self) -> typing.Type[CommandInterface]: def _interface_factory(self) -> typing.Type[CommandInterface]:
"""Create a :py:class:`royalnet.commands.CommandInterface` type and return it.
Returns:
The created :py:class:`royalnet.commands.CommandInterface` type."""
# noinspection PyAbstractClass,PyMethodParameters # noinspection PyAbstractClass,PyMethodParameters
class GenericInterface(CommandInterface): class GenericInterface(CommandInterface):
alchemy = self.alchemy alchemy = self.alchemy
@ -47,6 +39,9 @@ class GenericBot:
def register_net_handler(ci, message_type: str, network_handler: typing.Callable): def register_net_handler(ci, message_type: str, network_handler: typing.Callable):
self.network_handlers[message_type] = network_handler self.network_handlers[message_type] = network_handler
def unregister_net_handler(ci, message_type: str):
del self.network_handlers[message_type]
async def net_request(ci, request: Request, destination: str) -> dict: async def net_request(ci, request: Request, destination: str) -> dict:
if self.network is None: if self.network is None:
raise InvalidConfigError("Royalnet is not enabled on this bot") raise InvalidConfigError("Royalnet is not enabled on this bot")
@ -64,10 +59,13 @@ class GenericBot:
return GenericInterface return GenericInterface
def _data_factory(self) -> typing.Type[CommandData]:
raise NotImplementedError()
def _init_royalnet(self, royalnet_config: RoyalnetConfig): def _init_royalnet(self, royalnet_config: RoyalnetConfig):
"""Create a :py:class:`royalnet.network.RoyalnetLink`, and run it as a :py:class:`asyncio.Task`.""" """Create a :py:class:`royalnet.network.RoyalnetLink`, and run it as a :py:class:`asyncio.Task`."""
self.network: RoyalnetLink = RoyalnetLink(royalnet_config.master_uri, royalnet_config.master_secret, self.interface_name, self.network: RoyalnetLink = RoyalnetLink(royalnet_config.master_uri, royalnet_config.master_secret,
self._network_handler) self.interface_name, self._network_handler)
log.debug(f"Running RoyalnetLink {self.network}") log.debug(f"Running RoyalnetLink {self.network}")
self.loop.create_task(self.network.run()) self.loop.create_task(self.network.run())
@ -98,14 +96,16 @@ class GenericBot:
_, exc, _ = sys.exc_info() _, exc, _ = sys.exc_info()
log.debug(f"Exception {exc} in {network_handler}") log.debug(f"Exception {exc} in {network_handler}")
return ResponseError("exception_in_handler", return ResponseError("exception_in_handler",
f"An exception was raised in {network_handler} for {request.handler}. Check extra_info for details.", f"An exception was raised in {network_handler} for {request.handler}. Check "
f"extra_info for details.",
extra_info={ extra_info={
"type": exc.__class__.__name__, "type": exc.__class__.__name__,
"str": str(exc) "str": str(exc)
}).to_dict() }).to_dict()
def _init_database(self, commands: typing.List[typing.Type[Command]], database_config: DatabaseConfig): def _init_database(self, commands: typing.List[typing.Type[Command]], database_config: DatabaseConfig):
"""Create an :py:class:`royalnet.database.Alchemy` with the tables required by the commands. Then, find the chain that links the ``master_table`` to the ``identity_table``.""" """Create an :py:class:`royalnet.database.Alchemy` with the tables required by the commands. Then,
find the chain that links the ``master_table`` to the ``identity_table``. """
log.debug(f"Initializing database") log.debug(f"Initializing database")
required_tables = set() required_tables = set()
for command in commands: for command in commands:
@ -122,10 +122,7 @@ class GenericBot:
def __init__(self, *, def __init__(self, *,
royalnet_config: typing.Optional[RoyalnetConfig] = None, royalnet_config: typing.Optional[RoyalnetConfig] = None,
database_config: typing.Optional[DatabaseConfig] = None, database_config: typing.Optional[DatabaseConfig] = None,
command_prefix: str,
commands: typing.List[typing.Type[Command]] = None, commands: typing.List[typing.Type[Command]] = None,
missing_command: typing.Type[Command] = NullCommand,
error_command: typing.Type[Command] = NullCommand,
loop: asyncio.AbstractEventLoop = None): loop: asyncio.AbstractEventLoop = None):
if loop is None: if loop is None:
self.loop = asyncio.get_event_loop() self.loop = asyncio.get_event_loop()
@ -140,35 +137,12 @@ class GenericBot:
self._init_database(commands=commands, database_config=database_config) self._init_database(commands=commands, database_config=database_config)
if commands is None: if commands is None:
commands = [] commands = []
self._init_commands(command_prefix, commands, missing_command=missing_command, error_command=error_command) self._init_commands(commands)
self._Call = self._interface_factory()
if royalnet_config is None: if royalnet_config is None:
self.network = None self.network = None
else: else:
self._init_royalnet(royalnet_config=royalnet_config) self._init_royalnet(royalnet_config=royalnet_config)
async def call(self, command_name: str, channel, parameters: typing.List[str] = None, **kwargs):
"""Call the command with the specified name.
If it doesn't exist, call ``self.missing_command``.
If an exception is raised during the execution of the command, call ``self.error_command``."""
log.debug(f"Trying to call {command_name}")
if parameters is None:
parameters = []
try:
command: typing.Type[Command] = self.commands[command_name]
except KeyError:
log.debug(f"Calling missing_command because {command_name} does not exist")
command = self.missing_command
try:
await self._Call(channel, command, parameters, **kwargs).run()
except Exception as exc:
log.debug(f"Calling error_command because of an error in {command_name}")
await self._Call(channel, self.error_command,
exception=exc,
previous_command=command, **kwargs).run()
async def run(self): async def run(self):
"""A blocking coroutine that should make the bot start listening to commands and requests.""" """A blocking coroutine that should make the bot start listening to commands and requests."""
raise NotImplementedError() raise NotImplementedError()

65
royalnet/bots/telegram.py
View file

@ -3,11 +3,11 @@ import telegram.utils.request
import typing import typing
import logging as _logging import logging as _logging
from .generic import GenericBot from .generic import GenericBot
from ..utils import asyncify, telegram_escape from ..utils import *
from ..error import UnregisteredError, InvalidConfigError, RoyalnetResponseError from ..error import *
from ..network import RoyalnetConfig, Request, ResponseSuccess, ResponseError from ..network import *
from ..database import DatabaseConfig from ..database import *
from ..commands import CommandInterface from ..commands import *
log = _logging.getLogger(__name__) log = _logging.getLogger(__name__)
@ -30,28 +30,35 @@ class TelegramBot(GenericBot):
self._offset: int = -100 self._offset: int = -100
def _interface_factory(self) -> typing.Type[CommandInterface]: def _interface_factory(self) -> typing.Type[CommandInterface]:
# noinspection PyPep8Naming
GenericInterface = super()._interface_factory() GenericInterface = super()._interface_factory()
# noinspection PyMethodParameters # noinspection PyMethodParameters,PyAbstractClass
class TelegramInterface(GenericInterface): class TelegramInterface(GenericInterface):
name = "telegram" name = "telegram"
prefix = "/" prefix = "/"
alchemy = self.alchemy return TelegramInterface
async def reply(ci, extra: dict, text: str): def _data_factory(self) -> typing.Type[CommandData]:
await asyncify(ci.channel.send_message, telegram_escape(text), # noinspection PyMethodParameters,PyAbstractClass
class TelegramData(CommandData):
def __init__(data, interface: CommandInterface, update: telegram.Update):
data._interface = interface
data.update = update
async def reply(data, text: str):
await asyncify(data.update.effective_chat.send_message, telegram_escape(text),
parse_mode="HTML", parse_mode="HTML",
disable_web_page_preview=True) disable_web_page_preview=True)
async def get_author(ci, extra: dict, error_if_none=False): async def get_author(data, error_if_none=False):
update: telegram.Update = extra["update"] user: telegram.User = data.update.effective_user
user: telegram.User = update.effective_user
if user is None: if user is None:
if error_if_none: if error_if_none:
raise UnregisteredError("No author for this message") raise UnregisteredError("No author for this message")
return None return None
query = ci.session.query(self.master_table) query = data._interface.session.query(self.master_table)
for link in self.identity_chain: for link in self.identity_chain:
query = query.join(link.mapper.class_) query = query.join(link.mapper.class_)
query = query.filter(self.identity_column == user.id) query = query.filter(self.identity_column == user.id)
@ -60,22 +67,16 @@ class TelegramBot(GenericBot):
raise UnregisteredError("Author is not registered") raise UnregisteredError("Author is not registered")
return result return result
return TelegramCall return TelegramData
def __init__(self, *, def __init__(self, *,
telegram_config: TelegramConfig, telegram_config: TelegramConfig,
royalnet_config: typing.Optional[RoyalnetConfig] = None, royalnet_config: typing.Optional[RoyalnetConfig] = None,
database_config: typing.Optional[DatabaseConfig] = None, database_config: typing.Optional[DatabaseConfig] = None,
command_prefix: str = "/", commands: typing.List[typing.Type[Command]] = None):
commands: typing.List[typing.Type[Command]] = None,
missing_command: typing.Type[Command] = NullCommand,
error_command: typing.Type[Command] = NullCommand):
super().__init__(royalnet_config=royalnet_config, super().__init__(royalnet_config=royalnet_config,
database_config=database_config, database_config=database_config,
command_prefix=command_prefix, commands=commands)
commands=commands,
missing_command=missing_command,
error_command=error_command)
self._telegram_config = telegram_config self._telegram_config = telegram_config
self._init_client() self._init_client()
@ -92,15 +93,21 @@ class TelegramBot(GenericBot):
if text is None: if text is None:
return return
# Skip non-command updates # Skip non-command updates
if not text.startswith(self.command_prefix): if not text.startswith("/"):
return return
# Find and clean parameters # Find and clean parameters
command_text, *parameters = text.split(" ") command_text, *parameters = text.split(" ")
command_name = command_text.replace(f"@{self.client.username}", "").lower() command_name = command_text.replace(f"@{self.client.username}", "").lower()
# Send a typing notification # Send a typing notification
self.client.send_chat_action(update.message.chat, telegram.ChatAction.TYPING) self.client.send_chat_action(update.message.chat, telegram.ChatAction.TYPING)
# Call the command # Find the command
await self.call(command_name, update.message.chat, parameters, update=update) try:
command = self.commands[command_name]
except KeyError:
# Skip the message
return
# Run the command
await command.run(CommandArgs(parameters), self._Data(interface=command.interface, update=update))
async def run(self): async def run(self):
while True: while True:
@ -119,11 +126,3 @@ class TelegramBot(GenericBot):
except IndexError: except IndexError:
pass pass
@property
def botfather_command_string(self) -> str:
"""Generate a string to be pasted in the "Edit Commands" BotFather prompt."""
string = ""
for command_key in self.commands:
command = self.commands[command_key]
string += f"{command.command_name} - {command.command_description}\n"
return string

3
royalnet/commands/__init__.py
View file

@ -1,4 +1,5 @@
from .commandinterface import CommandInterface from .commandinterface import CommandInterface
from .command import Command from .command import Command
from .commanddata import CommandData
__all__ = ["CommandInterface", "Command"] __all__ = ["CommandInterface", "Command", "CommandData"]

3
royalnet/commands/command.py
View file

@ -2,6 +2,7 @@ import typing
from ..error import UnsupportedError from ..error import UnsupportedError
from .commandinterface import CommandInterface from .commandinterface import CommandInterface
from .commandargs import CommandArgs from .commandargs import CommandArgs
from .commanddata import CommandData
class Command: class Command:
@ -22,5 +23,5 @@ class Command:
def __init__(self, interface: CommandInterface): def __init__(self, interface: CommandInterface):
self.interface = interface self.interface = interface
async def run(self, args: CommandArgs, **extra) -> None: async def run(self, args: CommandArgs, data: CommandData) -> None:
raise UnsupportedError(f"Command {self.name} can't be called on {self.interface.name}.") raise UnsupportedError(f"Command {self.name} can't be called on {self.interface.name}.")

18
royalnet/commands/commanddata.py Normal file
View file

@ -0,0 +1,18 @@
class CommandData:
async def reply(self, text: str) -> None:
"""Send a text message to the channel where the call was made.
Parameters:
text: The text to be sent, possibly formatted in the weird undescribed markup that I'm using."""
raise NotImplementedError()
async def get_author(self, error_if_none: bool = False):
"""Try to find the identifier of the user that sent the message.
That probably means, the database row identifying the user.
Parameters:
error_if_none: Raise a :py:exc:`royalnet.error.UnregisteredError` if this is True and the call has no author.
Raises:
:py:exc:`royalnet.error.UnregisteredError` if ``error_if_none`` is set to True and no author is found."""
raise NotImplementedError()

28
royalnet/commands/commandinterface.py
View file

@ -10,38 +10,22 @@ class CommandInterface:
alchemy: "Alchemy" = NotImplemented alchemy: "Alchemy" = NotImplemented
bot: "GenericBot" = NotImplemented bot: "GenericBot" = NotImplemented
def __init__(self, alias: str): def __init__(self):
self.session = self.alchemy.Session() self.session = self.alchemy.Session()
def register_net_handler(self, message_type: str, network_handler: typing.Callable): def register_net_handler(self, message_type: str, network_handler: typing.Callable):
"""Register a new handler for messages received through Royalnet.""" """Register a new handler for messages received through Royalnet."""
raise NotImplementedError() raise NotImplementedError()
async def reply(self, extra: dict, text: str) -> None: def unregister_net_handler(self, message_type: str):
"""Send a text message to the channel where the call was made. """Remove a Royalnet handler."""
Parameters:
extra: The ``extra`` dict passed to the Command
text: The text to be sent, possibly formatted in the weird undescribed markup that I'm using."""
raise NotImplementedError() raise NotImplementedError()
async def net_request(self, extra: dict, message, destination: str) -> dict: async def net_request(self, message, destination: str) -> dict:
"""Send data through a :py:class:`royalnet.network.RoyalnetLink` and wait for a :py:class:`royalnet.network.Reply`. """Send data through a :py:class:`royalnet.network.RoyalnetLink` and wait for a
:py:class:`royalnet.network.Reply`.
Parameters: Parameters:
extra: The ``extra`` dict passed to the Command
message: The data to be sent. Must be :py:mod:`pickle`-able. message: The data to be sent. Must be :py:mod:`pickle`-able.
destination: The destination of the request, either in UUID format or node name.""" destination: The destination of the request, either in UUID format or node name."""
raise NotImplementedError() raise NotImplementedError()
async def get_author(self, extra: dict, error_if_none: bool = False):
"""Try to find the identifier of the user that sent the message.
That probably means, the database row identifying the user.
Parameters:
extra: The ``extra`` dict passed to the Command
error_if_none: Raise a :py:exc:`royalnet.error.UnregisteredError` if this is True and the call has no author.
Raises:
:py:exc:`royalnet.error.UnregisteredError` if ``error_if_none`` is set to True and no author is found."""
raise NotImplementedError()

32
royalnet/commands/royalgames/ping.py
View file

@ -1,21 +1,21 @@
import asyncio import typing
from ..utils import Command, Call from ..command import Command
from ..error import InvalidInputError from ..commandinterface import CommandInterface
from ..commandargs import CommandArgs
from ..commanddata import CommandData
class PingCommand(Command): class PingCommand(Command):
name: str = "ping"
command_name = "ping" description: str = "Replies with a Pong!"
command_description = "Ping pong dopo un po' di tempo!"
command_syntax = "[time_to_wait]"
@classmethod syntax: str = ""
async def common(cls, call: Call):
try: require_alchemy_tables: typing.Set = set()
time = int(call.args[0])
except InvalidInputError: def __init__(self, interface: CommandInterface):
time = 0 super().__init__(interface)
except ValueError:
raise InvalidInputError("time_to_wait is not a number") async def run(self, args: CommandArgs, data: CommandData) -> None:
await asyncio.sleep(time) await data.reply("Pong!")
await call.reply("🏓 Pong!")