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

Start work to remove interfaces

Browse source
This commit is contained in:
Steffo 2020-07-26 19:27:05 +02:00
parent 7845091bb7
commit f2ff31d155
14 changed files with 100 additions and 222 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

@ -5,7 +5,7 @@
[tool.poetry]
name = "royalnet"
version = "5.10.4"
version = "5.11.0"
description = "A multipurpose bot and web framework"
authors = ["Stefano Pigozzi <ste.pigozzi@gmail.com>"]
license = "AGPL-3.0+"

11
royalnet/backpack/commands/royalnetsync.py
View file

@ -2,6 +2,9 @@ from typing import *
import royalnet
import royalnet.commands as rc
import royalnet.utils as ru
import royalnet.serf.telegram
import royalnet.serf.discord
import royalnet.serf.matrix
from ..tables.telegram import Telegram
from ..tables.discord import Discord
@ -27,7 +30,7 @@ class RoyalnetsyncCommand(rc.Command):
if not successful:
raise rc.InvalidInputError(f"Invalid password!")
if self.interface.name == "telegram":
if isinstance(self.serf, royalnet.serf.telegram.TelegramSerf):
import telegram
message: telegram.Message = data.message
from_user: telegram.User = message.from_user
@ -53,7 +56,7 @@ class RoyalnetsyncCommand(rc.Command):
await data.session_commit()
await data.reply(f"↔️ Account {tg_user} synced to {author}!")
elif self.interface.name == "discord":
elif isinstance(self.serf, royalnet.serf.discord.DiscordSerf):
import discord
message: discord.Message = data.message
author: discord.User = message.author
@ -79,8 +82,8 @@ class RoyalnetsyncCommand(rc.Command):
await data.session_commit()
await data.reply(f"↔️ Account {ds_user} synced to {author}!")
elif self.interface.name == "matrix":
elif isinstance(self.serf, royalnet.serf.matrix.MatrixSerf):
raise rc.UnsupportedError(f"{self} hasn't been implemented for Matrix yet")
else:
raise rc.UnsupportedError(f"Unknown interface: {self.interface.name}")
raise rc.UnsupportedError(f"Unknown interface: {self.serf.__class__.__qualname__}")

8
royalnet/backpack/commands/royalnetversion.py
View file

@ -1,4 +1,4 @@
import royalnet.version
import pkg_resources
from royalnet.commands import *
@ -7,12 +7,16 @@ class RoyalnetversionCommand(Command):
description: str = "Display the current Royalnet version."
@property
def royalnet_version(self):
return pkg_resources.get_distribution("royalnet").version
async def run(self, args: CommandArgs, data: CommandData) -> None:
# noinspection PyUnreachableCode
if __debug__:
message = f"ℹ️ Royalnet [url=https://github.com/Steffo99/royalnet/]Unreleased[/url]\n"
else:
message = f"ℹ️ Royalnet [url=https://github.com/Steffo99/royalnet/releases/tag/{royalnet.version.semantic}]{royalnet.version.semantic}[/url]\n"
message = f"ℹ️ Royalnet [url=https://github.com/Steffo99/royalnet/releases/tag/{self.royalnet_version}]{self.royalnet_version}[/url]\n"
if "69" in royalnet.version.semantic:
message += "(Nice.)"
await data.reply(message)

4
royalnet/backpack/events/exception.py
View file

@ -5,6 +5,6 @@ class ExceptionEvent(Event):
name = "exception"
def run(self, **kwargs):
if not self.interface.config["exc_debug"]:
raise UserError(f"{self.interface.prefix}{self.name} is not enabled.")
if not self.config["exc_debug"]:
raise UserError(f"{self.__class__.__name__} is not enabled.")
raise Exception(f"{self.name} event was called")

2
royalnet/commands/__init__.py
View file

@ -1,6 +1,5 @@
"""The subpackage providing all classes related to Royalnet commands."""
from .commandinterface import CommandInterface
from .command import Command
from .commanddata import CommandData
from .commandargs import CommandArgs
@ -11,7 +10,6 @@ from .keyboardkey import KeyboardKey
from .configdict import ConfigDict
__all__ = [
"CommandInterface",
"Command",
"CommandData",
"CommandArgs",

26
royalnet/commands/command.py
View file

@ -1,10 +1,10 @@
import abc
from typing import *
from .commandinterface import CommandInterface
from .commandargs import CommandArgs
from .commanddata import CommandData
class Command:
class Command(metaclass=abc.ABCMeta):
name: str = NotImplemented
"""The main name of the command.
@ -24,31 +24,23 @@ class Command:
"""The syntax of the command, to be displayed when a :py:exc:`InvalidInputError` is raised,
in the format ``(required_arg) [optional_arg]``."""
def __init__(self, interface: CommandInterface):
self.interface = interface
def __init__(self, serf, config):
self.serf = serf
self.config = config
def __str__(self):
return f"[c]{self.interface.prefix}{self.name}[/c]"
@property
def serf(self):
"""A shortcut for :attr:`.interface.serf`."""
return self.interface.serf
return f"[c]{self.serf.prefix}{self.name}[/c]"
@property
def alchemy(self):
"""A shortcut for :attr:`.interface.alchemy`."""
return self.interface.alchemy
return self.serf.alchemy
@property
def loop(self):
"""A shortcut for :attr:`.interface.loop`."""
return self.interface.loop
@property
def config(self):
"""A shortcut for :attr:`.interface.config`."""
return self.interface.config
return self.serf.loop
@abc.abstractmethod
async def run(self, args: CommandArgs, data: CommandData) -> None:
raise NotImplementedError()

14
royalnet/commands/commanddata.py
View file

@ -9,27 +9,29 @@ from royalnet.backpack.tables.users import User
if TYPE_CHECKING:
from .keyboardkey import KeyboardKey
from royalnet.backpack.tables.users import User
log = logging.getLogger(__name__)
class CommandData:
def __init__(self, interface: CommandInterface, loop: aio.AbstractEventLoop):
self.loop: aio.AbstractEventLoop = loop
self._interface: CommandInterface = interface
def __init__(self, command):
self.command = command
self._session = None
# TODO: make this asyncronous... somehow?
@property
def session(self):
if self._session is None:
if self._interface.alchemy is None:
if self.command.serf.alchemy is None:
raise UnsupportedError("'alchemy' is not enabled on this Royalnet instance")
log.debug("Creating Session...")
self._session = self._interface.alchemy.Session()
self._session = self.command.serf.alchemy.Session()
return self._session
@property
def loop(self):
return self.command.serf.loop
async def session_commit(self):
"""Asyncronously commit the :attr:`.session` of this object."""
if self._session:

76
royalnet/commands/commandinterface.py
View file

@ -1,76 +0,0 @@
from typing import *
import asyncio as aio
from .errors import UnsupportedError
from .configdict import ConfigDict
if TYPE_CHECKING:
from .event import Event
from .command import Command
from ..alchemy import Alchemy
from ..serf import Serf
from ..constellation import Constellation
class CommandInterface:
name: str = NotImplemented
"""The name of the :class:`CommandInterface` that's being implemented.
Examples:
``telegram``, ``discord``, ``console``..."""
prefix: str = NotImplemented
"""The prefix used by commands on the interface.
Examples:
``/`` on Telegram, ``!`` on Discord."""
serf: Optional["Serf"] = None
"""A reference to the :class:`~royalnet.serf.Serf` that is implementing this :class:`CommandInterface`.
Example:
A reference to a :class:`~royalnet.serf.telegram.TelegramSerf`."""
constellation: Optional["Constellation"] = None
"""A reference to the Constellation that is implementing this :class:`CommandInterface`.
Example:
A reference to a :class:`~royalnet.constellation.Constellation`."""
def __init__(self, config: Dict[str, Any]):
self.config: ConfigDict[str, Any] = ConfigDict.convert(config)
"""The config section for the pack of the command."""
# Will be bound after the command/event has been created
self.command: Optional[Command] = None
self.event: Optional[Event] = None
@property
def alchemy(self) -> "Alchemy":
"""A shortcut for :attr:`.serf.alchemy`."""
return self.serf.alchemy
@property
def table(self) -> "Callable":
"""A shortcut for :func:`.serf.alchemy.get`.
Raises:
UnsupportedError: if :attr:`.alchemy` is :const:`None`."""
if self.alchemy is None:
raise UnsupportedError("'alchemy' is not enabled on this Royalnet instance")
return self.alchemy.get
@property
def loop(self) -> aio.AbstractEventLoop:
"""A shortcut for :attr:`.serf.loop`."""
if self.serf:
return self.serf.loop
raise UnsupportedError("This command is not being run in a serf.")
async def call_herald_event(self, destination: str, event_name: str, **kwargs) -> dict:
"""Call an event function on a different :class:`~royalnet.serf.Serf`.
Example:
You can run a function on a :class:`~royalnet.serf.discord.DiscordSerf` from a
:class:`~royalnet.serf.telegram.TelegramSerf`.
"""
raise UnsupportedError(f"{self.call_herald_event.__name__} is not supported on this platform.")

22
royalnet/commands/event.py
View file

@ -1,5 +1,4 @@
import asyncio as aio
from .commandinterface import CommandInterface
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@ -12,30 +11,19 @@ class Event:
name = NotImplemented
"""The event_name that will trigger this event."""
def __init__(self, interface: CommandInterface):
"""Bind the event to a :class:`~royalnet.serf.Serf`."""
self.interface: CommandInterface = interface
"""The :class:`CommandInterface` available to this :class:`Event`."""
@property
def serf(self) -> "Serf":
"""A shortcut for :attr:`.interface.serf`."""
return self.interface.serf
def __init__(self, serf, config):
self.serf = serf
self.config = config
@property
def alchemy(self):
"""A shortcut for :attr:`.interface.alchemy`."""
return self.interface.alchemy
return self.serf.alchemy
@property
def loop(self) -> aio.AbstractEventLoop:
"""A shortcut for :attr:`.interface.loop`."""
return self.interface.loop
@property
def config(self) -> dict:
"""A shortcut for :attr:`.interface.config`."""
return self.interface.config
return self.serf.loop
async def run(self, **kwargs):
raise NotImplementedError()

3
royalnet/commands/keyboardkey.py
View file

@ -1,15 +1,12 @@
from typing import *
from .commandinterface import CommandInterface
from .commanddata import CommandData
class KeyboardKey:
def __init__(self,
interface: CommandInterface,
short: str,
text: str,
callback: Callable[[CommandData], Awaitable[None]]):
self.interface: CommandInterface = interface
self.short: str = short
self.text: str = text
self.callback: Callable[[CommandData], Awaitable[None]] = callback

46
royalnet/serf/serf.py
View file

@ -20,6 +20,7 @@ class Serf(abc.ABC):
"""An abstract class, to be used as base to implement Royalnet bots on multiple interfaces (such as Telegram or
Discord)."""
interface_name = NotImplemented
prefix = NotImplemented
_master_table: type = rbt.User
_identity_table: type = NotImplemented
@ -91,9 +92,6 @@ class Serf(abc.ABC):
self.events: Dict[str, Event] = {}
"""A dictionary containing all :class:`Event` that can be handled by this :class:`Serf`."""
self.Interface: Type[CommandInterface] = self.interface_factory()
"""The :class:`CommandInterface` class of this Serf."""
self.commands: Dict[str, Command] = {}
"""The :class:`dict` connecting each command name to its :class:`Command` object."""
@ -138,15 +136,7 @@ class Serf(abc.ABC):
"""Find a relationship path starting from the master table and ending at the identity table, and return it."""
return ra.table_dfs(self.master_table, self.identity_table)
def interface_factory(self) -> Type[CommandInterface]:
"""Create the :class:`CommandInterface` class for the Serf."""
# noinspection PyMethodParameters
class GenericInterface(CommandInterface):
alchemy: ra.Alchemy = self.alchemy
serf: "Serf" = self
async def call_herald_event(ci, destination: str, event_name: str, **kwargs) -> Dict:
async def call_herald_event(self, destination: str, event_name: str, **kwargs) -> Dict:
"""Send a :class:`royalherald.Request` to a specific destination, and wait for a
:class:`royalherald.Response`."""
if self.herald is None:
@ -187,42 +177,34 @@ class Serf(abc.ABC):
raise ProgramError(f"Other Herald Link returned unknown response:\n"
f"[p]{response}[/p]")
return GenericInterface
def register_commands(self, commands: List[Type[Command]], pack_cfg: Dict[str, Any]) -> None:
"""Initialize and register all commands passed as argument."""
# Instantiate the Commands
for SelectedCommand in commands:
# Create a new interface
interface = self.Interface(config=pack_cfg)
# Try to instantiate the command
try:
command = SelectedCommand(interface)
command = SelectedCommand(serf=self, config=pack_cfg)
except Exception as e:
log.error(f"Skipping: "
f"{SelectedCommand.__qualname__} - {e.__class__.__qualname__} in the initialization.")
ru.sentry_exc(e)
continue
# Link the interface to the command
interface.command = command
# Warn if the command would be overriding something
if f"{self.Interface.prefix}{SelectedCommand.name}" in self.commands:
if SelectedCommand.name in self.commands:
log.info(f"Overriding (already defined): "
f"{SelectedCommand.__qualname__} -> {self.Interface.prefix}{SelectedCommand.name}")
f"{SelectedCommand.__qualname__} -> {SelectedCommand.name}")
else:
log.debug(f"Registering: "
f"{SelectedCommand.__qualname__} -> {self.Interface.prefix}{SelectedCommand.name}")
f"{SelectedCommand.__qualname__} -> {SelectedCommand.name}")
# Register the command in the commands dict
self.commands[f"{interface.prefix}{SelectedCommand.name}"] = command
self.commands[SelectedCommand.name] = command
# Register aliases, but don't override anything
for alias in SelectedCommand.aliases:
if f"{interface.prefix}{alias}" not in self.commands:
log.debug(f"Aliasing: {SelectedCommand.__qualname__} -> {interface.prefix}{alias}")
self.commands[f"{interface.prefix}{alias}"] = \
self.commands[f"{interface.prefix}{SelectedCommand.name}"]
if alias not in self.commands:
log.debug(f"Aliasing: {SelectedCommand.__qualname__} -> {alias}")
self.commands[alias] = self.commands[SelectedCommand.name]
else:
log.warning(
f"Ignoring (already defined): {SelectedCommand.__qualname__} -> {interface.prefix}{alias}")
log.warning(f"Ignoring (already defined): {SelectedCommand.__qualname__} -> {alias}")
def init_herald(self, herald_cfg: Dict[str, Any]):
"""Create a :class:`Link` and bind :class:`Event`."""
@ -231,11 +213,9 @@ class Serf(abc.ABC):
def register_events(self, events: List[Type[Event]], pack_cfg: Dict[str, Any]):
for SelectedEvent in events:
# Create a new interface
interface = self.Interface(config=pack_cfg)
# Initialize the event
try:
event = SelectedEvent(interface)
event = SelectedEvent(serf=self, config=pack_cfg)
except Exception as e:
log.error(f"Skipping: "
f"{SelectedEvent.__qualname__} - {e.__class__.__qualname__} in the initialization.")
@ -285,7 +265,7 @@ class Serf(abc.ABC):
await command.run(CommandArgs(parameters), data)
except InvalidInputError as e:
await data.reply(f"⚠️ {e.message}\n"
f"Syntax: [c]{command.interface.prefix}{command.name} {command.syntax}[/c]")
f"Syntax: [c]{self.prefix}{command.name} {command.syntax}[/c]")
except UserError as e:
await data.reply(f"⚠️ {e.message}")
except UnsupportedError as e:

4
royalnet/serf/telegram/escape.py
View file

@ -1,11 +1,13 @@
from typing import *
import re
def escape(string: str) -> str:
def escape(string: Optional[str]) -> Optional[str]:
"""Escape a string to be sent through Telegram (as HTML), and format it using RoyalCode.
Warning:
Currently escapes everything, even items in code blocks."""
url_pattern = re.compile(r"\[url=(.*?)](.*?)\[/url]")
url_replacement = r'<a href="\1">\2</a>'

26
royalnet/serf/telegram/telegramserf.py
View file

@ -24,6 +24,7 @@ log = logging.getLogger(__name__)
class TelegramSerf(Serf):
"""A Serf that connects to `Telegram <https://telegram.org/>`_ as a bot."""
interface_name = "telegram"
prefix = "/"
_identity_table = rbt.Telegram
_identity_column = "tg_id"
@ -90,25 +91,13 @@ class TelegramSerf(Serf):
break
return None
def interface_factory(self) -> Type[rc.CommandInterface]:
# noinspection PyPep8Naming
GenericInterface = super().interface_factory()
# noinspection PyMethodParameters
class TelegramInterface(GenericInterface):
name = self.interface_name
prefix = "/"
return TelegramInterface
def message_data_factory(self) -> Type[rc.CommandData]:
# noinspection PyMethodParameters
class TelegramMessageData(rc.CommandData):
def __init__(data,
interface: rc.CommandInterface,
loop: aio.AbstractEventLoop,
command: rc.Command,
message: telegram.Message):
super().__init__(interface=interface, loop=loop)
super().__init__(command=command)
data.message: telegram.Message = message
async def reply(data, text: str):
@ -171,10 +160,9 @@ class TelegramSerf(Serf):
# noinspection PyMethodParameters
class TelegramKeyboardData(rc.CommandData):
def __init__(data,
interface: rc.CommandInterface,
loop: aio.AbstractEventLoop,
command: rc.Command,
cbq: telegram.CallbackQuery):
super().__init__(interface=interface, loop=loop)
super().__init__(command=command)
data.cbq: telegram.CallbackQuery = cbq
async def reply(data, text: str):
@ -250,7 +238,7 @@ class TelegramSerf(Serf):
# Send a typing notification
await self.api_call(message.chat.send_action, telegram.ChatAction.TYPING)
# Prepare data
data = self.MessageData(interface=command.interface, loop=self.loop, message=message)
data = self.MessageData(command=command, message=message)
# Call the command
await self.call(command, data, parameters)
@ -260,7 +248,7 @@ class TelegramSerf(Serf):
await self.api_call(cbq.answer, text="⚠️ This keyboard has expired.", show_alert=True)
return
key: rc.KeyboardKey = self.key_callbacks[uid]
data: rc.CommandData = self.CallbackData(interface=key.interface, loop=self.loop, cbq=cbq)
data: rc.CommandData = self.CallbackData(command=command, cbq=cbq)
await self.press(key, data)
def register_keyboard_key(self, identifier: str, key: rc.KeyboardKey):

2
royalnet/version.py
View file

@ -1 +1 @@
semantic = "5.10.4"
semantic = "5.11.0"