reasonspace-bot

Официальный Python SDK для ботов Reason Space: бот получает события (длинным опросом — без домена и вебхуков, или вебхуком), отвечает через REST-клиент и читает и пишет текст каналов в том же шифровании, что и люди.

pip install reasonspace-bot

Нужен Python 3.10+.

Бот за вечер

Домен, сертификат и вебхук не нужны: бот сам забирает события с сервера.

1. В Reason Space откройте Настройки → Боты → Создать бота и скопируйте токен.
2. Там же — Пригласить в Space: выберите пространство, права и каналы, которые бот видит. События из других каналов к нему не приходят. Для примера ниже права не нужны.
3. Сохраните код в bot.py:

from reasonspace_bot import Bot, DMMessageCreatedEvent

bot = Bot()  # токен — из переменной окружения BOT_TOKEN


@bot.on("dm.message.created")
async def echo(event: DMMessageCreatedEvent) -> None:
    await event.reply(f"Вы написали: {event.content}")


bot.run_polling()

4. Запустите BOT_TOKEN=bot_... python bot.py, найдите бота среди участников пространства, откройте его профиль и напишите ему.

Остановить — Ctrl+C. Чтобы после перезапуска не получать уже обработанные события заново, дайте боту файл для позиции: bot.run_polling(state_file="bot.seq").

Статистика канала

Бот считает сообщения по авторам и отвечает на /top. Права: messages.read, members.read, commands.

import asyncio
from collections import Counter

from reasonspace_bot import Bot, CommandInvokedEvent, MessageCreatedEvent

bot = Bot()
messages_by_author: Counter[str] = Counter()


@bot.on("message.created")
async def count(event: MessageCreatedEvent) -> None:
    if event.author_user_id and not event.author_is_bot:
        messages_by_author[event.author_user_id] += 1


@bot.command("top", description="Кто пишет больше всех")
async def top(event: CommandInvokedEvent) -> None:
    members = await bot.client.list_members(event.space_id)
    names = {m["user_id"]: m.get("nickname") or m["user_id"][:8] for m in members}
    lines = [f"{names.get(uid, uid[:8])} — {n}" for uid, n in messages_by_author.most_common(5)]
    await bot.respond(event, "\n".join(lines) or "Пока никто не писал")


async def main() -> None:
    await bot.register_all_commands()  # объявить /top на платформе
    await bot.poll_forever()


asyncio.run(main())

Счётчик живёт в памяти; для настоящей статистики храните его в своей базе.

Текст сообщений

В событии о сообщении человека текста нет: сервер не читает переписку ради ботов. Бот с правом messages.read берёт сообщение из истории канала и расшифровывает его ключом канала — SDK получает ключ у сервера один раз и держит в памяти.

from reasonspace_bot import Bot, MessageCreatedEvent

bot = Bot()


@bot.on("message.created")
async def watch(event: MessageCreatedEvent) -> None:
    if event.author_is_bot:
        return
    history = await bot.client.list_messages(event.channel_id, space_id=event.space_id, limit=20)
    for msg in history:
        if msg["id"] == event.message_id:
            text = await bot.client.decrypt_message(event.space_id, msg)
            if "помогите" in text.lower():
                await bot.send_message(
                    event.channel_id,
                    "Позвал модератора",
                    space_id=event.space_id,
                    reply_to=event.message_id,
                )


bot.run_polling(state_file="bot.seq")

send_message шифрует текст сам, если у бота есть ключ канала. Нет права messages.read или у канала ещё нет ключа — сообщение уходит открытым текстом, как раньше, а в лог один раз на канал пишется предупреждение. Свои сообщения бот получает в message.created с текстом: await event.text() вернёт его расшифрованным. Шифровать и расшифровывать вручную можно функциями encrypt, decrypt и is_encrypted из reasonspace_bot.crypto.

Личка

Человек пишет боту — приходит dm.message.created с текстом (личка не шифруется), event.reply(...) отвечает ему же, как в примере «Бот за вечер»; quote=True — ответом на его сообщение. Право для лички не нужно.

Первым бот пишет только своему владельцу. Остальным — await bot.client.send_dm(user_id, "...") после того, как человек написал боту сам, иначе AuthError (403). Лимит — 20 сообщений в час одному человеку (RateLimitError). История переписки — await bot.client.list_dm(user_id, limit=50, before=None).

События

@bot.on("тип") или @bot.on(["тип", "тип"]). Хендлер получает событие нужного класса; общие поля всех событий — id, seq, type, space_id, created_at, raw_data. Чего сервер не прислал — None. Событие нового типа, которого SDK ещё не знает, приходит как RawEvent.

Хендлеры вызываются по очереди: долгую работу выносите в asyncio.create_task. Упавший хендлер пишет ошибку в лог и не мешает остальным. Повторно доставленное событие (тот же id) хендлеры не увидят.

Опрос сам переживает сбои: при обрыве сети и ошибках сервера ждёт 1, 2, 4… до 30 секунд, при лимите — сколько скажет сервер. Если тот же бот уже запущен в другом месте, сервер отвечает 409, и опрос ждёт 5 секунд. Неверный токен останавливает бота с AuthError.

Ошибки

Все ошибки API — BotError (старое имя BotAPIError тоже работает), у каждой есть status_code и detail:

Клиент сам повторяет запрос до трёх раз при сбое сети, ответах 500/502/503/504 и 429 (если ждать не дольше 30 секунд). Сбой сети после повторов выходит наружу как исключение httpx.

Вебхук вместо опроса

Если у бота есть публичный адрес, события можно принимать вебхуком — хендлеры те же:

from reasonspace_bot import Bot, DMMessageCreatedEvent

bot = Bot()  # BOT_TOKEN и WEBHOOK_SECRET — из переменных окружения


@bot.on("dm.message.created")
async def echo(event: DMMessageCreatedEvent) -> None:
    await event.reply(f"Вы написали: {event.content}")


if __name__ == "__main__":
    bot.run(port=8765)

bot.app — ASGI-приложение (POST /webhook и POST /): проверяет подпись X-Reasonspace-Signature и отбрасывает повторные доставки. Для prod:

pip install 'reasonspace-bot[server]' gunicorn
gunicorn bot:bot.app --workers 4 --worker-class uvicorn.workers.UvicornWorker

Свой приёмник вместо bot.app — verify_signature(secret, body, header) и parse_event(envelope).

Переменные окружения

Изменения по версиям — в CHANGELOG.md. Документация платформы: https://docs.reasonspace.ru