Initial commit

.gitignore

@@ -0,0 +1,8 @@
+**/__pycache__/
+*.pyc
+.venv
+**/*.egg-info/
+
+.vscode/
+
+secrets.toml

README.md

@@ -0,0 +1,28 @@
+# Rollbot v4
+
+- document store
+    - index on chat, channel, etc.
+    - defined by class
+    - back by sqlite
+- proper API for representing a source and response logic
+- liberal use of dependency injection
+- use asyncio/allow commands to be async
+
+Things that can be accessed in a command, regardless of platform:
+    - Message
+        - Source
+        - Channel
+        - Sender
+        - Message Text
+            - Parsed and converted arguments
+        - Timestamp
+        - Admin? (group vs global)
+        - Attachments?
+    - Configuration
+        - Secrets
+    - Database
+        - Query by source, channel, user, etc.
+        - Inject more specific queries
+    - Response
+        - Abstracted way to respond manually
+        - Should support attachment of some kind? Maybe just images?

lib/rollbot/__init__.py

@@ -0,0 +1,2 @@
+from .bot import Rollbot
+from .types import CommandConfiguration, Message, Response

lib/rollbot/bot.py

@@ -0,0 +1,53 @@
+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
+        ))

lib/rollbot/types.py

@@ -0,0 +1,58 @@
+from dataclasses import dataclass
+from datetime import datetime
+from collections.abc import Callable, Coroutine, Container
+from typing import Union, Any, Optional
+
+
+@dataclass
+class Attachment:
+    name: str
+    body: Union[str, bytes]
+
+
+@dataclass
+class Message:
+    origin_id: str
+    channel_id: str
+    sender_id: str
+    timestamp: datetime
+    origin_admin: bool
+    channel_admin: bool
+    text: Optional[str]
+    attachments: list[Attachment]
+
+
+@dataclass
+class Response:
+    origin_id: str
+    channel_id: str
+    text: str
+    attachments: list[Attachment]
+
+    @staticmethod
+    def from_message(msg: Message, text: str, attachments: list[Attachment] = None) -> "Response":
+        return Response(
+            origin_id=msg.origin_id,
+            channel_id=msg.channel_id,
+            text=text,
+            attachments=attachments or [],
+        )
+
+
+@dataclass
+class Context:
+    message: Message
+    config: Callable[[str], Any]
+    respond: Callable[[], Coroutine[None, None, None]]
+    # database: Callable # TODO proper type
+
+
+CommandType = Callable[[Context], Coroutine[None, None, None]]
+
+
+@dataclass
+class CommandConfiguration:
+    commands: dict[str, CommandType]
+    call_and_response: dict[str, str]
+    aliases: dict[str, str]
+    bangs: Container[str] = ("!",)

lib/setup.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python
+
+from distutils.core import setup
+
+setup(
+    name="rollbot",
+    version="4.0.0",
+    description="Rollbot framework library",
+    author="Kirk Trombley",
+    author_email="ktrom3894@gmail.com",
+    packages=["rollbot"],
+    install_requires=[
+        "aiosqlite",
+    ]
+)

repl_driver.py

@@ -0,0 +1,53 @@
+from datetime import datetime
+import asyncio
+
+import rollbot
+
+
+class MyBot(rollbot.Rollbot[str]):
+    def read_config(self, key):
+        return key
+
+    def parse(self, raw):
+        return rollbot.Message(
+            origin_id="REPL",
+            channel_id=".",
+            sender_id=".",
+            timestamp=datetime.now(),
+            origin_admin=True,
+            channel_admin=True,
+            text=raw,
+            attachments=[],
+        )
+
+    async def respond(self, res):
+        print(res, flush=True)
+
+
+async def goodbye_command(context):
+    await context.respond(rollbot.Response.from_message(context.message, "Goodbye!"))
+
+
+config = rollbot.CommandConfiguration(
+    bangs=("/",),
+    commands={
+        "goodbye": goodbye_command,
+    },
+    call_and_response={
+        "hello": "Hello!",
+    },
+    aliases={
+        "hi": "hello",
+        "bye": "goodbye",
+    }
+)
+
+bot = MyBot(config, "/tmp/my.db")
+
+
+async def run():
+    while True:
+        await bot.on_message(input("> "))
+
+
+asyncio.run(run())