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

192 lines
7.6 KiB
Python

# 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 import utf16_surrogate
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 import KnownChatType
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:
prefix = suffix = ""
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"]},
)
)
return self
elif entity_type == EntityType.BOLD:
prefix = suffix = "*"
elif entity_type == EntityType.ITALIC:
prefix = suffix = "_"
elif entity_type == EntityType.STRIKETHROUGH:
prefix = suffix = "~"
elif entity_type == EntityType.URL:
if kwargs["url"] != self.text:
suffix = f" ({kwargs['url']})"
elif entity_type == EntityType.PREFORMATTED:
prefix = f"```{kwargs['language']}\n"
suffix = "\n```"
elif entity_type == EntityType.INLINE_CODE:
prefix = suffix = "`"
elif entity_type == EntityType.BLOCKQUOTE:
children = self.trim().split("\n")
children = [child.prepend("> ") for child in children]
return self.join(children, "\n")
elif entity_type == EntityType.HEADER:
prefix = "#" * kwargs["size"] + " "
else:
return self
self._offset_entities(len(prefix))
self.text = f"{prefix}{self.text}{suffix}"
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,
skip_reply: bool = False
) -> 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 and not skip_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:
src_event = await intent.get_event(room_id, message.mxid)
except:
log.exception(f"Failed to find Matrix event for reply target {message.mxid}")
raise
src_kt_sender = await _get_id_from_mxid(src_event.sender)
if src_kt_sender is None:
raise ValueError(
f"Found no KakaoTalk user ID for reply target sender {src_event.sender}"
)
content.trim_reply_fallback()
src_converted = await matrix_to_kakaotalk(src_event.content, room_id, log, intent, skip_reply=True)
if src_event.content.relates_to.rel_type == RelationType.REPLY:
src_type = KnownChatType.REPLY
src_message = src_converted.text
else:
src_type = TO_MSGTYPE_MAP[src_event.content.msgtype]
if src_type == KnownChatType.FILE:
src_message = _media_type_reply_body_map[KnownChatType.FILE] + src_converted.text
else:
src_message = _media_type_reply_body_map.get(src_type, src_converted.text)
reply_to = ReplyAttachment(
# NOTE mentions will be merged into this later
# TODO Set this for emoticon reply, but must first support them
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),
# TODO Confirm why official client sets this to 0, and whether this should be left as None instead
attach_type=0,
src_logId=message.ktid,
src_mentions=src_converted.mentions or [],
src_message=src_message,
src_type=src_type,
src_userId=src_kt_sender,
)
else:
reply_to = None
if content.get("format", None) == Format.HTML and content["formatted_body"] and content.msgtype.is_text:
parsed = await ToKakaoTalkParser().parse(utf16_surrogate.add(content["formatted_body"]))
text = utf16_surrogate.remove(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()) if mentions_by_user else None
else:
text = content.body
mentions = None
return SendParams(text=text, mentions=mentions, reply_to=reply_to)
_media_type_reply_body_map: dict[KnownChatType, str] = {
KnownChatType.PHOTO: "Photo",
KnownChatType.VIDEO: "Video",
KnownChatType.AUDIO: "Voice Note",
KnownChatType.FILE: "File: ",
}