Fixed BodyRegexFilter returning re.Match

- Improved README.md
- Added examples
- Added `session_storage/` to .gitignore so that examples do not introduce leftover files after execution

.gitignore

@@ -1,4 +1,6 @@
 __pycache__/
+session_storage/
+*.vscode
 .venv/
 dist/
 *.egg-info/

README.md

@@ -1,58 +1,61 @@
-# 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`)**
+
+## 🚀 Usage
+
+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.4.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
+Please inspect [`examples/shell_bot.py`](examples/shell_bot.py),
+[`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
+## 🏷️ Versioning
-import asyncio
-from mab import MatrixBot, MatrixBotConfig
-from nio import MatrixRoom, MatrixMessageText
 
-bot: MatrixBot
+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.
 
-async def on_room_message_text(room: MatrixRoom, event: RoomMessageText) -> None:
+## 🛠️ Development
-    global bot
-    # do not respond to ourselves
-    if event.sender == bot.get_client().user_id:
-        return
-    text = f"You message contains {len(event.body)} symbols"
-    await bot.send_text_to_room(room.room_id, text)
 
-
+Here's the list of commands you should execute to get started with development
-async def main() -> None:
+(including cloning the repository and installing required packages). Please note
-    global bot
+that your workflow may use something other than `venv`.
-    # create and start the bot
+```bash
-    cfg = MatrixBotConfig(
+apt install libmagic1-dev libolm-dev
-        matrix_homeserver_url="matrix.domain.su",
+git clone https://git.tyukalov.su/nikita/mab
-        matrix_username_localpart="nagibator666",
+cd mab
-        storage_directory=Path("storage_nagibator666")
+python3 -m venv .venv
-    )
+. .venv/bin/activate
-    bot = MatrixBot(matrix_bot_config)
+pip install -e .
-    bot.start()
-    # wait for Ctrl+C
-    try:
-        while True:
-            await asyncio.sleep(1)
-    except:
-        pass
-    # stop the bot
-    await bot.stop()
-
-if __name__ == "__main__":
-    asyncio.run(main())
 ```

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/echo_bot.py

@@ -0,0 +1,51 @@
+"""
+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 (
+    MatrixBot,
+    MatrixBotConfig,
+    RoomEventData,
+    BodyExistsFilter,
+    MessageTypeFilter,
+    SenderIsBotFilter,
+    MessageType
+)
+
+from _environment import check_environment
+
+async def on_text_message(data: RoomEventData) -> None:
+    """This callback is called when a text message arrives."""
+    await data.bot.send_text(data.room, data.event.body) # type: ignore
+
+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,103 @@
+"""
+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 (
+    MatrixBot,
+    MatrixBotConfig,
+    RoomEventData,
+    MessageTypeFilter,
+    BodyCommandFilter,
+    SenderIsBotFilter,
+    MessageType
+)
+
+from _environment import check_environment
+
+async def on_gen_command(data: RoomEventData) -> 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.event.command_args] # type: ignore
+    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: RoomEventData) -> 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.0.2"
+version = "0.4.0"
 authors = [
     { name = "Tyukalov Nikita", email = "nikita@tyukalov.su" }
 ]
@@ -47,6 +47,8 @@ dependencies = [
     "unpaddedbase64==2.1.0",
     "vodozemac==0.10.0",
     "yarl==1.24.5",
+    "python-magic==0.4.27",
+    "pillow==12.3.0"
 ]
 
 [project.urls]

src/mab/__init__.py

@@ -1,10 +1,14 @@
 from . import bot
 from . import types
 
-from .types import MatrixBotConfig
+from .types import MatrixBotConfig, RoomEventData, MessageType
 
 from .bot import MatrixBot
 
+from .filters.base import *
+from .filters.message import *
+from .filters.body import *
+
 __all__ = [
     # module names
     "bot",
@@ -12,7 +16,29 @@ __all__ = [
 
     # .types
     "MatrixBotConfig",
+    "RoomEventData",
+    "MessageType",
 
     # .bot
-    "MatrixBot"
+    "MatrixBot",
+
+    # .filters.base
+    "BaseEventFilter",
+    "EventTypeFilter",
+
+    # .filters.body
+    "BodyExistsFilter",
+    "BodyContainsFilter",
+    "BodyStartsWithFilter",
+    "BodyEndsWithFilter",
+    "BodyCommandFilter",
+    "BodyRegexFilter",
+
+    # .filters.message
+    "MessageTypeFilter",
+    "NewMessageFilter",
+    "EditedMessageFilter",
+    "RedactedMessageFilter",
+    "SenderIsFilter",
+    "SenderIsBotFilter",
 ]

src/mab/bot.py

@@ -1,514 +0,0 @@
-import asyncio
-import aiofiles
-import aioconsole
-import traceback
-import logging
-import time
-import json
-import os
-import re
-from html.parser import HTMLParser
-from urllib.parse import urlparse
-from typing import Any, Callable, Awaitable
-
-from nio import AsyncClient, AsyncClientConfig, SyncResponse
-from nio import LoginResponse, LoginError, WhoamiResponse, WhoamiError
-
-from nio import RoomSendResponse, RoomSendError
-
-from nio import OlmUnverifiedDeviceError
-
-from nio import MatrixInvitedRoom, InviteMemberEvent
-from nio import JoinResponse
-
-import nio.events
-
-from .types import *
-
-
-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 _check_and_remove_html(text_to_check: str) -> tuple[bool, str]:
-        """Checks if `text_to_check` is HTML and sanitizes it.
-
-        Returns:
-        tuple[bool, str] where the first element is True if `text_to_check` contains
-        valid HTML, and the second element is text without HTML (or just copy of
-        `text_to_check` if it does not contain HTML)
-        """
-        has_tags = False
-        text_fragments = []
-        class Extractor(HTMLParser):
-            def handle_starttag(self, tag, attrs):
-                nonlocal has_tags
-                has_tags = True
-            def handle_data(self, data):
-                text_fragments.append(data)
-        parser = Extractor(convert_charrefs=True)
-        parser.feed(text_to_check)
-        try:
-            if has_tags:
-                return (True, " ".join("".join(text_fragments).split()))
-        except:
-            traceback.print_exc()
-        return (False, text_to_check)
-
-    @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
-
-    #
-    # 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())
-
-    #
-    # LIFECYCLE
-    #
-    def _setup_client_callbacks(self) -> None:
-        """Setup internal client callbacks"""
-        self._client.add_response_callback(self._callback_sync, SyncResponse) # 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
-        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
-
-    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[[Any], Awaitable[None]] | None, event_class: nio.events.Event) -> None:
-        """Added event callback for events of specified class.
-            Use `None` instead of callback to print parameter types you need to use in your callback."""
-        if callback is None:
-            callback = self._debug_event_callback
-        self._client.add_event_callback(callback, event_class) # type: ignore
-
-    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, **kwargs) -> 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.
-        """
-        original_text = text
-        # guess if text is HTML
-        if is_html is None:
-            is_html, text = self._check_and_remove_html(text)
-        # text is HTML
-        elif is_html:
-            _, text = self._check_and_remove_html(text)
-        # create `content` for `room_send()`
-        content = {
-            "msgtype": "m.text",
-            "body": text,
-            **kwargs
-        }
-        if is_html:
-            content["format"] = "org.matrix.custom.html"
-            content["formatted_body"] = original_text
-        # 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_text_to_room(room_id, original_text, is_html, **kwargs)
-            else:
-                raise
-        # success
-        if type(result) is RoomSendResponse:
-            return result.event_id
-        # error
-        elif type(result) is RoomSendError:
-            raise RuntimeError(result)
-        # unknown error
-        else:
-            raise RuntimeError("Unknown error has occured", result)

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, RoomEventData
+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[[RoomEventData], 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
+            try:
+                if not await callback_info.filter(room, event, self._client):
+                    continue
+            except asyncio.CancelledError:
+                raise
+            except:
+                self._logger.error(traceback.format_exc())
+                continue
+            event_data = RoomEventData(
+                room=room,
+                event=event,
+                filter=callback_info.filter,
+                bot=self._matrix_bot
+            )
+            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[[RoomEventData], 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,246 @@
+import asyncio
+import logging
+
+from typing import Callable, Coroutine, Any
+
+from nio import AsyncClient
+
+from ..filters.base import BaseEventFilter
+from ..types import *
+
+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[[RoomEventData], 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/filters/base.py

@@ -0,0 +1,156 @@
+from abc import ABC, abstractmethod
+import logging
+from typing import Any, Type
+
+from nio import AsyncClient
+from nio import MatrixRoom, Event
+
+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):
+            raise TypeError(f"BaseEventFilter can't be ANDed against {type(other)}")
+        return CompoundEventFilter(
+            CompoundEventFilter.OPERATOR_AND,
+            [self, other]
+        )
+
+    def __rand__(self, other):
+        return self.__and__(other)
+
+    # OR
+    def __or__(self, other):
+        if not isinstance(other, BaseEventFilter):
+            raise TypeError(f"BaseEventFilter can't be ORed against {type(other)}")
+        return CompoundEventFilter(
+            CompoundEventFilter.OPERATOR_OR,
+            [self, other]
+        )
+
+    def __ror__(self, other):
+        return self.__or__(other)
+
+    # XOR
+    def __xor__(self, other):
+        if not isinstance(other, BaseEventFilter):
+            raise TypeError(f"BaseEventFilter can't be XORed against {type(other)}")
+        return CompoundEventFilter(
+            CompoundEventFilter.OPERATOR_XOR,
+            [self, other]
+        )
+
+    def __rxor__(self, other):
+        return self.__xor__(other)
+
+    # INVERT
+    def __invert__(self):
+        return CompoundEventFilter(
+            CompoundEventFilter.OPERATOR_INVERT,
+            [self]
+        )
+
+    # PAYLOAD
+    def __repr__(self) -> str:
+        """
+        This method may be redefined in derived classes to improve debugging
+        experience.
+        """
+        return str(self.__class__.__name__)
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        """
+        This abstract method must be redefined in derived classes so that the
+        filter operates according to its description. This method must not raise
+        exceptions. In case of exception it should log it using `self._logger`
+        and return False
+        
+        Args:
+            - room - room the event has happened in
+            - event - the event to check againts this filter
+            - client - the client
+        
+        Returns:
+            - True if the event satisfies this filter
+            - False if the event does not satisfy this filter
+        """
+        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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        return isinstance(event, self._type)
+
+class CompoundEventFilter(BaseEventFilter):
+    """Event filter that consists of multiple filters"""
+    OPERATOR_AND = "and"
+    OPERATOR_OR = "or"
+    OPERATOR_XOR = "xor"
+    OPERATOR_INVERT = "invert"
+
+    @staticmethod
+    def _is_operator_valid(op: str) -> bool:
+        """Returns True if operator `op` is a valid operator"""
+        return op in [
+            CompoundEventFilter.OPERATOR_AND,
+            CompoundEventFilter.OPERATOR_OR,
+            CompoundEventFilter.OPERATOR_XOR,
+            CompoundEventFilter.OPERATOR_INVERT
+        ]
+
+    @staticmethod
+    def _is_elements_count_valid(op: str, count: int) -> bool:
+        """Returns True if operator `op` may take `count` arguments"""
+        return count in {
+            CompoundEventFilter.OPERATOR_AND: [2],
+            CompoundEventFilter.OPERATOR_OR: [2],
+            CompoundEventFilter.OPERATOR_XOR: [2],
+            CompoundEventFilter.OPERATOR_INVERT: [1],
+        }[op]
+
+    def __init__(self, operator: str, arguments: list[BaseEventFilter], **kwargs):
+        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)):
+            raise RuntimeError(f"Operator `{operator}` does not take `{len(arguments)}` arguments")
+        self._operator = operator
+        self._arguments = list(arguments)
+
+    def __repr__(self) -> str:
+        expression = "False"
+        reprs = [repr(a) for a in self._arguments]
+        if self._operator == CompoundEventFilter.OPERATOR_AND:
+            expression = " & ".join(reprs)
+        elif self._operator == CompoundEventFilter.OPERATOR_OR:
+            expression = " | ".join(reprs)
+        elif self._operator == CompoundEventFilter.OPERATOR_XOR:
+            expression = " ^ ".join(reprs)
+        elif self._operator == CompoundEventFilter.OPERATOR_INVERT:
+            expression = f"~{reprs[0]}"
+        return f"({expression})"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        evaluated = [await arg(room, event, client) for arg in self._arguments]
+        if self._operator == self.OPERATOR_AND:
+            return all(evaluated)
+        elif self._operator == self.OPERATOR_OR:
+            return any(evaluated)
+        elif self._operator == self.OPERATOR_XOR:
+            result = evaluated[0]
+            for v in evaluated[1:]:
+                result ^= v
+            return result
+        elif self._operator == self.OPERATOR_INVERT:
+            return not evaluated[0]
+        return False

src/mab/filters/body.py

@@ -0,0 +1,196 @@
+import re
+import traceback
+from .message import NewMessageFilter
+
+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.
+    """
+    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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        if not hasattr(event, "body"):
+            return False
+        if not isinstance(event.body, str): # type: ignore
+            return False
+        if not event.body.strip(): # type: ignore
+            return False
+        if self._ignore_filename_in_body:
+            content = event.source["content"]
+            if "filename" in content and content["filename"] == event.body: # type: ignore
+                return False
+        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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        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
+
+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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        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
+        
+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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        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
+
+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.
+
+    If this filter is matched, 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 = "!", **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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        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
+
+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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        try:
+            return self._regex.match(event.body) is not None # type: ignore
+        except:
+            self._logger.error(traceback.format_exc())
+            return False

src/mab/filters/message.py

@@ -0,0 +1,112 @@
+import traceback
+from .base import BaseEventFilter, EventTypeFilter
+from ..types import MessageType
+
+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.
+    """
+    def __init__(self, types: list[MessageType] | MessageType, **kwargs):
+        super().__init__(**kwargs)
+        if isinstance(types, MessageType):
+            types = [types]
+        self._types = types
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        if "msgtype" not in event.source["content"]:
+            return False
+        return (
+            event.source["content"]["msgtype"] in [t.value for t in self._types]
+        )
+
+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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        return "m.new_content" not in 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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        return "m.new_content" in event.source["content"]
+
+class RedactedMessageFilter(EventTypeFilter):
+    """
+    This filter returns True if the event is a RedactionEvent.
+    """
+    def __init__(self, **kwargs):
+        super().__init__(RedactionEvent, **kwargs)
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        return await super().__call__(room, event, client)
+
+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.
+    """
+    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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        sender = event.sender.lower() if self._any_case else event.sender
+        for s in self._sender:
+            if sender == s:
+                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, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not await super().__call__(room, event, client):
+            return False
+        return client.user_id == event.sender

src/mab/filters/room.py

@@ -0,0 +1,17 @@
+from .base import BaseEventFilter
+
+from nio import AsyncClient
+from nio import MatrixRoom, Event
+
+class RoomEncryptedFilter(BaseEventFilter):
+    """
+    This filter returns True if the room is encrypted.
+    """
+    def __init__(self):
+        super().__init__()
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            return room.encrypted
+        except:
+            return False

src/mab/types.py

@@ -2,6 +2,16 @@
 
 from pathlib import Path
 from dataclasses import dataclass
+from enum import Enum
+
+from nio import MatrixRoom, Event
+from nio import UploadResponse
+
+from .filters.base import BaseEventFilter
+
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from .bot import MatrixBot
 
 @dataclass
 class MatrixBotConfig:
@@ -13,7 +23,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
@@ -21,3 +31,81 @@ class MatrixBotConfig:
 
     auto_join_any_room_on_invite: bool = True
     """Enable automatic join for invites (implemented by automatically adding internal callback on bot creation)"""
+
+    auto_ffprobe: bool = True
+    """Enable automatic `ffprobe` execution to get some media properties"""
+
+    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"""
+
+    width_px: int
+    """Width in pixels"""
+
+    height_px: int
+    """Height in pixels"""
+
+    duration_ms: int
+    """Duration in milliseconds"""
+
+    thumbnail: Path | str | bytes | None = None
+    """Path to the thumbnail or the raw JPEG thumbnail data"""
+
+@dataclass
+class RoomEventData:
+    """Dataclass that hold information about 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"""
+
+    filter: BaseEventFilter
+    """The filter that invoked this event"""
+
+    bot: "MatrixBot"
+    """The bot that is the source of the event"""
+
+@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"

src/mab/utils.py

@@ -0,0 +1,133 @@
+"""Utility things"""
+
+import asyncio
+import json
+from pathlib import Path
+from urllib.parse import urlparse
+
+from nio import AsyncClient
+
+from .types import *
+
+class Utils:
+    @staticmethod
+    async def get_video_props_using_ffprobe(path: Path | str) -> VideoFileProperties:
+        """Returns video properties for file at `path`.
+
+        Consider using `asyncio.timeout` to prevent deadlocking.
+        """
+        # prepare arguments for ffprobe
+        args = [
+            "ffprobe",
+            "-v", "error",
+            "-select_streams", "v:0",
+            "-show_entries", "stream=width,height:format=duration",
+            "-of", "json",
+            str(path)
+        ]
+        # start the process
+        proc = await asyncio.create_subprocess_exec(
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        # communicate `stdout` and `stderr`
+        try:
+            stdout, stderr = await proc.communicate()
+        except asyncio.CancelledError:
+            if proc.returncode is None:
+                proc.kill()
+            await proc.wait()
+            raise
+        # error
+        if proc.returncode != 0:
+            raise RuntimeError(f"`ffprobe` exits with code {proc.returncode}: {stderr.decode(errors="replace")}")
+        # success
+        result = json.loads(stdout)
+        return VideoFileProperties(
+            width_px=int(result["streams"][0]["width"]),
+            height_px=int(result["streams"][0]["height"]),
+            duration_ms=int(float(result["format"]["duration"]) * 1000)
+        )
+
+    @staticmethod
+    async def get_video_thumbnail(path: Path | str, *, timestamp: float = 0, max_size: int = 300) -> bytes:
+        """Generates video thumbnail for file at `path`.
+
+            Args:
+                - path - path to the video
+                - timestamp - timestamp to use as the thumbnail
+                - max_size - max width/height
+            
+            Returns:
+                Raw bytes of JPEG thumbnail
+        """
+        # prepare arguments for ffmpeg
+        args = [
+            "ffmpeg",
+            "-v", "error",
+            "-ss", "%.2f" % timestamp,
+            "-i", str(path),
+            "-map", "0:v:0",
+            "-vf",
+            (
+                f"scale='min(iw,{max_size})':'min(ih,{max_size})':"
+                "force_original_aspect_ratio=decrease"
+            ),
+            "-frames:v", "1",
+            "-q:v", "3",
+            "-f", "image2pipe",
+            "-vcodec", "mjpeg",
+            "pipe:1"
+        ]
+        # start the process
+        proc = await asyncio.create_subprocess_exec(
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        # communicate `stdout` and `stderr`
+        try:
+            stdout, stderr = await proc.communicate()
+        except asyncio.CancelledError:
+            if proc.returncode is None:
+                proc.kill()
+            await proc.wait()
+            raise
+        # error
+        if proc.returncode != 0 or not stdout:
+            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