📁 Структура кода

Основные директории для понимания архитектуры проекта:

🔒 Безопасность в Sekai

Sekai — это не сейф, но у него есть трёхслойная защита.

1. Pre-Load Scan (Static Analysis)

При загрузке community модуля его исходный код сканируется через AST до компиляции.

Чёрный список

• Session control: login, logout, stop, start
• Sensitive data: crypto, api, device_id
• Sneaky tricks: обход через getattr(client, "crypto")

2. Runtime Firewall

Если модуль прошёл сканирование, он всё равно наблюдается в реальном времени.

Блокировка в реальном времени:

• Запись файлов — можно читать, но нельзя писать/удалять
• Core hijacking — community модули не могут импортировать Core
• Memory hacks ctypes запрещены

Золотое правило

Устанавливайте только модули от людей, которым доверяете. Читайте код, если сомневаетесь.

🔧 Справочник core/utils.py

Все публичные функции файла src/mxuserbot/core/utils.py.

Что помнить на практике

• answer() по умолчанию редактирует текущую команду
• Для новых модулей аргументы лучше разбирать через Pydantic payload
• request() возвращает None при ошибке
• send_image() — строгий helper, кидает исключение при отсутствии данных

Общие helpers

get_platform() → str

• Назначение: сбор информации о системе
• Возвращает: HTML-строку с hostname, ОС, RAM, CPU
• Caveats: строка содержит HTML-теги

get_commands(cls) → dict

• Назначение: поиск методов с @loader.command()
• Возвращает: словарь {command_name: function}

Работа с reply и аргументами

await get_reply_text(mx, event) → str | None | bool

• Достаёт текст из reply с автодешифровкой
• Возвращает: False (не reply), None (ошибка), str (успех)

await get_args_raw(mx, event) → str

• Достаёт аргументы как "сырой" текст
• При отсутствии аргументов подтягивает текст из reply

await get_args(mx, event) → list

• Достаёт аргументы списком
• Парсит через shlex.split()

Отправка сообщений

await answer(mx, text, html=True, room_id=None, event=None, edit_id=-1, **kwargs)

• Отправляет или редактирует сообщение
• edit_id=-1 = попробовать редактировать текущую команду
• edit_id=None = отправить новое сообщение

await send_image(mx, room_id, url=None, file_bytes=None, ...)

• Отправляет картинку по URL, mxc:// или bytes
• При шифровании комнаты использует encrypt_attachment()
• Без bytes кидает ValueError

HTTP и RPC

await request(url, method="GET", return_type="json", ...)

• Универсальный HTTP helper поверх aiohttp
• На ошибке возвращает None
• Создаёт новую ClientSession на каждый вызов

await set_rpc_media(mx, artist, album, track, ...)

• Ставит статус rich presence m.rpc.media
• Автоматически загружает обложку в Matrix store

await clear_rpc(mx)

• Полностью удаляет RPC-статус из профиля

Экранирование

escape_html(text) → str

Экранирует &, <, >

escape_quotes(text) → str

Экранирует HTML + двойные кавычки для атрибутов

🚀 Разработка модулей

Если вы впервые открыли проект и мозг плавится от слов вроде loader, state, Pydantic, watcher — это нормально.

Самая короткая карта

У вас есть 4 основных инструмента:

Ключевая идея проекта

То есть вы пишете не "как распарсить строку", а "что делать с уже готовыми данными".

Валидация аргументов команды

С версии 2.1 функция get_args удалена — теперь используется автоматическая валидация через Pydantic.

1. Прямое указание типов

@loader.command()
async def example(self, mx, event: MessageEvent, age: int = None):
    """[age] - возраст"""
    await utils.answer(mx, f"Age: {age}")

2. Через Pydantic модели

class AFKPayload(BaseModel):
    reason: str = Field(default="", description="Причина AFK")

    @model_validator(mode='before')
    @classmethod
    def parse_payload(cls, v):
        if isinstance(v, str):
            return {"reason": v.strip()}
        return v

@loader.command()
async def afk(self, mx, event, payload: AFKPayload):
    """[reason] - Установить статус AFK"""
    # payload.reason уже валидирован

Минимальный модуль

from ...core import loader, utils

class Meta:
    name = "HelloModule"
    description = "Пример модуля"
    version = "1.0.0"
    tags = ["example"]

class HelloPayload(BaseModel):
    name: str = Field(min_length=1)

@loader.tds
class HelloModule(loader.Module):
    @loader.command()
    async def hello(self, mx, event: MessageEvent, payload: HelloPayload):
        """<name> - сказать привет"""
        await utils.answer(mx, f"Hello, {payload.name}!")

Что обязательно в модуле

class Meta

• name — название
• description — описание
• version — версия
• tags — теги

Класс модуля

• Имя должно содержать Module
• Наследуется от loader.Module
• Рекомендуется использовать @loader.tds

⚙️ ConfigValue

ConfigValue — это настройки модуля. Всё, что:

• Вы хотите менять без переписывания кода
• Должно сохраняться между перезагрузками
• Относится к поведению модуля

Когда использовать

Используйте ConfigValue если у вас есть:

• API ключ
• Лимит
• Переключатель вкл/выкл
• Список комнат
• Режим работы

Не используйте если значение:

• Живёт только внутри одной команды
• Нужно 2 секунды и забыть
• Относится к одному FSM диалогу

Синтаксис

config = {
    "api_key": loader.ConfigValue("NONE", "API ключ", required=True),
    "limit":   loader.ConfigValue(10, "Сколько загружать", lambda x: x > 0),
    "enabled": loader.ConfigValue(True, "Включить модуль"),
}

Параметры ConfigValue

default

Значение по умолчанию. Тип очень важен — по нему система понимает во что преобразовывать строку:

• "25" станет int, если default был 10
• "true" станет bool, если default был False

description

Человеческое описание. Нужно для .help, .cfg и чтобы самому не забыть.

validator

"limit": loader.ConfigValue(10, "Лимит", lambda x: x > 0)

required=True

Означает "без этой настройки модуль не должен работать".

Как читать настройки

# Через квадратные скобки
limit = self.config["limit"]

# Через .get()
token = self.config.get("api_key")

Как изменять настройки

ok = self.config.set("limit", "25")
# Возвращает True/False

📥 Обработка событий (on)

@loader.on(...) нужен когда модуль должен реагировать на событие — не на команду, не на FSM шаг, именно на событие.

Простая метафора

• Команда — пользователь стучится и говорит "сделай это"
• on — дверь сама открылась, и вы хотите среагировать

Базовый пример

from mautrix.types import EventType, MessageEvent
from ...core import loader

@loader.tds
class DemoModule(loader.Module):
    @loader.on(EventType.ROOM_MESSAGE)
    async def handle_message(self, mx, event: MessageEvent):
        body = getattr(event.content, "body", "") or ""
        self.log.info(f"Message: {body}")

Доступные события

• EventType.ROOM_MESSAGE
• EventType.REACTION
• EventType.ROOM_REDACTION
• EventType.ROOM_TOMBSTONE
• EventType.ROOM_MEMBER

Зачем это нужно

Без on можно реагировать только на команды. Но многие модули должны жить "сами по себе":

• Автоответчик
• Логгер
• Антиспам
• Статистика
• Приветствие

Хелперы для ROOM_MESSAGE

Для сообщений добавляются удобные методы:

• await event.reply(text)
• await event.react(key)
• await event.get_reply_text()

🔍 Watcher

Watcher — это "перехватывать сообщения по паттерну".

Простая метафора

Watcher — это как поставить фильтр на почту. Письмо проходит через фильтр → если совпало → вы получаете уведомление.

Базовый пример

from ...core import loader

@loader.tds
class DemoModule(loader.Module):
    @loader.watcher(lambda msg: "тест" in msg.lower())
    async def on_test(self, mx, event):
        await event.reply("Это было слово 'тест'!")

Типичные паттерны

Проверка на слово

@loader.watcher(lambda msg: "help" in msg.lower())

Проверка по регулярке

import re
pattern = re.compile(r"^\s*(\d+)\s*$")

@loader.watcher(lambda msg: pattern.match(msg))

Проверка нескольких условий

@loader.watcher(lambda msg: "hello" in msg.lower() and len(msg) < 50)

Чем отличается от on

Пример: детектор ссылок

@loader.tds
class LinkDetector(loader.Module):
    @loader.watcher(lambda msg: "http" in msg.lower() or "www." in msg.lower())
    async def on_link(self, mx, event):
        await event.reply("Обнаружена ссылка!")

🔄 FSM

FSM (Finite State Machine) — это "сценарий из нескольких шагов".

Простая метафора

FSM — это как разговор с оператором. Бот задаёт вопрос → ждёт ответ → переходит к следующему шагу.

Базовый пример

from ...core import loader, utils

@loader.tds
class FeedbackModule(loader.Module):
    states = {"wait_name": 1, "wait_feedback": 2}

    @loader.command()
    async def feedback(self, mx, event):
        await utils.answer(mx, "Как вас зовут?")
        return self.states["wait_name"]

    @loader.state(states={"wait_name": 1})
    async def state_wait_name(self, mx, event):
        name = event.content.body
        await utils.answer(mx, f"{name}, напишите ваш отзыв")
        await mx.save_state("name", name)
        return self.states["wait_feedback"]

    @loader.state(states={"wait_feedback": 2})
    async def state_wait_feedback(self, mx, event):
        feedback = event.content.body
        name = await mx.get_state("name")
        await utils.answer(mx, f"Спасибо, {name}!")
        return None

Синтаксис

Определение состояний

states = {
    "state_name": numeric_id,
    "another_state": numeric_id,
}

Декоратор состояния

@loader.state(states={"state_name": id})
async def handler(self, mx, event):
    return next_state_id  # или None для сброса

Как работает

1. Пользователь вызывает команду .feedback
2. Бот переходит в состояние wait_name (id=1)
3. Следующее сообщение попадает в state_wait_name
4. После обработки бот переходит в wait_feedback (id=2)
5. После финального шага возвращаем None

Работа с состоянием

Сохранение данных

await mx.save_state("key", value)

Получение данных

value = await mx.get_state("key")

Очистка

await mx.clear_state()