import logging import re from markdown import markdown from nio import AsyncClient, MatrixRoom from .exceptions import NoRoomException from .utils import setup_logger class Message(object): def __init__(self, _type: str = 'm.room.message', loglevel: int = 10) -> None: self.__type = _type self.__ignore_unverified_devices = True self.__logger = setup_logger(__name__) async def send(self, client: AsyncClient, room: (MatrixRoom, None)) -> None: if room: self.logger.debug('Sending message: %s', self.content) await client.room_send( room.room_id, self.type, self.content, ignore_unverified_devices=self.ignore_unverified_devices) else: raise NoRoomException() @property def type(self) -> str: return self.__type @property def ignore_unverified_devices(self) -> bool: return self.__ignore_unverified_devices @property def logger(self) -> logging.Logger: return self.__logger @property def content(self) -> str: raise NotImplementedError('content must be implemented') class TextMessage(Message): def __init__(self, body: str, msgtype: str = 'm.notice', _type: str = 'm.room.message') -> None: super().__init__(_type) self.body = body self.msgtype = msgtype @property def content(self) -> dict: content = {'msgtype': self.msgtype, 'body': self.body} return content class MarkdownMessage(Message): clean_regexp = re.compile('<.*?>') def __init__(self, body: str, msgtype: str = 'm.notice', _type: str = 'm.room.message') -> None: super().__init__(_type) self.formatted_body = markdown(body) self.format = 'org.matrix.custom.html' self.body = self.clean_html(self.formatted_body) self.msgtype = msgtype @classmethod def clean_html(cls, raw_html: str) -> str: cleantext = re.sub(cls.clean_regexp, '', raw_html) return cleantext @property def content(self) -> dict: content = { 'msgtype': self.msgtype, 'body': self.body, 'formatted_body': self.formatted_body, 'format': self.format } return content