"""Callbacks for matrix messages"""
import html
import time
import traceback
from mab import MatrixBot
from nio import MatrixRoom, RoomMessageText
import util
#
# PRIVATE
#
_bot: MatrixBot = None # type: ignore
def _generate_help_message() -> str:
"""Generates help message in HTML markup"""
result = "Как использовать
"
result += "- Используя
!auth, создайте токен "
result += "- Добавьте бота в комнату для уведомлений
"
result += "- Выполните
!info в комнате, чтобы узнать код канала уведомений "
result += "- Используя полученные токен и код канала, отправьте уведомление через веб-запрос
"
result += "
"
result += "
Команды"
for aliases in _COMMANDS:
result += f"
!{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"""
response = "Сведения о комнате
"
response += f"channel: {html.escape(room.room_id)}"
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"""
TEST_LIST = {
"t0K3n": {
"created": time.time() - 60,
"last_use": time.time() - 30,
"total_uses": 534,
"service_name": "Резервное копирование",
},
"T35t_t0KeN": {
"created": time.time() - 12341,
"last_use": time.time() - 123,
"total_uses": 342,
"service_name": None,
},
}
response = "Список токенов"
for token in TEST_LIST:
data = TEST_LIST[token]
response += f"
{html.escape(token)}
"
if data["service_name"] is not None:
response += f"- Имя службы:
{data["service_name"]} "
else:
response += f"- Имя службы: указывается в запросе
"
response += f"- Создан: {util.date_to_text(data["created"])}
"
response += f"- Последнее использование: {util.date_to_text(data["last_use"])}
"
response += f"- Всего использований: {data["total_uses"]}
"
response += "
"
await _bot.send_text_to_room(room.room_id, response, is_html=True)
async def _on_cmd_auth(room: MatrixRoom, args: list[str]) -> None:
"""!auth"""
new_token = "inr3i4evfds"
new_token_name = " ".join(args) if args else None
if type(new_token_name) is str and not new_token_name.strip():
new_token_name = None
response = "Создан новый токен"
response += f"
Код: {new_token}"
response += f"
Имя сервиса: "
if new_token_name:
response += f"{html.escape(new_token_name)}"
else:
response += "указывается в запросе"
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"""
if len(args) != 1:
await _bot.send_text_to_room(room.room_id, "Укажите токен, который надо удалить (должен быть ровно один аргумент)", is_html=True)
return
token = args[0]
# TODO
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"""
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
# TODO
if new_name:
response = f"Новое имя для токена {token}: {new_name}"
else:
response = f"Удалено имя для токена {token}"
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) -> None:
"""Setup the callbacks"""
global _bot
_bot = bot
_bot.add_event_callback(_on_text, RoomMessageText) # type: ignore