Added example for commands

examples/command_bot.py

@@ -0,0 +1,130 @@
+"""
+This example implements Matrix bot that can execute some commands.
+
+It uses environment variables to specify authorization data. Use Ctrl+C to stop
+the bot.
+"""
+
+import asyncio
+import time
+import html
+import os
+import traceback
+import logging
+
+from mab import *
+
+from _environment import check_environment
+
+async def on_help_command(ctx: EventContext) -> None:
+    """!help"""
+    HELP_MESSAGE = (
+        "<strong>Here is the list of the commands:</strong><br>"
+        "<ul>"
+        "<li><code>!help</code> - this help message</li>"
+        "<li><code>!time</code> - get UNIX timestamp</li>"
+        "<li><code>!raise</code> - raise <code>RuntimeError()</code></li>"
+        "<li><code>!assert</code> - perform <code>assert</code> that will fail</li>"
+        "<li><code>!mul A B [C] [D]...</code> - multiply A, B... and so on</li>"
+        "<li><code>!args arg1 [arg2] ... [arg5]</code> - command that takes 1..5 arguments</li>"
+        "</ul>"
+    )
+    await ctx.bot.send_text(ctx.room, HELP_MESSAGE)
+
+async def on_time_command(ctx: EventContext) -> None:
+    """!time"""
+    await ctx.bot.send_text(
+        ctx.room,
+        f"Current UNIX timestamp is <strong>{int(time.time())}</strong>"
+    )
+
+async def on_raise_command(ctx: EventContext) -> None:
+    """!raise"""
+    await ctx.bot.send_text(
+        ctx.room,
+        "<strong>Executing <code>raise RuntimeError()</code>...</strong>"
+    )
+    raise RuntimeError()
+
+async def on_assert_command(ctx: EventContext) -> None:
+    """!assert"""
+    await ctx.bot.send_text(
+        ctx.room,
+        "<strong>Executing <code>assert False</code>...</strong>"
+    )
+    assert False
+
+async def on_mul_command(ctx: EventContext) -> None:
+    """!mul"""
+    try:
+        numbers = [float(v) for v in ctx[CTX_CMD_ARGS]]
+        v = numbers[0]
+        for n in numbers[1:]:
+            v *= n
+        response = " * ".join(html.escape("%.2f" % n) for n in numbers)
+        response += f" = <strong>{html.escape(str(v))}<strong>"
+        await ctx.bot.send_text(ctx.room, response)
+    except Exception as e:
+        await ctx.bot.send_text(ctx.room, f"Could not process the command: {e}")
+
+async def on_args_command(ctx: EventContext) -> None:
+    """!args"""
+    try:
+        response = (
+            f"Prefix: <code>{ctx[CTX_CMD_PREFIX]}</code><br>"
+            f"Verb: <code>{ctx[CTX_CMD_VERB]}</code><br>"
+            f"Arguments: <code>{len(ctx[CTX_CMD_ARGS])}</code><br>"
+            f"Arguments are:<br><ol>"
+        )
+        for arg in ctx[CTX_CMD_ARGS]:
+            response += f"<li><code>{html.escape(arg)}</code></li>"
+        response += "</ol>"
+        await ctx.bot.send_text(
+            ctx.room,
+            response
+        )
+    except:
+        await ctx.bot.send_text(
+            ctx.room,
+            f"Could not process the command: {traceback.format_exc()}"
+        )
+
+async def invalid_usage(ctx: EventContext) -> None:
+    """This callback is called when the bot used incorrectly."""
+    await ctx.bot.send_text(ctx.room, "Use <code>!help</code>")
+
+
+async def main() -> None:
+    """Application entry point"""
+    logging.basicConfig(level=logging.INFO)
+    logging.getLogger("nio").setLevel(logging.CRITICAL+1)
+    check_environment()
+
+    config = MatrixBotConfig(
+        matrix_homeserver_url=os.environ["MATRIX_HOMESERVER"],
+        matrix_username_localpart=os.environ["MATRIX_USERNAME"],
+        storage_directory="session_storage"
+    )
+    bot = MatrixBot(config)
+
+    COMMANDS = {
+        on_help_command: BodyCommandFilter(["help", "?"]),
+        on_time_command: BodyCommandFilter("time"),
+        on_raise_command: BodyCommandFilter("raise"),
+        on_assert_command: BodyCommandFilter("assert"),
+        on_mul_command: BodyCommandFilter("mul", min_args=2),
+        on_args_command: BodyCommandFilter("args", min_args=1, max_args=5),
+    }
+    for callback, filter in COMMANDS.items():
+        f = ~SenderIsBotFilter() & filter
+        bot.add_callback(f, callback)
+    bot.add_callback(~SenderIsBotFilter() & NewMessageFilter(), invalid_usage)
+
+    # run until Ctrl+C
+    try:
+        await bot.run()
+    except asyncio.CancelledError:
+        pass
+
+if __name__ == "__main__":
+    asyncio.run(main())