mirror of
https://gitlab.com/ceda_ei/Quadnite-Bot
synced 2026-05-12 08:40:06 +02:00
Add commands to Command, logging and improve command check
This commit is contained in:
@@ -1,3 +1,5 @@
|
||||
import logging
|
||||
|
||||
from deltabot_cli import BotCli
|
||||
from deltachat2 import events
|
||||
|
||||
@@ -6,6 +8,8 @@ from quadnite_bot.command_registry import dispatcher
|
||||
def main():
|
||||
cli = BotCli("quadnite-bot")
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
cli.on(events.NewMessage)(dispatcher)
|
||||
|
||||
cli.start()
|
||||
|
||||
@@ -1,12 +1,28 @@
|
||||
from abc import ABC, abstractmethod
|
||||
import logging
|
||||
import re
|
||||
from deltachat2 import Bot, MsgData, NewMsgEvent
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class Context:
|
||||
def __init__(self, bot: Bot, accid: int, event: NewMsgEvent) -> None:
|
||||
match: re.Match[str] | None
|
||||
params: str
|
||||
args: list[str]
|
||||
|
||||
def __init__(self, bot: Bot, accid: int, event: NewMsgEvent, command: "Command") -> None:
|
||||
self.bot = bot
|
||||
self.accid = accid
|
||||
self.event = event
|
||||
if command.command:
|
||||
self.params = re.sub(fr"^/{command.command}\s+", "", event.msg.text)
|
||||
else:
|
||||
self.params = event.msg.text
|
||||
self.args = re.split(r"\s+", self.params)
|
||||
self.match = None
|
||||
if command.regex:
|
||||
self.match = command.regex.search(event.msg.text)
|
||||
|
||||
def reply(self, text: str):
|
||||
self.bot.rpc.send_msg(
|
||||
@@ -18,14 +34,21 @@ class Context:
|
||||
|
||||
class Command(ABC):
|
||||
command: str|None = None
|
||||
commands: list[str]|None = None
|
||||
regex: re.Pattern|None = None
|
||||
run_next: bool = False
|
||||
|
||||
def handles_message(self, event: NewMsgEvent) -> bool:
|
||||
if self.command:
|
||||
return event.msg.text.lower().startswith(f"/{self.command.lower()} ")
|
||||
commands = [self.command] if self.command else self.commands
|
||||
if commands:
|
||||
for command in commands:
|
||||
user_input = event.msg.text.lower()
|
||||
expected_input = f"/{command.lower()}"
|
||||
if user_input == expected_input or user_input.startswith(expected_input + " "):
|
||||
return True
|
||||
return False
|
||||
if self.regex:
|
||||
return self.regex.search(event.msg) is not None
|
||||
return self.regex.search(event.msg.text) is not None
|
||||
raise NotImplementedError(f"The command has neither 'command' nor 'regex' set. Required one of the two.")
|
||||
|
||||
@abstractmethod
|
||||
@@ -33,7 +56,7 @@ class Command(ABC):
|
||||
pass
|
||||
|
||||
def __call__(self, bot: Bot, accid: int, event: NewMsgEvent) -> None:
|
||||
self.process_event(Context(bot, accid, event), event)
|
||||
self.process_event(Context(bot, accid, event, self), event)
|
||||
|
||||
|
||||
class CommandDispatcher:
|
||||
@@ -44,8 +67,11 @@ class CommandDispatcher:
|
||||
self._commands.append(command)
|
||||
|
||||
def __call__(self, bot: Bot, accid: int, event: NewMsgEvent) -> None:
|
||||
logger.info("Received new message")
|
||||
for command in self._commands:
|
||||
logger.debug("Checking against %s", command)
|
||||
if command.handles_message(event):
|
||||
logger.info("Match found. Running the command %s", command)
|
||||
command(bot, accid, event)
|
||||
if not command.run_next:
|
||||
break
|
||||
|
||||
Reference in New Issue
Block a user