1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 |
- from dataclasses import dataclass
- from typing import Any, Generic, TypeVar
- from .types import CommandConfiguration, Message, Response, Context
- # TODO logging
- RawMsg = TypeVar('RawMsg')
- @dataclass
- class Rollbot(Generic[RawMsg]):
- command_config: CommandConfiguration
- database_file: str
- def read_config(self, key: str) -> Any:
- raise NotImplemented("Must be implemented by driver")
- def parse(self, incoming: RawMsg) -> Message:
- raise NotImplemented("Must be implemented by driver")
- async def respond(self, response: Response):
- raise NotImplemented("Must be implemented by driver")
- async def on_message(self, incoming: RawMsg):
- message = self.parse(incoming)
- if message.text is None:
- return
- cleaned = message.text.lstrip()
- if len(cleaned) == 0 or cleaned[0] not in self.command_config.bangs:
- return
- parts = cleaned[1:].lstrip().split(maxsplit=1)
- if len(parts) == 0:
- return
- command = self.command_config.aliases.get(parts[0], parts[0])
- res = self.command_config.call_and_response.get(command, None)
- if res is not None:
- await self.respond(Response.from_message(message, res))
- return
- command_call = self.command_config.commands.get(command, None)
- if command is None:
- await self.respond(Response.from_message(message, f"Sorry! I don't know the command {command}."))
- return
- await command_call(Context(
- message=message,
- config=self.read_config,
- respond=self.respond,
- # database=..., # TODO database
- ))
|