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

Maybe this was a bad idea afterall

Browse source
This commit is contained in:
Steffo 2019-08-18 01:10:01 +03:00
parent e1c3270527
commit d2c1f87798
6 changed files with 59 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
royalnet/bots/discord.py
View file

@ -2,7 +2,6 @@ import discord
import typing
import logging as _logging
from .generic import GenericBot
from royalnet.commands import NullCommand
from ..utils import asyncify, Call, Command, discord_escape
from ..error import UnregisteredError, NoneFoundError, TooManyFoundError, InvalidConfigError, RoyalnetResponseError
from ..network import RoyalnetConfig, Request, ResponseSuccess, ResponseError
@ -32,7 +31,7 @@ class DiscordBot(GenericBot):
log.debug(f"Creating music_data dict")
self.music_data: typing.Dict[discord.Guild, playmodes.PlayMode] = {}
def _call_factory(self) -> typing.Type[Call]:
def _interface_factory(self) -> typing.Type[Call]:
log.debug(f"Creating DiscordCall")
# noinspection PyMethodParameters

42
royalnet/bots/generic.py
View file

@ -2,10 +2,11 @@ import sys
import typing
import asyncio
import logging
from ..utils import Command, NetworkHandler, Call
from royalnet.commands import NullCommand
from ..network import RoyalnetLink, Request, Response, ResponseError, RoyalnetConfig
from ..utils import NetworkHandler
from ..network import RoyalnetLink, Request, Response, ResponseSuccess, ResponseError, RoyalnetConfig
from ..database import Alchemy, DatabaseConfig, relationshiplinkchain
from ..commands import Command, CommandInterface
from ..error import *
log = logging.getLogger(__name__)
@ -28,17 +29,40 @@ class GenericBot:
for command in commands:
lower_command_name = command.command_name.lower()
self.commands[f"{command_prefix}{lower_command_name}"] = command
self.network_handlers = {**self.network_handlers, **command.network_handler_dict()}
self.missing_command: typing.Type[Command] = missing_command
self.error_command: typing.Type[Command] = error_command
log.debug(f"Successfully generated commands")
def _call_factory(self) -> typing.Type[Call]:
"""Create the TelegramCall class, representing a command call. It should inherit from :py:class:`royalnet.utils.Call`.
def _interface_factory(self) -> typing.Type[CommandInterface]:
"""Create a :py:class:`royalnet.commands.CommandInterface` type and return it.
Returns:
The created TelegramCall class."""
raise NotImplementedError()
The created :py:class:`royalnet.commands.CommandInterface` type."""
# noinspection PyAbstractClass,PyMethodParameters
class GenericInterface(CommandInterface):
alchemy = self.alchemy
bot = self
def register_net_handler(ci, message_type: str, network_handler: typing.Callable):
self.network_handlers[message_type] = network_handler
async def net_request(ci, request: Request, destination: str) -> dict:
if self.network is None:
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
return GenericInterface
def _init_royalnet(self, royalnet_config: RoyalnetConfig):
"""Create a :py:class:`royalnet.network.RoyalnetLink`, and run it as a :py:class:`asyncio.Task`."""
@ -117,7 +141,7 @@ class GenericBot:
if commands is None:
commands = []
self._init_commands(command_prefix, commands, missing_command=missing_command, error_command=error_command)
self._Call = self._call_factory()
self._Call = self._interface_factory()
if royalnet_config is None:
self.network = None
else:

41
royalnet/bots/telegram.py
View file

@ -3,11 +3,11 @@ import telegram.utils.request
import typing
import logging as _logging
from .generic import GenericBot
from royalnet.commands import NullCommand
from ..utils import asyncify, Call, Command, telegram_escape
from ..utils import asyncify, telegram_escape
from ..error import UnregisteredError, InvalidConfigError, RoyalnetResponseError
from ..network import RoyalnetConfig, Request, ResponseSuccess, ResponseError
from ..database import DatabaseConfig
from ..commands import CommandInterface
log = _logging.getLogger(__name__)
@ -21,7 +21,6 @@ class TelegramConfig:
class TelegramBot(GenericBot):
"""A bot that connects to `Telegram <https://telegram.org/>`_."""
interface_name = "telegram"
def _init_client(self):
"""Create the :py:class:`telegram.Bot`, and set the starting offset."""
@ -30,43 +29,29 @@ class TelegramBot(GenericBot):
self.client = telegram.Bot(self._telegram_config.token, request=request)
self._offset: int = -100
def _call_factory(self) -> typing.Type[Call]:
def _interface_factory(self) -> typing.Type[CommandInterface]:
GenericInterface = super()._interface_factory()
# noinspection PyMethodParameters
class TelegramCall(Call):
interface_name = self.interface_name
interface_obj = self
interface_prefix = "/"
class TelegramInterface(GenericInterface):
name = "telegram"
prefix = "/"
alchemy = self.alchemy
async def reply(call, text: str):
await asyncify(call.channel.send_message, telegram_escape(text),
async def reply(ci, extra: dict, text: str):
await asyncify(ci.channel.send_message, telegram_escape(text),
parse_mode="HTML",
disable_web_page_preview=True)
async def net_request(call, request: Request, destination: str) -> dict:
if self.network is None:
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):
update: telegram.Update = call.kwargs["update"]
async def get_author(ci, extra: dict, error_if_none=False):
update: telegram.Update = extra["update"]
user: telegram.User = update.effective_user
if user is None:
if error_if_none:
raise UnregisteredError("No author for this message")
return None
query = call.session.query(self.master_table)
query = ci.session.query(self.master_table)
for link in self.identity_chain:
query = query.join(link.mapper.class_)
query = query.filter(self.identity_column == user.id)

6
royalnet/commands/command.py
View file

@ -1,6 +1,7 @@
import typing
from ..error import UnsupportedError
from .commandinterface import CommandInterface
from .commandargs import CommandArgs
class Command:
@ -21,8 +22,5 @@ class Command:
def __init__(self, interface: CommandInterface):
self.interface = interface
async def common(self) -> None:
async def run(self, args: CommandArgs, **extra) -> None:
raise UnsupportedError(f"Command {self.name} can't be called on {self.interface.name}.")
def __getattr__(self, item) -> typing.Callable:
return self.common

0
royalnet/utils/commandargs.py → royalnet/commands/commandargs.py
View file

17
royalnet/commands/commandinterface.py
View file

@ -13,32 +13,35 @@ class CommandInterface:
def __init__(self, alias: str):
self.session = self.alchemy.Session()
async def reply(self, text: str) -> None:
def register_net_handler(self, message_type: str, network_handler: typing.Callable):
"""Register a new handler for messages received through Royalnet."""
raise NotImplementedError()
async def reply(self, extra: dict, text: str) -> None:
"""Send a text message to the channel where the call was made.
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()
async def net_request(self, message, destination: str) -> dict:
async def net_request(self, extra: dict, message, destination: str) -> dict:
"""Send data through a :py:class:`royalnet.network.RoyalnetLink` and wait for a :py:class:`royalnet.network.Reply`.
Parameters:
extra: The ``extra`` dict passed to the Command
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."""
raise NotImplementedError()
async def get_author(self, error_if_none: bool = False):
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()
def register_net_handler(self, message_type: str, network_handler: typing.Callable):
"""Register a new handler for messages received through Royalnet."""
raise NotImplementedError()