"""Callbacks for matrix messages""" import html import time import traceback from mab import MatrixBot, TextCommandFilter from nio import MatrixRoom, RoomMessageText import util from database import Database from datatypes import * # # PRIVATE # _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 = "Как использовать
    " result += "
  1. Используя !auth, создайте токен
  2. " result += "
  3. Добавьте бота в комнату для уведомлений
  4. " result += "
  5. Выполните !info в комнате, чтобы узнать код канала уведомений
  6. " result += "
  7. Используя полученные токен и код канала, отправьте уведомление через веб-запрос
  8. " result += "
" result += "

Команды" for key in _COMMANDS: result += f"
!{key} - {html.escape(_COMMANDS[key])}" return result # # COMMAND HANDLERS # 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, event: RoomMessageText) -> None: """!info""" # get room info (or add it) room_info = await _db.room_get(room.room_id) if room_info is None: room_info = await _db.room_create(room.room_id) # failure if room_info is None: await _bot.send_text_to_room( room.room_id, "Нет информации о комнате", is_html=True ) return # respond response = "Сведения о комнате
" 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, event: RoomMessageText) -> None: """!info""" # get all tokens and check if there are none tokens = await _db.token_get_all() if not tokens: await _bot.send_text_to_room( room.room_id, "Нет токенов, используйте !auth", is_html=True ) return # create the list of tokens response = "Список токенов" for token in tokens: response += f"
{html.escape(token.code)}
" # respond await _bot.send_text_to_room(room.room_id, response, is_html=True) 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 event.command_args: token.name = " ".join(event.command_args) await _db.token_set_name(token.code, token.name) # create the response response = "Создан новый токен" response += f"
Код: {token.code}" response += f"
Имя сервиса: " if token.name: response += f"{html.escape(token.name)}" else: response += "указывается в запросе" # respond await _bot.send_text_to_room(room.room_id, response, is_html=True) async def _on_cmd_deauth(room: MatrixRoom, event: RoomMessageText) -> None: """!deauth""" # no token provided if len(event.command_args) != 1: await _bot.send_text_to_room(room.room_id, "Укажите токен, который надо удалить (должен быть ровно один аргумент)", is_html=True) return 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) return # remove the token await _db.token_delete(token) # respond response = f"Удалён токен {token}" await _bot.send_text_to_room(room.room_id, response, is_html=True) async def _on_cmd_name(room: MatrixRoom, event: RoomMessageText) -> None: """!name""" # check arguments if len(event.command_args) < 1: error = "Требуется указать как минимум токен. Если хотите убрать имя, то имя указывать не надо. Если имя нужно назначить или изменить, то после токена укажите новое имя." await _bot.send_text_to_room(room.room_id, error, is_html=True) return token = event.command_args[0] new_name = " ".join(event.command_args[1:]) if not new_name.strip(): new_name = None # check if token exists if await _db.token_get(token) is None: await _bot.send_text_to_room(room.room_id, "Токен не существует", is_html=True) return # set new name await _db.token_set_name(token, new_name) # prepare the response if new_name: response = f"Новое имя для токена {token}: {new_name}" else: response = f"Удалено имя для токена {token}" # respond await _bot.send_text_to_room(room.room_id, response, is_html=True) async def _on_cmd_ban(room: MatrixRoom, event: RoomMessageText) -> None: """!ban""" # check arguments 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 event.command_args try: 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) return # ban try: await _db.ban_create(ip, time.time() + duration, reason) 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_unban(room: MatrixRoom, event: RoomMessageText) -> None: """!unban""" # check arguments 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(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, event: RoomMessageText) -> None: """!bans""" # list bans = _db.ban_get_all() if not bans: await _bot.send_text_to_room(room.room_id, "Нет заблокированных IP адресов", is_html=True) return # create the response result = "Список заблокированных IP
" await _bot.send_text_to_room(room.room_id, result, is_html=True) # # PUBLIC # def setup(bot: MatrixBot, db: Database) -> None: """Setup the callbacks""" global _bot, _db _db = db _bot = bot _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"]))