"""Callbacks for matrix messages"""
import html
import time
import traceback
from mab import MatrixBot
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
def _generate_help_message() -> str:
"""Generates help message in HTML markup"""
result = "Как использовать
!auth, создайте токен!info в комнате, чтобы узнать код канала уведомений!{aliases[0]} - {html.escape(_COMMANDS[aliases][1])}"
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:
"""!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:
"""!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, args: list[str]) -> 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, args: list[str]) -> None:
"""!deauth"""
# no token provided
if len(args) != 1:
await _bot.send_text_to_room(room.room_id, "Укажите токен, который надо удалить (должен быть ровно один аргумент)", is_html=True)
return
token = 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, args: list[str]) -> None:
"""!name"""
# check arguments
if len(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:])
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_leave(room: MatrixRoom, args: list[str]) -> None:
"""!leave"""
await _bot.send_text_to_room(room.room_id, "Команда не реализована", 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, "Задать (или удалить) имя для токена"),
("leave", "l"): (_on_cmd_leave, "Покинуть комнату, в которой получена эта команда"),
}
#
# PUBLIC
#
def setup(bot: MatrixBot, db: Database) -> None:
"""Setup the callbacks"""
global _bot, _db
_db = db
_bot = bot
_bot.add_event_callback(_on_text, RoomMessageText) # type: ignore