Deleted logic.py

logic.py

@@ -1,148 +0,0 @@
-"""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