Fix Matrix->Telegram name mentions
This commit is contained in:
@@ -1,5 +1,4 @@
|
||||
from .from_matrix import (matrix_reply_to_telegram, matrix_to_telegram, matrix_text_to_telegram,
|
||||
init_mx)
|
||||
from .from_matrix import matrix_reply_to_telegram, matrix_to_telegram, init_mx
|
||||
from .from_telegram import telegram_reply_to_matrix, telegram_to_matrix
|
||||
from .. import context as c
|
||||
|
||||
|
||||
@@ -18,10 +18,12 @@ import re
|
||||
import logging
|
||||
|
||||
from telethon.tl.types import (MessageEntityMention, MessageEntityMentionName, MessageEntityItalic,
|
||||
TypeMessageEntity)
|
||||
TypeMessageEntity, InputMessageEntityMentionName)
|
||||
from telethon.helpers import add_surrogate, del_surrogate
|
||||
from telethon import TelegramClient
|
||||
|
||||
from mautrix.types import RoomID, MessageEventContent
|
||||
from mautrix.util.logging import TraceLogger
|
||||
|
||||
from ... import puppet as pu
|
||||
from ...types import TelegramID
|
||||
@@ -31,30 +33,19 @@ from .parser import ParsedMessage, parse_html
|
||||
if TYPE_CHECKING:
|
||||
from ...context import Context
|
||||
|
||||
log: logging.Logger = logging.getLogger("mau.fmt.mx")
|
||||
log: TraceLogger = logging.getLogger("mau.fmt.mx")
|
||||
should_bridge_plaintext_highlights: bool = False
|
||||
|
||||
command_regex: Pattern = re.compile(r"^!([A-Za-z0-9@]+)")
|
||||
not_command_regex: Pattern = re.compile(r"^\\(![A-Za-z0-9@]+)")
|
||||
plain_mention_regex: Optional[Pattern] = None
|
||||
|
||||
|
||||
def plain_mention_to_html(match: Match) -> str:
|
||||
puppet = pu.Puppet.find_by_displayname(match.group(2))
|
||||
if puppet:
|
||||
return (f"{match.group(1)}"
|
||||
f"<a href='https://matrix.to/#/{puppet.mxid}'>"
|
||||
f"{puppet.displayname}"
|
||||
"</a>")
|
||||
return "".join(match.groups())
|
||||
|
||||
|
||||
MAX_LENGTH = 4096
|
||||
CUTOFF_TEXT = " [message cut]"
|
||||
CUT_MAX_LENGTH = MAX_LENGTH - len(CUTOFF_TEXT)
|
||||
|
||||
|
||||
def cut_long_message(message: str, entities: List[TypeMessageEntity]) -> ParsedMessage:
|
||||
def _cut_long_message(message: str, entities: List[TypeMessageEntity]) -> ParsedMessage:
|
||||
if len(message) > MAX_LENGTH:
|
||||
message = message[0:CUT_MAX_LENGTH] + CUTOFF_TEXT
|
||||
new_entities = []
|
||||
@@ -73,23 +64,6 @@ class FormatError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def matrix_to_telegram(html: str) -> ParsedMessage:
|
||||
try:
|
||||
html = command_regex.sub(r"<command>\1</command>", html)
|
||||
html = html.replace("\t", " " * 4)
|
||||
html = not_command_regex.sub(r"\1", html)
|
||||
if should_bridge_plaintext_highlights:
|
||||
html = plain_mention_regex.sub(plain_mention_to_html, html)
|
||||
|
||||
text, entities = parse_html(add_surrogate(html))
|
||||
text = del_surrogate(text.strip())
|
||||
text, entities = cut_long_message(text, entities)
|
||||
|
||||
return text, entities
|
||||
except Exception as e:
|
||||
raise FormatError(f"Failed to convert Matrix format: {html}") from e
|
||||
|
||||
|
||||
def matrix_reply_to_telegram(content: MessageEventContent, tg_space: TelegramID,
|
||||
room_id: Optional[RoomID] = None) -> Optional[TelegramID]:
|
||||
event_id = content.get_reply_to()
|
||||
@@ -103,19 +77,61 @@ def matrix_reply_to_telegram(content: MessageEventContent, tg_space: TelegramID,
|
||||
return None
|
||||
|
||||
|
||||
def matrix_text_to_telegram(text: str) -> ParsedMessage:
|
||||
async def matrix_to_telegram(client: TelegramClient, *, text: Optional[str] = None,
|
||||
html: Optional[str] = None) -> ParsedMessage:
|
||||
if html is not None:
|
||||
text, entities = _matrix_html_to_telegram(html)
|
||||
elif text is not None:
|
||||
text, entities = _matrix_text_to_telegram(text)
|
||||
else:
|
||||
raise ValueError("text or html must be provided to convert formatting")
|
||||
await _fix_name_mentions(client, entities)
|
||||
return text, entities
|
||||
|
||||
|
||||
def _matrix_html_to_telegram(html: str) -> ParsedMessage:
|
||||
try:
|
||||
html = command_regex.sub(r"<command>\1</command>", html)
|
||||
html = html.replace("\t", " " * 4)
|
||||
html = not_command_regex.sub(r"\1", html)
|
||||
if should_bridge_plaintext_highlights:
|
||||
html = plain_mention_regex.sub(_plain_mention_to_html, html)
|
||||
|
||||
text, entities = parse_html(add_surrogate(html))
|
||||
text = del_surrogate(text.strip())
|
||||
text, entities = _cut_long_message(text, entities)
|
||||
|
||||
return text, entities
|
||||
except Exception as e:
|
||||
raise FormatError(f"Failed to convert Matrix format: {html}") from e
|
||||
|
||||
|
||||
def _matrix_text_to_telegram(text: str) -> ParsedMessage:
|
||||
text = command_regex.sub(r"/\1", text)
|
||||
text = text.replace("\t", " " * 4)
|
||||
text = not_command_regex.sub(r"\1", text)
|
||||
if should_bridge_plaintext_highlights:
|
||||
entities, pmr_replacer = plain_mention_to_text()
|
||||
entities, pmr_replacer = _plain_mention_to_text()
|
||||
text = plain_mention_regex.sub(pmr_replacer, text)
|
||||
else:
|
||||
entities = []
|
||||
return text, entities
|
||||
|
||||
|
||||
def plain_mention_to_text() -> Tuple[List[TypeMessageEntity], Callable[[Match], str]]:
|
||||
async def _fix_name_mentions(client: TelegramClient, entities: List[TypeMessageEntity]) -> None:
|
||||
for index in reversed(range(len(entities))):
|
||||
entity = entities[index]
|
||||
if isinstance(entity, (MessageEntityMentionName, InputMessageEntityMentionName)):
|
||||
try:
|
||||
user = await client.get_input_entity(entity.user_id)
|
||||
except (ValueError, TypeError) as e:
|
||||
log.trace(f"Dropping mention of {entity.user_id}: {e}")
|
||||
del entities[index]
|
||||
else:
|
||||
entities[index] = InputMessageEntityMentionName(entity.offset, entity.length, user)
|
||||
|
||||
|
||||
def _plain_mention_to_text() -> Tuple[List[TypeMessageEntity], Callable[[Match], str]]:
|
||||
entities = []
|
||||
|
||||
def replacer(match: Match) -> str:
|
||||
@@ -136,6 +152,16 @@ def plain_mention_to_text() -> Tuple[List[TypeMessageEntity], Callable[[Match],
|
||||
return entities, replacer
|
||||
|
||||
|
||||
def _plain_mention_to_html(match: Match) -> str:
|
||||
puppet = pu.Puppet.find_by_displayname(match.group(2))
|
||||
if puppet:
|
||||
return (f"{match.group(1)}"
|
||||
f"<a href='https://matrix.to/#/{puppet.mxid}'>"
|
||||
f"{puppet.displayname}"
|
||||
"</a>")
|
||||
return "".join(match.groups())
|
||||
|
||||
|
||||
def init_mx(context: "Context") -> None:
|
||||
global plain_mention_regex, should_bridge_plaintext_highlights
|
||||
config = context.config
|
||||
|
||||
Reference in New Issue
Block a user