1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-27 13:34:28 +00:00

Do some more stuff

This commit is contained in:
Steffo 2019-03-15 11:28:48 +01:00
parent 35235d5ad9
commit c0dfcae156
7 changed files with 42 additions and 14 deletions

2
.gitignore vendored
View file

@ -1,2 +1,4 @@
.idea/ .idea/
.vscode/ .vscode/
__pycache__
test.py

View file

@ -1,25 +1,28 @@
import telegram import telegram
import asyncio import asyncio
import typing import typing
from ..commands import PingCommand from ..commands import NullCommand
from ..utils import asyncify, Call from ..utils import asyncify, Call, Command
class TelegramBot: class TelegramBot:
def __init__(self, api_key: str): def __init__(self, api_key: str, commands: typing.List[Command], *, missing_command: Command=NullCommand):
self.bot = telegram.Bot(api_key) self.bot = telegram.Bot(api_key)
self.should_run = False self.should_run = False
self.offset = -100 self.offset = -100
self.commands = { self.commands = commands
"/ping": PingCommand self.missing_command: typing.Callable = missing_command
} # Generate commands
self._commands = {}
for command in self.commands:
self._commands[f"/{command.command_name}"] = command
class TelegramCall(Call): class TelegramCall(Call):
interface_name = "telegram" interface_name = "telegram"
interface_obj = self interface_obj = self
async def reply(call, text: str): async def reply(self, text: str):
await asyncify(call.channel.send_message, text, parse_mode="HTML") await asyncify(self.channel.send_message, text, parse_mode="HTML")
self.Call = TelegramCall self.Call = TelegramCall
async def run(self): async def run(self):
@ -53,9 +56,9 @@ class TelegramBot:
command_text.replace(f"@{self.bot.username}", "") command_text.replace(f"@{self.bot.username}", "")
# Find the function # Find the function
try: try:
command = self.commands[command_text] command = self._commands[command_text]
except KeyError: except KeyError:
# Skip inexistent commands # Skip inexistent commands
return command = self.missing_command
# Call the command # Call the command
return await self.Call(message.chat, command).run() return await self.Call(message.chat, command, parameters).run()

View file

@ -1,3 +1,4 @@
from .null import NullCommand
from .ping import PingCommand from .ping import PingCommand
__all__ = ["PingCommand"] __all__ = ["NullCommand", "PingCommand"]

10
royalnet/commands/null.py Normal file
View file

@ -0,0 +1,10 @@
from ..utils import Command, Call
class NullCommand(Command):
command_name = "null"
command_title = "Do nothing"
async def common(self, call: Call, *args, **kwargs):
pass

View file

@ -2,5 +2,9 @@ from ..utils import Command, Call
class PingCommand(Command): class PingCommand(Command):
command_name = "ping"
command_title = "Ping pong!"
async def common(self, call: Call, *args, **kwargs): async def common(self, call: Call, *args, **kwargs):
await call.reply("Pong!") await call.reply("Pong!")

View file

@ -1,3 +1,4 @@
import typing
from .command import Command from .command import Command
@ -8,14 +9,18 @@ class Call:
interface_name = NotImplemented interface_name = NotImplemented
interface_obj = NotImplemented interface_obj = NotImplemented
async def reply(cls, text: str): async def reply(self, text: str):
"""Send a text message to the channel the call was made.""" """Send a text message to the channel the call was made."""
raise NotImplementedError() raise NotImplementedError()
# These parameters / methods should be left alone # These parameters / methods should be left alone
def __init__(self, channel, command: Command): def __init__(self, channel, command: Command, parameters: typing.List[str]=None):
self.channel = channel self.channel = channel
self.command = command self.command = command
if parameters is None:
self.parameters = []
else:
self.parameters = parameters
async def run(self, *args, **kwargs): async def run(self, *args, **kwargs):
try: try:

View file

@ -6,5 +6,8 @@ if typing.TYPE_CHECKING:
class Command: class Command:
"""A generic command, called from any source.""" """A generic command, called from any source."""
command_name: str = NotImplemented
command_title: str = NotImplemented
async def common(self, call: "Call", *args, **kwargs): async def common(self, call: "Call", *args, **kwargs):
raise NotImplementedError() raise NotImplementedError()