matrix-appservice-kakaotalk/matrix_appservice_kakaotalk/formatter/from_matrix.py

# matrix-appservice-kakaotalk - A Matrix-KakaoTalk puppeting bridge.
# Copyright (C) 2022 Tulir Asokan, Andrew Ferrazzutti
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
from __future__ import annotations

from typing import NamedTuple

from mautrix.appservice import IntentAPI
from mautrix.types import Format, MessageEventContent, RelationType, RoomID, UserID
from mautrix.util.formatter import (
    EntityString,
    EntityType,
    MarkdownString,
    MatrixParser,
    SimpleEntity,
)
from mautrix.util.logging import TraceLogger

from ..kt.types.bson import Long
from ..kt.types.chat.attachment import ReplyAttachment, MentionStruct

from ..kt.client.types import TO_MSGTYPE_MAP

from .. import puppet as pu, user as u
from ..db import Message as DBMessage


class SendParams(NamedTuple):
    text: str
    mentions: list[MentionStruct] | None
    reply_to: ReplyAttachment


class KakaoTalkFormatString(EntityString[SimpleEntity, EntityType], MarkdownString):
    def format(self, entity_type: EntityType, **kwargs) -> KakaoTalkFormatString:
        if entity_type == EntityType.USER_MENTION:
            self.entities.append(
                SimpleEntity(
                    type=entity_type,
                    offset=0,
                    length=len(self.text),
                    extra_info={"user_id": kwargs["user_id"]},
                )
            )
            self.text = f"@{self.text}"
        return self


class ToKakaoTalkParser(MatrixParser[KakaoTalkFormatString]):
    fs = KakaoTalkFormatString


async def _get_id_from_mxid(mxid: UserID) -> Long | None:
    user = await u.User.get_by_mxid(mxid, create=False)
    if user and user.ktid:
        return user.ktid
    else:
        puppet = await pu.Puppet.get_by_mxid(mxid, create=False)
        return puppet.ktid if puppet else None


async def matrix_to_kakaotalk(
    content: MessageEventContent, room_id: RoomID, log: TraceLogger, intent: IntentAPI
) -> SendParams:
    # NOTE By design, this *throws* if user intent can't be matched (i.e. if a reply can't be created)
    if content.relates_to.rel_type == RelationType.REPLY:
        message = await DBMessage.get_by_mxid(content.relates_to.event_id, room_id)
        if not message:
            raise ValueError(
                f"Couldn't find reply target {content.relates_to.event_id}"
                " to bridge text message reply metadata to KakaoTalk"
            )
        try:
            mx_event = await intent.get_event(room_id, message.mxid)
        except:
            log.exception(f"Failed to find Matrix event for reply target {message.mxid}")
            raise
        kt_sender = await _get_id_from_mxid(mx_event.sender)
        if kt_sender is None:
            raise ValueError(
                f"Found no KakaoTalk user ID for reply target sender {mx_event.sender}"
            )
        content.trim_reply_fallback()
        reply_to = ReplyAttachment(
            # TODO
            #mentions=[],
            # TODO What are reply URLs for?
            #urls=[],
            # TODO Set this for emoticon reply, but must first support those
            attach_only=False,
            # TODO If replying with media works, must set type AND all attachment properties
            #      But then, the reply object must be an intersection of a ReplyAttachment and something else
            #attach_type=TO_MSGTYPE_MAP.get(content.msgtype),
            src_logId=message.ktid,
            # TODO
            src_mentions=[],
            # TODO Check if source message needs to be formatted
            src_message=mx_event.content.body,
            src_type=TO_MSGTYPE_MAP[mx_event.content.msgtype],
            src_userId=kt_sender,
        )
    else:
        reply_to = None
    if content.get("format", None) == Format.HTML and content["formatted_body"]:
        parsed = await ToKakaoTalkParser().parse(content["formatted_body"])
        text = parsed.text
        mentions_by_user: dict[Long, MentionStruct] = {}
        # Make sure to not create remote mentions for any remote user not in the room
        if parsed.entities:
            joined_members = set(await intent.get_room_members(room_id))
            last_offset = 0
            at = 0
            for mention in sorted(parsed.entities, key=lambda entity: entity.offset):
                mxid = mention.extra_info["user_id"]
                if mxid not in joined_members:
                    continue
                ktid = await _get_id_from_mxid(mxid)
                if ktid is None:
                    continue
                at += text[last_offset:mention.offset+1].count("@")
                last_offset = mention.offset+1
                mention = mentions_by_user.setdefault(ktid, MentionStruct(
                    at=[],
                    len=mention.length,
                    user_id=ktid,
                ))
                mention.at.append(at)
        mentions = list(mentions_by_user.values())
    else:
        text = content.body
        mentions = None
    return SendParams(text=text, mentions=mentions, reply_to=reply_to)