diff --git a/bot_callbacks.py b/bot_callbacks.py index 8f91c2c..7ddcba2 100644 --- a/bot_callbacks.py +++ b/bot_callbacks.py @@ -4,7 +4,7 @@ import html import time import traceback -from mab import MatrixBot +from mab import MatrixBot, TextCommandFilter from nio import MatrixRoom, RoomMessageText import util @@ -17,6 +17,18 @@ from datatypes import * _bot: MatrixBot = None # type: ignore _db: Database = None # type: ignore +_COMMANDS = { + "help": "Получить справку", + "info": "Получить сведения о комнате", + "tokens": "Получить список токенов", + "auth": "Создать новый токен", + "deauth": "Удалить существующий токен", + "name": "Задать (или удалить) имя для токена", + "ban": "Забанить IP адрес", + "unban": "Разбанить IP адрес", + "bans": "Получить список забаненных IP адресов", +} + def _generate_help_message() -> str: """Generates help message in HTML markup""" result = "Как использовать
    " @@ -27,48 +39,18 @@ def _generate_help_message() -> str: result += "
" result += "

Команды" - for aliases in _COMMANDS: - result += f"
!{aliases[0]} - {html.escape(_COMMANDS[aliases][1])}" + for key in _COMMANDS: + result += f"
!{key} - {html.escape(_COMMANDS[key])}" return result -# -# GENERIC EVENT HANDLERS -# -async def _on_text(room: MatrixRoom, event: RoomMessageText) -> None: - """This callback is called when text message is received""" - if event.sender == _bot.get_client().user_id: - return - text = event.body - parts = [p for p in text.split() if p.strip()] - if len(parts) < 1: - return - cmd = parts[0].lower() - if not cmd.startswith("!"): - return - cb = None - for aliases in _COMMANDS: - for alias in aliases: - if alias == cmd.lstrip("!"): - cb = _COMMANDS[aliases][0] - break - if cb: - break - if cb is None: - await _bot.send_text_to_room(room.room_id, "Используйте !help") - return - try: - await cb(room, parts[1:]) - except: - traceback.print_exc() - # # COMMAND HANDLERS # -async def _on_cmd_help(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_help(room: MatrixRoom, event: RoomMessageText) -> None: """!help""" await _bot.send_text_to_room(room.room_id, _generate_help_message(), is_html=True) -async def _on_cmd_info(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_info(room: MatrixRoom, event: RoomMessageText) -> None: """!info""" # get room info (or add it) room_info = await _db.room_get(room.room_id) @@ -87,7 +69,7 @@ async def _on_cmd_info(room: MatrixRoom, args: list[str]) -> None: response += f"Канал: {html.escape(room_info.code)}" await _bot.send_text_to_room(room.room_id, response, is_html=True) -async def _on_cmd_tokens(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_tokens(room: MatrixRoom, event: RoomMessageText) -> None: """!info""" # get all tokens and check if there are none tokens = await _db.token_get_all() @@ -112,13 +94,13 @@ async def _on_cmd_tokens(room: MatrixRoom, args: list[str]) -> None: # respond await _bot.send_text_to_room(room.room_id, response, is_html=True) -async def _on_cmd_auth(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_auth(room: MatrixRoom, event: RoomMessageText) -> None: """!auth""" # create new token token = await _db.token_create() # set the name if it is provided - if args: - token.name = " ".join(args) + if event.command_args: + token.name = " ".join(event.command_args) await _db.token_set_name(token.code, token.name) # create the response response = "Создан новый токен" @@ -131,13 +113,13 @@ async def _on_cmd_auth(room: MatrixRoom, args: list[str]) -> None: # respond await _bot.send_text_to_room(room.room_id, response, is_html=True) -async def _on_cmd_deauth(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_deauth(room: MatrixRoom, event: RoomMessageText) -> None: """!deauth""" # no token provided - if len(args) != 1: + if len(event.command_args) != 1: await _bot.send_text_to_room(room.room_id, "Укажите токен, который надо удалить (должен быть ровно один аргумент)", is_html=True) return - token = args[0] + token = event.command_args[0] # check if token does not exist if await _db.token_get(token) is None: await _bot.send_text_to_room(room.room_id, "Токен не найден", is_html=True) @@ -148,15 +130,15 @@ async def _on_cmd_deauth(room: MatrixRoom, args: list[str]) -> None: response = f"Удалён токен {token}" await _bot.send_text_to_room(room.room_id, response, is_html=True) -async def _on_cmd_name(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_name(room: MatrixRoom, event: RoomMessageText) -> None: """!name""" # check arguments - if len(args) < 1: + if len(event.command_args) < 1: error = "Требуется указать как минимум токен. Если хотите убрать имя, то имя указывать не надо. Если имя нужно назначить или изменить, то после токена укажите новое имя." await _bot.send_text_to_room(room.room_id, error, is_html=True) return - token = args[0] - new_name = " ".join(args[1:]) + token = event.command_args[0] + new_name = " ".join(event.command_args[1:]) if not new_name.strip(): new_name = None # check if token exists @@ -173,18 +155,18 @@ async def _on_cmd_name(room: MatrixRoom, args: list[str]) -> None: # respond await _bot.send_text_to_room(room.room_id, response, is_html=True) -async def _on_cmd_ban(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_ban(room: MatrixRoom, event: RoomMessageText) -> None: """!ban""" # check arguments - if len(args) < 2: + if len(event.command_args) < 2: error = "Формат: !ban <IP> <SECONDS> [REASON]" await _bot.send_text_to_room(room.room_id, error, is_html=True) return - # get args + # get event.command_args try: - ip = args[0] - duration = float(args[1]) - reason = " ".join(args[2:]) if args[2:] else "Manual ban" + ip = event.command_args[0] + duration = float(event.command_args[1]) + reason = " ".join(event.command_args[2:]) if event.command_args[2:] else "Manual ban" except: error = "Возникла ошибка. Наверняка неправильно указаны секунды." await _bot.send_text_to_room(room.room_id, error, is_html=True) @@ -196,21 +178,21 @@ async def _on_cmd_ban(room: MatrixRoom, args: list[str]) -> None: except: await _bot.send_text_to_room(room.room_id, "Возникла ошибка", is_html=True) -async def _on_cmd_unban(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_unban(room: MatrixRoom, event: RoomMessageText) -> None: """!unban""" # check arguments - if len(args) != 1: + if len(event.command_args) != 1: error = "Формат: !ban <IP>" await _bot.send_text_to_room(room.room_id, error, is_html=True) return # unban try: - await _db.ban_delete(args[0]) + await _db.ban_delete(event.command_args[0]) await _bot.send_text_to_room(room.room_id, "IP адрес разблокирован (если он был заблокирован)", is_html=True) except: await _bot.send_text_to_room(room.room_id, "Возникла ошибка", is_html=True) -async def _on_cmd_bans(room: MatrixRoom, args: list[str]) -> None: +async def _on_cmd_bans(room: MatrixRoom, event: RoomMessageText) -> None: """!bans""" # list bans = _db.ban_get_all() @@ -228,18 +210,6 @@ async def _on_cmd_bans(room: MatrixRoom, args: list[str]) -> None: await _bot.send_text_to_room(room.room_id, result, is_html=True) -_COMMANDS = { - ("help", "h", "?"): (_on_cmd_help, "Получить справку"), - ("info", "room", "i", "r"): (_on_cmd_info, "Получить сведения о комнате"), - ("tokens", "t"): (_on_cmd_tokens, "Получить список токенов"), - ("auth", "create", "a", "c"): (_on_cmd_auth, "Создать новый токен"), - ("deauth", "delete", "d"): (_on_cmd_deauth, "Удалить существующий токен"), - ("name", "n"): (_on_cmd_name, "Задать (или удалить) имя для токена"), - ("ban", "b"): (_on_cmd_ban, "Забанить IP адрес"), - ("unban", "u"): (_on_cmd_unban, "Разбанить IP адрес"), - ("bans", "l"): (_on_cmd_bans, "Получить список забаненных IP адресов"), -} - # # PUBLIC # @@ -248,4 +218,21 @@ def setup(bot: MatrixBot, db: Database) -> None: global _bot, _db _db = db _bot = bot - _bot.add_event_callback(_on_text, RoomMessageText) # type: ignore + _bot.add_event_callback( + _on_cmd_help, TextCommandFilter(["help", "h", "?"])) + _bot.add_event_callback( + _on_cmd_info, TextCommandFilter(["info", "room", "i", "r"])) + _bot.add_event_callback( + _on_cmd_tokens, TextCommandFilter(["tokens", "t"])) + _bot.add_event_callback( + _on_cmd_auth, TextCommandFilter(["auth", "create", "a", "c"])) + _bot.add_event_callback( + _on_cmd_deauth, TextCommandFilter(["deauth", "delete", "d"])) + _bot.add_event_callback( + _on_cmd_name, TextCommandFilter(["name", "n"])) + _bot.add_event_callback( + _on_cmd_ban, TextCommandFilter(["ban", "b"])) + _bot.add_event_callback( + _on_cmd_unban, TextCommandFilter(["unban", "u"])) + _bot.add_event_callback( + _on_cmd_bans, TextCommandFilter(["bans", "l"])) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 6b26564..1dea17d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -34,7 +34,7 @@ idna==3.19 Jinja2==3.1.6 jsonschema==4.26.0 jsonschema-specifications==2025.9.1 -mab @ git+https://git.tyukalov.su/nikita/mab@v0.0.2 +mab @ git+https://git.tyukalov.su/nikita/mab@v0.1.0 markdown-it-py==4.2.0 MarkupSafe==3.0.3 matrix-nio==0.26.0