nikita
2867582c01
- Updated README.md - Updated examples - `MatrixBot` does not redefine methods from classes it is composed of anymore
130 lines
4.1 KiB
Python
130 lines
4.1 KiB
Python
"""
|
|
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()) |