149 lines
4.5 KiB
Python
149 lines
4.5 KiB
Python
"""Main logic implementation"""
|
|
|
|
import traceback
|
|
from typing import Any
|
|
import html
|
|
|
|
from nio import AsyncClient
|
|
from nio import JoinResponse, RoomSendResponse, RoomSendError
|
|
from nio import MatrixInvitedRoom, InviteMemberEvent
|
|
from nio import MatrixRoom, RoomMessageText
|
|
|
|
from nio import OlmUnverifiedDeviceError
|
|
|
|
from datatypes import *
|
|
import util
|
|
|
|
|
|
#
|
|
# DATA
|
|
#
|
|
_client: AsyncClient
|
|
|
|
|
|
|
|
#
|
|
# PRIVATE
|
|
#
|
|
async def _verify_all_devices() -> None:
|
|
"""Verifies all known devices"""
|
|
for user_id in _client.device_store.users:
|
|
for device_id, olm_device in _client.device_store[user_id].items():
|
|
# can't trust ourselves
|
|
if device_id == _client.device_id and user_id == _client.user_id:
|
|
continue
|
|
# they are already verified
|
|
if olm_device.verified:
|
|
continue
|
|
# verify them
|
|
_client.verify_device(olm_device)
|
|
|
|
def _handle_html_in_kwargs(kwargs: dict[str, Any]) -> None:
|
|
"""Modified `kwargs` in-place so that `formatted_body` appears if needed"""
|
|
if "formatted_body" in kwargs or "body" not in kwargs:
|
|
return
|
|
is_html, text_without_html = util.check_and_remove_html(kwargs["body"])
|
|
if not is_html:
|
|
return
|
|
kwargs["format"] = "org.matrix.custom.html"
|
|
kwargs["formatted_body"] = kwargs["body"]
|
|
kwargs["body"] = text_without_html
|
|
|
|
async def _send_message_to(room_id: str, message_type: MessageType, **kwargs) -> str:
|
|
"""Sends a message to the room and returns event_id.
|
|
|
|
This function automatically detects `body` key in `kwargs` and checks
|
|
if it is a valid HTML. If it is a valid HTML, it will send it as such.
|
|
Moreover, `body` attribute will be cleaned from any HTML tags, so that
|
|
the text will be looking well. `formatted_body` attribute is added
|
|
automatically and you should not add it manually.
|
|
"""
|
|
try:
|
|
# handle HTML
|
|
_handle_html_in_kwargs(kwargs)
|
|
# try to send the message
|
|
result = await _client.room_send(
|
|
room_id=room_id,
|
|
message_type="m.room.message",
|
|
content={
|
|
"msgtype": message_type.value,
|
|
**kwargs
|
|
}
|
|
)
|
|
# success
|
|
if type(result) is RoomSendResponse:
|
|
return result.event_id
|
|
# error
|
|
elif type(result) is RoomSendError:
|
|
raise Exception(result)
|
|
# unknown error
|
|
else:
|
|
raise RuntimeError()
|
|
except OlmUnverifiedDeviceError as e:
|
|
# verify everyone and retry
|
|
await _verify_all_devices()
|
|
return await _send_message_to(room_id, message_type, **kwargs)
|
|
except:
|
|
raise
|
|
|
|
async def _send_text_to(room_id: str, text: str) -> str:
|
|
"""Sends a text message to the room. `text` may be HTML"""
|
|
return await _send_message_to(
|
|
room_id=room_id,
|
|
message_type=MessageType.TEXT,
|
|
body=text
|
|
)
|
|
|
|
|
|
|
|
#
|
|
# CALLBACKS
|
|
#
|
|
async def _message_callback(room: MatrixRoom, event: RoomMessageText) -> None:
|
|
"""Handle commands received from Matrix"""
|
|
try:
|
|
# do not process messages sent by ourselves
|
|
if event.sender == _client.user_id:
|
|
return
|
|
# prepare response
|
|
response = "<b>Получено сообщение</b>"
|
|
response += f"<br><br><b>Room ID:</b> {html.escape(room.room_id)}"
|
|
response += f"<br><b>Sender:</b> {html.escape(event.sender)}"
|
|
print(await _send_text_to(room.room_id, response))
|
|
except:
|
|
traceback.print_exc()
|
|
|
|
async def _invite_callback(room: MatrixInvitedRoom, event: InviteMemberEvent) -> None:
|
|
"""Happens when the bot is invited to somewhere"""
|
|
try:
|
|
result = await _client.join(room.room_id)
|
|
if type(result) is JoinResponse:
|
|
util.log_info(f"Joined the room {room.room_id}")
|
|
else:
|
|
util.log_error(f"Can't join room {room.room_id}")
|
|
except:
|
|
traceback.print_exc()
|
|
|
|
async def _generic_test_callback(*args, **kwargs) -> None:
|
|
"""Use this callback to check argument types"""
|
|
print("GENERIC TEST CALLBACK")
|
|
for a in args:
|
|
print(f" - {type(a)}")
|
|
for k in kwargs:
|
|
print(f" * {k} = {kwargs[k]}")
|
|
|
|
|
|
|
|
#
|
|
# PUBLIC
|
|
#
|
|
async def setup(client: AsyncClient) -> None:
|
|
global _client
|
|
_client = client
|
|
client.add_event_callback(_message_callback, RoomMessageText) # type: ignore
|
|
client.add_event_callback(_invite_callback, InviteMemberEvent) # type: ignore
|
|
|
|
async def stop() -> None:
|
|
"""Stop all ongoing processes"""
|
|
pass
|