2026-matrix-csonac/bot_callbacks.py

"""Callbacks for matrix messages"""

import html
import time
import traceback

from mab import * # type: ignore

import util
from database import Database
from datatypes import *

#
# PRIVATE
#
_db: Database = None # type: ignore

_COMMANDS = {
    "help": "Получить справку",
    "info": "Получить сведения о комнате",
    "tokens": "Получить список токенов",
    "auth": "Создать новый токен",
    "deauth": "Удалить существующий токен",
    "name": "Задать (или удалить) имя для токена",
    "ban": "Забанить IP адрес",
    "unban": "Разбанить IP адрес",
    "bans": "Получить список забаненных IP адресов",
    "short": "(Небезопасно) Создать токен-имя",
}

def _generate_help_message() -> str:
    """Generates help message in HTML markup"""
    result = "<strong><i>Как использовать</i></strong><br><ol>"
    result += "<li>Используя <code>!auth</code>, создайте токен</li>"
    result += "<li>Добавьте бота в комнату для уведомлений</li>"
    result += "<li>Выполните <code>!info</code> в комнате, чтобы узнать код канала уведомений</li>"
    result += "<li>Используя полученные токен и код канала, отправьте уведомление через веб-запрос</li>"
    result += "</ol>"

    result += "<br><br><strong><i>Команды</i></strong>"
    for key in _COMMANDS:
        result += f"<br><code>!{key}</code> - <i>{html.escape(_COMMANDS[key])}</i>"
    return result

#
# COMMAND HANDLERS
#
async def _on_cmd_help(ctx: EventContext) -> None:
    """!help"""
    await ctx.bot.send_text(ctx.room, _generate_help_message())

async def _on_cmd_info(ctx: EventContext) -> None:
    """!info"""
    # get room info (or add it)
    room_info = await _db.room_get(ctx.room.room_id)
    if room_info is None:
        room_info = await _db.room_create(ctx.room.room_id)
    # failure
    if room_info is None:
        await ctx.bot.send_text(
            ctx.room, "<strong>Нет информации о комнате</strong>"
        )
        return
    # respond
    response = "<strong><i>Сведения о комнате</i></strong><br>"
    response += f"<strong>Канал:</strong> <code>{html.escape(room_info.code)}</code>"
    await ctx.bot.send_text(ctx.room, response)

async def _on_cmd_tokens(ctx: EventContext) -> None:
    """!info"""
    # get all tokens and check if there are none
    tokens = await _db.token_get_all()
    if not tokens:
        await ctx.bot.send_text(
            ctx.room,
            "<strong>Нет токенов, используйте <code>!auth</code></strong>"
        )
        return
    # create the list of tokens
    response = "<strong><i>Список токенов</i></strong>"
    for token in tokens:
        response += f"<br><strong><code>{html.escape(token.code)}</code></strong><br><ul>"
        if token.name is not None:
            response += f"<li><strong>Имя службы:</strong> <code>{token.name}</code></li>"
        else:
            response += f"<li><strong>Имя службы:</strong> <i>указывается в запросе</i></li>"
        response += f"<li><strong>Создан:</strong> <i>{util.date_to_text(token.created_at)}</i></li>"
        response += f"<li><strong>Последнее использование:</strong> <i>{util.date_to_text(token.last_access_at)}</i></li>"
        response += "</ul>"
    # respond
    await ctx.bot.send_text(ctx.room, response)

async def _on_cmd_auth(ctx: EventContext) -> None:
    """!auth"""
    # create new token
    token = await _db.token_create()
    # set the name if it is provided
    if ctx[CTX_CMD_ARGS]:
        token.name = " ".join(ctx[CTX_CMD_ARGS])
        await _db.token_set_name(token.code, token.name)
    # create the response
    response = "<strong>Создан новый токен</strong>"
    response += f"<br><strong>Код:</strong> <code>{token.code}</code>"
    response += f"<br><strong>Имя сервиса:</strong> "
    if token.name:
        response += f"<code>{html.escape(token.name)}</code>"
    else:
        response += "указывается в запросе"
    # respond
    await ctx.bot.send_text(ctx.room, response)

async def _on_cmd_deauth(ctx: EventContext) -> None:
    """!deauth"""
    # no token provided
    if len(ctx[CTX_CMD_ARGS]) != 1:
        await ctx.bot.send_text(ctx.room, "<strong>Укажите токен, который надо удалить (должен быть ровно один аргумент)</strong>")
        return
    token = ctx[CTX_CMD_ARGS][0]
    # check if token does not exist
    if await _db.token_get(token) is None:
        await ctx.bot.send_text(ctx.room, "<strong>Токен не найден</strong>")
        return
    # remove the token
    await _db.token_delete(token)
    # respond
    response = f"<strong>Удалён токен <code>{token}</code></strong>"
    await ctx.bot.send_text(ctx.room, response)

async def _on_cmd_name(ctx: EventContext) -> None:
    """!name"""
    # check arguments
    if len(ctx[CTX_CMD_ARGS]) < 1:
        error = "<strong>Требуется указать как минимум токен. Если хотите убрать имя, то имя указывать не надо. Если имя нужно назначить или изменить, то после токена укажите новое имя.</strong>"
        await ctx.bot.send_text(ctx.room, error)
        return
    token = ctx[CTX_CMD_ARGS][0]
    new_name = " ".join(ctx[CTX_CMD_ARGS][1:])
    if not new_name.strip():
        new_name = None
    # check if token exists
    if await _db.token_get(token) is None:
        await ctx.bot.send_text(ctx.room, "<strong>Токен не существует</strong>")
        return
    # set new name
    await _db.token_set_name(token, new_name)
    # prepare the response
    if new_name:
        response = f"<strong>Новое имя для токена <code>{token}</code>: <code>{new_name}</code></strong>"
    else:
        response = f"<strong>Удалено имя для токена <code>{token}</code></strong>"
    # respond
    await ctx.bot.send_text(ctx.room, response)

async def _on_cmd_ban(ctx: EventContext) -> None:
    """!ban"""
    # check arguments
    if len(ctx[CTX_CMD_ARGS]) < 2:
        error = "<strong>Формат: <code>!ban &lt;IP&gt; &lt;SECONDS&gt; [REASON]</code></strong>"
        await ctx.bot.send_text(ctx.room, error)
        return
    # get event.command_args
    try:
        ip = ctx[CTX_CMD_ARGS][0]
        duration = float(ctx[CTX_CMD_ARGS][1])
        reason = " ".join(ctx[CTX_CMD_ARGS][2:]) if ctx[CTX_CMD_ARGS][2:] else "Manual ban"
    except:
        error = "<strong>Возникла ошибка. Наверняка неправильно указаны секунды.</strong>"
        await ctx.bot.send_text(ctx.room, error)
        return
    # ban
    try:
        await _db.ban_create(ip, time.time() + duration, reason)
        await ctx.bot.send_text(ctx.room, "<strong>IP адрес заблокирован</strong>")
    except:
        await ctx.bot.send_text(ctx.room, "<strong>Возникла ошибка</strong>")

async def _on_cmd_unban(ctx: EventContext) -> None:
    """!unban"""
    # check arguments
    if len(ctx[CTX_CMD_ARGS]) != 1:
        error = "<strong>Формат: <code>!ban &lt;IP&gt;</code></strong>"
        await ctx.bot.send_text(ctx.room, error)
        return
    # unban
    try:
        await _db.ban_delete(ctx[CTX_CMD_ARGS][0])
        await ctx.bot.send_text(ctx.room, "<strong>IP адрес разблокирован (если он был заблокирован)</strong>")
    except:
        await ctx.bot.send_text(ctx.room, "<strong>Возникла ошибка</strong>")

async def _on_cmd_bans(ctx: EventContext) -> None:
    """!bans"""
    # list
    bans = _db.ban_get_all()
    if not bans:
        await ctx.bot.send_text(ctx.room, "<strong>Нет заблокированных IP адресов</strong>")
        return
    # create the response
    result = "<strong>Список заблокированных IP</strong><br><ul>"
    for ban in bans:
        result += f"<li><strong>{html.escape(ban["ip"])}</strong><ul>"
        result += f"<li><strong>Действует до:</strong> <code>{util.date_to_text(ban["expires_at"])}</code></li>"
        result += f"<li><strong>Причина:</strong> <code>{html.escape(ban["reason"])}</code></li>"
        result += "</ul></li>"
    result += "</ul>"
    await ctx.bot.send_text(ctx.room, result)

async def _on_cmd_short(ctx: EventContext) -> None:
    """!auth"""
    # get token to use
    if len(ctx[CTX_CMD_ARGS]) != 1:
        error = "<strong>Формат: <code>!short &lt;token&gt;</code></strong>"
        await ctx.bot.send_text(ctx.room, error)
        return
    # create new token
    try:
        token = await _db.token_create(ctx[CTX_CMD_ARGS][0])
    except Exception as e:
        await ctx.bot.send_text(ctx.room, f"Не удалось создать токен: {" ".join(e.args)}")
        return
    # create the response
    response = "<strong>Создан новый токен</strong>"
    response += f"<br><strong>Код:</strong> <code>{token.code}</code>"
    response += f"<br><strong>Имя сервиса:</strong> указывается в запросе"
    # respond
    await ctx.bot.send_text(ctx.room, response)


#
# PUBLIC
#
def setup(bot: MatrixBot, db: Database) -> None:
    """Setup the callbacks"""
    global _db
    _db = db
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["info", "room", "i", "r"]),
        _on_cmd_info
    )
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["tokens", "t"]),
        _on_cmd_tokens
    )
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["auth", "create", "a", "c"]),
        _on_cmd_auth
    )
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["deauth", "delete", "d"]),
        _on_cmd_deauth
    )
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["name", "n"]),
        _on_cmd_name
    )
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["ban", "b"]),
        _on_cmd_ban
    )
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["unban", "u"]),
        _on_cmd_unban
    )
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["bans", "l"]),
        _on_cmd_bans
    )
    bot.add_callback(
        ~SenderIsBotFilter() & BodyCommandFilter(["short", "s"]),
        _on_cmd_short
    )
    bot.add_callback(
        ~SenderIsBotFilter() & NewMessageFilter(),
        _on_cmd_help
    )