"""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 = "Как использовать
!auth, создайте токен!info в комнате, чтобы узнать код канала уведомений!{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 = "Сведения о комнате{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)}{token.name}{token.code}"
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{util.date_to_text(ban["expires_at"])}{html.escape(ban["reason"])}