Fixed _callbacks.py did not reraise CancelledError

- **kwargs are propagated to base classes in text filters from now on
- `_callbacks.py` prints filter exceptions from now on

.gitignore

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

README.md

@@ -6,14 +6,24 @@ because I wasn't satisfied by simplicity and usage of other libraries. So
 this library does not aim to be "the best matrix bot library", it only aims to
 be good enough for me.
 
+## Features
+
+The package supports the following features:
+- **Filter-based callback system**
+- **Images sending**
+- **Videos sending with automatic thumbnail generation (requires `ffmpeg`)**
+
 ## Installation
 
 Use `pip` to install this package:
 
 ```bash
-python -m pip install git+https://git.tyukalov.su/nikita/mab
+python -m pip install git+https://git.tyukalov.su/nikita/mab@v0.3.0
 ```
 
+You should specify package version you want to use, because `main` without tags
+contain unstable code.
+
 ## Basic usage
 
 This is the most simple bot you can create. It would respond to any message
@@ -23,21 +33,16 @@ that starts with `!test`, `!hello` or `!hi`.
 import asyncio
 from mab import MatrixBot, MatrixBotConfig
 from mab import TextCommandFilter
+from mab.types import RoomEventData
 
-from nio import MatrixRoom, MatrixMessageText
-
-bot: MatrixBot
-
-async def on_valid_command(room: MatrixRoom, event: RoomMessageText) -> None:
-    global bot
+async def on_valid_command(data: RoomEventData) -> None:
     # do not respond to ourselves
-    if event.sender == bot.get_client().user_id:
+    if event.sender == data.bot.get_client().user_id:
         return
-    text = f"Your message contains {len(event.body)} symbols"
-    await bot.send_text_to_room(room.room_id, text)
+    text = f"Your message contains {len(data.event.body)} symbols"
+    await data.bot.send_text_to_room(data.room, text)
 
 async def main() -> None:
-    global bot
     # create and start the bot
     cfg = MatrixBotConfig(
         matrix_homeserver_url="matrix.domain.su",
@@ -45,11 +50,11 @@ async def main() -> None:
         storage_directory=Path("storage_nagibator666")
     )
     bot = MatrixBot(matrix_bot_config)
-    bot.add_event_callback(
-        on_valid_command,
-        TextCommandFilter(["test", "hello", "hi"])
+    bot.add_callback(
+        TextCommandFilter(["test", "hello", "hi"]),
+        on_valid_command
     )
-    bot.start()
+    await bot.start()
     # wait for Ctrl+C
     try:
         while True:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "mab"
-version = "0.1.0"
+version = "0.3.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/bot.py

@@ -1,537 +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
-
-from .filters.base import BaseEventFilter
-
-from nio.events.room_events import Event as RoomEvemt
-
-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())
-
-    async def _callback_filter_router(self, *args, **kwargs):
-        if len(args) != 2:
-            self._logger.debug("Can't process the event, not enough positional args")
-            self._debug_event_callback(*args, **kwargs)
-            return
-        room = args[0]
-        event = args[1]
-        for filter in self._filters:
-            filter_object = filter[0]
-            filter_callback = filter[1]
-            filter_stop_after_this = filter[2]
-            if filter_object(room, event):
-                self._logger.debug(f"Filter {repr(filter_object)} matched")
-                try:
-                    await filter_callback(room, event)
-                except:
-                    self._logger.error(traceback.format_exc())
-                if filter_stop_after_this:
-                    self._logger.debug(f"Filter {repr(filter_object)} stops matching")
-                    break
-
-    #
-    # LIFECYCLE
-    #
-    def _setup_client_callbacks(self) -> None:
-        """Setup internal client callbacks"""
-        self._client.add_response_callback(self._callback_sync, SyncResponse) # type: ignore
-        self._client.add_event_callback(self._callback_filter_router, RoomEvemt) # type: ignore
-        if self._config.auto_join_any_room_on_invite:
-            self._client.add_event_callback(self._callback_autojoin, InviteMemberEvent) # type: ignore
-
-    async def _client_login_session_data(self, session_data: dict[str, Any]) -> None:
-        """Login using session data. Raises and exception on failure."""
-        self._logger.debug("Using stored session data to log in")
-        # build user id
-        username = self._build_matrix_username(self._config)
-        self._client.restore_login(
-            user_id=username,
-            **session_data
-        )
-        self._client.load_store()
-        result = await self._client.whoami()
-        if type(result) is WhoamiError:
-            self._logger.error(f"Can't log in using stored session data: '{result.message}'")
-            raise RuntimeError(result.message)
-        elif type(result) is not WhoamiResponse:
-            self._logger.error("Can't log in using stored session data, unknown error")
-            raise RuntimeError("Unknown response for whoami request")
-        self._logger.debug("Logged in using stored session data")
-
-    async def _client_login_password(self) -> None:
-        """Login using password and save result to disk on success.
-            Raises an exception on failure.
-        """
-        self._logger.debug("Using password to log in")
-        # get the password
-        password = await self._cb_password()
-        result = await self._client.login(password=password)
-        if type(result) is LoginResponse:
-            self._logger.debug("Logged in using password")
-            await self._write_session_data(
-                access_token=result.access_token,
-                device_id=result.device_id
-            )
-        elif type(result) is LoginError:
-            self._logger.error(f"Can't log in using password: '{result.message}'")
-            raise RuntimeError(result.message)
-        else:
-            self._logger.error(f"Can't log in using password, unknown error")
-            raise RuntimeError("Unknown login result")
-
-    async def _client_login(self) -> None:
-        """This function logs in."""
-        # check if we have session data stored on the disk
-        session_data = await self._read_session_data()
-        # session data is present, try to log in
-        if session_data is not None:
-            self._logger.debug("Some session data found on the disk")
-            try:
-                await self._client_login_session_data(session_data)
-                return
-            except asyncio.CancelledError:
-                raise
-            except:
-                pass
-        # no session data - login using password
-        try:
-            self._logger.debug("No session data found on the disk OR invalid data")
-            await self._client_login_password()
-            return
-        except asyncio.CancelledError:
-            raise
-        except:
-            pass
-        # can't login
-        self._logger.error("Can't log in using available methods")
-        raise RuntimeError("All login methods have failed, can't continue")
-
-    async def _client_destroy(self) -> None:
-        """Gracefully destroys the client."""
-        try:
-            self._logger.debug("Closing the client")
-            await self._client.close()
-            if self._last_next_batch is not None:
-                self._logger.debug("Saving next_batch")
-                await self._write_next_batch(self._last_next_batch)
-        except:
-            traceback.print_exc()
-
-    async def _client_cancellable_sync_forever(self, *args, **kwargs) -> Any:
-        """Behaves exactly like AsyncClient.sync_forever, but supports task cancellation"""
-        sync_forever_task = asyncio.create_task(
-            self._client.sync_forever(*args, **kwargs)
-        )
-        try:
-            return await sync_forever_task
-        except asyncio.CancelledError:
-            try:
-                self._client.stop_sync_forever()
-            except Exception:
-                pass
-            finally:
-                sync_forever_task.cancel()
-                await asyncio.gather(sync_forever_task, return_exceptions=True)
-            raise
-
-    async def _background_coroutine(self) -> None:
-        """This function implements bot lifecycle."""
-        # we should stop when this task stops
-        self._logger.debug("_background_coroutine is started")
-        stop_wait_task = asyncio.create_task(self._stop_event.wait()) # type: ignore
-        # perform login
-        login_task = asyncio.create_task(self._client_login())
-        try:
-            await self._wait_for_task_and_stop_event(login_task, stop_wait_task)
-        except asyncio.CancelledError:
-            self._logger.debug("Background task is cancelled during login")
-            await self._client_destroy()
-            return
-        except:
-            traceback.print_exc()
-        self._logger.info("Succesfully logged in")
-        # sync forever
-        self._logger.info("Syncing forever")
-        sync_task = asyncio.create_task(
-            self._client_cancellable_sync_forever(
-                timeout=self.MATRIX_SYNC_PERIOD,
-                since=(await self._read_next_batch()),
-                full_state=True
-            )
-        )
-        try:
-            await self._wait_for_task_and_stop_event(sync_task, stop_wait_task)
-        except asyncio.CancelledError:
-            self._logger.debug("Sync task is cancelled")
-            await self._client_destroy()
-            return
-        except:
-            traceback.print_exc()
-
-
-    #
-    # PUBLIC
-    #
-    def __init__(self, config: MatrixBotConfig) -> None:
-        # check if config is valid
-        self._validate_bot_config(config) # may raise an Exception
-        # save the config
-        self._config: MatrixBotConfig = config
-        # create the logger
-        self._logger = logging.getLogger(self._build_matrix_username(config))
-        self._logger.setLevel(logging.DEBUG)
-        # create the client
-        self._client: AsyncClient = self._build_client(self._config)
-        self._setup_client_callbacks()
-
-        # prepare some private data
-        self._background_task: asyncio.Task | None = None
-        self._last_next_batch_dump: float = 0.0
-        self._last_next_batch: str | None = None
-        self._cb_password = self._default_password_callback
-        self._filters = []
-
-    def start(self) -> None:
-        """Start the bot.
-            Starts the bot in background task. Raises an exception if there are
-            problems (for example, the bot is already started). The bot will
-            do everything to keep itself running, including restarts. Use
-            `stop()` to stop the bot.
-        """
-        if self._background_task is not None:
-            raise RuntimeError("The bot is already started!")
-        self._stop_event = asyncio.Event()
-        self._background_task = asyncio.create_task(
-            self._background_coroutine()
-        )
-
-    async def stop(self) -> None:
-        """Stop the bot and wait for the bot stop."""
-        if self._background_task is None or self._stop_event is None:
-            return
-        self._stop_event.set()
-        try:
-            await self._background_task
-        except:
-            traceback.print_exc()
-        self._stop_event = None
-        self._background_task = None
-
-    def verify_all_known_devices(self) -> bool:
-        """Verifies all known devices.
-        
-        Returns:
-            True if there were unverified devices that are verified now.
-        """
-        result = False
-        for user_id in self._client.device_store.users:
-            for device_id, olm_device in self._client.device_store[user_id].items():
-                # can't trust ourselves
-                if device_id == self._client.device_id and user_id == self._client.user_id:
-                    continue
-                # they are already verified
-                if olm_device.verified:
-                    continue
-                # verify them
-                self._client.verify_device(olm_device)
-                result = True
-        return result
-
-    def add_event_callback(self, callback: Callable[[Any], Awaitable[None]] | None, filter: BaseEventFilter, stop_after_this: bool = True) -> None:
-        """Add event callback for events that pass the filter."""
-        if callback is None:
-            callback = self._debug_event_callback
-        self._filters.append((filter, callback, stop_after_this))
-
-    def get_client(self) -> AsyncClient:
-        """Get AsyncClient in use"""
-        return self._client
-
-    async def send_text_to_room(self, room_id: str, text: str, is_html: bool | None = None, **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,202 @@
+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:
+            asyncio.shield(self._background_task)
+        except:
+            self._logger.error(traceback.format_exc())
+        try:
+            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,300 @@
+import asyncio
+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_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,198 @@
+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()
+
+    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_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

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
 import logging
 
+from nio import AsyncClient
 from nio import MatrixRoom, Event
 
 class BaseEventFilter(ABC):
@@ -59,14 +60,16 @@ class BaseEventFilter(ABC):
         pass
 
     @abstractmethod
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
+    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
@@ -112,18 +115,19 @@ class CompoundEventFilter(BaseEventFilter):
 
     def __repr__(self) -> str:
         expression = "False"
+        reprs = [repr(a) for a in self._arguments]
         if self._operator == CompoundEventFilter.OPERATOR_AND:
-            expression = " & ".join(self._arguments)
+            expression = " & ".join(reprs)
         elif self._operator == CompoundEventFilter.OPERATOR_OR:
-            expression = " | ".join(self._arguments)
+            expression = " | ".join(reprs)
         elif self._operator == CompoundEventFilter.OPERATOR_XOR:
-            expression = " ^ ".join(self._arguments)
+            expression = " ^ ".join(reprs)
         elif self._operator == CompoundEventFilter.OPERATOR_INVERT:
-            expression = f"~{self._arguments[0]}"
+            expression = f"~{reprs[0]}"
         return f"({expression})"
 
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        evaluated = [arg(room, event) for arg in self._arguments]
+    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:

src/mab/filters/room.py

@@ -0,0 +1,107 @@
+from .base import BaseEventFilter
+
+from nio import AsyncClient
+from nio import MatrixRoom, Event
+
+class RoomIdContainsFilter(BaseEventFilter):
+    """
+        This filter returns True if the `room.room_id` contains `needle` (or
+        any of needles from the list). The check will be case insensetive if
+        `any_case` is True.
+    """
+    def __init__(self, needle: str | list[str], *, any_case: bool = True):
+        super().__init__()
+        if type(needle) is str:
+            needle = [needle]
+        self._any_case = any_case
+        if self._any_case:
+            self._needle = [s.lower() for s in needle]
+        else:
+            self._needle = list(needle)
+
+    def __repr__(self):
+        return f"RoomIdContainsFilter({repr(self._needle)}, any_case={repr(self._any_case)})"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            room_id = room.room_id.lower() if self._any_case else room.room_id
+            for s in self._needle:
+                if s in room_id:
+                    return True
+            return False
+        except:
+            return False
+
+class RoomIdStartsWithFilter(BaseEventFilter):
+    """
+        This filter returns True if the `room.room_id` starts with `substring`
+        (or any of substrings from the list). The check will be case insensetive
+        if `any_case` is True.
+    """
+    def __init__(self, substring: str | list[str], *, any_case: bool = True):
+        super().__init__()
+        if type(substring) is str:
+            substring = [substring]
+        self._any_case = any_case
+        if self._any_case:
+            self._substring = [s.lower() for s in substring]
+        else:
+            self._substring = list(substring)
+
+    def __repr__(self):
+        return f"RoomIdStartsWithFilter({repr(self._substring)}, any_case={repr(self._any_case)})"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            room_id = room.room_id.lower() if self._any_case else room.room_id
+            for s in self._substring:
+                if room_id.startswith(s):
+                    return True
+            return False
+        except:
+            return False
+
+class RoomIdEndsWithFilter(BaseEventFilter):
+    """
+        This filter returns True if the `room.room_id` ends with `substring` (or
+        any of substrings from the list). The check will be case insensetive
+        if `any_case` is True.
+    """
+    def __init__(self, substring: str | list[str], *, any_case: bool = True):
+        super().__init__()
+        if type(substring) is str:
+            substring = [substring]
+        self._any_case = any_case
+        if self._any_case:
+            self._substring = [s.lower() for s in substring]
+        else:
+            self._substring = list(substring)
+
+    def __repr__(self):
+        return f"RoomIdEndsWithFilter({repr(self._substring)}, any_case={repr(self._any_case)})"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            room_id = room.room_id.lower() if self._any_case else room.room_id
+            for s in self._substring:
+                if room_id.endswith(s):
+                    return True
+            return False
+        except:
+            return False
+
+class RoomEncryptedFilter(BaseEventFilter):
+    """
+    This filter returns True if the room is encrypted.
+    """
+    def __init__(self):
+        super().__init__()
+
+    def __repr__(self):
+        return f"RoomEncryptedFilter()"
+
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        try:
+            return room.encrypted
+        except:
+            return False

src/mab/filters/text.py

@@ -1,124 +1,190 @@
+import re
+import traceback
 from .base import BaseEventFilter
 
+from nio import AsyncClient
 from nio import MatrixRoom, Event
 
 class TextFilter(BaseEventFilter):
     """
-        This filter returns True if the event contains `body` attribute.
-        `body` attribute contains unformatted text, string.
+    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.
     """
-    def __init__(self):
-        super().__init__()
+    def __init__(self, *, ignore_filename_in_body: bool = True, **kwargs):
+        super().__init__(**kwargs)
+        self._ignore_filename_in_body = ignore_filename_in_body
 
-    def __repr__(self):
-        return "TextFilter"
+    def __repr__(self) -> str:
+        return "TextFilter()"
 
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        return hasattr(event, "body") and type(event.body) is str
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        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 FormattedTextFilter(BaseEventFilter):
     """
-        This filter returns True if the event contains valid `formatted_body`
-        attribute. `formatted_body` attribute contains formatted text, string.
+    This filter returns True if all conditions are met:
+    1. `event` has attribute `formatted_body`
+    2. `event.formatted_body` is instance of `str`
+    3. `event.formatted_body.strip()` evaluates to True
+
+    If this filter matches, you can access `event.formatted_body` and it stores
+    formatted text of the message.
     """
-    def __init__(self):
-        super().__init__()
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
 
     def __repr__(self):
-        return "FormattedTextFilter"
+        return "FormattedTextFilter()"
 
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        return hasattr(event, "formatted_body") and type(event.formatted_body) is str
+    async def __call__(self, room: MatrixRoom, event: Event, client: AsyncClient) -> bool:
+        if not hasattr(event, "formatted_body"):
+            return False
+        if not isinstance(event.formatted_body, str): # type: ignore
+            return False
+        if not event.formatted_body.strip(): # type: ignore
+            return False
+        return True
 
-class TextContainsFilter(BaseEventFilter):
+class TextContainsFilter(TextFilter):
     """
-        This filter returns True if the `event.body` contains `needle`
-        substring (or any of neddle from the list).
+    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]):
-        super().__init__()
+    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
 
-    def __repr__(self):
-        return f"TextContainsFilter({repr(self._neddle)})"
+    def __repr__(self) -> str:
+        return f"TextContainsFilter({repr(self._needle)}, any_case={repr(self._any_case)})"
 
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        try:
+    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 event.body:
+            if n in body:
                 return True
         return False
-        except:
-            return False
 
-class TextStartsWithFilter(BaseEventFilter):
+class TextStartsWithFilter(TextFilter):
     """
-        This filter returns True if the `event.body` starts with `substring` (or
-        any of substrings from the list).
+    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]):
-        super().__init__()
+    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
 
     def __repr__(self):
-        return f"TextStartsWithFilter({repr(self._substring)})"
+        return f"TextStartsWithFilter({repr(self._substring)}, any_case={repr(self._any_case)})"
 
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        try:
+    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 event.body.startswith(s):
+            if body.startswith(s):
                 return True
-            print(self._substring)
-            return False
-        except:
         return False
         
-class TextEndsWithFilter(BaseEventFilter):
+class TextEndsWithFilter(TextFilter):
     """
-        This filter returns True if the `event.body` ends with `substring` (or
-        any of substrings from the list).
+    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]):
-        super().__init__()
+    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
 
     def __repr__(self):
-        return f"TextEndsWithFilter({repr(self._substring)})"
+        return f"TextEndsWithFilter({repr(self._substring)}, any_case={repr(self._any_case)})"
 
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        try:
+    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 event.body.endswith(s):
+            if body.endswith(s):
                 return True
         return False
-        except:
-            return False
 
-class TextCommandFilter(BaseEventFilter):
+class TextCommandFilter(TextFilter):
     """
-        This filter returns True if all conditions are True:
+    This filter returns True if all conditions are met:
     1. `event.body` contains at least `min_args + 1` words after split()
-        2. `event.body` conrains at most `max_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 stripping `prefix`)
-           starts with any of strings in `verbs` list (case-insensitive)
+    4. First element of splitted `event.body` (after lstripping `prefix`) starts
+       with any of strings in `verbs` list
     
-        Remarks:
-        - If this filter is satified, then it will set a new attribute for the
-          event: `event.command_args: list[str]`. You may use this attribute in
-          your callback for this event.
+    `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 = "!"):
-        super().__init__()
+    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]
-        verbs = [v.lower() for v in verbs]
         self._verbs = verbs
         self._min_args = min_args
         self._max_args = max_args
@@ -127,9 +193,10 @@ class TextCommandFilter(BaseEventFilter):
     def __repr__(self):
         return f"TextCommandFilter({repr(self._verbs)}, {repr(self._min_args)}, {repr(self._max_args)}, {repr(self._prefix)})"
 
-    def __call__(self, room: MatrixRoom, event: Event) -> bool:
-        try:
-            parts = [p.strip() for p in event.body.split() if p.strip()]
+    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
@@ -143,5 +210,25 @@ class TextCommandFilter(BaseEventFilter):
                 setattr(event, "command_args", parts[1:])
                 return True
         return False
-        except:
+
+class TextRegexFilter(TextFilter):
+    """
+    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
+
+    def __repr__(self):
+        return f"TextRegexFilter({repr(self._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) # type: ignore
+        except:
+            self._logger.error(traceback.format_exc())
             return False

src/mab/types.py

@@ -3,6 +3,15 @@
 from pathlib import Path
 from dataclasses import dataclass
 
+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:
     """Configuration for MatrixBot"""
@@ -13,7 +22,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 +30,71 @@ 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"""

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