mirror of
https://gitlab.com/ceda_ei/Quadnite-Bot
synced 2026-05-06 05:40:04 +02:00
Add core structure for DeltaChat port
This commit is contained in:
3
src/quadnite_bot/__init__.py
Normal file
3
src/quadnite_bot/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
from quadnite_bot.__main__ import main
|
||||
|
||||
__all__ = ["main"]
|
||||
11
src/quadnite_bot/__main__.py
Normal file
11
src/quadnite_bot/__main__.py
Normal file
@@ -0,0 +1,11 @@
|
||||
from deltabot_cli import BotCli
|
||||
from deltachat2 import events
|
||||
|
||||
from quadnite_bot.command_registry import dispatcher
|
||||
|
||||
def main():
|
||||
cli = BotCli("quadnite-bot")
|
||||
|
||||
cli.on(events.NewMessage)(dispatcher)
|
||||
|
||||
cli.start()
|
||||
51
src/quadnite_bot/command.py
Normal file
51
src/quadnite_bot/command.py
Normal file
@@ -0,0 +1,51 @@
|
||||
from abc import ABC, abstractmethod
|
||||
import re
|
||||
from deltachat2 import Bot, MsgData, NewMsgEvent
|
||||
|
||||
class Context:
|
||||
def __init__(self, bot: Bot, accid: int, event: NewMsgEvent) -> None:
|
||||
self.bot = bot
|
||||
self.accid = accid
|
||||
self.event = event
|
||||
|
||||
def reply(self, text: str):
|
||||
self.bot.rpc.send_msg(
|
||||
self.accid,
|
||||
self.event.msg.chat_id,
|
||||
MsgData(text),
|
||||
)
|
||||
|
||||
|
||||
class Command(ABC):
|
||||
command: 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()} ")
|
||||
if self.regex:
|
||||
return self.regex.search(event.msg) is not None
|
||||
raise NotImplementedError(f"The command has neither 'command' nor 'regex' set. Required one of the two.")
|
||||
|
||||
@abstractmethod
|
||||
def process_event(self, ctx: Context, event: NewMsgEvent) -> None:
|
||||
pass
|
||||
|
||||
def __call__(self, bot: Bot, accid: int, event: NewMsgEvent) -> None:
|
||||
self.process_event(Context(bot, accid, event), event)
|
||||
|
||||
|
||||
class CommandDispatcher:
|
||||
def __init__(self):
|
||||
self._commands: list[Command] = []
|
||||
|
||||
def add_command(self, command: Command):
|
||||
self._commands.append(command)
|
||||
|
||||
def __call__(self, bot: Bot, accid: int, event: NewMsgEvent) -> None:
|
||||
for command in self._commands:
|
||||
if command.handles_message(event):
|
||||
command(bot, accid, event)
|
||||
if not command.run_next:
|
||||
break
|
||||
6
src/quadnite_bot/command_registry.py
Normal file
6
src/quadnite_bot/command_registry.py
Normal file
@@ -0,0 +1,6 @@
|
||||
from quadnite_bot.command import CommandDispatcher
|
||||
from quadnite_bot.commands.echo import EchoCommand
|
||||
|
||||
dispatcher = CommandDispatcher()
|
||||
|
||||
dispatcher.add_command(EchoCommand())
|
||||
0
src/quadnite_bot/commands/__init__.py
Normal file
0
src/quadnite_bot/commands/__init__.py
Normal file
8
src/quadnite_bot/commands/echo.py
Normal file
8
src/quadnite_bot/commands/echo.py
Normal file
@@ -0,0 +1,8 @@
|
||||
from deltachat2 import NewMsgEvent
|
||||
from quadnite_bot.command import Command, Context
|
||||
|
||||
class EchoCommand(Command):
|
||||
command = "echo"
|
||||
|
||||
def process_event(self, ctx: Context, event: NewMsgEvent) -> None:
|
||||
ctx.reply(event.msg.text)
|
||||
Reference in New Issue
Block a user