Updated to v0.5.0

- `RoomEventData` is renamed to `EventContext` and moved to context.py
- `EventContext.filter` is removed
- Added context variables system which improves type hints and simplifies callbacks code
- `BodyCommandFilter` sets context variables from now on
- Added `super().__call__` invocation to filters implements in base.py
- Examples are updated to include required changes

.gitignore

@@ -1,4 +1,5 @@
 __pycache__/
+session_storage/
 *.vscode
 .venv/
 dist/

README.md

@@ -1,64 +1,121 @@
-# mab
+# 🤖 mab
 
 **mab** *(MAtrix Bot)* is a **very** simple Python package that can be used to
-develop **very** simple Matrix bots. I have decided to make something like this
+develop **very** simple Matrix bots. It does not aim to be the best library out
-because I wasn't satisfied by simplicity and usage of other libraries. So
+there, but it aims to be convenient and usable for relatively serious projects.
-this library does not aim to be "the best matrix bot library", it only aims to
-be good enough for me.
 
-## Installation
+## ✨ Features
 
-Use `pip` to install this package:
+The library supports the following features:
+- **Completely `asyncio` based**
+- **Filter-based callback system**
+- **Sending images**
+- **Sending videos with automatic thumbnail generation (requires `ffmpeg`)**
+
+## 📦 Installation
+
+Use `apt` to install required system packages and `pip` to install the package.
+You may need to use `root` privileges to use `apt`. It's highly recommended you
+use `venv` or another Python virtual environment. Here are the commands to
+install the latest version of the library:
 
 ```bash
-python -m pip install git+https://git.tyukalov.su/nikita/mab
+apt install libmagic1-dev libolm-dev
+python -m pip install git+https://git.tyukalov.su/nikita/mab@v0.5.0
 ```
 
-## Basic usage
+`libmagic1-dev` is needed for automatic file MIME type detection, `libolm-dev`
+is needed for E2EE to work.
 
-This is the most simple bot you can create. It would respond to any message
+Please inspect [`examples/image_gen_bot.py`](examples/image_gen_bot.py),
-that starts with `!test`, `!hello` or `!hi`.
+[`examples/echo_bot.py`](examples/echo_bot.py) or open [`examples/`](examples/)
+directory to find usage examples. Examples require that you set
+`MATRIX_HOMESERVER` and `MATRIX_USERNAME` environment variables. Examples create
+`session_storage` directory in working directory.
 
-```python
+## 🚀 Usage
-import asyncio
-from mab import MatrixBot, MatrixBotConfig
-from mab import TextCommandFilter
 
-from nio import MatrixRoom, MatrixMessageText
+If you use `mab`, your application will *most likely* be using **callbacks** to
+react to user actions. `mab` uses filter-based callback system to avoid exposing
+raw `nio-matrix` event objects.
 
-bot: MatrixBot
+This is the workflow you will most likely follow:
+1. **Define the callback as `async` function that take 1 argument of type
+   `EventContext`.** For example, this callback would print the caption of the
+   message:
+   ```python
+   from mab import *
 
-async def on_valid_command(room: MatrixRoom, event: RoomMessageText) -> None:
+   async def on_media_with_body(ctx: EventContext):
-    global bot
+       """To be called when a message with image/video and caption is received."""
-    # do not respond to ourselves
+       print(ctx[CTX_BODY])
-    if event.sender == bot.get_client().user_id:
+   ```
-        return
+2. **Define the conditions your callback must be called on.** For example, you
-    text = f"Your message contains {len(event.body)} symbols"
+   may want your callback to be called when `the sender is not the bot` and
-    await bot.send_text_to_room(room.room_id, text)
+   `the message contains textual body` and (`the message is an image` or
+   `the message is a video`).
+3. **Define the conditions as `filters`.** Most of them are pretty
+   straightforward. For example, if you want to use the conditions from above:
+   ```python
+   from mab import *
 
-async def main() -> None:
+   filters = (
-    global bot
+       ~SenderIsBotFilter()
-    # create and start the bot
+       & MessageTypeFilter([MessageType.IMAGE, MessageType.VIDEO])
-    cfg = MatrixBotConfig(
+       & BodyExistsFilter()
-        matrix_homeserver_url="matrix.domain.su",
-        matrix_username_localpart="nagibator666",
-        storage_directory=Path("storage_nagibator666")
    )
-    bot = MatrixBot(matrix_bot_config)
+   ```
-    bot.add_event_callback(
+4. **Add the callback to your `MatrixBot` instance.** For example, if you would
-        on_valid_command,
+   have used everything from above, then your code would look something like
-        TextCommandFilter(["test", "hello", "hi"])
+   this:
-    )
+   ```python
-    bot.start()
+   from mab import *
-    # wait for Ctrl+C
-    try:
-        while True:
-            await asyncio.sleep(1)
-    except:
-        pass
-    # stop the bot
-    await bot.stop()
 
-if __name__ == "__main__":
+   # let's assume you create your MatrixBot as `bot` variable here
-    asyncio.run(main())
+
+   async def on_media_with_body(ctx: EventContext):
+       """To be called when a message with image/video and caption is received."""
+       print(ctx[CTX_BODY])
+   
+   filters = (
+       ~SenderIsBotFilter()
+       & MessageTypeFilter([MessageType.IMAGE, MessageType.VIDEO])
+       & BodyExistsFilter()
+   )
+   bot.add_callback(filters, on_media_with_body)
+
+   ...
+   ```
+
+Filters support bitwise operators to implement complex matching logic. Some
+filters set context variables which can be accessed by
+`context[CTX_KEY_NAME]`-like syntax. Possible variables are defined in
+[this file](src/mab/context.py). Filters are implemented in
+[files of this directory](src/mab/filters/).
+
+## 🏷️ Versioning
+
+Releases are tagged in this repository using the `vX.Y.Z` format. If the commit
+is not tagged, it must be treated as versionless and should not be used for your
+application.
+- `X` **(Major)**: Breaking architectiral changes or complete rewrites. Existing
+  code will break. Note that `0.Y.Z` versions are considered **very unstable**,
+  the API may change at any time and some features do not work as expected.
+- `Y` **(Minor)**: Breaking API changes, feature removals, or behavioral
+  modifications. Existing code will likely break.
+- `Z` **(Patch)**: Backward-compatible feature additions, bug fixes, or internal
+  changes. Existing code will not break.
+
+## 🛠️ Development
+
+Here's the list of commands you should execute to get started with development
+(including cloning the repository and installing required packages). Please note
+that your workflow may use something other than `venv`.
+```bash
+apt install libmagic1-dev libolm-dev
+git clone https://git.tyukalov.su/nikita/mab
+cd mab
+python3 -m venv .venv
+. .venv/bin/activate
+pip install -e .
 ```

examples/_environment.py

@@ -0,0 +1,18 @@
+import os
+import sys
+
+def check_environment() -> None:
+    """
+    This function checks if required environment variables are set. It prints
+    problem resolution guide and exits using `sys.exit(1)` on problem.
+    """
+    if "MATRIX_HOMESERVER" not in os.environ:
+        print("Please set `MATRIX_HOMESERVER` environment variable!")
+        print("P.S. use something like this in your shell:")
+        print("    export MATRIX_HOMESERVER=\"https://matrix.server.net\"")
+        sys.exit(1)
+    if "MATRIX_USERNAME" not in os.environ:
+        print("Please set `MATRIX_USERNAME` environment variable!")
+        print("P.S. use something like this in your shell:")
+        print("    export MATRIX_USERNAME=\"megakiller228\"")
+        sys.exit(1)

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())

examples/echo_bot.py

@@ -0,0 +1,43 @@
+"""
+This example implements Matrix bot that echoes all text messages it receives.
+
+It uses environment variables to specify authorization data. Use Ctrl+C to stop
+the bot.
+"""
+
+import asyncio
+import os
+import logging
+
+from mab import *
+
+from _environment import check_environment
+
+async def on_text_message(ctx: EventContext) -> None:
+    """This callback is called when a text message arrives."""
+    await ctx.bot.send_text(ctx.room, ctx[CTX_BODY])
+
+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)
+    bot.add_callback(
+        ~SenderIsBotFilter() & BodyExistsFilter() & MessageTypeFilter(MessageType.TEXT),
+        on_text_message)
+
+    # run until Ctrl+C
+    try:
+        await bot.run()
+    except asyncio.CancelledError:
+        pass
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/image_gen_bot.py

@@ -0,0 +1,95 @@
+"""
+This example implements Matrix bot that generates a pixelized noise image with
+specified maximum R, G and B values.
+
+It uses environment variables to specify authorization data. Use Ctrl+C to stop
+the bot.
+"""
+
+import asyncio
+from io import BytesIO
+import os
+import logging
+import random
+from PIL import Image
+
+from mab import *
+
+from _environment import check_environment
+
+async def on_gen_command(data: EventContext) -> None:
+    """This callback is called when `!gen R G B` command is received."""
+    # convert R, G and B to floats
+    try:
+        r, g, b = [float(v) for v in data[CTX_CMD_ARGS]]
+    except:
+        await data.bot.send_text(data.room, "Invalid arguments")
+        return
+    await data.bot.send_text(data.room, "Generating the noise...")
+    # create the basic noise
+    img = Image.new("RGB", (16, 16))
+    for x in range(img.width):
+        for y in range(img.height):
+            col = (random.random() * r, random.random() * g, random.random() * b)
+            img.putpixel(
+                (x, y),
+                tuple(int(c * 255) for c in col)
+            )
+    # pixelized upscale
+    img = img.resize((2048, 2048), resample=Image.Resampling.NEAREST)
+    # save to buffer
+    buf = BytesIO()
+    img.save(buf, format="PNG")
+    buf.seek(0)
+    buf = buf.read()
+    # send
+    await data.bot.send_image_bytes(data.room, buf, "noise.png")
+
+async def on_wrong_message(data: EventContext) -> None:
+    """This callback is called when a wrong message is received."""
+    await data.bot.send_text(
+        data.room,
+        "Text me something like <code>!gen 0.1 0.7 1.0</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)
+    command_filter = MessageTypeFilter(MessageType.TEXT) & BodyCommandFilter(
+        verbs=["gen"],
+        min_args=3,
+        max_args=3
+    )
+
+    # callback for message that
+    # 1. are sent not by this bot
+    # 2. do match the command filter
+    bot.add_callback(
+        ~SenderIsBotFilter() & command_filter,
+        on_gen_command)
+    
+    # callback for message that
+    # 1. are sent not by this bot
+    # 2. do NOT match the command filter
+    bot.add_callback(
+        ~SenderIsBotFilter() & ~command_filter,
+        on_wrong_message
+    )
+
+    # run until Ctrl+C
+    try:
+        await bot.run()
+    except asyncio.CancelledError:
+        pass
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "mab"
-version = "0.1.0"
+version = "0.5.0"
 authors = [
     { name = "Tyukalov Nikita", email = "nikita@tyukalov.su" }
 ]

src/mab/__init__.py

@@ -1,12 +1,14 @@
 from . import bot
 from . import types
 
-from .types import MatrixBotConfig
+from .types import MatrixBotConfig, MessageType
+from .context import *
 
 from .bot import MatrixBot
 
 from .filters.base import *
-from .filters.text import *
+from .filters.message import *
+from .filters.body import *
 
 __all__ = [
     # module names
@@ -15,18 +17,37 @@ __all__ = [
 
     # .types
     "MatrixBotConfig",
+    "MessageType",
+
+    # .context
+    "EventContext",
+    "CTX_BODY",
+    "CTX_MESSAGE_TYPE",
+    "CTX_SENDER",
+    "CTX_CMD_PREFIX",
+    "CTX_CMD_VERB",
+    "CTX_CMD_ARGS",
 
     # .bot
     "MatrixBot",
 
     # .filters.base
     "BaseEventFilter",
+    "EventTypeFilter",
 
-    # .filters.text
+    # .filters.body
-    "TextFilter",
+    "BodyExistsFilter",
-    "FormattedTextFilter",
+    "BodyContainsFilter",
-    "TextContainsFilter",
+    "BodyStartsWithFilter",
-    "TextStartsWithFilter",
+    "BodyEndsWithFilter",
-    "TextEndsWithFilter",
+    "BodyCommandFilter",
-    "TextCommandFilter",
+    "BodyRegexFilter",
+
+    # .filters.message
+    "MessageTypeFilter",
+    "NewMessageFilter",
+    "EditedMessageFilter",
+    "RedactedMessageFilter",
+    "SenderIsFilter",
+    "SenderIsBotFilter",
 ]

src/mab/bot.py

@@ -1,827 +0,0 @@
-import asyncio
-import aiofiles, aiofiles.os
-import aioconsole
-import traceback
-import logging
-import time
-import json
-import os
-import re
-import io
-import magic
-from html.parser import HTMLParser
-from urllib.parse import urlparse
-from typing import Any, Callable, Awaitable
-
-from PIL import Image
-
-from nio import AsyncClient, AsyncClientConfig, SyncResponse
-from nio import LoginResponse, LoginError, WhoamiResponse, WhoamiError
-
-from nio import RoomSendResponse, RoomSendError
-from nio import UploadResponse, UploadError
-
-from nio import OlmUnverifiedDeviceError
-
-from nio import MatrixInvitedRoom, InviteMemberEvent
-from nio import JoinResponse
-
-from .filters.base import BaseEventFilter
-
-from nio.events.room_events import Event as RoomEvemt
-
-from .types import *
-from .utils import Utils
-
-
-class MatrixBot:
-    """Asynchronous Matrix Bot Implementation.
-
-        Use objects of this class to build your bots. Manage the event loop
-        by yourself.
-    """
-    NEXT_BATCH_DUMP_PERIOD = 120.0
-    MATRIX_SYNC_PERIOD = 5000
-
-    #
-    # PRIVATE
-    #
-    @staticmethod
-    def _validate_matrix_homeserver_url(url: str) -> None:
-        """Checks if `url` is a valid matrix homeserver URL.
-            Raises an Exception if it is not.
-        """
-        parsed = urlparse(url)
-        if parsed.scheme not in ("http", "https"):
-            raise RuntimeError(
-                f"Scheme {parsed.scheme} is not a valid scheme for matrix homeserver URL"
-            )
-        if not parsed.netloc:
-            raise RuntimeError(f"{url} is not a valid matrix homeserver URL")
-        if parsed.path != "":
-            raise RuntimeError(f"{url} must have empty path (remove `{parsed.path}` after the hostname)")
-
-    @staticmethod
-    def _validate_matrix_username_localpart(localpart: str) -> None:
-        """Checks if `username` is a valid localpart of matrix username.
-            Raises an Exception if it is not.
-        """
-        pattern = r"^[a-z0-9._=\-]+$"
-        if not bool(re.fullmatch(pattern, localpart)) or len(localpart) > 255:
-            raise RuntimeError(f"{localpart} is not a valid matrix username localpart")
-
-    @staticmethod
-    def _validate_storage_directory(path: Path) -> None:
-        """Checks if `path` is a valid storage directory and creates it.
-            Raises an Exception if it is not.
-        """
-        path.mkdir(parents=True, exist_ok=True)
-        if not path.is_dir():
-            raise RuntimeError(f"Could not create directory {path}")
-
-    @staticmethod
-    def _validate_bot_config(config: MatrixBotConfig) -> None:
-        """Checks if `config` has errors.
-            Raises an Exception if it does.
-        """
-        MatrixBot._validate_matrix_homeserver_url(config.matrix_homeserver_url)
-        MatrixBot._validate_matrix_username_localpart(config.matrix_username_localpart)
-        MatrixBot._validate_storage_directory(config.storage_directory)
-
-    @staticmethod
-    def _build_client(config: MatrixBotConfig) -> AsyncClient:
-        """Builds `nio.AsyncClient` from `MatrixBotConfig`"""
-        # create the config for the client
-        client_config = AsyncClientConfig(
-            store_name="nio_store_file",
-            encryption_enabled=True,
-            store_sync_tokens=False
-        )
-        # create the client
-        client = AsyncClient(
-            homeserver=config.matrix_homeserver_url,
-            user=config.matrix_username_localpart,
-            store_path=str(config.storage_directory),
-            config=client_config
-        )
-        return client
-
-    @staticmethod
-    def _build_matrix_username(config: MatrixBotConfig) -> str:
-        """Builds full matrix username."""
-        homeserver_name = urlparse(config.matrix_homeserver_url).hostname
-        localpart = config.matrix_username_localpart
-        return f"@{localpart}:{homeserver_name}"
-
-    @staticmethod
-    def _process_html_text(text: str, is_html: bool | None) -> dict:
-        """Processes `text` and `is_html` arguments of `send_*` functions and
-           prepares `content` keys related to text data inside the message.
-        
-            Args:
-                text - `text` as passed to the `send_*` function
-                is_html - `is_html` as passed to the `send_*` function
-            
-            Returns:
-                `dict` that contains keys required for the text message to be
-                displayed correctly.
-        """
-        # text w/ and w/o HTML
-        text_w_html: str | None = None
-        text_wo_html: str | None = None
-        # check if HTML is present
-        if is_html or is_html is None:
-            # will be True if `text` contains HTML tags
-            has_tags = False
-            # text fragments without tags
-            text_fragments = []
-            # temporary extractor class
-            class Extractor(HTMLParser):
-                def handle_starttag(self, tag, attrs):
-                    nonlocal has_tags
-                    has_tags = True
-                def handle_data(self, data):
-                    text_fragments.append(data)
-            # parse
-            parser = Extractor(convert_charrefs=True)
-            try:
-                parser.feed(text)
-            except:
-                traceback.print_exc()
-                text_wo_html = text
-            is_html = has_tags
-            if is_html:
-                text_wo_html = " ".join(text_fragments)
-                text_w_html = text
-        if not is_html:
-            text_wo_html = text
-        # return the content
-        if text_w_html:
-            return {
-                "body": text_wo_html,
-                "format": "org.matrix.custom.html",
-                "formatted_body": text_w_html
-            }
-        else:
-            return {
-                "body": text_wo_html
-            }
-    
-    @staticmethod
-    async def _wait_for_task_and_stop_event(payload_task: asyncio.Task, stop_wait_task: asyncio.Task, print_exc: bool = False) -> Any:
-        """Waits for `payload_task` to finish. Uses `stop_wait_task` to check if payload task
-            should be cancelled. Returns value returned by `payload_task` task. Raises exception
-            raised by `payload_task` task.
-
-            If `stop_wait_task` finishes, then `asyncio.CancelledError` is raised.
-        """
-        try:
-            done, _ = await asyncio.wait(
-                [payload_task, stop_wait_task],
-                return_when=asyncio.FIRST_COMPLETED
-            )
-            if stop_wait_task in done:
-                payload_task.cancel()
-                await payload_task
-                raise asyncio.CancelledError()
-            return payload_task.result()
-        except asyncio.CancelledError:
-            raise
-        except:
-            if print_exc:
-                traceback.print_exc()
-            raise
-
-    @staticmethod
-    async def _default_password_callback() -> str:
-        """Gets password from `MATRIX_PASSWORD` envvar if it is set. Asks
-            the user for the password otherwise."""
-        if "MATRIX_PASSWORD" in os.environ:
-            return os.environ["MATRIX_PASSWORD"]
-        print("--- A password is required (btw you might use MATRIX_PASSWORD envvar) ---")
-        return await aioconsole.ainput("Password: ")
-
-    @staticmethod
-    async def _debug_event_callback(*args, **kwargs) -> None:
-        """Just prints types of arguments"""
-        try:
-            print(f"_debug_event_callback ({len(args)} args, {len(kwargs)} kwargs)")
-            for a in args:
-                print(f"  - {type(a)}")
-            for k in kwargs:
-                print(f"  * {k} = {kwargs[k]}")
-        except:
-            traceback.print_exc()
-
-
-
-    async def _write_next_batch(self, next_batch: str) -> None:
-        """Writes `next_batch` value to disk."""
-        path = self._config.storage_directory / "next_batch"
-        async with aiofiles.open(path, "w") as f:
-            await f.write(next_batch)
-        self._logger.debug("next_batch value is written to the disk")
-
-    async def _read_next_batch(self) -> str | None:
-        """Reads `next_batch` value from disk. Returns None if file does not exist."""
-        path = self._config.storage_directory / "next_batch"
-        if not path.is_file():
-            return None
-        async with aiofiles.open(path, "r") as f:
-            return (await f.read()).strip()
-
-    async def _write_session_data(self, *, access_token: str, device_id: str) -> None:
-        """Write session data to disk."""
-        path = self._config.storage_directory / "session_data.json"
-        data = {
-            "access_token": access_token,
-            "device_id": device_id
-        }
-        async with aiofiles.open(path, "w") as f:
-            await f.write(json.dumps(data, indent=4))
-        self._logger.debug("Session data is writter to the disk")
-
-    async def _read_session_data(self) -> dict[str, Any] | None:
-        """Read session data from disk."""
-        path = self._config.storage_directory / "session_data.json"
-        if not path.is_file():
-            return None
-        async with aiofiles.open(path, "r") as f:
-            j = json.loads(await f.read())
-        return j
-
-    async def _upload_file(self, path: Path | str, mime_type: str, *, filename: str | None = None) -> tuple[UploadResponse, dict, int]:
-        """Uploads the file with encryption enabled. UploadError will be raised
-           as RuntimeError.
-        
-            Args:
-                path - path to the file to upload
-                mime_type - mime-type to use
-                filename - filename to use (base name of the `path` will be used if not set)
-            
-            Returns:
-                1. UploadResponse as returned by `_upload_binary_data(...)`
-                2. dict as returned by `_upload_binary_data(...)`
-                3. file size in bytes
-        """
-        # get basic file info
-        file_info = await aiofiles.os.stat(path)
-        if not filename:
-            filename = os.path.basename(path)
-        # upload
-        async with aiofiles.open(path, "r+b") as f:
-            response, keys = await self._upload_provider(
-                f,
-                mime_type,
-                filename=filename,
-                filesize=file_info.st_size
-            )
-            return response, keys, file_info.st_size
-
-    async def _upload_provider(self, provider: Any, mime_type: str, *, filename: str, filesize: int) -> tuple[UploadResponse, dict]:
-        """Uploads the file with encryption enabled. UploadError will be raised
-           as RuntimeError.
-        
-            Args:
-                provider - data provider as described in `AsyncClient.upload(...)` docs
-                mime_type - mime-type to use
-                filename - filename to use
-            
-            Returns:
-                1. UploadResponse as returned by `AsyncClient.upload(...)`
-                2. dict as returned by `AsyncClient.upload(...)`
-        """
-        # upload
-        response, keys = await self._client.upload(
-            provider,
-            content_type=mime_type,
-            filename=filename,
-            encrypt=True,
-            monitor=None,
-            filesize=filesize
-        )
-        if isinstance(response, UploadError):
-            raise RuntimeError(response.message, response)
-        if not isinstance(response, UploadResponse):
-            raise RuntimeError(
-                "Upload result must be `UploadRespond`, but it's not",
-                response
-            )
-        return response, keys # type: ignore
-
-    async def _upload_thumbnail(self, thumbnail: Path | str | bytes | None) -> dict:
-        """Upload thumbnail and return fields required to use it for video.
-        
-            Args:
-                thumbnail - path to the thumbnail file or raw `bytes` of the image.
-                            You can actually pass `None`, empty dict will be returned
-                            in this case.  
-            
-            Returns:
-                `dict` you should embed into your `content["info"]`
-        """
-        if thumbnail is None:
-            return {}
-        # raw data
-        raw_bytes_io: io.BytesIO | None = None
-        if type(thumbnail) is bytes:
-            thumb_mime = magic.from_buffer(thumbnail, mime=True)
-            raw_bytes_io = io.BytesIO(thumbnail)
-        # file path
-        else:
-            thumb_mime = magic.from_file(thumbnail, mime=True)
-        # check if invalid MIME
-        if not thumb_mime.startswith("image/"):
-            raise RuntimeError(f"Thumbnail MIME is not image: `{thumb_mime}`")
-        # get thumbnail size
-        with Image.open(raw_bytes_io or thumbnail) as f:
-            width, height = f.size
-        # seek BytesIO to zero so we can reuse it
-        if raw_bytes_io is not None:
-            raw_bytes_io.seek(0)
-        # upload
-        if type(thumbnail) is bytes:
-            thumb_size = len(thumbnail)
-            thumb_resp, thumb_keys = await self._upload_provider(
-                raw_bytes_io,
-                thumb_mime,
-                filename=f"preview.{thumb_mime.split("/")[-1]}",
-                filesize=thumb_size
-            )
-        else:
-            thumb_resp, thumb_keys, thumb_size = await self._upload_file(
-                thumbnail, # type: ignore
-                mime_type=thumb_mime,
-                filename=os.path.basename(str(thumbnail))
-            )
-        # return the result
-        return {
-            "thumbnail_file": {
-                "url": thumb_resp.content_uri,
-                "mimetype": thumb_mime,
-                **thumb_keys
-            },
-            "thumbnail_info": {
-                "w": width,
-                "h": height,
-                "size": thumb_size,
-                "mimetype": thumb_mime
-            }
-        }
-
-    async def _send_content_to_room(self, room_id: str, content: dict) -> RoomSendResponse:
-        """Sends `m.room.message` event to the room. Returns RoomSendResponse
-           as returned by `AsyncClient.room_send(...)`. Automatically verifies
-           all devices if this feature is enabled. Return values other than
-           `RoomSendResponse` are raised as `RuntimeError`."""
-        # try to send the message
-        try:
-            result = await self._client.room_send(
-                room_id=room_id,
-                message_type="m.room.message",
-                content=content
-            )
-        except OlmUnverifiedDeviceError:
-            if self._config.auto_verify_all_known_devices:
-                if not self.verify_all_known_devices():
-                    raise
-                return await self._send_content_to_room(room_id, content)
-            else:
-                raise
-        # success
-        if type(result) is RoomSendResponse:
-            return result
-        # error
-        elif type(result) is RoomSendError:
-            raise RuntimeError(result)
-        # unknown error
-        else:
-            raise RuntimeError("Unknown error has occured", result)
-
-    #
-    # CALLBACKS
-    #
-    async def _callback_sync(self, response: SyncResponse) -> None:
-        """This callback is called when AsyncClient syncs with the server"""
-        current_time = time.time()
-        delta_time = current_time - self._last_next_batch_dump
-        self._last_next_batch = response.next_batch
-        if delta_time >= self.NEXT_BATCH_DUMP_PERIOD:
-            self._last_next_batch_dump = current_time
-            try:
-                await self._write_next_batch(self._last_next_batch)
-            except:
-                traceback.print_exc()
-
-    async def _callback_autojoin(self, room: MatrixInvitedRoom, event: InviteMemberEvent):
-        try:
-            # event.state_key must be our username
-            if event.state_key != self._client.user_id:
-                return
-            # membership status must be invite
-            if event.membership != "invite":
-                return
-            result = await self._client.join(room.room_id)
-            if type(result) is JoinResponse:
-                self._logger.info(f"Autojoined the room {room.room_id}")
-            else:
-                self._logger.error(f"Can't autojoin the room {room.room_id}")
-        except:
-            self._logger.error(traceback.format_exc())
-
-    async def _callback_filter_router(self, *args, **kwargs):
-        if len(args) != 2:
-            self._logger.debug("Can't process the event, not enough positional args")
-            await self._debug_event_callback(*args, **kwargs)
-            return
-        room = args[0]
-        event = args[1]
-        for filter in self._filters:
-            filter_object = filter[0]
-            filter_callback = filter[1]
-            filter_stop_after_this = filter[2]
-            if filter_object(room, event):
-                self._logger.debug(f"Filter {repr(filter_object)} matched")
-                try:
-                    await filter_callback(room, event)
-                except:
-                    self._logger.error(traceback.format_exc())
-                if filter_stop_after_this:
-                    self._logger.debug(f"Filter {repr(filter_object)} stops matching")
-                    break
-
-    #
-    # LIFECYCLE
-    #
-    def _setup_client_callbacks(self) -> None:
-        """Setup internal client callbacks"""
-        self._client.add_response_callback(self._callback_sync, SyncResponse) # type: ignore
-        self._client.add_event_callback(self._callback_filter_router, RoomEvemt) # type: ignore
-        if self._config.auto_join_any_room_on_invite:
-            self._client.add_event_callback(self._callback_autojoin, InviteMemberEvent) # type: ignore
-
-    async def _client_login_session_data(self, session_data: dict[str, Any]) -> None:
-        """Login using session data. Raises and exception on failure."""
-        self._logger.debug("Using stored session data to log in")
-        # build user id
-        username = self._build_matrix_username(self._config)
-        self._client.restore_login(
-            user_id=username,
-            **session_data
-        )
-        self._client.load_store()
-        result = await self._client.whoami()
-        if type(result) is WhoamiError:
-            self._logger.error(f"Can't log in using stored session data: '{result.message}'")
-            raise RuntimeError(result.message)
-        elif type(result) is not WhoamiResponse:
-            self._logger.error("Can't log in using stored session data, unknown error")
-            raise RuntimeError("Unknown response for whoami request")
-        self._logger.debug("Logged in using stored session data")
-
-    async def _client_login_password(self) -> None:
-        """Login using password and save result to disk on success.
-            Raises an exception on failure.
-        """
-        self._logger.debug("Using password to log in")
-        # get the password
-        password = await self._cb_password()
-        result = await self._client.login(password=password)
-        if type(result) is LoginResponse:
-            self._logger.debug("Logged in using password")
-            await self._write_session_data(
-                access_token=result.access_token,
-                device_id=result.device_id
-            )
-        elif type(result) is LoginError:
-            self._logger.error(f"Can't log in using password: '{result.message}'")
-            raise RuntimeError(result.message)
-        else:
-            self._logger.error(f"Can't log in using password, unknown error")
-            raise RuntimeError("Unknown login result")
-
-    async def _client_login(self) -> None:
-        """This function logs in."""
-        # check if we have session data stored on the disk
-        session_data = await self._read_session_data()
-        # session data is present, try to log in
-        if session_data is not None:
-            self._logger.debug("Some session data found on the disk")
-            try:
-                await self._client_login_session_data(session_data)
-                return
-            except asyncio.CancelledError:
-                raise
-            except:
-                pass
-        # no session data - login using password
-        try:
-            self._logger.debug("No session data found on the disk OR invalid data")
-            await self._client_login_password()
-            return
-        except asyncio.CancelledError:
-            raise
-        except:
-            pass
-        # can't login
-        self._logger.error("Can't log in using available methods")
-        raise RuntimeError("All login methods have failed, can't continue")
-
-    async def _client_destroy(self) -> None:
-        """Gracefully destroys the client."""
-        try:
-            self._logger.debug("Closing the client")
-            await self._client.close()
-            if self._last_next_batch is not None:
-                self._logger.debug("Saving next_batch")
-                await self._write_next_batch(self._last_next_batch)
-        except:
-            traceback.print_exc()
-
-    async def _client_cancellable_sync_forever(self, *args, **kwargs) -> Any:
-        """Behaves exactly like AsyncClient.sync_forever, but supports task cancellation"""
-        sync_forever_task = asyncio.create_task(
-            self._client.sync_forever(*args, **kwargs)
-        )
-        try:
-            return await sync_forever_task
-        except asyncio.CancelledError:
-            try:
-                self._client.stop_sync_forever()
-            except Exception:
-                pass
-            finally:
-                sync_forever_task.cancel()
-                await asyncio.gather(sync_forever_task, return_exceptions=True)
-            raise
-
-    async def _background_coroutine(self) -> None:
-        """This function implements bot lifecycle."""
-        # we should stop when this task stops
-        self._logger.debug("_background_coroutine is started")
-        stop_wait_task = asyncio.create_task(self._stop_event.wait()) # type: ignore
-        # perform login
-        login_task = asyncio.create_task(self._client_login())
-        try:
-            await self._wait_for_task_and_stop_event(login_task, stop_wait_task)
-        except asyncio.CancelledError:
-            self._logger.debug("Background task is cancelled during login")
-            await self._client_destroy()
-            return
-        except:
-            traceback.print_exc()
-        self._logger.info("Succesfully logged in")
-        # sync forever
-        self._logger.info("Syncing forever")
-        sync_task = asyncio.create_task(
-            self._client_cancellable_sync_forever(
-                timeout=self.MATRIX_SYNC_PERIOD,
-                since=(await self._read_next_batch()),
-                full_state=True
-            )
-        )
-        try:
-            await self._wait_for_task_and_stop_event(sync_task, stop_wait_task)
-        except asyncio.CancelledError:
-            self._logger.debug("Sync task is cancelled")
-            await self._client_destroy()
-            return
-        except:
-            traceback.print_exc()
-
-
-    #
-    # PUBLIC
-    #
-    def __init__(self, config: MatrixBotConfig) -> None:
-        # check if config is valid
-        if type(config.storage_directory) is str:
-            config.storage_directory = Path(config.storage_directory)
-        self._validate_bot_config(config) # may raise an Exception
-        # save the config
-        self._config: MatrixBotConfig = config
-        # create the logger
-        self._logger = logging.getLogger(self._build_matrix_username(config))
-        self._logger.setLevel(logging.DEBUG)
-        # create the client
-        self._client: AsyncClient = self._build_client(self._config)
-        self._setup_client_callbacks()
-
-        # prepare some private data
-        self._background_task: asyncio.Task | None = None
-        self._last_next_batch_dump: float = 0.0
-        self._last_next_batch: str | None = None
-        self._cb_password = self._default_password_callback
-        self._filters = []
-
-    def start(self) -> None:
-        """Start the bot.
-            Starts the bot in background task. Raises an exception if there are
-            problems (for example, the bot is already started). The bot will
-            do everything to keep itself running, including restarts. Use
-            `stop()` to stop the bot.
-        """
-        if self._background_task is not None:
-            raise RuntimeError("The bot is already started!")
-        self._stop_event = asyncio.Event()
-        self._background_task = asyncio.create_task(
-            self._background_coroutine()
-        )
-
-    async def stop(self) -> None:
-        """Stop the bot and wait for the bot stop."""
-        if self._background_task is None or self._stop_event is None:
-            return
-        self._stop_event.set()
-        try:
-            await self._background_task
-        except:
-            traceback.print_exc()
-        self._stop_event = None
-        self._background_task = None
-
-    def verify_all_known_devices(self) -> bool:
-        """Verifies all known devices.
-        
-        Returns:
-            True if there were unverified devices that are verified now.
-        """
-        result = False
-        for user_id in self._client.device_store.users:
-            for device_id, olm_device in self._client.device_store[user_id].items():
-                # can't trust ourselves
-                if device_id == self._client.device_id and user_id == self._client.user_id:
-                    continue
-                # they are already verified
-                if olm_device.verified:
-                    continue
-                # verify them
-                self._client.verify_device(olm_device)
-                result = True
-        return result
-
-    def add_event_callback(self, callback: Callable[..., Awaitable[None]] | None, filter: BaseEventFilter, stop_after_this: bool = True) -> None:
-        """Add event callback for events that pass the filter."""
-        if callback is None:
-            callback = self._debug_event_callback
-        self._filters.append((filter, callback, stop_after_this))
-
-    def get_client(self) -> AsyncClient:
-        """Get AsyncClient in use"""
-        return self._client
-
-    async def send_text_to_room(self, room_id: str, text: str, is_html: bool | None = None) -> str:
-        """Sends a text message to the room and handle HTML as specified.
-        
-        Args:
-            text - text to send
-            is_html - True if text is HTML; False if text is not HTML; None if the value should be guessed
-            kwargs - passed as `m.room.message` content keys
-        
-        Returns:
-            event_id of the message on success. Raises an exception on error.
-        """
-        text_data = self._process_html_text(text, is_html)
-        content = {
-            "msgtype": "m.text",
-            **text_data
-        }
-        return (await self._send_content_to_room(room_id, content)).event_id
-
-    async def send_image_to_room(self,
-                                room_id: str,
-                                path: Path | str,
-                                caption: str | None = None,
-                                is_html: bool | None = None) -> str:
-        """Send an image to the room.
-        
-            This method sends specified image file to the specified room. It
-            will use mime-type deduced using `magic` module. Image dimension is
-            deduced using Pillow. The file is read asynchronously, so this
-            method won't block your event loop.
-
-            Please note that HTML is incorrectly interpreted for images in
-            some clients (as of September 6th, 2026).
-
-            Args:
-                room_id - room_id to send the image to
-                path - path to the image file
-                caption - caption to use (will be hidden if it is the same as file basename)
-                is_html - whether the caption is HTML (autodetected if None)
-            
-            Returns:
-                event_id
-        """
-        # caption must not actually be empty
-        if caption is None or not caption.strip():
-            caption = os.path.basename(path)
-            is_html = False
-        # check if the file is image
-        mime_type: str = magic.from_file(path, mime=True)
-        if not mime_type.startswith("image/"):
-            raise RuntimeError(f"File {path} has non-image mime-type")
-        # get image size
-        with Image.open(path) as image:
-            width, height = image.size
-        # upload
-        response, keys, filesize = await self._upload_file(path, mime_type)
-        # prepare the content
-        content = {
-            "msgtype": "m.image",
-            "filename": os.path.basename(path),
-            **self._process_html_text(caption, is_html),
-            "file": {
-                "url": response.content_uri,
-                "mimetype": mime_type,
-                **keys
-            },
-            "info": {
-                "mimetype": mime_type,
-                "size": filesize,
-                "w": width,
-                "h": height
-            }
-        }
-        # send
-        return (await self._send_content_to_room(room_id, content)).event_id
-
-    async def send_video_to_room(self,
-                                room_id: str,
-                                path: Path | str,
-                                props: VideoFileProperties | None = None,
-                                caption: str | None = None,
-                                is_html: bool | None = None) -> str:
-        """Send a video to the room.
-        
-            This method sends specified video file to the specified room. It
-            will use mime-type deduced using `magic` module. Most video
-            properties must be passed using `props` argument.
-            
-            If you pass `None` as `props` then system `ffprobe` will be called
-            to get video dimensions and duration. You can disable this behavior
-            by disabling `auto_ffprobe` in `MatrixBotConfig`.
-
-            If you pass `None` as `props.thumbnail_path` then thumbnail will be
-            generated automatically. You can disable this behavior by disabling
-            `auto_ffmpeg` in `MatrixBotConfig`.
-
-            Please note that HTML is incorrectly interpreted for videos in
-            some clients (as of September 6th, 2026).
-
-            Args:
-                room_id - room_id to send the video to
-                path - path to the video file
-                props - video properties
-                caption - caption to use (will be hidden if it is the same as file basename)
-                is_html - whether the caption is HTML (autodetected if None)
-            
-            Returns:
-                event_id
-        """
-        # caption must not actually be empty
-        if caption is None or not caption.strip():
-            caption = os.path.basename(path)
-            is_html = False
-        # check if the file is video
-        mime_type: str = magic.from_file(path, mime=True)
-        if not mime_type.startswith("video/"):
-            raise RuntimeError(f"File {path} has non-video mime-type")
-        # properties must not be empty
-        if props is None:
-            if not self._config.auto_ffprobe:
-                raise RuntimeError("No `props` are supplied and `auto_ffprobe` is disabled")
-            props = await Utils.get_video_props_using_ffprobe(path)
-        # thumbnail data to use
-        thumbnail_data = {}
-        # autogenerate thumbnail if allowed
-        if not props.thumbnail and self._config.auto_ffmpeg:
-            props.thumbnail = await Utils.get_video_thumbnail(
-                path,
-                timestamp=props.duration_ms / 1000 / 2,
-                max_size=400
-            )
-        # upload the thumbnail if specified
-        thumbnail_data = await self._upload_thumbnail(props.thumbnail)
-        # upload
-        response, keys, filesize = await self._upload_file(path, mime_type)
-        # prepare the content
-        content = {
-            "msgtype": "m.video",
-            "filename": os.path.basename(path),
-            **self._process_html_text(caption, is_html),
-            "file": {
-                "url": response.content_uri,
-                "mimetype": mime_type,
-                **keys
-            },
-            "info": {
-                "duration": props.duration_ms,
-                "mimetype": mime_type,
-                "size": filesize,
-                "w": props.width_px,
-                "h": props.height_px,
-                **thumbnail_data
-            }
-        }
-        # send
-        return (await self._send_content_to_room(room_id, content)).event_id

src/mab/bot/__init__.py

@@ -0,0 +1,5 @@
+from .bot import *
+
+__all__ = [
+    "MatrixBot"
+]

src/mab/bot/_callbacks.py

@@ -0,0 +1,177 @@
+import asyncio
+import logging
+import traceback
+from typing import Callable, Any, Coroutine, TYPE_CHECKING
+from dataclasses import dataclass
+
+from nio import AsyncClient
+from nio import MatrixRoom
+from nio import MatrixInvitedRoom, InviteMemberEvent, JoinResponse
+from nio.events.room_events import Event as RoomEvent
+
+from ._storage import Storage
+from ..types import MatrixBotConfig
+from ..context import EventContext
+from ..filters.base import BaseEventFilter
+
+if TYPE_CHECKING:
+    from .bot import MatrixBot
+
+class Callbacks:
+    """
+    This class implements user callbacks routing and some basic internal
+    callbacks. You should use it to add callbacks to the bot.
+    """
+    #
+    # DATACLASSES
+    #
+    @dataclass
+    class _FilterBasedCallback:
+        """Data structure that holds information about filter-based callback"""
+
+        filter: BaseEventFilter
+        """Filter to use for matching"""
+
+        callback: Callable[[EventContext], Coroutine[Any, Any, None]] | None
+        """Callback that will be called if the filter matches"""
+
+        stop_matching: bool
+        """Whether to stop matching if the filter matches"""
+
+    #
+    # PRIVATE
+    #
+    async def _callback_room_event_router(self,
+                                          room: MatrixRoom,
+                                          event: RoomEvent) -> None:
+        """
+        This callback implements routing of filter-based event callbacks.
+        """
+        if self._config is None or self._client is None:
+            raise RuntimeError("Callbacks were not set up")
+        for callback_info in self._filters:
+            if not isinstance(callback_info, self._FilterBasedCallback):
+                continue
+            event_data = EventContext(
+                room=room,
+                event=event,
+                bot=self._matrix_bot
+            )
+            try:
+                if not await callback_info.filter(event_data):
+                    continue
+            except asyncio.CancelledError:
+                raise
+            except:
+                self._logger.error(traceback.format_exc())
+                continue
+            try:
+                # dump argument types
+                if callback_info.callback is None:
+                    self._logger.critical(
+                        f"Filter {repr(callback_info.filter)} has no callback. "
+                        f"`room` type is `{type(event_data.room)}`. "
+                        f"`event` type is `{type(event_data.event)}`."
+                    )
+                    continue
+                # create the coroutine
+                cor = callback_info.callback(event_data)
+                # multitasking
+                if self._config.enable_multitasking_callbacks:
+                    t = asyncio.create_task(cor)
+                    t.add_done_callback(self._multitasking_set.discard)
+                    self._multitasking_set.add(t)
+                # no multitasking
+                else:
+                    await cor
+            except asyncio.CancelledError:
+                raise
+            except:
+                self._logger.error(traceback.format_exc())
+            if callback_info.stop_matching:
+                break
+
+    async def _callback_autojoin(self,
+                                 room: MatrixInvitedRoom,
+                                 event: InviteMemberEvent) -> None:
+        """
+        This callback implements automatic room joining.
+        """
+        if self._client is None:
+            raise RuntimeError("Callbacks were not set up")
+        try:
+            # event.state_key must be our username
+            if event.state_key != self._client.user_id:
+                return
+            # membership status must be invite
+            if event.membership != "invite":
+                return
+            result = await self._client.join(room.room_id)
+            if type(result) is JoinResponse:
+                self._logger.info(f"Autojoined the room {room.room_id}")
+            else:
+                self._logger.error(f"Can't autojoin the room {room.room_id}")
+        except:
+            self._logger.error(traceback.format_exc())
+
+    async def _setup_internal_callbacks(self) -> None:
+        """
+        Setup internal callbacks.
+        """
+        if self._config is None or self._client is None:
+            raise RuntimeError("Callbacks were not set up")
+        # filter-based callback routing
+        self._client.add_event_callback(
+            self._callback_room_event_router, RoomEvent)
+        # auto-join
+        if self._config.auto_join_any_room_on_invite:
+            self._client.add_event_callback(
+                self._callback_autojoin, InviteMemberEvent) # type: ignore
+
+    #
+    # PUBLIC
+    #
+    def __init__(self, storage: Storage, matrix_bot: "MatrixBot"):
+        self._logger = logging.getLogger("Callbacks")
+        self._storage = storage
+        self._matrix_bot = matrix_bot
+
+        self._config: MatrixBotConfig | None = None
+        self._client: AsyncClient | None = None
+        self._filters: list = []
+        self._multitasking_set: set[asyncio.Task] = set()
+
+    def add_room_event_callback(
+            self,
+            filter: BaseEventFilter,
+            callback: Callable[[EventContext], Coroutine[Any, Any, None]] | None,
+            *,
+            stop_matching: bool = True) -> None:
+        """
+        Add filter-based callback.
+
+        Args:
+            - filter - filter to use for the callback
+            - callback - the callback that must be called if the filter matches
+            - stop_matching - whether to stop matching if the filter matches
+        """
+        callback_info = self._FilterBasedCallback(
+            filter=filter,
+            callback=callback,
+            stop_matching=stop_matching
+        )
+        self._filters.append(callback_info)
+
+
+    async def setup(self, config: MatrixBotConfig, client: AsyncClient) -> None:
+        """
+        Setup the callbacks. Must be called by `ClientManager`.
+
+        This method must be called AFTER required callbacks were added by using
+        `add_room_event_callback(...)`.
+        """
+        self._config = config
+        self._client = client
+        await self._setup_internal_callbacks()
+
+    

src/mab/bot/_client_auth.py

@@ -0,0 +1,142 @@
+import asyncio
+import aioconsole
+import logging
+import os
+from ._storage import Storage
+
+from ..utils import Utils
+from ..types import MatrixBotConfig
+
+from nio import AsyncClient
+from nio import WhoamiError, WhoamiResponse
+from nio import LoginError, LoginResponse
+
+class ClientAuth:
+    """This class is used by to authenticate AsyncClient."""
+
+    #
+    # PRIVATE
+    #
+    async def _default_password_callback(self) -> str:
+        if self._config is None:
+            raise RuntimeError("No config")
+        if "MATRIX_PASSWORD" in os.environ:
+            return os.environ["MATRIX_PASSWORD"]
+        if self._config.allow_ainput_password:
+            return await aioconsole.ainput("Matrix password: ")
+        raise RuntimeError("Can't get password")
+
+    async def _login_using_session_data(self, client: AsyncClient) -> None:
+        """
+        Login using session data from `Storage`.
+
+        Args:
+            - client - client that needs to get logged in
+
+        Returns:
+            - Does not return anything on success
+            - Raises an exception on error
+        """
+        if self._full_matrix_username is None:
+            raise RuntimeError("ClientAuth is not set up")
+        self._logger.info("Trying to log in using session data from disk...")
+        session_data = await self._storage.read_session_data()
+        if session_data is None:
+            self._logger.error("Can't read session data from disk")
+            raise RuntimeError("Can't read session data from disk")
+        client.restore_login(
+            user_id=self._full_matrix_username,
+            access_token=session_data["access_token"],
+            device_id=session_data["device_id"]
+        )
+        client.load_store()
+        result = await client.whoami()
+        if type(result) is WhoamiError:
+            self._logger.error(
+                f"Can't log in using stored session data: '{result.message}'"
+            )
+            raise RuntimeError(result.message)
+        elif type(result) is not WhoamiResponse:
+            self._logger.error(
+                "Can't log in using stored session data, unknown error"
+            )
+            raise RuntimeError(
+                "Unknown response for whoami request"
+            )
+        self._logger.info("Logged in using stored session data")
+
+    async def _login_using_password(self, client: AsyncClient) -> None:
+        """
+        Login using password.
+
+        Args:
+            - client - client that needs to get logged in
+        
+        Returns:
+            - Does not return anything on success
+            - Raises an exception on error
+        """
+        if self._full_matrix_username is None:
+            raise RuntimeError("ClientAuth is not set up")
+        self._logger.info("Trying to log in using password")
+        password = await self._default_password_callback()
+        result = await client.login(password=password)
+        if type(result) is LoginResponse:
+            self._logger.info("Logged in using password")
+            await self._storage.write_session_data(
+                access_token=result.access_token,
+                device_id=result.device_id
+            )
+        elif type(result) is LoginError:
+            self._logger.error(f"Can't log in using password: '{result.message}'")
+            raise RuntimeError(result.message)
+        else:
+            self._logger.error(f"Can't log in using password, unknown error")
+            raise RuntimeError("Unknown login result")
+
+    #
+    # PUBLIC
+    #
+    def __init__(self, storage: Storage):
+        self._logger = logging.getLogger("ClientAuth")
+        self._storage = storage
+        self._full_matrix_username: str | None = None
+        self._config: MatrixBotConfig | None = None
+
+    async def setup(self, config: MatrixBotConfig) -> None:
+        """
+        Setup `ClientAuth` object using `config`.
+        """
+        self._config = config
+        self._full_matrix_username = Utils.build_full_matrix_username(config)
+
+    async def login(self, client: AsyncClient) -> None:
+        """
+        Login. Saves authorization data if it is not saved yet. Must be called
+        by `ClientManager`.
+
+        Args:
+            - client - client that needs to get logged in
+
+        Returns:
+            - Does not return anything on success
+            - Raises an exception on error
+        """
+        # try session data
+        try:
+            await self._login_using_session_data(client)
+            return
+        except asyncio.CancelledError:
+            raise
+        except:
+            pass
+        # try password
+        try:
+            await self._login_using_password(client)
+            return
+        except asyncio.CancelledError:
+            raise
+        except:
+            pass
+        # failed
+        raise RuntimeError("Failed to login")

src/mab/bot/_client_manager.py

@@ -0,0 +1,204 @@
+import asyncio
+import logging
+import time
+import traceback
+from typing import Any
+
+from nio import AsyncClient, AsyncClientConfig, SyncResponse
+from ..types import MatrixBotConfig
+
+from ._storage import Storage
+from ._client_auth import ClientAuth
+
+class ClientManager:
+    """This class manages basic of `AsyncClient`."""
+
+    #
+    # PRIVATE
+    #
+    async def _close_client(self) -> None:
+        """
+        Closes the client correctly and saves data.
+        """
+        if self._client is None:
+            raise RuntimeError("The client does not exist")
+        self._logger.debug("Closing the client...")
+        await self._client.close()
+        self._logger.debug("Dumping 'next_batch' value...")
+        if self._last_next_batch is not None:
+            await self._storage.write_next_batch(self._last_next_batch)
+
+    async def _cancellable_sync_forever(self, *args, **kwargs) -> Any:
+        """
+        Behaves exactly like AsyncClient.sync_forever, but supports proper task
+        cancellation.
+        """
+        if self._client is None:
+            raise RuntimeError("Client does not exist")
+        task = asyncio.create_task(self._client.sync_forever(*args, **kwargs))
+        try:
+            return await task
+        except asyncio.CancelledError:
+            try:
+                self._client.stop_sync_forever()
+            except Exception:
+                pass
+            finally:
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+            raise
+
+    async def _background_coroutine(self) -> None:
+        """
+        Coroutine that manages the `AsyncClient` in background. It supports
+        cancellation and you should use it to stop the operation of the bot.
+        """
+        # refuse to start without config
+        if self._config is None:
+            raise RuntimeError("Config is not set up")
+        # refuse to start without client
+        if self._client is None:
+            raise RuntimeError("Client does not exist")
+        # try to login
+        self._logger.info("Logging in...")
+        try:
+            await self._client_auth.login(self._client)
+            self._logger.info(f"Logged in as {self._client.user_id}")
+        except asyncio.CancelledError:
+            raise
+        except:
+            self._logger.error(traceback.format_exc())
+            raise
+        # sync forever
+        self._logger.info("Syncing forever...")
+        await self._cancellable_sync_forever(
+            timeout=self._config.client_sync_period_ms,
+            since=(await self._storage.read_next_batch()),
+            full_state=True
+        )
+
+    async def _callback_sync(self, response: SyncResponse) -> None:
+        """
+        This callback is called when AsyncClient syncs with the server
+        """
+        if self._config is None:
+            raise RuntimeError("ClientManager is not set up")
+        # save last `next_batch`
+        self._last_next_batch = response.next_batch
+        # dump if needed
+        current_time = time.time()
+        delta_time = current_time - self._last_next_batch_write_time
+        if delta_time >= self._config.client_next_batch_dump_period_s:
+            try:
+                await self._storage.write_next_batch(self._last_next_batch)
+                self._last_next_batch_write_time = current_time
+            except:
+                self._logger.error(traceback.format_exc())
+
+    async def _setup_callbacks(self, config: MatrixBotConfig) -> None:
+        """
+        Setup callbacks.
+        """
+        if self._client is None:
+            raise RuntimeError("The client does not exist")
+        self._client.add_response_callback(
+            self._callback_sync, SyncResponse) # type: ignore
+
+    #
+    # PUBLIC
+    #
+    def __init__(self, client_auth: ClientAuth, storage: Storage):
+        self._logger = logging.getLogger("ClientManager")
+        self._client_auth = client_auth
+        self._storage = storage
+
+        self._config: MatrixBotConfig | None = None
+        self._client: AsyncClient | None = None
+        self._was_started_ever: bool = False
+        self._was_built_ever: bool = False
+        self._background_task: asyncio.Task | None = None
+        self._last_next_batch_write_time: float = 0.0
+        self._last_next_batch: str | None = None
+    
+    async def setup(self, config: MatrixBotConfig) -> None:
+        """
+        Build `AsyncClient` from `config`.
+
+        Returns:
+            - Does not return anything, raises an exception on error.
+        """
+        if self._was_built_ever:
+            raise RuntimeError("The client was already built")
+        client_config = AsyncClientConfig(
+            store_name="nio_store_file",
+            encryption_enabled=True,
+            store_sync_tokens=False
+        )
+        self._client = AsyncClient(
+            homeserver=config.matrix_homeserver_url,
+            user=config.matrix_username_localpart,
+            store_path=str(config.storage_directory),
+            config=client_config
+        )
+        try:
+            await self._setup_callbacks(config)
+        except:
+            self._client = None
+            raise
+        self._config = config
+        self._was_built_ever = True
+
+    def get_client(self) -> AsyncClient:
+        """
+        Get AsyncClient.
+
+        Returns:
+            - `AsyncClient` you may use to access Matrix. Raises a RuntimeError
+              if the client does not exist or is not operational.
+        """
+        if self._client is None:
+            raise RuntimeError("The client does not exist")
+        return self._client
+
+    async def start(self) -> None:
+        """
+        Start client operation in background. This method can be called
+        succesfully only once for `ClientManager` object. Any subsequent calls
+        will raise an exception.
+        """
+        if self._client is None:
+            raise RuntimeError("Client does not exist")
+        if self._was_started_ever:
+            raise RuntimeError("Client can't be started several times")
+        self._background_task = asyncio.create_task(
+            self._background_coroutine()
+        )
+        self._was_started_ever = True
+        self._logger.info("Bot background is started")
+
+    async def stop(self) -> None:
+        """
+        Stop client operation in background. This method can be called only if
+        the client was `start(...)`ed previously. This method will wait for the
+        bot to completely stop.
+        """
+        if self._client is None:
+            raise RuntimeError("Client does not exist")
+        if self._background_task is None:
+            if self._was_started_ever:
+                raise RuntimeError("The bot was stopped previously")
+            else:
+                raise RuntimeError("The bot was never started")
+        self._background_task.cancel()
+        try:
+            await asyncio.shield(self._background_task)
+        except asyncio.CancelledError:
+            pass
+        except:
+            self._logger.error(traceback.format_exc())
+        try:
+            await asyncio.shield(self._close_client())
+        except:
+            self._logger.error(traceback.format_exc())
+        self._background_task = None
+        self._logger.info("Bot background task is stopped")

src/mab/bot/_client_sender.py

@@ -0,0 +1,364 @@
+import asyncio
+from io import BytesIO
+import logging
+from html.parser import HTMLParser
+from pathlib import Path
+import traceback
+import os
+
+import magic
+from PIL import Image
+from nio import AsyncClient, MatrixRoom, RoomSendResponse, RoomSendError
+from nio import OlmUnverifiedDeviceError
+
+from ._client_uploader import ClientUploader
+from ..types import MatrixBotConfig, VideoFileProperties
+from ..utils import Utils
+
+class ClientSender:
+    """This class sends messages"""
+    #
+    # PRIVATE
+    #
+    def _process_html_text(self, text: str, is_html: bool | None) -> dict:
+        """
+        Processes `text` and `is_html` arguments of `send_*` functions and
+        prepares `content` keys related to text data inside the message.
+        
+        Args:
+            - text - `text` as passed to the `send_*` function
+            - is_html - `is_html` as passed to the `send_*` function
+        
+        Returns:
+            `dict` that contains keys required for the text message to be
+            displayed correctly.
+        """
+        # text w/ and w/o HTML
+        text_w_html: str | None = None
+        text_wo_html: str | None = None
+        # check if HTML is present
+        if is_html or is_html is None:
+            # will be True if `text` contains HTML tags
+            has_tags = False
+            # text fragments without tags
+            text_fragments = []
+            # temporary extractor class
+            class Extractor(HTMLParser):
+                def handle_starttag(self, tag, attrs):
+                    nonlocal has_tags
+                    has_tags = True
+                def handle_data(self, data):
+                    text_fragments.append(data)
+            # parse
+            parser = Extractor(convert_charrefs=True)
+            try:
+                parser.feed(text)
+            except:
+                self._logger.error(traceback.format_exc())
+                text_wo_html = text
+            is_html = has_tags
+            if is_html:
+                text_wo_html = " ".join(text_fragments)
+                text_w_html = text
+        if not is_html:
+            text_wo_html = text
+        # return the content
+        if text_w_html:
+            return {
+                "body": text_wo_html,
+                "format": "org.matrix.custom.html",
+                "formatted_body": text_w_html
+            }
+        else:
+            return {
+                "body": text_wo_html
+            }
+
+    #
+    # PUBLIC
+    #
+    def __init__(self):
+        self._logger = logging.getLogger("ClientSender")
+        self._config: MatrixBotConfig | None = None
+        self._client: AsyncClient | None = None
+
+    async def setup(self,
+                    config: MatrixBotConfig,
+                    client: AsyncClient,
+                    uploader: ClientUploader) -> None:
+        """
+        Setup the sender.
+
+        Args:
+            - config - config to use
+            - client - client to use
+            - uploader - file uploaded to use
+        """
+        self._config = config
+        self._client = client
+        self._uploader = uploader
+
+    async def send_content(self,
+                           room: MatrixRoom | str,
+                           content: dict) -> RoomSendResponse:
+        """
+        Send raw `content` JSON to `room`.
+
+        Args:
+            - room - the room to send the content to
+            - content - the content to send to the room
+        
+        Returns:
+            - `RoomSendResponse` returned by `AsyncClient.room_send`
+            - Raises an exception on error
+        """
+        if self._client is None or self._config is None:
+            raise RuntimeError("ClientSender is not set up")
+        try:
+            result = await self._client.room_send(
+                room_id=room.room_id if isinstance(room, MatrixRoom) else room,
+                message_type="m.room.message",
+                content=content
+            )
+        except OlmUnverifiedDeviceError:
+            if self._config.auto_verify_all_known_devices:
+                if not Utils.verify_all_known_devices(self._client):
+                    raise
+                return await self.send_content(room, content)
+            else:
+                raise
+        if type(result) is RoomSendResponse:
+            return result
+        elif type(result) is RoomSendError:
+            raise RuntimeError(result)
+        else:
+            raise RuntimeError("Unknown error has occured", result)
+
+    async def send_text(self,
+                        room: MatrixRoom | str,
+                        text: str,
+                        *,
+                        is_html: bool | None = None) -> str:
+        """
+        Send text message to `room`.
+
+        Args:
+            - room - the room to send the text to
+            - text - the text to send to the room
+            - is_html - whether the text is HTML-formatted. Use `None` for auto
+        
+        Returns:
+            - `event_id` of sent message on success
+            - Raises an exception on error
+        """
+        text_data = self._process_html_text(text, is_html)
+        content = {
+            "msgtype": "m.text",
+            **text_data
+        }
+        return (await self.send_content(room, content)).event_id
+
+    async def send_image(self,
+                         room: MatrixRoom | str,
+                         path: Path | str, *,
+                         text: str | None = None,
+                         is_html: bool | None = None,
+                         filename: str | None = None,
+                         timeout: float | None = 60 * 60) -> str:
+        """
+        Send the image to `room`. Please note that formatted text is displayed
+        incorrectly in some clients as of September 8th, 2026
+
+        Args:
+            - room - the room to send the text to
+            - path - path to the image file
+            - text - image caption to use (`None` to disable)
+            - is_html - whether the text is HTML-formatted (`None` for auto)
+            - filename - filename to use for the file (`None` for auto)
+            - timeout - upload timeout in seconds (`None` to disable)
+        
+        Returns:
+            - `event_id` of sent message on success
+            - Raises an exception on error
+        """
+        # caption must not actually be empty
+        if text is None or not text.strip():
+            text = os.path.basename(path)
+            is_html = False
+        # check if the file is image
+        mime_type: str = magic.from_file(path, mime=True)
+        if not mime_type.startswith("image/"):
+            raise RuntimeError(f"File {path} has non-image mime-type")
+        # get image size
+        with Image.open(path) as image:
+            width, height = image.size
+        # upload
+        async with asyncio.timeout(timeout):
+            upload_result = await self._uploader.upload_file(
+                path, mime_type=mime_type, filename=filename)
+        # prepare the content and send
+        content = {
+            "msgtype": "m.image",
+            "filename": os.path.basename(path),
+            **self._process_html_text(text, is_html),
+            "file": {
+                "url": upload_result.response.content_uri,
+                "mimetype": mime_type,
+                **upload_result.keys
+            },
+            "info": {
+                "mimetype": mime_type,
+                "size": upload_result.filesize,
+                "w": width,
+                "h": height
+            }
+        }
+        return (await self.send_content(room, content)).event_id
+
+    async def send_image_bytes(self,
+                               room: MatrixRoom | str,
+                               data: bytes,
+                               filename: str,
+                               *,
+                               text: str | None = None,
+                               is_html: bool | None = None,
+                               timeout: float | None = 60 * 60) -> str:
+        """
+        Send the image to `room`. Please note that formatted text is displayed
+        incorrectly in some clients as of September 8th, 2026
+
+        Args:
+            - room - the room to send the text to
+            - bytes - the image to send
+            - filename - filename to use for the file
+            - text - image caption to use (`None` to disable)
+            - is_html - whether the text is HTML-formatted (`None` for auto)
+            - timeout - upload timeout in seconds (`None` to disable)
+        
+        Returns:
+            - `event_id` of sent message on success
+            - Raises an exception on error
+        """
+        # caption must not actually be empty
+        if text is None or not text.strip():
+            text = filename
+            is_html = False
+        # check if the file is image
+        mime_type: str = magic.from_buffer(data, mime=True)
+        if not mime_type.startswith("image/"):
+            raise RuntimeError(f"Data has non-image mime-type")
+        # get image size
+        buffer = BytesIO(data)
+        with Image.open(buffer) as image:
+            width, height = image.size
+        buffer.seek(0)
+        # upload
+        async with asyncio.timeout(timeout):
+            upload_result = await self._uploader.upload_using_provider(
+                provider=buffer,
+                mime_type=mime_type,
+                filename=filename,
+                filesize=len(data))
+        # prepare the content and send
+        content = {
+            "msgtype": "m.image",
+            "filename": filename,
+            **self._process_html_text(text, is_html),
+            "file": {
+                "url": upload_result.response.content_uri,
+                "mimetype": mime_type,
+                **upload_result.keys
+            },
+            "info": {
+                "mimetype": mime_type,
+                "size": upload_result.filesize,
+                "w": width,
+                "h": height
+            }
+        }
+        return (await self.send_content(room, content)).event_id
+
+    async def send_video(self,
+                         room: MatrixRoom | str,
+                         path: Path | str,
+                         *,
+                         props: VideoFileProperties | None = None,
+                         text: str | None = None,
+                         is_html: bool | None = None,
+                         timeout: float | None = 60 * 60) -> str:
+        """
+        Send the video to `room`. Please note that formatted text is displayed
+        incorrectly in some clients as of September 8th, 2026. Unknown video
+        properties will be automatically deduced as configured in
+        `MatrixBotConfig`.
+
+        Args:
+            - room - the room to send the text to
+            - path - path to the video file
+            - props - video properties (`None` for auto, if the feature is ON)
+            - text - video caption to use (`None` to disable)
+            - is_html - whether the text is HTML-formatted (`None` for auto)
+            - timeout - upload timeout in seconds (`None` to disable)
+        
+        Returns:
+            - `event_id` of sent message on success
+            - Raises an exception on error
+        """
+        if self._client is None or self._config is None:
+            raise RuntimeError("ClientSender is not set up")
+        # caption must not actually be empty
+        if text is None or not text.strip():
+            text = os.path.basename(path)
+            is_html = False
+        # check if the file is video
+        mime_type: str = magic.from_file(path, mime=True)
+        if not mime_type.startswith("video/"):
+            raise RuntimeError(f"File {path} has non-video mime-type")
+        # properties must not be empty
+        if props is None:
+            if not self._config.auto_ffprobe:
+                raise RuntimeError(
+                    "No `props` are supplied and `auto_ffprobe` is disabled"
+                )
+            async with asyncio.timeout(30):
+                props = await Utils.get_video_props_using_ffprobe(path)
+        # autogenerate thumbnail if allowed to do so
+        if not props.thumbnail and self._config.auto_ffmpeg:
+            try:
+                async with asyncio.timeout(30):
+                    props.thumbnail = await Utils.get_video_thumbnail(
+                        path,
+                        timestamp=props.duration_ms / 1000 / 2,
+                        max_size=400
+                    )
+            except asyncio.TimeoutError:
+                pass
+        async with asyncio.timeout(timeout):
+            # upload the thumbnail if specified
+            thumbnail_data = await self._uploader.upload_thumbnail(props.thumbnail)
+            # upload
+            upload_result = await self._uploader.upload_file(
+                path,
+                mime_type=mime_type
+            )
+        # prepare the content
+        content = {
+            "msgtype": "m.video",
+            "filename": os.path.basename(path),
+            **self._process_html_text(text, is_html),
+            "file": {
+                "url": upload_result.response.content_uri,
+                "mimetype": mime_type,
+                **upload_result.keys
+            },
+            "info": {
+                "duration": props.duration_ms,
+                "mimetype": mime_type,
+                "size": upload_result.filesize,
+                "w": props.width_px,
+                "h": props.height_px,
+                **thumbnail_data
+            }
+        }
+        # send
+        return (await self.send_content(room, content)).event_id

src/mab/bot/_client_uploader.py

@@ -0,0 +1,187 @@
+from typing import Any
+from pathlib import Path
+import io
+import os
+
+import aiofiles
+import aiofiles.os
+import magic
+from PIL import Image
+
+from nio import AsyncClient
+from nio import UploadError
+
+from ._storage import Storage
+from ..types import MatrixBotConfig, UploadResult
+
+class ClientUploader:
+    """This class uploads files"""
+    def __init__(self, storage: Storage):
+        self._storage = storage
+        self._config: MatrixBotConfig | None = None
+        self._client: AsyncClient | None = None
+
+    async def setup(self, config: MatrixBotConfig, client: AsyncClient) -> None:
+        """
+        Setup the uploader.
+
+        Args:
+            - config - config to take some settings from.
+        """
+        self._config = config
+        self._client = client
+
+    async def upload_using_provider(self,
+                                    provider: Any,
+                                    *,
+                                    mime_type: str,
+                                    filename: str,
+                                    filesize: int) -> UploadResult:
+        """
+        Upload data using provider (as defined in `AsyncClient.upload` docs).
+        You better use `upload_file(...)`, `upload_bytes(...)`. Data is always
+        encrypted.
+
+        Args:
+            - provider - data provider (e.g. `aiofiles.open(...)`)
+            - mime_type - mime-type of the data
+            - filename - filename to declare for the file
+            - filesize - size of the file
+        
+        Returns:
+            - `UploadResult` that contains all data you need on success
+            - Raises an exception on failure
+        """
+        if self._config is None or self._client is None:
+            raise RuntimeError("ClientUploader is not set up")
+        response, keys = await self._client.upload(
+            provider,
+            content_type=mime_type,
+            filename=filename,
+            encrypt=True,
+            monitor=None,
+            filesize=filesize
+        )
+        if isinstance(response, UploadError):
+            raise RuntimeError(f"Upload failed: {response.message}", response)
+        return UploadResult(
+            response=response,
+            keys=keys, # type: ignore
+            filename=filename,
+            filesize=filesize
+        )
+
+    async def upload_file(self,
+                          path: Path | str,
+                          *,
+                          mime_type: str | None = None,
+                          filename: str | None = None) -> UploadResult:
+        """
+        Upload data from file. Data is always encrypted.
+
+        Args:
+            - path - path to the file that needs to be uploaded
+            - mime_type - mime-type to use for the file (`None` for auto)
+            - filename - filename to use for the file (`None` for auto)
+        """
+        if isinstance(path, str):
+            path = Path(path)
+        if not path.is_file():
+            raise RuntimeError(f"{path} is not a file")
+        if not filename:
+            filename = os.path.basename(path)
+        if not mime_type:
+            mime_type = magic.from_file(path, mime=True)
+        filesize = (await aiofiles.os.stat(path)).st_size
+        async with aiofiles.open(path, "r+b") as f:
+            return await self.upload_using_provider(
+                f,
+                mime_type=mime_type,
+                filename=filename,
+                filesize=filesize
+            )
+
+    async def upload_bytes(self,
+                           data: bytes,
+                           *,
+                           filename: str,
+                           mime_type: str | None = None) -> UploadResult:
+        """
+        Upload data. Data is always encrypted.
+
+        Args:
+            - data - data that needs to be uploaded
+            - filename - filename to use for the file
+            - mime_type - mime-type to use for the file (`None` for auto)
+        """
+        if not mime_type:
+            mime_type = magic.from_buffer(data, mime=True)
+        buffer = io.BytesIO(data)
+        return await self.upload_using_provider(
+            buffer,
+            mime_type=mime_type,
+            filename=filename,
+            filesize=len(data)
+        )
+
+    async def upload_thumbnail(self,
+                               thumbnail: Path | str | bytes | None) -> dict:
+        """
+        Upload thumbnail and return fields required to use it for the video.
+        
+        Args:
+            - thumbnail - path to the thumbnail file or raw `bytes` of the
+                          image. You can actually pass `None`, empty dict will
+                          be returned in this case.  
+        
+        Returns:
+            `dict` you should embed into your `content["info"]`
+        """
+        if thumbnail is None:
+            return {}
+        # raw data
+        raw_bytes_io: io.BytesIO | None = None
+        if type(thumbnail) is bytes:
+            thumb_mime = magic.from_buffer(thumbnail, mime=True)
+            raw_bytes_io = io.BytesIO(thumbnail)
+        # file path
+        else:
+            thumb_mime = magic.from_file(thumbnail, mime=True)
+        # check if invalid MIME
+        if not thumb_mime.startswith("image/"):
+            raise RuntimeError(f"Thumbnail MIME is not image: `{thumb_mime}`")
+        # get thumbnail size
+        with Image.open(raw_bytes_io or thumbnail) as f:
+            width, height = f.size
+        # seek BytesIO to zero so we can reuse it
+        if raw_bytes_io is not None:
+            raw_bytes_io.seek(0)
+        # upload
+        if type(thumbnail) is bytes:
+            thumb_size = len(thumbnail)
+            upload_result = await self.upload_using_provider(
+                raw_bytes_io,
+                mime_type=thumb_mime,
+                filename=f"preview.{thumb_mime.split("/")[-1]}",
+                filesize=thumb_size
+            )
+        else:
+            upload_result = await self.upload_file(
+                thumbnail, # type: ignore
+                mime_type=thumb_mime,
+                filename=os.path.basename(str(thumbnail))
+            )
+        # return the result
+        return {
+            "thumbnail_file": {
+                "url": upload_result.response.content_uri,
+                "mimetype": thumb_mime,
+                **upload_result.keys
+            },
+            "thumbnail_info": {
+                "w": width,
+                "h": height,
+                "size": upload_result.filesize,
+                "mimetype": thumb_mime
+            }
+        }

src/mab/bot/_storage.py

@@ -0,0 +1,103 @@
+from pathlib import Path
+import aiofiles
+import logging
+import json
+from typing import Any
+
+from ..types import MatrixBotConfig
+
+class Storage:
+    """This class manages internal storage of the MatrixBot"""
+
+    def __init__(self):
+        self._logger = logging.getLogger("Storage")
+        self._config: MatrixBotConfig | None = None
+
+    async def setup(self, config: MatrixBotConfig) -> None:
+        """
+        Setup the internal storage of the MatrixBot.
+
+        Args:
+            - config - the config to use
+        
+        Returns:
+            - Does not return anything
+            - Raises an exception on error
+        """
+        self._config = config
+
+    async def write_next_batch(self, next_batch: str) -> None:
+        """
+        Write `next_batch` value to the disk.
+        
+        Args:
+            - next_batch - value to save
+        
+        Returns:
+            - Does not return anything
+            - Raises an exception on error
+        """
+        if self._config is None:
+            raise RuntimeError("Storage is not set up")
+        path = Path(self._config.storage_directory) / "next_batch"
+        async with aiofiles.open(path, "w") as f:
+            await f.write(next_batch)
+        self._logger.debug("next_batch value is written to the disk")
+
+    async def read_next_batch(self) -> str | None:
+        """
+        Read `next_batch` value from the disk.
+        
+        Returns:
+            - Returns `str` if there is `next_batch` value saved on disk
+            - Returns `None` if `next_batch` was never saved yet
+            - Raises an exception on error
+        """
+        if self._config is None:
+            raise RuntimeError("Storage is not set up")
+        path = Path(self._config.storage_directory) / "next_batch"
+        if not path.is_file():
+            return None
+        async with aiofiles.open(path, "r") as f:
+            return (await f.read()).strip()
+    
+    async def write_session_data(self, *, access_token: str, device_id: str) -> None:
+        """
+        Write session data to disk.
+        
+        Args:
+            - access_token - access_token to write to the disk
+            - device_id - device_id to write to disk
+
+        Returns:
+            - Does not return anything
+            - Raises an exception on error
+        """
+        if self._config is None:
+            raise RuntimeError("Storage is not set up")
+        path = Path(self._config.storage_directory) / "session_data.json"
+        data = {
+            "access_token": access_token,
+            "device_id": device_id
+        }
+        async with aiofiles.open(path, "w") as f:
+            await f.write(json.dumps(data, indent=4))
+        self._logger.debug("Session data is writter to the disk")
+
+    async def read_session_data(self) -> dict[str, Any] | None:
+        """
+        Read session data from disk.
+        
+        Returns:
+            - Dictionary with `access_token` and `device_id` keys if these
+              values were saved using `write_session_data`
+            - `None` if session data was never saved yet
+        """
+        if self._config is None:
+            raise RuntimeError("Storage is not set up")
+        path = Path(self._config.storage_directory) / "session_data.json"
+        if not path.is_file():
+            return None
+        async with aiofiles.open(path, "r") as f:
+            j = json.loads(await f.read())
+        return j

src/mab/bot/_validation.py

@@ -0,0 +1,84 @@
+import re
+from pathlib import Path
+from urllib.parse import urlparse
+
+from ..types import MatrixBotConfig
+
+class Validator:
+    """This class is used inside MatrixBot to validate data"""
+
+    def validate_matrix_homeserver_url(self, url: str) -> None:
+        """
+        Check if `url` is a valid matrix homeserver URL.
+        
+        Args:
+            - url - URL to check (only format is checked)
+        
+        Returns:
+            - Does not return anything
+            - Raises an exception on error
+        """
+        parsed = urlparse(url)
+        if parsed.scheme not in ("http", "https"):
+            raise RuntimeError(
+                f"Scheme {parsed.scheme} is not a valid scheme for matrix "
+                "homeserver URL"
+            )
+        if not parsed.netloc:
+            raise RuntimeError(
+                f"{url} is not a valid matrix homeserver URL"
+            )
+        if parsed.path != "":
+            raise RuntimeError(
+                f"{url} must have empty path (remove `{parsed.path}` after the "
+                "hostname)"
+            )
+
+    def validate_matrix_username_localpart(self, localpart: str) -> None:
+        """
+        Check if `localpart` is a valid localpart of matrix username.
+
+        Args:
+            - localpart - localpart of matrix username
+
+        Returns:
+            - Does not return anything
+            - Raises an exception on error
+        """
+        pattern = r"^[a-z0-9._=\-]+$"
+        if not bool(re.fullmatch(pattern, localpart)) or len(localpart) > 255:
+            raise RuntimeError(
+                f"{localpart} is not a valid matrix username localpart"
+            )
+
+    def ensure_directory_exists(self, path: Path | str) -> None:
+        """
+        Ensure that `path` is an existing directory. Creates it if it does not
+        exist. The entire directory tree is created if needed.
+
+        Args:
+            - path - directory path
+        
+        Returns:
+            - Does not return anything
+            - Raises an exception on error
+        """
+        if isinstance(path, str):
+            path = Path(path)
+        if not path.is_dir():
+            path.mkdir(parents=True, exist_ok=True)
+
+    def validate_bot_config(self, config: MatrixBotConfig) -> None:
+        """
+        Check if `config` is a valid MatrixBot config.
+
+        Args:
+            - config - configuration to check
+        
+        Returns:
+            - Does not return anything
+            - Raises an exception on error
+        """
+        self.validate_matrix_homeserver_url(config.matrix_homeserver_url)
+        self.validate_matrix_username_localpart(config.matrix_username_localpart)
+        self.ensure_directory_exists(config.storage_directory)

src/mab/bot/bot.py

@@ -0,0 +1,247 @@
+import asyncio
+import logging
+
+from typing import Callable, Coroutine, Any
+
+from nio import AsyncClient, MatrixRoom
+
+from ..filters.base import BaseEventFilter
+from ..types import *
+from ..context import EventContext
+
+from ._validation import Validator
+from ._storage import Storage
+from ._client_auth import ClientAuth
+from ._client_manager import ClientManager
+from ._client_uploader import ClientUploader
+from ._client_sender import ClientSender
+from ._callbacks import Callbacks
+
+
+class MatrixBot:
+    """
+    Asynchronous Matrix Bot Implementation.
+    """
+
+    def __init__(self, config: MatrixBotConfig) -> None:
+        # create the logger
+        self._logger = logging.getLogger("MatrixBot")
+        self._was_started = False
+        # create classes that compose MatrixBot
+        self._validator = Validator()
+        self._storage = Storage()
+        self._client_auth = ClientAuth(self._storage)
+        self._client_manager = ClientManager(self._client_auth, self._storage)
+        self._client_uploader = ClientUploader(self._storage)
+        self._client_sender = ClientSender()
+        self._callbacks = Callbacks(self._storage, self)
+        # validate the config and save it
+        try:
+            self._validator.validate_bot_config(config)
+            self._config: MatrixBotConfig = config
+        except Exception as e:
+            self._logger.error(f"Couldn't validate bot config: {e}")
+            raise
+
+    def add_callback(self,
+                           filter: BaseEventFilter,
+                           callback: Callable[[EventContext], Coroutine[Any, Any, None]] | None,
+                           *,
+                           stop_matching: bool = True) -> None:
+        """
+        Add filter-based callback.
+
+        Args:
+            - filter - filter to use for the callback
+            - callback - the callback that must be called if the filter matches.
+                         Use `None` to log the type of the Event that will
+                         happen for specified filter.
+            - stop_matching - whether to stop matching if the filter matches
+        """
+        self._callbacks.add_room_event_callback(
+            filter=filter,
+            callback=callback,
+            stop_matching=stop_matching
+        )
+
+    async def start(self) -> None:
+        """
+        Start client operation in background. This method can be called
+        succesfully only once for `MatrixBot` object. Any subsequent calls
+        will raise an exception.
+        """
+        if self._was_started:
+            raise RuntimeError("The bot was already started")
+        self._was_started = True
+        await self._client_manager.setup(self._config)
+        await self._storage.setup(self._config)
+        await self._client_auth.setup(self._config)
+        await self._client_uploader.setup(
+            self._config,
+            self._client_manager.get_client()
+        )
+        await self._client_sender.setup(
+            self._config,
+            self._client_manager.get_client(),
+            self._client_uploader
+        )
+        await self._callbacks.setup(
+            self._config,
+            self._client_manager.get_client()
+        )
+        await self._client_manager.start()
+
+    async def stop(self) -> None:
+        """
+        Stop client operation in background. This method can be called only if
+        the client was `start(...)`ed previously. This method will wait for the
+        bot to completely stop.
+        """
+        await self._client_manager.stop()
+
+    async def run(self) -> None:
+        """
+        Start bot operation in foreground. You may cancel task running this
+        method to stop the bot.
+
+        Warning: calling `stop()` is not a supported way to stop the bot. You
+        should cancel this task instead.
+        """
+        await self.start()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            await self.stop()
+
+    def get_client(self) -> AsyncClient:
+        """
+        Get AsyncClient.
+
+        Returns:
+            - `AsyncClient` you may use to access Matrix. Raises a RuntimeError
+              if the client does not exist or is not operational.
+        """
+        return self._client_manager.get_client()
+
+    async def send_text(self,
+                        room: MatrixRoom | str,
+                        text: str,
+                        *,
+                        is_html: bool | None = None) -> str:
+        """
+        Send text message to `room`.
+
+        Args:
+            - room - the room to send the text to
+            - text - the text to send to the room
+            - is_html - whether the text is HTML-formatted. Use `None` for auto
+        
+        Returns:
+            - `event_id` of sent message on success
+            - Raises an exception on error
+        """
+        return await self._client_sender.send_text(
+            room=room,
+            text=text,
+            is_html=is_html
+        )
+
+    async def send_image(self,
+                         room: MatrixRoom | str,
+                         path: Path | str, *,
+                         text: str | None = None,
+                         is_html: bool | None = None,
+                         filename: str | None = None,
+                         timeout: float | None = 60 * 60) -> str:
+        """
+        Send the image to `room`. Please note that formatted text is displayed
+        incorrectly in some clients as of September 8th, 2026
+
+        Args:
+            - room - the room to send the text to
+            - path - path to the image file
+            - text - image caption to use (`None` to disable)
+            - is_html - whether the text is HTML-formatted (`None` for auto)
+            - filename - filename to use for the file (`None` for auto)
+            - timeout - upload timeout in seconds (`None` to disable)
+        
+        Returns:
+            - `event_id` of sent message on success
+            - Raises an exception on error
+        """
+        return await self._client_sender.send_image(
+            room=room,
+            path=path,
+            text=text,
+            is_html=is_html,
+            filename=filename,
+            timeout=timeout
+        )
+
+    async def send_image_bytes(self,
+                               room: MatrixRoom | str,
+                               data: bytes,
+                               filename: str,
+                               *,
+                               text: str | None = None,
+                               is_html: bool | None = None,
+                               timeout: float | None = 60 * 60) -> str:
+        """
+        Send the image to `room`. Please note that formatted text is displayed
+        incorrectly in some clients as of September 8th, 2026
+
+        Args:
+            - room - the room to send the text to
+            - bytes - the image to send
+            - filename - filename to use for the file
+            - text - image caption to use (`None` to disable)
+            - is_html - whether the text is HTML-formatted (`None` for auto)
+            - timeout - upload timeout in seconds (`None` to disable)
+        
+        Returns:
+            - `event_id` of sent message on success
+            - Raises an exception on error
+        """
+        return await self._client_sender.send_image_bytes(
+            room=room,
+            data=data,
+            filename=filename,
+            text=text,
+            is_html=is_html,
+            timeout=timeout
+        )
+
+    async def send_video(self,
+                         room: MatrixRoom | str,
+                         path: Path | str,
+                         *,
+                         props: VideoFileProperties | None = None,
+                         text: str | None = None,
+                         is_html: bool | None = None,
+                         timeout: float | None = 60 * 60) -> str:
+        """
+        Send the video to `room`. Please note that formatted text is displayed
+        incorrectly in some clients as of September 8th, 2026. Unknown video
+        properties will be automatically deduced as configured in
+        `MatrixBotConfig`.
+
+        Args:
+            - room - the room to send the text to
+            - path - path to the video file
+            - props - video properties (`None` for auto, if the feature is ON)
+            - text - video caption to use (`None` to disable)
+            - is_html - whether the text is HTML-formatted (`None` for auto)
+            - timeout - upload timeout in seconds (`None` to disable)
+        
+        Returns:
+            - `event_id` of sent message on success
+            - Raises an exception on error
+        """
+        return await self._client_sender.send_video(
+            room=room,
+            path=path,
+            props=props,
+            text=text,
+            is_html=is_html,
+            timeout=timeout
+        )

src/mab/context.py

@@ -0,0 +1,79 @@
+"""This module implements logic for event context"""
+
+from dataclasses import dataclass
+from typing import Any, TYPE_CHECKING
+
+from .types import ContextDataKey, MessageType
+
+from nio import MatrixRoom, Event
+
+if TYPE_CHECKING:
+    from .bot import MatrixBot
+
+#
+# Possible context variables
+#
+CTX_BODY = ContextDataKey[str]("CTX_BODY")
+"""Value of `event.body`"""
+
+CTX_MESSAGE_TYPE = ContextDataKey[MessageType]("CTX_MESSAGE_TYPE")
+"""Value of `msgtype` for the event"""
+
+CTX_SENDER = ContextDataKey[str]("CTX_SENDER")
+"""Value of `event.sender`"""
+
+CTX_CMD_PREFIX = ContextDataKey[str]("CTX_CMD_PREFIX")
+"""Command prefix that was used when matching"""
+
+CTX_CMD_VERB = ContextDataKey[str]("CTX_CMD_VERB")
+"""The verb that was used to execute the command"""
+
+CTX_CMD_ARGS = ContextDataKey[list[str]]("CTX_CMD_ARGS")
+"""Arguments that were passed with the command"""
+
+CTX_ROOM_ENCRYPTED = ContextDataKey[bool]("CTX_ROOM_ENCRYPTED")
+"""True if the room is encrypted"""
+
+
+#
+# EventContext implementation
+#
+@dataclass
+class EventContext:
+    """The class holding information about an event that happened in the room"""
+
+    room: MatrixRoom
+    """The room the event has happened in"""
+
+    event: Event
+    """The event that has happened in the room"""
+
+    bot: "MatrixBot"
+    """The bot that is the source of the event"""
+
+    def __setitem__[T](self, key: ContextDataKey[T], value: T | None) -> None:
+        """Set a value inside the context data storage. `None` removes it"""
+        if not hasattr(self, "_datastore"):
+            self._datastore: dict[ContextDataKey, Any] = {}
+        if value is None:
+            del self._datastore[key]
+        else:
+            self._datastore[key] = value
+
+    def __getitem__[T](self, key: ContextDataKey[T]) -> T:
+        """
+        Get a value inside the context data storage.
+        
+        Raises RuntimeError if the value is not present.
+        """
+        if not hasattr(self, "_datastore") or key not in self._datastore:
+            raise RuntimeError(f"Context does not contain {repr(key)}")
+        return self._datastore[key]
+
+    def __contains__[T](self, key: ContextDataKey[T]) -> bool:
+        """Check if context data storage contains the value"""
+        if not hasattr(self, "_datastore"):
+            return False
+        if key not in self._datastore:
+            return False
+        return True

src/mab/filters/base.py

@@ -1,12 +1,20 @@
 from abc import ABC, abstractmethod
 import logging
+from typing import Any, Type
 
+from nio import AsyncClient
 from nio import MatrixRoom, Event
 
+from ..context import EventContext
+
 class BaseEventFilter(ABC):
     """Base class for all message filters"""
     _logger = logging.Logger("EventFilter")
 
+    def __init__(self, **kwargs):
+        for a in kwargs:
+            self._logger.critical(f"Unknown keyword argument for some filter is used: '{a}'={repr(kwargs[a])}")
+
     # AND
     def __and__(self, other):
         if not isinstance(other, BaseEventFilter):
@@ -29,7 +37,7 @@ class BaseEventFilter(ABC):
         )
 
     def __ror__(self, other):
-        return self.__ror__(other)
+        return self.__or__(other)
 
     # XOR
     def __xor__(self, other):
@@ -51,28 +59,41 @@ class BaseEventFilter(ABC):
         )
 
     # PAYLOAD
-    @abstractmethod
     def __repr__(self) -> str:
-        """This method must be redefined in derived classes to improve
-           debugging experience.
         """
-        pass
+        This method may be redefined in derived classes to improve debugging
+        experience.
+        """
+        return str(self.__class__.__name__)
 
-    @abstractmethod
+    async def __call__(self, context: EventContext) -> bool:
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
+        """
-        """This abstract method must be redefined in derived classes so that
+        This abstract method must be redefined in derived classes so that the
-           the filter operates according to its description. This method must
+        filter operates according to its description. This method must not raise
-           not raise exceptions. In case of exception it should log it using
+        exceptions. In case of exception it should log it using `self._logger`
-           `self._logger` and return False
+        and return False
         
         Args:
-                event - the event to check againts this filter
+            - context - event context; your derived classes may add variables
+                        to it (see `message.MessageTypeFilter` implementation
+                        for reference)
         
         Returns:
-                True if the event satisfies this filter
+            - True if the event satisfies this filter
-                False if the event does not satisfy this filter
+            - False if the event does not satisfy this filter
         """
-        pass
+        return True
+
+class EventTypeFilter(BaseEventFilter):
+    """Event filter that checks if the event is an instance of some class"""
+    def __init__(self, type: Type, **kwargs):
+        super().__init__(**kwargs)
+        self._type = type
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        return isinstance(context.event, self._type)
 
 class CompoundEventFilter(BaseEventFilter):
     """Event filter that consists of multiple filters"""
@@ -101,8 +122,8 @@ class CompoundEventFilter(BaseEventFilter):
             CompoundEventFilter.OPERATOR_INVERT: [1],
         }[op]
 
-    def __init__(self, operator: str, arguments: list[BaseEventFilter]):
+    def __init__(self, operator: str, arguments: list[BaseEventFilter], **kwargs):
-        super().__init__()
+        super().__init__(**kwargs)
         if not self._is_operator_valid(operator):
             raise RuntimeError(f"Invalid operator `{operator}`")
         if not self._is_elements_count_valid(operator, len(arguments)):
@@ -123,8 +144,10 @@ class CompoundEventFilter(BaseEventFilter):
             expression = f"~{reprs[0]}"
         return f"({expression})"
 
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
+    async def __call__(self, context: EventContext) -> bool:
-        evaluated = [arg(room, event) for arg in self._arguments]
+        if not await super().__call__(context):
+            return False
+        evaluated = [await arg(context) for arg in self._arguments]
         if self._operator == self.OPERATOR_AND:
             return all(evaluated)
         elif self._operator == self.OPERATOR_OR:

src/mab/filters/body.py

@@ -0,0 +1,209 @@
+import re
+import traceback
+from .message import NewMessageFilter
+from ..context import EventContext
+from ..context import (
+    CTX_BODY,
+    CTX_CMD_PREFIX,
+    CTX_CMD_VERB,
+    CTX_CMD_ARGS
+)
+
+from nio import AsyncClient
+from nio import MatrixRoom, Event
+
+class BodyExistsFilter(NewMessageFilter):
+    """
+    This filter returns True if all conditions are met:
+    1. `event` has attribute `body`
+    2. `event.body` is instance of `str`
+    3. `event.body.strip()` evaluates to True
+
+    If this filter matches, you can access `event.body` and it stores
+    unformatted text of the message.
+
+    If `event.body` value equals to `event.source["content"]["filename"]` (if it
+    is present, of course) then this filter will not match it by default. You
+    may disable `ignore_filename_in_body` to disable this feature.
+
+    This filter will match any message that has `body` in it, including images,
+    videos, files, etc.
+
+    This filter sets `CTX_BODY` context variable.
+    """
+    def __init__(self, *, ignore_filename_in_body: bool = True, **kwargs):
+        super().__init__(**kwargs)
+        self._ignore_filename_in_body = ignore_filename_in_body
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        if not hasattr(context.event, "body"):
+            return False
+        if not isinstance(context.event.body, str): # type: ignore
+            return False
+        if not context.event.body.strip(): # type: ignore
+            return False
+        if self._ignore_filename_in_body:
+            content = context.event.source["content"]
+            if "filename" in content and content["filename"] == context.event.body: # type: ignore
+                return False
+        context[CTX_BODY] = context.event.body # type: ignore
+        return True
+
+class BodyContainsFilter(BodyExistsFilter):
+    """
+    This filter returns True if `event.body` contains `needle` substring (or any
+    of neddle from the list). `event.body` will be converted to lower case if
+    `any_case` is True.
+
+    `needle` list is stored by reference so you can dynamically edit behavior of
+    this filter.
+
+    Please note that only `event.body` is converted to lower case if `any_case`
+    is True. That means you must ensure that `needle` is lower case. The filter
+    will never match otherwise.
+    """
+    def __init__(self, needle: str | list[str], *, any_case: bool = True, **kwargs):
+        super().__init__(**kwargs)
+        if type(needle) is str:
+            needle = [needle]
+        self._any_case = any_case
+        self._needle = needle
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        body = context.event.body.lower() if self._any_case else context.event.body # type: ignore
+        for n in self._needle:
+            if n in body:
+                return True
+        return False
+
+class BodyStartsWithFilter(BodyExistsFilter):
+    """
+    This filter returns True if `event.body` starts with `substring` (or any of
+    substrings from the list). The check will be case insensetive if `any_case`
+    is True.
+
+    `substring` list is stored by reference so you can dynamically edit behavior
+    of this filter.
+
+    Please note that only `event.body` is converted to lower case if `any_case`
+    is True. That means you must ensure that `needle` is lower case. The filter
+    will never match otherwise.
+    """
+    def __init__(self, substring: str | list[str], *, any_case: bool = True, **kwargs):
+        super().__init__(**kwargs)
+        if type(substring) is str:
+            substring = [substring]
+        self._any_case = any_case
+        self._substring = substring
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        body = context.event.body.lower() if self._any_case else context.event.body # type: ignore
+        for s in self._substring:
+            if body.startswith(s):
+                return True
+        return False
+        
+class BodyEndsWithFilter(BodyExistsFilter):
+    """
+    This filter returns True if `event.body` ends with `substring` (or any of
+    substrings from the list). The check will be case insensetive if `any_case`
+    is True.
+
+    `substring` list is stored by reference so you can dynamically edit behavior
+    of this filter.
+
+    Please note that only `event.body` is converted to lower case if `any_case`
+    is True. That means you must ensure that `needle` is lower case. The filter
+    will never match otherwise.
+    """
+    def __init__(self, substring: str | list[str], *, any_case: bool = True, **kwargs):
+        super().__init__(**kwargs)
+        if type(substring) is str:
+            substring = [substring]
+        self._any_case = any_case
+        self._substring = substring
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        body = context.event.body.lower() if self._any_case else context.event.body # type: ignore
+        for s in self._substring:
+            if body.endswith(s):
+                return True
+        return False
+
+class BodyCommandFilter(BodyExistsFilter):
+    """
+    This filter returns True if all conditions are met:
+    1. `event.body` contains at least `min_args + 1` words after split()
+    2. `event.body` contains at most `max_args + 1` words after split()
+    3. First element of splitted `event.body` starts with `prefix`
+    4. First element of splitted `event.body` (after lstripping `prefix`) starts
+       with any of strings in `verbs` list
+    
+    `verbs` list is stored by reference so you can dynamically edit behavior of
+    this filter.
+    
+    Please note that prefix is checked case sensetively. However, event.body is
+    converted to lower case when `verbs` matching is performed. So you must
+    store all verbs in lower case. This filter will not match any verbs that
+    use mixed case of upper case.
+
+    This filter sets the following context variables:
+    - `CTX_CMD_PREFIX` - prefix that was used
+    - `CTX_CMD_VERB` - verb that was used
+    - `CTX_CMD_ARGS` - arguments that were passed
+    """
+    def __init__(self, verbs: str | list[str], min_args: int = 0, max_args: int | None = None, prefix: str = "!", **kwargs):
+        super().__init__(**kwargs)
+        if type(verbs) is str:
+            verbs = [verbs]
+        self._verbs = verbs
+        self._min_args = min_args
+        self._max_args = max_args
+        self._prefix = prefix
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        parts = [p.strip() for p in context.event.body.split() if p.strip()] # type: ignore
+        args_count = len(parts) - 1
+        if args_count < self._min_args:
+            return False
+        if self._max_args is not None and args_count > self._max_args:
+            return False
+        if not parts[0].startswith(self._prefix):
+            return False
+        cmd = parts[0][len(self._prefix):].lower()
+        for verb in self._verbs:
+            if cmd == verb:
+                context[CTX_CMD_PREFIX] = self._prefix
+                context[CTX_CMD_VERB] = verb
+                context[CTX_CMD_ARGS] = parts[1:]
+                return True
+        return False
+
+class BodyRegexFilter(BodyExistsFilter):
+    """
+    This filter returns True if the `event.body` passes the regex.
+    """
+    def __init__(self, regex: re.Pattern | str, **kwargs):
+        super().__init__(**kwargs)
+        if isinstance(regex, str):
+            regex = re.compile(regex)
+        self._regex = regex
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        try:
+            return self._regex.match(context.event.body) is not None # type: ignore
+        except:
+            self._logger.error(traceback.format_exc())
+            return False

src/mab/filters/message.py

@@ -0,0 +1,121 @@
+import traceback
+from .base import BaseEventFilter, EventTypeFilter
+from ..types import MessageType
+from ..context import EventContext, CTX_MESSAGE_TYPE, CTX_SENDER
+
+from nio import AsyncClient
+from nio import MatrixRoom, Event
+
+from nio import RedactionEvent
+
+class MessageTypeFilter(BaseEventFilter):
+    """
+    This filter should be used to match specific message types (text-only,
+    images, videos, files, etc) based on `event.source["content"]["msgtype"]`
+    value.
+
+    `types` list is stored by reference so you may modify the behavior of this
+    filter dynamically.
+
+    This filter sets `CTX_MESSAGE_TYPE` variable in the context.
+    """
+    def __init__(self, types: list[MessageType] | MessageType, **kwargs):
+        super().__init__(**kwargs)
+        if isinstance(types, MessageType):
+            types = [types]
+        self._types = types
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        if "msgtype" not in context.event.source["content"]:
+            return False
+        msgtype = context.event.source["content"]["msgtype"]
+        if not msgtype in [t.value for t in self._types]:
+            return False
+        context[CTX_MESSAGE_TYPE] = MessageType(msgtype)
+        return True
+
+class NewMessageFilter(BaseEventFilter):
+    """
+    This filter returns True if the event is a new message. Most filters are
+    derived from this base class because it ignores events about edited
+    messages.
+    """
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        return "m.new_content" not in context.event.source["content"]
+
+class EditedMessageFilter(BaseEventFilter):
+    """
+    This filter returns True if the event is an edited message. You may use this
+    filter to create callbacks that are called if the message gets edited.
+    """
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        return "m.new_content" in context.event.source["content"]
+
+class RedactedMessageFilter(EventTypeFilter):
+    """
+    This filter returns True if the event is a RedactionEvent.
+    """
+    def __init__(self, **kwargs):
+        super().__init__(RedactionEvent, **kwargs)
+        raise NotImplementedError()
+
+    async def __call__(self, context: EventContext) -> bool:
+        raise NotImplementedError()
+
+class SenderIsFilter(BaseEventFilter):
+    """
+    This filter returns True if `event.sender` is any of specified senders.
+    `event.sender` is converted to lower case if `any_case` is True (default).
+    Supplied sender list is NEVER converted to lower case, so it is your duty to
+    use lower case if `any_case` is True.
+
+    `senders` list is stored by reference so you can modify behavior of this
+    filter dynamically.
+
+    This filter sets `CTX_SENDER` variable in the context.
+    """
+    def __init__(self, sender: list[str] | str, *, any_case: bool = True, **kwargs):
+        super().__init__(**kwargs)
+        if isinstance(sender, str):
+            sender = [sender]
+        self._sender = sender
+        self._any_case = any_case
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        sender = context.event.sender.lower() if self._any_case else context.event.sender
+        for s in self._sender:
+            if sender == s:
+                context[CTX_SENDER] = sender
+                return True
+        return False
+
+class SenderIsBotFilter(BaseEventFilter):
+    """
+    This filter returns True if `event.sender` is the client that has received
+    the event. You may use this filter to set callbacks for messages sent by
+    other users by using the following syntax:
+    ```py
+    ~SenderIsBotFilter()
+    ```
+    """
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        return context.bot.get_client().user_id == context.event.sender

src/mab/filters/room.py

@@ -0,0 +1,19 @@
+from .base import BaseEventFilter
+
+from ..context import EventContext, CTX_ROOM_ENCRYPTED
+
+class RoomEncryptedFilter(BaseEventFilter):
+    """
+    This filter returns True if the room is encrypted.
+    """
+    def __init__(self):
+        super().__init__()
+
+    async def __call__(self, context: EventContext) -> bool:
+        if not await super().__call__(context):
+            return False
+        try:
+            context[CTX_ROOM_ENCRYPTED] = context.room.encrypted
+            return context.room.encrypted
+        except:
+            return False

src/mab/filters/text.py

@@ -1,164 +0,0 @@
-from .base import BaseEventFilter
-
-from nio import MatrixRoom, Event
-
-class TextFilter(BaseEventFilter):
-    """
-        This filter returns True if the event contains `body` attribute.
-        `body` attribute contains unformatted text, string.
-    """
-    def __init__(self):
-        super().__init__()
-
-    def __repr__(self):
-        return "TextFilter()"
-
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        return hasattr(event, "body") and type(event.body) is str # type: ignore
-
-class FormattedTextFilter(BaseEventFilter):
-    """
-        This filter returns True if the event contains valid `formatted_body`
-        attribute. `formatted_body` attribute contains formatted text, string.
-    """
-    def __init__(self):
-        super().__init__()
-
-    def __repr__(self):
-        return "FormattedTextFilter()"
-
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        return hasattr(event, "formatted_body") and type(event.formatted_body) is str # type: ignore
-
-class TextContainsFilter(BaseEventFilter):
-    """
-        This filter returns True if the `event.body` contains `needle`
-        substring (or any of neddle from the list). The check will be case
-        insensetive if `any_case` is True.
-    """
-    def __init__(self, needle: str | list[str], *, any_case: bool = True):
-        super().__init__()
-        if type(needle) is str:
-            needle = [needle]
-        self._any_case = any_case
-        if self._any_case:
-            self._needle = [s.lower() for s in needle]
-        else:
-            self._needle = list(needle)
-
-    def __repr__(self):
-        return f"TextContainsFilter({repr(self._needle)}, any_case={repr(self._any_case)})"
-
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        try:
-            body = event.body.lower() if self._any_case else event.body # type: ignore
-            for n in self._needle:
-                if n in body:
-                    return True
-            return False
-        except:
-            return False
-
-class TextStartsWithFilter(BaseEventFilter):
-    """
-        This filter returns True if the `event.body` starts with `substring` (or
-        any of substrings from the list). The check will be case insensetive if
-        `any_case` is True.
-    """
-    def __init__(self, substring: str | list[str], *, any_case: bool = True):
-        super().__init__()
-        if type(substring) is str:
-            substring = [substring]
-        self._any_case = any_case
-        if self._any_case:
-            self._substring = [s.lower() for s in substring]
-        else:
-            self._substring = list(substring)
-
-    def __repr__(self):
-        return f"TextStartsWithFilter({repr(self._substring)}, any_case={repr(self._any_case)})"
-
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        try:
-            body = event.body.lower() if self._any_case else event.body # type: ignore
-            for s in self._substring:
-                if body.startswith(s):
-                    return True
-            return False
-        except:
-            return False
-        
-class TextEndsWithFilter(BaseEventFilter):
-    """
-        This filter returns True if the `event.body` ends with `substring` (or
-        any of substrings from the list). The check will be case insensetive if
-        `any_case` is True.
-    """
-    def __init__(self, substring: str | list[str], *, any_case: bool = True):
-        super().__init__()
-        if type(substring) is str:
-            substring = [substring]
-        self._any_case = any_case
-        if self._any_case:
-            self._substring = [s.lower() for s in substring]
-        else:
-            self._substring = list(substring)
-
-    def __repr__(self):
-        return f"TextEndsWithFilter({repr(self._substring)}, any_case={repr(self._any_case)})"
-
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        try:
-            body = event.body.lower() if self._any_case else event.body # type: ignore
-            for s in self._substring:
-                if body.endswith(s):
-                    return True
-            return False
-        except:
-            return False
-
-class TextCommandFilter(BaseEventFilter):
-    """
-        This filter returns True if all conditions are True:
-        1. `event.body` contains at least `min_args + 1` words after split()
-        2. `event.body` conrains at most `max_args + 1` words after split()
-        3. First element of splitted `event.body` starts with `prefix`
-        4. First element of splitted `event.body` (after stripping `prefix`)
-           starts with any of strings in `verbs` list (case-insensitive)
-        
-        Remarks:
-        - If this filter is satified, then it will set a new attribute for the
-          event: `event.command_args: list[str]`. You may use this attribute in
-          your callback for this event.
-    """
-    def __init__(self, verbs: str | list[str], min_args: int = 0, max_args: int | None = None, prefix: str = "!"):
-        super().__init__()
-        if type(verbs) is str:
-            verbs = [verbs]
-        verbs = [v.lower() for v in verbs]
-        self._verbs = verbs
-        self._min_args = min_args
-        self._max_args = max_args
-        self._prefix = prefix
-
-    def __repr__(self):
-        return f"TextCommandFilter({repr(self._verbs)}, {repr(self._min_args)}, {repr(self._max_args)}, {repr(self._prefix)})"
-
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        try:
-            parts = [p.strip() for p in event.body.split() if p.strip()] # type: ignore
-            args_count = len(parts) - 1
-            if args_count < self._min_args:
-                return False
-            if self._max_args is not None and args_count > self._max_args:
-                return False
-            if not parts[0].startswith(self._prefix):
-                return False
-            cmd = parts[0][len(self._prefix):].lower()
-            for verb in self._verbs:
-                if cmd == verb:
-                    setattr(event, "command_args", parts[1:])
-                    return True
-            return False
-        except:
-            return False

src/mab/types.py

@@ -2,6 +2,9 @@
 
 from pathlib import Path
 from dataclasses import dataclass
+from enum import Enum
+
+from nio import UploadResponse
 
 @dataclass
 class MatrixBotConfig:
@@ -13,7 +16,7 @@ class MatrixBotConfig:
     matrix_username_localpart: str
     """Localpart of matrix username (without homeserver), for example: `valid-username`"""
 
-    storage_directory: Path
+    storage_directory: Path | str
     """Path to the storage directory (will be created if needed)"""
 
     auto_verify_all_known_devices: bool = True
@@ -28,6 +31,20 @@ class MatrixBotConfig:
     auto_ffmpeg: bool = True
     """Enabled automatic `ffmpeg` execution to generate some media (e.g. thumbnails)"""
 
+    allow_ainput_password: bool = True
+    """Allow `aioconsole.ainput` usage to get user password"""
+
+    client_sync_period_ms: int = 5000
+    """Sync period for `nio.AsyncClient` in milliseconds"""
+
+    client_next_batch_dump_period_s: float = 120.0
+    """`next_batch` value dumping period in seconds"""
+
+    enable_multitasking_callbacks: bool = True
+    """Filter-based callbacks will be started each in separate task so that
+       long callback execution will not prevent other callbacks from executting
+    """
+
 @dataclass
 class VideoFileProperties:
     """Properties of the video for sending"""
@@ -43,3 +60,46 @@ class VideoFileProperties:
 
     thumbnail: Path | str | bytes | None = None
     """Path to the thumbnail or the raw JPEG thumbnail data"""
+
+@dataclass
+class UploadResult:
+    """Result of data upload"""
+
+    response: UploadResponse
+    """Response, as returned by `AsyncClient.upload(...)`"""
+
+    keys: dict
+    """Encryptions keys, as returned by `AsyncClient.upload(...)`"""
+
+    filename: str
+    """Name of the uploaded file"""
+
+    filesize: int
+    """Size of uploaded file"""
+
+class MessageType(Enum):
+    TEXT = "m.text"
+    EMOTE = "m.emote"
+    NOTICE = "m.notice"
+    IMAGE = "m.image"
+    FILE = "m.file"
+    AUDIO = "m.audio"
+    LOCATION = "m.location"
+    VIDEO = "m.video"
+
+class ContextDataKey[T]:
+    """
+    Instances of this class represent a single possible data key that can be
+    stored inside EventContext.
+    """
+    def __init__(self, name: str) -> None:
+        """
+        Initialize a ContextDataKey
+        
+        Args:
+            - name - name that will be used internally
+        """
+        self._name = name
+
+    def __repr__(self) -> str:
+        return f"ContextDataKey[{type(T)}]({repr(self._name)})"

src/mab/utils.py

@@ -3,6 +3,9 @@
 import asyncio
 import json
 from pathlib import Path
+from urllib.parse import urlparse
+
+from nio import AsyncClient
 
 from .types import *
 
@@ -96,3 +99,35 @@ class Utils:
             raise RuntimeError(f"`ffmpeg` exits with code {proc.returncode}: {stderr.decode(errors="replace")}", args)
         # success
         return stdout
+
+    @staticmethod
+    def build_full_matrix_username(config: MatrixBotConfig) -> str:
+        """Build complete matrix username using data from MatrixBotConfig."""
+        homeserver_name = urlparse(config.matrix_homeserver_url).hostname
+        localpart = config.matrix_username_localpart
+        return f"@{localpart}:{homeserver_name}"
+
+    @staticmethod
+    def verify_all_known_devices(client: AsyncClient) -> bool:
+        """
+        Verify all known devices.
+
+        Args:
+            - client - client to verify all known devices for
+        
+        Returns:
+            - True if there were unverified devices that are verified now
+        """
+        result = False
+        for user_id in client.device_store.users:
+            for device_id, olm_device in client.device_store[user_id].items():
+                # can't trust ourselves
+                if device_id == client.device_id and user_id == client.user_id:
+                    continue
+                # they are already verified
+                if olm_device.verified:
+                    continue
+                # verify them
+                client.verify_device(olm_device)
+                result = True
+        return result